feat(perception): add scoped stream control and bounded clock observations

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

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),
+        )

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

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(

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)

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