diff --git a/experiments/perception/worker/grpc_control_probe.py b/experiments/perception/worker/grpc_control_probe.py new file mode 100644 index 0000000..baa7889 --- /dev/null +++ b/experiments/perception/worker/grpc_control_probe.py @@ -0,0 +1,312 @@ +"""CPU-only clock/grant route evidence; NOT a model/freshness qualification. + +One bootstrap ticket/certificate copied via authenticated SSH, subsequent data +grants delivered only by TLS Poll. Source events/resume proof remain synthetic; +real monotonic clock observations use explicit CONDITIONAL error/rate budgets. +""" + +import argparse +import asyncio +import json +import os +import resource +import secrets +import subprocess +import sys +import threading +import time +from dataclasses import asdict +from datetime import UTC, datetime +from hashlib import sha256 +from pathlib import Path + +import grpc +from grpc_transport_probe import document, identity, until + +from k1link.compute.live_perception import LiveIngressEvent +from k1link.perception.realtime_contract import StreamStart +from k1link.perception.streaming_clock import ClockMappingError, ClockWindow +from k1link.perception.streaming_continuity import ResumeEvidence +from k1link.perception.streaming_control_grpc import ( + ControlTicket, + StreamControlClient, + StreamControlEndpoint, +) +from k1link.perception.streaming_grpc import GrpcStreamClient, GrpcStreamEndpoint +from k1link.perception.streaming_lifecycle import StreamingLifecycle +from k1link.perception.streaming_queue import StreamMailbox + + +async def server(root): + private = root / "private" + private.mkdir(mode=0o700, exist_ok=True) + cert, key, bootstrap = [private / x for x in ("cert.pem", "key.pem", "bootstrap.json")] + subprocess.run( + [ + "openssl", + "req", + "-x509", + "-newkey", + "rsa:2048", + "-nodes", + "-days", + "1", + "-subj", + "/CN=localhost", + "-keyout", + str(key), + "-out", + str(cert), + ], + check=True, + capture_output=True, + timeout=10, + ) + run = StreamingLifecycle( + identity(), + root / "lease", + StreamMailbox(), + threading.Event(), + recover_input=True, + source_clock_ns=lambda: 1_000_000_000, + ) + resident = run.spawn( + lambda: subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(180)"], + start_new_session=True, + ) + ) + run.ready() + pulse_stop = threading.Event() + + def pulse(): + while not pulse_stop.wait(0.1): + run.renew(identity()) + + pulse_thread = threading.Thread(target=pulse, daemon=True) + pulse_thread.start() + received = [] + + def consume(event): + item = { + "sequence": event.ingress_sequence, + "sha256": sha256(event.payload).hexdigest(), + "bytes": len(event.payload), + "resident_pid": resident.pid, + } + received.append(item) + if len(received) > 8: + raise ValueError("bounded count exceeded") + endpoint.publish( + run.continuity.epoch, event.ingress_sequence, json.dumps(item, sort_keys=True).encode() + ) + + endpoint = GrpcStreamEndpoint(run, consume, lambda _: None) + pending = [endpoint.issue(run.continuity.epoch, "synthetic-capture", 1)] + ticket = ControlTicket(identity(), secrets.token_hex(32)) + clock_id = "worker-process-" + secrets.token_hex(12) + control = StreamControlEndpoint( + ticket, + lambda: pending[0] if endpoint.grant is not None else None, + clock_id=clock_id, + ) + report = { + "utc": datetime.now(UTC).isoformat(), + "started_monotonic_ns": str(time.monotonic_ns()), + "model_count": 0, + "synthetic_source_and_resume": True, + "real_time_qualified": False, + "actuation_allowed": False, + "received": received, + "clock_id": clock_id, + "grpc_version": grpc.__version__, + } + rpc = None + try: + rpc, _ = await endpoint.serve( + "0.0.0.0:50061", + certificate=cert.read_bytes(), + private_key=key.read_bytes(), + control_handlers=(control.handler(),), + ) + document( + bootstrap, + { + "activation": identity().to_dict(), + "token": ticket.token, + "remote_clock_id": clock_id, + }, + ) + bootstrap.chmod(0o600) + await until(lambda: len(received) == 4 and endpoint.active is None, seconds=90) + report["first_transport"] = endpoint.last + report["before_gap"] = run.snapshot() + began = time.monotonic_ns() + await asyncio.sleep(2.2) + report["gap_ns"] = str(time.monotonic_ns() - began) + assert resident.poll() is None and run.continuity.phase == "waiting" + report["after_gap"] = run.snapshot() + epoch = run.begin_input(identity()) + run.resume_input(epoch, ResumeEvidence(*([1_000_000_000] * 4), True), lambda: None) + pending[0] = endpoint.issue(epoch, "synthetic-capture", 1) + await until(lambda: len(received) == 8 and run.mailbox.done, seconds=30) + endpoint.finish_results(epoch) + await until(lambda: endpoint.active is None) + report["second_transport"] = endpoint.last + report["passed"] = resident.poll() is None + report["peak_mailbox_bytes"] = run.mailbox.peak_bytes + finally: + if rpc: + await rpc.stop(0) + pulse_stop.set() + pulse_thread.join(1) + report["closed"] = run.close() + report["resident_reaped"] = resident.poll() is not None + report["final_mailbox_bytes"] = run.mailbox.bytes + report["control_accepted"] = control.accepted + report["control_rejected"] = control.rejected + report["rss_max_kib"] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + report["finished_monotonic_ns"] = str(time.monotonic_ns()) + document(root / "server-report.json", report) + for path in (key, bootstrap): + path.unlink(missing_ok=True) + + +async def client(root, target): + bootstrap = root / "bootstrap.json" + value = json.loads(bootstrap.read_text()) + ticket = ControlTicket(StreamStart.from_dict(value["activation"]), value["token"]) + local_clock_id = "mac-process-" + secrets.token_hex(12) + control = StreamControlClient( + target, (root / "cert.pem").read_bytes(), ticket, clock_id=local_clock_id + ) + mapping = ClockWindow( + local_clock_id, + value["remote_clock_id"], + rate_ppm=500, + timestamp_error_ns=50_000, + maximum_age_ns=2_000_000_000, + ) + rows, grants, replies, errors = [], [], [], [] + report = { + "utc": datetime.now(UTC).isoformat(), + "started_monotonic_ns": str(time.monotonic_ns()), + "schema_version": "missioncore.grpc-control-clock-probe/v1", + "observations": rows, + "grants": grants, + "replies": replies, + "errors": errors, + "model_count": 0, + "synthetic_source_and_resume": True, + "real_time_qualified": False, + "one_way_frame_age_measured": False, + "actuation_allowed": False, + "conditional_relative_rate_ppm": 500, + "timestamp_error_budget_ns": "50000", + "maximum_mapping_age_ns": "2000000000", + "admission_uncertainty_ns": "5000000", + "path": "TLS gRPC over existing Mac-Worker authenticated SSH/Tailscale route", + } + + async def poll(phase): + for _ in range(4): + try: + probe, access = await control.poll() + break + except grpc.aio.AioRpcError as exc: + errors.append({"phase": phase, "code": exc.code().name}) + await asyncio.sleep(0.1) + else: + raise RuntimeError("bounded control polling failed") + bounds = mapping.add(probe) + now = time.monotonic_ns() + uncertainty = bounds.uncertainty_ns(now) + raw = {k: str(v) if type(v) is int else v for k, v in asdict(probe).items()} + rows.append( + { + "phase": phase, + "probe": raw, + "bounds": bounds.to_dict(), + "evaluated_at_ns": str(now), + "uncertainty_ns": str(uncertainty), + "admitted_5ms": uncertainty <= 5_000_000, + } + ) + return access + + try: + previous = None + for phase in (1, 2): + if phase == 2: + await asyncio.sleep(2.3) + try: + mapping.current(time.monotonic_ns()) + except ClockMappingError: + report["mapping_expired_during_gap"] = True + else: + raise AssertionError("old clock mapping survived expiry") + access = None + for _ in range(32): + candidate = await poll(phase) + if candidate is not None: + access = candidate + await asyncio.sleep(0.05) + assert access is not None and (previous is None or access.epoch != previous.epoch) + assert previous is None or access.token != previous.token + grants.append( + { + "phase": phase, + "epoch": access.epoch.to_dict(), + "token_changed": previous is not None, + "delivery": "authenticated-control-poll", + } + ) + stream = GrpcStreamClient(target, (root / "cert.pem").read_bytes(), access) + try: + await stream.open() + for sequence in range((phase - 1) * 4 + 1, phase * 4 + 1): + payload = bytes([sequence]) * 32768 + event = LiveIngressEvent( + sequence, + "synthetic-capture", + 1, + "lidar", + "lidar", + sequence, + 1_799_999_999_123_456_789, + 1_000_000_000, + payload, + ) + started = time.monotonic_ns() + await stream.send(event) + result = await stream.receive() + elapsed = time.monotonic_ns() - started + returned = json.loads(result[1]) + assert ( + result[0] == sequence and returned["sha256"] == sha256(payload).hexdigest() + ) + replies.append({**returned, "round_trip_ns": str(elapsed)}) + assert await poll(phase) is None # Control still works during Exchange. + await asyncio.sleep(0.05) + if phase == 2: + await stream.end() + assert await stream.receive() is None + finally: + await stream.close() + previous = access + report["passed"] = True + finally: + await control.close() + report["finished_monotonic_ns"] = str(time.monotonic_ns()) + document(root / "client-report.json", report) + bootstrap.unlink(missing_ok=True) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("mode", choices=("server", "client")) + parser.add_argument("root", type=Path) + parser.add_argument("--target", default="localhost:18561") + args = parser.parse_args() + os.umask(0o077) + asyncio.run(server(args.root) if args.mode == "server" else client(args.root, args.target)) diff --git a/src/k1link/perception/streaming_clock.py b/src/k1link/perception/streaming_clock.py new file mode 100644 index 0000000..8fa3ee3 --- /dev/null +++ b/src/k1link/perception/streaming_clock.py @@ -0,0 +1,225 @@ +"""Bounded, conditional mapping of two process-scoped monotonic clocks. + +Four timestamps bound remote-minus-local offset WITHOUT symmetric-path claims. +Intersect recent intervals, widened for an explicit relative-rate/error budget. +Neither OS clock is adjusted. The rate/error envelope is an assumption supplied +by the controller, not a hardware accuracy certificate inferred from low RTT. +""" + +from __future__ import annotations + +from collections import deque +from dataclasses import dataclass + +from .streaming_wire import identifier + + +class ClockMappingError(ValueError): + pass + + +def _stamp(value: int) -> None: + if type(value) is not int or not 0 <= value < 2**63: + raise ClockMappingError("invalid monotonic timestamp") + + +def _drift(duration_ns: int, ppm: int) -> int: + return (duration_ns * ppm + 999_999) // 1_000_000 + + +@dataclass(frozen=True) +class ClockProbe: + local_clock_id: str + remote_clock_id: str + nonce: str + local_send_ns: int + remote_receive_ns: int + remote_send_ns: int + local_receive_ns: int + + def __post_init__(self) -> None: + for value in (self.local_clock_id, self.remote_clock_id, self.nonce): + identifier(value) + for stamp in ( + self.local_send_ns, + self.remote_receive_ns, + self.remote_send_ns, + self.local_receive_ns, + ): + _stamp(stamp) + if ( + not 0 <= self.local_receive_ns - self.local_send_ns <= 2_000_000_000 + or self.remote_send_ns < self.remote_receive_ns + ): + raise ClockMappingError("clock probe ordering/deadline violated") + + def bounds_at( + self, local_ns: int, *, rate_ppm: int, timestamp_error_ns: int + ) -> tuple[int, int]: + _stamp(local_ns) + if local_ns < self.local_receive_ns: + raise ClockMappingError("clock observation is from the future") + # Any outbound/inbound asymmetry is inside these bounds. Remote receive + # and send occur between local send and receive; widen across the whole + # interval, not just elapsed time after receipt. + widen = _drift(local_ns - self.local_send_ns, rate_ppm) + 2 * timestamp_error_ns + return ( + self.remote_send_ns - self.local_receive_ns - widen, + self.remote_receive_ns - self.local_send_ns + widen, + ) + + +@dataclass(frozen=True) +class ClockBounds: + local_clock_id: str + remote_clock_id: str + measured_at_ns: int + expires_at_ns: int + offset_lower_ns: int + offset_upper_ns: int + rate_ppm: int + timestamp_error_ns: int + samples: int + + def __post_init__(self) -> None: + for value in (self.local_clock_id, self.remote_clock_id): + identifier(value) + _stamp(self.measured_at_ns) + _stamp(self.expires_at_ns) + if ( + self.measured_at_ns >= self.expires_at_ns + 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 type(self.rate_ppm) is not int + or not 1 <= self.rate_ppm <= 1000 + or type(self.timestamp_error_ns) is not int + or not 1 <= self.timestamp_error_ns <= 1_000_000 + or type(self.samples) is not int + or not 1 <= self.samples <= 16 + ): + raise ClockMappingError("invalid clock bounds") + + def offset_at(self, local_ns: int) -> tuple[int, int]: + _stamp(local_ns) + if not self.measured_at_ns <= local_ns < self.expires_at_ns: + raise ClockMappingError("clock mapping expired or local clock regressed") + extra = _drift(local_ns - self.measured_at_ns, self.rate_ppm) + return self.offset_lower_ns - extra, self.offset_upper_ns + extra + + def uncertainty_ns(self, local_ns: int) -> int: + lower, upper = self.offset_at(local_ns) + return (upper - lower + 1) // 2 + + def require(self, local_ns: int, maximum_uncertainty_ns: int = 5_000_000) -> None: + if type(maximum_uncertainty_ns) is not int or maximum_uncertainty_ns <= 0: + raise ClockMappingError("invalid uncertainty gate") + if self.uncertainty_ns(local_ns) > maximum_uncertainty_ns: + raise ClockMappingError("clock uncertainty exceeds admission budget") + + def to_dict(self) -> dict[str, object]: + return { + "schema_version": "missioncore.monotonic-clock-bounds/v1", + "local_clock_id": self.local_clock_id, + "remote_clock_id": self.remote_clock_id, + "measured_at_ns": str(self.measured_at_ns), + "expires_at_ns": str(self.expires_at_ns), + "offset_lower_ns": str(self.offset_lower_ns), + "offset_upper_ns": str(self.offset_upper_ns), + "relative_rate_budget_ppm": self.rate_ppm, + "timestamp_error_budget_ns": str(self.timestamp_error_ns), + "samples": self.samples, + "conditional_rate_error_envelope": True, + "symmetric_network_assumed": False, + } + + +class ClockWindow: + def __init__( + self, + local_clock_id: str, + remote_clock_id: str, + *, + rate_ppm: int, + timestamp_error_ns: int, + maximum_age_ns: int = 2_000_000_000, + ) -> None: + for value in (local_clock_id, remote_clock_id): + identifier(value) + if ( + type(rate_ppm) is not int + or not 1 <= rate_ppm <= 1000 + or type(timestamp_error_ns) is not int + or not 1 <= timestamp_error_ns <= 1_000_000 + or type(maximum_age_ns) is not int + or not 0 < maximum_age_ns <= 5_000_000_000 + ): + raise ClockMappingError("explicit bounded clock envelope is required") + self.local_clock_id, self.remote_clock_id = local_clock_id, remote_clock_id + self.rate_ppm, self.timestamp_error_ns, self.maximum_age_ns = ( + rate_ppm, + timestamp_error_ns, + maximum_age_ns, + ) + self.samples: deque[ClockProbe] = deque(maxlen=16) + self.failed = False + self.last_receive_ns = -1 + + def add(self, sample: ClockProbe) -> ClockBounds: + if self.failed: + raise ClockMappingError("clock session quarantined; explicit new session required") + if ( + (sample.local_clock_id, sample.remote_clock_id) + != (self.local_clock_id, self.remote_clock_id) + or sample.local_send_ns <= self.last_receive_ns + or any(old.nonce == sample.nonce for old in self.samples) + ): + raise ClockMappingError("foreign, overlapping or replayed clock probe") + while ( + self.samples + and sample.local_receive_ns - self.samples[0].local_receive_ns >= self.maximum_age_ns + ): + self.samples.popleft() + # A new sample must intersect the still-valid old evidence. Do not + # silently forget a contradictory sample to manufacture a narrower fit. + bounds = [ + old.bounds_at( + sample.local_receive_ns, + rate_ppm=self.rate_ppm, + timestamp_error_ns=self.timestamp_error_ns, + ) + for old in (*self.samples, sample) + ] + lower, upper = max(x[0] for x in bounds), min(x[1] for x in bounds) + if lower > upper: + self.failed = True + raise ClockMappingError("clock jump or violated rate/error envelope") + self.samples.append(sample) + self.last_receive_ns = sample.local_receive_ns + return self.current(sample.local_receive_ns) + + def current(self, local_ns: int) -> ClockBounds: + _stamp(local_ns) + if self.failed or not self.samples or local_ns < self.last_receive_ns: + raise ClockMappingError("no valid monotonic clock mapping") + recent = [x for x in self.samples if local_ns - x.local_receive_ns < self.maximum_age_ns] + if not recent: + raise ClockMappingError("clock mapping expired") + bounds = [ + x.bounds_at( + local_ns, rate_ppm=self.rate_ppm, timestamp_error_ns=self.timestamp_error_ns + ) + for x in recent + ] + lower, upper = max(x[0] for x in bounds), min(x[1] for x in bounds) + return ClockBounds( + self.local_clock_id, + self.remote_clock_id, + local_ns, + min(x.local_receive_ns for x in recent) + self.maximum_age_ns, + lower, + upper, + self.rate_ppm, + self.timestamp_error_ns, + len(recent), + ) diff --git a/src/k1link/perception/streaming_control_grpc.py b/src/k1link/perception/streaming_control_grpc.py new file mode 100644 index 0000000..3f6b55f --- /dev/null +++ b/src/k1link/perception/streaming_control_grpc.py @@ -0,0 +1,269 @@ +"""Capability-scoped clock observation and pending-grant delivery over TLS. + +Poll cannot acquire/renew a GPU lease, pick an epoch, reset models or cancel a +run. Only the trusted local controller supplies a pending grant. A request nonce +binds each clock response; authenticated polling is rate/buffer bounded. The +per-activation bootstrap ticket is provisioned separately, never in CLI/logs. +""" + +from __future__ import annotations + +import hmac +import secrets +import threading +import time +from collections.abc import Callable +from dataclasses import dataclass, field, replace +from hashlib import sha256 +from typing import Any + +import grpc # type: ignore[import-untyped] + +from . import streaming_wire as wire +from .realtime_contract import StreamStart +from .streaming_clock import ClockProbe +from .streaming_grpc import OPTIONS, StreamAccess + +SERVICE = "missioncore.perception.v1.StreamControl" +METHOD = f"/{SERVICE}/Poll" +MAX_CONTROL = 8192 + + +def _document(raw: bytes) -> dict[str, Any]: + if type(raw) is not bytes or not 0 < len(raw) <= MAX_CONTROL: + raise wire.StreamWireError("control message exceeds bound") + return wire.parse_header(bytearray(raw)) + + +def _token(value: object) -> None: + if ( + not isinstance(value, str) + or len(value) != 64 + or any(c not in "0123456789abcdef" for c in value) + ): + raise ValueError("invalid capability") + + +def _offer(access: StreamAccess, activation: StreamStart) -> None: + if replace(access.epoch, epoch_id=activation.epoch_id) != activation: + raise ValueError("pending grant belongs to another activation") + wire.identifier(access.session_id) + wire.decimal(access.session_generation) + if access.session_generation < 1: + raise ValueError("invalid capture generation") + _token(access.token) + + +@dataclass(frozen=True) +class ControlTicket: + activation: StreamStart + token: str = field(repr=False) + + def __post_init__(self) -> None: + _token(self.token) + + def metadata(self) -> tuple[tuple[str, str], ...]: + return ( + ("authorization", "Bearer " + self.token), + ("x-stream-binding", wire.binding(self.activation)), + ) + + +class StreamControlEndpoint: + def __init__( + self, + ticket: ControlTicket, + pending: Callable[[], StreamAccess | None], + *, + clock_id: str, + clock_ns: Callable[[], int] = time.monotonic_ns, + ) -> None: + wire.identifier(clock_id) + self.activation, self.pending, self.clock_id, self.clock_ns = ( + ticket.activation, + pending, + clock_id, + clock_ns, + ) + self._token_hash = sha256(ticket.token.encode()).digest() + self._lock = threading.Lock() + self._last_poll_ns = -1 + self.accepted = self.rejected = 0 + + def handler(self) -> Any: + return grpc.method_handlers_generic_handler( + SERVICE, {"Poll": grpc.unary_unary_rpc_method_handler(self.poll)} + ) + + def _authenticate(self, context: Any) -> None: + if b"ssl" not in context.auth_context().get("transport_security_type", ()): + raise wire.StreamWireError("TLS control capability required") + metadata = list(context.invocation_metadata()) + relevant = [(k, v) for k, v in metadata if k in ("authorization", "x-stream-binding")] + values = dict(relevant) + token = values.get("authorization") + if ( + len(relevant) != 2 + or len(values) != 2 + or not isinstance(token, str) + or len(token) != 71 + or not token.startswith("Bearer ") + or values.get("x-stream-binding") != wire.binding(self.activation) + or not hmac.compare_digest(sha256(token[7:].encode()).digest(), self._token_hash) + ): + raise wire.StreamWireError("control capability rejected") + + async def poll(self, raw: bytes, context: Any) -> bytes: + received = self.clock_ns() + try: + self._authenticate(context) + request = _document(raw) + if set(request) != {"nonce", "local_clock_id", "local_send_ns"}: + raise wire.StreamWireError("unsupported control request") + wire.identifier(request["nonce"]) + wire.identifier(request["local_clock_id"]) + if wire.uint64(request["local_send_ns"]) >= 2**63: + raise ValueError("invalid monotonic timestamp") + except (ValueError, TypeError): + self.rejected += 1 + await context.abort( + grpc.StatusCode.UNAUTHENTICATED, "control capability/request rejected" + ) + return b"" + with self._lock: + limited = self._last_poll_ns >= 0 and received - self._last_poll_ns < 25_000_000 + if not limited: + self._last_poll_ns = received + if limited: + self.rejected += 1 + await context.abort( + grpc.StatusCode.RESOURCE_EXHAUSTED, "bounded control poll rate exceeded" + ) + return b"" + try: + offer = self.pending() # Trusted constant-time snapshot; no model/control work. + if offer is not None: + _offer(offer, self.activation) + response = { + "schema_version": "missioncore.stream-control-poll/v1", + "activation_binding": wire.binding(self.activation), + **request, + "remote_clock_id": self.clock_id, + "remote_receive_ns": str(received), + "remote_send_ns": str(self.clock_ns()), + "grant": None + if offer is None + else { + "epoch": offer.epoch.to_dict(), + "session_id": offer.session_id, + "session_generation": str(offer.session_generation), + "token": offer.token, + }, + } + encoded = wire.canonical(response) + if len(encoded) > MAX_CONTROL: + raise ValueError("trusted control snapshot exceeds bound") + except (ValueError, RuntimeError): + await context.abort( + grpc.StatusCode.UNAVAILABLE, "local controller has no current offer" + ) + return b"" + self.accepted += 1 + return encoded + + +class StreamControlClient: + def __init__( + self, + target: str, + roots: bytes, + ticket: ControlTicket, + *, + clock_id: str, + clock_ns: Callable[[], int] = time.monotonic_ns, + ) -> None: + if not roots: + raise ValueError("trusted TLS roots required") + wire.identifier(clock_id) + self.ticket, self.clock_id, self.clock_ns = ticket, clock_id, clock_ns + self.channel = grpc.aio.secure_channel( + target, + grpc.ssl_channel_credentials(root_certificates=roots), + options=OPTIONS + + ( + ("grpc.max_send_message_length", MAX_CONTROL), + ("grpc.max_receive_message_length", MAX_CONTROL), + ), + ) + self.call = self.channel.unary_unary(METHOD) + self.polling = False + + async def poll(self) -> tuple[ClockProbe, StreamAccess | None]: + if self.polling: + raise ValueError("one outstanding clock/control probe per client") + self.polling = True + nonce = secrets.token_hex(16) + sent = self.clock_ns() + try: + raw = await self.call( + wire.canonical( + {"nonce": nonce, "local_clock_id": self.clock_id, "local_send_ns": str(sent)} + ), + metadata=self.ticket.metadata(), + timeout=0.5, + wait_for_ready=False, + ) + received = self.clock_ns() + response = _document(raw) + if ( + set(response) + != { + "schema_version", + "activation_binding", + "nonce", + "local_clock_id", + "local_send_ns", + "remote_clock_id", + "remote_receive_ns", + "remote_send_ns", + "grant", + } + or response["schema_version"] != "missioncore.stream-control-poll/v1" + or response["activation_binding"] != wire.binding(self.ticket.activation) + or response["nonce"] != nonce + or response["local_clock_id"] != self.clock_id + or response["local_send_ns"] != str(sent) + ): + raise wire.StreamWireError("clock/control response binding mismatch") + probe = ClockProbe( + self.clock_id, + response["remote_clock_id"], + nonce, + sent, + wire.uint64(response["remote_receive_ns"]), + wire.uint64(response["remote_send_ns"]), + received, + ) + grant = response["grant"] + access = None + if grant is not None: + if not isinstance(grant, dict) or set(grant) != { + "epoch", + "session_id", + "session_generation", + "token", + }: + raise wire.StreamWireError("invalid control grant shape") + access = StreamAccess( + StreamStart.from_dict(grant["epoch"]), + grant["session_id"], + wire.uint64(grant["session_generation"]), + grant["token"], + ) + _offer(access, self.ticket.activation) + return probe, access + finally: + self.polling = False + + async def close(self) -> None: + await self.channel.close() diff --git a/src/k1link/perception/streaming_grpc.py b/src/k1link/perception/streaming_grpc.py index 471eede..d51e8cd 100644 --- a/src/k1link/perception/streaming_grpc.py +++ b/src/k1link/perception/streaming_grpc.py @@ -149,7 +149,14 @@ class GrpcStreamEndpoint: if self.active is not None and self.active[0] == epoch: self.active[1].finish() - async def serve(self, address: str, *, certificate: bytes, private_key: bytes) -> Any: + async def serve( + self, + address: str, + *, + certificate: bytes, + private_key: bytes, + control_handlers: tuple[Any, ...] = (), + ) -> Any: if not certificate or not private_key: raise ValueError("TLS identity is required") server = grpc.aio.server( @@ -158,7 +165,11 @@ class GrpcStreamEndpoint: ("grpc.max_receive_message_length", INPUT_CHUNK), ("grpc.max_send_message_length", MAX_REPLY), ), - maximum_concurrent_rpcs=2, + # aio reserves its next accept before awaiting a request. Leave + # one bounded accept slot beside Exchange + a completing Poll; + # otherwise the next Poll can inherit a stale "limit exceeded". + # Data ownership still admits exactly one Exchange, never 2 GPUs. + maximum_concurrent_rpcs=3 if control_handlers else 2, compression=grpc.Compression.NoCompression, ) server.add_generic_rpc_handlers( @@ -166,6 +177,7 @@ class GrpcStreamEndpoint: grpc.method_handlers_generic_handler( SERVICE, {"Exchange": grpc.stream_stream_rpc_method_handler(self.exchange)} ), + *control_handlers, ) ) port = server.add_secure_port( diff --git a/tests/test_perception_streaming_clock.py b/tests/test_perception_streaming_clock.py new file mode 100644 index 0000000..ee4c00c --- /dev/null +++ b/tests/test_perception_streaming_clock.py @@ -0,0 +1,134 @@ +"""Conditional clock intervals, never a symmetric-network offset assertion.""" + +from dataclasses import replace + +import pytest + +from k1link.perception.streaming_clock import ClockMappingError, ClockProbe, ClockWindow + +BASE = 10**16 # Deliberately above IEEE754 exact-integer range. + + +def probe(number=1, *, outbound=1_000_000, inbound=2_000_000, offset=-(10**12)): + sent = BASE + number * 100_000_000 + return ClockProbe( + "mac", + "worker", + str(number), + sent, + sent + offset + outbound, + sent + offset + outbound + 100_000, + sent + outbound + inbound + 100_000, + ) + + +def window(**kwargs): + return ClockWindow("mac", "worker", rate_ppm=500, timestamp_error_ns=50_000, **kwargs) + + +@pytest.mark.parametrize("offset", [-(10**12), 0, 10**12]) +@pytest.mark.parametrize("outbound,inbound", [(1, 9_000_000), (9_000_000, 1), (1, 1)]) +def test_contains_true_offset_without_symmetric_path_assumption(offset, outbound, inbound): + sample = probe(offset=offset, outbound=outbound, inbound=inbound) + mapping = window().add(sample) + now = sample.local_receive_ns + assert mapping.offset_at(now)[0] <= offset <= mapping.offset_at(now)[1] + assert mapping.uncertainty_ns(now + 1_000_000) > mapping.uncertainty_ns(now) + value = mapping.to_dict() + assert value["measured_at_ns"] == str(now) + assert not value["symmetric_network_assumed"] + assert value["timestamp_error_budget_ns"] == "50000" + + +def test_intersection_requires_evidence_and_drift_is_not_averaged_away(): + mapping = window() + first = mapping.add(probe(outbound=1_000_000, inbound=20_000_000)) + with pytest.raises(ClockMappingError, match="budget"): + first.require(first.measured_at_ns) + second = mapping.add(probe(2, outbound=20_000_000, inbound=1_000_000)) + second.require(second.measured_at_ns) + assert second.uncertainty_ns(second.measured_at_ns) < 2_000_000 + assert ( + second.offset_at(second.measured_at_ns)[0] + < -(10**12) + < second.offset_at(second.measured_at_ns)[1] + ) + + +def test_expired_mapping_waits_and_new_observation_can_restore_it(): + mapping = window() + first = mapping.add(probe()) + with pytest.raises(ClockMappingError, match="expired"): + first.require(first.expires_at_ns) + with pytest.raises(ClockMappingError, match="expired"): + mapping.current(first.expires_at_ns) + resumed = mapping.add(probe(30)) + assert resumed.samples == 1 + resumed.require(resumed.measured_at_ns) + + +def test_contradictory_samples_quarantine_not_cherry_pick(): + mapping = window() + mapping.add(probe()) + with pytest.raises(ClockMappingError, match="envelope"): + mapping.add(probe(2, offset=0)) + with pytest.raises(ClockMappingError, match="quarantined"): + mapping.add(probe(3)) + with pytest.raises(ClockMappingError): + mapping.current(BASE + 3_000_000_000) + + +@pytest.mark.parametrize("fault", ["local", "remote", "nonce", "overlap"]) +def test_foreign_or_replayed_observation_cannot_refresh_mapping(fault): + mapping = window() + original = probe() + mapping.add(original) + second = probe(2) + changes = { + "local": {"local_clock_id": "other"}, + "remote": {"remote_clock_id": "rebooted"}, + "nonce": {"nonce": original.nonce}, + "overlap": {"local_send_ns": original.local_receive_ns}, + } + with pytest.raises(ClockMappingError): + mapping.add(replace(second, **changes[fault])) + assert mapping.last_receive_ns == original.local_receive_ns + + +def test_history_is_bounded_and_exact_values_survive(): + mapping = window(maximum_age_ns=5_000_000_000) + for number in range(1, 35): + bounds = mapping.add(probe(number)) + assert len(mapping.samples) == bounds.samples == 16 + assert bounds.to_dict()["measured_at_ns"] == str(probe(34).local_receive_ns) + + +@pytest.mark.parametrize("value", [-1, 2**63, True, 1.0]) +def test_invalid_monotonic_timestamp_rejected(value): + with pytest.raises(ClockMappingError): + replace(probe(), local_send_ns=value) + + +def test_deadline_remote_order_and_backward_reads(): + sample = probe() + for changes in ( + {"local_receive_ns": sample.local_send_ns - 1}, + {"local_receive_ns": sample.local_send_ns + 2_000_000_001}, + {"remote_send_ns": sample.remote_receive_ns - 1}, + ): + with pytest.raises(ClockMappingError): + replace(sample, **changes) + mapping = window().add(sample) + with pytest.raises(ClockMappingError): + mapping.offset_at(mapping.measured_at_ns - 1) + with pytest.raises(ClockMappingError): + replace(mapping, offset_lower_ns=mapping.offset_upper_ns + 1) + + +@pytest.mark.parametrize( + "rate,error,age", + [(0, 1, 1), (1001, 1, 1), (True, 1, 1), (1, 0, 1), (1, 1_000_001, 1), (1, 1, 0)], +) +def test_controller_must_supply_bounded_envelope(rate, error, age): + with pytest.raises(ClockMappingError): + ClockWindow("mac", "worker", rate_ppm=rate, timestamp_error_ns=error, maximum_age_ns=age) diff --git a/tests/test_perception_streaming_control_grpc.py b/tests/test_perception_streaming_control_grpc.py new file mode 100644 index 0000000..1887a8a --- /dev/null +++ b/tests/test_perception_streaming_control_grpc.py @@ -0,0 +1,206 @@ +"""Small TLS clock/grant checks; no model workload or lease authority in Poll.""" + +# ruff: noqa: E402, F811 -- imported shared pytest TLS fixture. + +import asyncio +import threading +import time +from contextlib import asynccontextmanager +from dataclasses import replace +from hashlib import sha256 + +import pytest + +grpc = pytest.importorskip("grpc") + +from test_perception_streaming_grpc import event, eventually, identity, tls # noqa: F401 + +from k1link.perception import streaming_wire as wire +from k1link.perception.streaming_clock import ClockWindow +from k1link.perception.streaming_continuity import ResumeEvidence +from k1link.perception.streaming_control_grpc import ( + MAX_CONTROL, + ControlTicket, + StreamControlClient, + StreamControlEndpoint, + _offer, +) +from k1link.perception.streaming_grpc import GrpcStreamClient, GrpcStreamEndpoint, StreamAccess +from k1link.perception.streaming_lifecycle import StreamingLifecycle +from k1link.perception.streaming_queue import StreamMailbox + + +@asynccontextmanager +async def controlled(tmp_path, tls): + run = StreamingLifecycle( + identity(), + tmp_path, + StreamMailbox(), + threading.Event(), + recover_input=True, + source_clock_ns=lambda: 1_000_000_000, + ) + run.ready() + endpoint = GrpcStreamEndpoint( + run, + lambda value: endpoint.publish( + run.continuity.epoch, value.ingress_sequence, sha256(value.payload).digest() + ), + lambda _: None, + ) + pending = [endpoint.issue(identity(), "capture", 1)] + ticket = ControlTicket(identity(), "b" * 64) + control = StreamControlEndpoint(ticket, lambda: pending[0], clock_id="worker") + server, port = await endpoint.serve( + "localhost:0", + certificate=tls[0], + private_key=tls[1], + control_handlers=(control.handler(),), + ) + client = StreamControlClient(f"localhost:{port}", tls[0], ticket, clock_id="mac") + try: + yield run, endpoint, control, client, pending, port + finally: + await client.close() + await server.stop(0) + await eventually(lambda: endpoint.active is None) + assert run.close() + + +def test_control_delivery_coexists_with_stream_without_lease_or_model_authority(tmp_path, tls): + async def check(): + async with controlled(tmp_path, tls) as (run, endpoint, control, client, pending, port): + probe, access = await client.poll() + assert access == pending[0] + mapping = ClockWindow("mac", "worker", rate_ppm=500, timestamp_error_ns=50_000) + assert mapping.add(probe).uncertainty_ns(probe.local_receive_ns) > 0 + first = access + for phase in range(2): + stream = GrpcStreamClient(f"localhost:{port}", tls[0], access) + try: + await stream.open() + pending[0] = None + await stream.send(event(phase + 1)) + assert await stream.receive() == (phase + 1, sha256(event().payload).digest()) + await asyncio.sleep(0.03) + _, absent = await client.poll() + assert absent is None and endpoint.active is not None + finally: + await stream.close() + await eventually(lambda: endpoint.active is None) + assert run.continuity.phase == "waiting" + assert run.lease.start == identity() + if phase == 0: + run.renew(identity()) # Only independent trusted owner renews lease. + epoch = run.begin_input(identity()) + run.resume_input( + epoch, ResumeEvidence(*([1_000_000_000] * 4), True), lambda: None + ) + pending[0] = endpoint.issue(epoch, "capture", 1) + await asyncio.sleep(0.03) + _, access = await client.poll() + assert access.epoch != first.epoch and access.token != first.token + assert control.accepted == 4 and run.lease.start.lease_generation == 1 + + asyncio.run(check()) + + +@pytest.mark.parametrize("fault", ["secret", "binding", "duplicate", "oversize", "int64", "extra"]) +def test_bad_control_cannot_claim_data_or_change_owner(tmp_path, tls, fault): + async def check(): + async with controlled(tmp_path, tls) as (run, endpoint, control, client, pending, _): + metadata = client.ticket.metadata() + request = {"nonce": "n", "local_clock_id": "mac", "local_send_ns": "1"} + if fault == "secret": + metadata = ControlTicket(identity(), "0" * 64).metadata() + elif fault == "binding": + metadata = ControlTicket( + replace(identity(), source_id="other"), "b" * 64 + ).metadata() + elif fault == "duplicate": + metadata = (*metadata, *metadata) + elif fault == "int64": + request["local_send_ns"] = str(2**63) + elif fault == "extra": + request["renew_lease"] = True + raw = b"x" * (MAX_CONTROL + 1) if fault == "oversize" else wire.canonical(request) + with pytest.raises(grpc.aio.AioRpcError): + await client.call(raw, metadata=metadata, timeout=1) + assert endpoint.active is None and endpoint.grant is not None + assert run.lease.start == identity() and run.mailbox.bytes == 0 + assert control.accepted == 0 + + asyncio.run(check()) + + +def test_control_rate_and_single_outstanding_are_bounded(tmp_path, tls): + async def check(): + async with controlled(tmp_path, tls) as (_, _, control, client, _, _): + control.clock_ns = lambda: 10**16 + await client.poll() + with pytest.raises(grpc.aio.AioRpcError) as exc: + await client.poll() + assert exc.value.code() == grpc.StatusCode.RESOURCE_EXHAUSTED + client.polling = True + with pytest.raises(ValueError, match="outstanding"): + await client.poll() + assert control.accepted == control.rejected == 1 + + asyncio.run(check()) + + +@pytest.mark.parametrize( + "changes", + [ + {"source_id": "other"}, + {"clock_domain_id": "other"}, + {"input_mode": "recorded-source-paced"}, + {"lease_generation": 2}, + ], +) +def test_offer_is_bound_to_every_activation_field(changes): + access = StreamAccess(replace(identity(), **changes), "capture", 1, "a" * 64) + with pytest.raises(ValueError): + _offer(access, identity()) + + +@pytest.mark.parametrize("value", [None, 1, "x" * 64, "a" * 63]) +def test_invalid_capabilities_fail_closed_without_leaking_in_repr(value): + with pytest.raises(ValueError): + ControlTicket(identity(), value) + assert "b" * 64 not in repr(ControlTicket(identity(), "b" * 64)) + + +def test_malicious_response_nonce_or_offer_rejected(tls): + async def check(): + ticket = ControlTicket(identity(), "a" * 64) + client = StreamControlClient("localhost:1", tls[0], ticket, clock_id="mac") + for mode in ("nonce", "offer"): + + async def fake(raw, mode=mode, **kwargs): + request = wire.parse_header(bytearray(raw)) + return wire.canonical( + { + "schema_version": "missioncore.stream-control-poll/v1", + "activation_binding": wire.binding(identity()), + **request, + "nonce": "wrong" if mode == "nonce" else request["nonce"], + "remote_clock_id": "worker", + "remote_receive_ns": str(time.monotonic_ns()), + "remote_send_ns": str(time.monotonic_ns()), + "grant": { + "epoch": replace(identity(), source_id="foreign").to_dict(), + "session_id": "capture", + "session_generation": "1", + "token": "a" * 64, + }, + } + ) + + client.call = fake + with pytest.raises(ValueError): + await client.poll() + assert not client.polling + await client.close() + + asyncio.run(check())