test(perception): localize transport tails with bounded TLS record witnesses

experiments/perception/worker/streaming_profile_stage1/pilot_clock_route_probe.py

@@ -11,6 +11,7 @@ from pathlib import Path
 
 import grpc
 from pilot_network_control import read_control, write_control
+from pilot_tls_record_trace import TlsRecordRelay
 
 from k1link.perception.realtime_contract import StreamStart
 from k1link.perception.streaming_clock import ClockMappingError, ClockWindow
@@ -65,8 +66,14 @@ def assessment(own, peer, now):
 async def client(args, report):
     bootstrap = read_control(Path(args.bootstrap))
     ticket = ControlTicket(StreamStart.from_dict(bootstrap["activation"]), bootstrap["token"])
+    relay = None
+    target = args.target
+    if getattr(args, "trace_relay", False):
+        host, port = target.rsplit(":", 1)
+        relay = TlsRecordRelay(host, int(port))
+        target = f"localhost:{await relay.start()}"
     client = StreamControlClient(
-        args.target,
+        target,
         Path(args.certificate).read_bytes(),
         ticket,
         clock_id="route-source-" + secrets.token_hex(12),
@@ -74,6 +81,30 @@ async def client(args, report):
     window = ClockWindow(
         client.clock_id, bootstrap["remote_clock_id"], rate_ppm=500, timestamp_error_ns=50_000
     )
+    calls = report.setdefault("rpc_calls", [])
+
+    def traced(call, method):
+        async def invoke(raw, **kwargs):
+            value = json.loads(raw)
+            nonce = value["nonce"] if method == "poll" else value["probe"]["nonce"]
+            if len(calls) >= 256:
+                raise ValueError("bounded RPC trace exceeded")
+            row = {"method": method, "nonce": nonce, "request_bytes": len(raw)}
+            calls.append(row)
+            row["invoked_ns"] = str(time.monotonic_ns())
+            try:
+                response = await call(raw, **kwargs)
+                row["completed_ns"] = str(time.monotonic_ns())
+                row["response_bytes"] = len(response)
+                return response
+            except BaseException:
+                row["failed_ns"] = str(time.monotonic_ns())
+                raise
+
+        return invoke
+
+    client.call = traced(client.call, "poll")
+    client.report_call = traced(client.report_call, "report")
     try:
         for index in range(args.samples):
             probe, grant = await client.poll()
@@ -102,6 +133,40 @@ async def client(args, report):
                 await asyncio.sleep(0.05)
     finally:
         await client.close()
+        if relay:
+            await relay.close()
+            report["tls_records"] = relay.snapshot()
+
+
+class TracedEndpoint(StreamControlEndpoint):
+    def __init__(self, *args, rpc_trace, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.rpc_trace = rpc_trace
+
+    async def _traced(self, method, handler, raw, context):
+        entered = time.monotonic_ns()
+        response = await handler(raw, context)
+        finished = time.monotonic_ns()
+        value = json.loads(response)
+        if len(self.rpc_trace) >= 256:
+            raise ValueError("bounded handler trace exceeded")
+        self.rpc_trace.append(
+            {
+                "method": method,
+                "nonce": value["nonce"],
+                "entered_ns": str(entered),
+                "returned_ns": str(finished),
+                "request_bytes": len(raw),
+                "response_bytes": len(response),
+            }
+        )
+        return response
+
+    async def poll(self, raw, context):
+        return await self._traced("poll", super().poll, raw, context)
+
+    async def report_clock(self, raw, context):
+        return await self._traced("report", super().report_clock, raw, context)
 
 
 async def server(args, report):
@@ -137,7 +202,13 @@ async def server(args, report):
         )
         return result
 
-    endpoint = StreamControlEndpoint(ticket, lambda: None, clock_id=clock_id, observe=observe)
+    endpoint = TracedEndpoint(
+        ticket,
+        lambda: None,
+        clock_id=clock_id,
+        observe=observe,
+        rpc_trace=report.setdefault("rpc_calls", []),
+    )
     rpc = grpc.aio.server(
         options=(
             ("grpc.max_send_message_length", MAX_CONTROL),
@@ -149,10 +220,18 @@ async def server(args, report):
     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):
+    traced = getattr(args, "trace_relay", False)
+    port = rpc.add_secure_port("127.0.0.1:0" if traced else args.target, credentials)
+    if not port:
         raise ValueError("clock diagnostic bind failed")
     await rpc.start()
+    relay = None
     try:
+        if traced:
+            host, external = args.target.rsplit(":", 1)
+            relay = TlsRecordRelay("127.0.0.1", port)
+            port = await relay.start(host, int(external))
+        report["listen_port"] = port
         write_control(
             Path(args.bootstrap),
             {
@@ -169,6 +248,9 @@ async def server(args, report):
     finally:
         Path(args.bootstrap).unlink(missing_ok=True)
         await rpc.stop(1)
+        if relay:
+            await relay.close()
+            report["tls_records"] = relay.snapshot()
 
 
 async def run(args):
@@ -188,6 +270,7 @@ async def run(args):
         "actuation_allowed": False,
         "samples": [],
         "error": None,
+        "trace_relay": getattr(args, "trace_relay", False),
     }
     witness = LoopWitness()
     task = asyncio.create_task(witness.run())
@@ -219,6 +302,7 @@ if __name__ == "__main__":
     parser.add_argument("--private-key")
     parser.add_argument("--samples", type=int, default=96)
     parser.add_argument("--output", required=True)
+    parser.add_argument("--trace-relay", action="store_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")

experiments/perception/worker/streaming_profile_stage1/pilot_tls_record_trace.py

@@ -0,0 +1,143 @@
+"""Bounded diagnostic relay: correlate encrypted TLS records, never decrypt/log payloads.
+
+For isolated CPU attribution only. This relay is NOT a runtime transport replacement.
+Receipt and drain timestamps describe application boundaries, not kernel ACK/wire time.
+"""
+
+import asyncio
+import hashlib
+import time
+from contextlib import suppress
+
+
+class RecordWitness:
+    def __init__(self, *, max_bytes=8 * 1024 * 1024, max_records=4096):
+        self.max_bytes, self.max_records = max_bytes, max_records
+        self.buffer = bytearray()
+        self.rows = []
+        self.total = self.offset = 0
+        self.first_ns = None
+
+    def feed(self, raw, received_ns, drained_ns):
+        if self.total + len(raw) > self.max_bytes:
+            raise ValueError("bounded encrypted stream exceeded")
+        self.total += len(raw)
+        self.buffer.extend(raw)
+        if self.first_ns is None:
+            self.first_ns = received_ns
+        while len(self.buffer) >= 5:
+            kind, major, minor = self.buffer[:3]
+            size = int.from_bytes(self.buffer[3:5], "big")
+            if kind not in (20, 21, 22, 23) or major != 3 or minor > 3 or size > 18432:
+                raise ValueError("invalid bounded TLS record")
+            length = size + 5
+            if len(self.buffer) < length:
+                return
+            if len(self.rows) >= self.max_records:
+                raise ValueError("bounded TLS record ledger exceeded")
+            self.rows.append(
+                {
+                    "offset": self.offset,
+                    "bytes": length,
+                    "type": kind,
+                    "sha256": hashlib.sha256(memoryview(self.buffer)[:length]).hexdigest(),
+                    "first_received_ns": str(self.first_ns),
+                    "received_ns": str(received_ns),
+                    "drained_ns": str(drained_ns),
+                }
+            )
+            del self.buffer[:length]
+            self.offset += length
+            self.first_ns = received_ns if self.buffer else None
+
+    def snapshot(self):
+        return {"bytes": self.total, "incomplete_bytes": len(self.buffer), "records": self.rows}
+
+
+class TlsRecordRelay:
+    """At most two local diagnostic connections; no listener outside its owner scope."""
+
+    def __init__(self, target_host, target_port):
+        self.target = (target_host, target_port)
+        self.server = None
+        self.connections = []
+        self.tasks = set()
+        self.writers = set()
+
+    async def start(self, host="127.0.0.1", port=0):
+        self.server = await asyncio.start_server(self._accept, host, port, limit=64 * 1024)
+        return self.server.sockets[0].getsockname()[1]
+
+    async def _accept(self, reader, writer):
+        task = asyncio.current_task()
+        self.tasks.add(task)
+        self.writers.add(writer)
+        row = {"request": RecordWitness(), "response": RecordWitness(), "error": None}
+        if len(self.connections) >= 2:
+            writer.close()
+            self.tasks.remove(task)
+            self.writers.remove(writer)
+            return
+        self.connections.append(row)
+        peer = None
+        jobs = []
+        try:
+            remote, peer = await asyncio.wait_for(asyncio.open_connection(*self.target), 2)
+            self.writers.add(peer)
+
+            async def pipe(source, sink, witness):
+                while True:
+                    raw = await asyncio.wait_for(source.read(64 * 1024), 10)
+                    received = time.monotonic_ns()
+                    if not raw:
+                        if sink.can_write_eof():
+                            sink.write_eof()
+                        return
+                    sink.write(raw)
+                    await asyncio.wait_for(sink.drain(), 0.5)
+                    drained = time.monotonic_ns()
+                    witness.feed(raw, received, drained)
+
+            jobs = [
+                asyncio.create_task(pipe(reader, peer, row["request"])),
+                asyncio.create_task(pipe(remote, writer, row["response"])),
+            ]
+            await asyncio.gather(*jobs)
+        except asyncio.CancelledError:
+            row["error"] = "cancelled-on-close"
+            raise
+        except Exception as exc:
+            row["error"] = type(exc).__name__
+        finally:
+            for job in jobs:
+                job.cancel()
+            await asyncio.gather(*jobs, return_exceptions=True)
+            for stream in (writer, peer):
+                if stream is not None:
+                    stream.close()
+                    with suppress(OSError, TimeoutError):
+                        await asyncio.wait_for(stream.wait_closed(), 1)
+                    self.writers.discard(stream)
+            self.tasks.discard(task)
+
+    async def close(self):
+        if self.server:
+            self.server.close()
+            await self.server.wait_closed()
+        if self.tasks:
+            _, pending = await asyncio.wait(self.tasks, timeout=0.3)
+            for task in pending:
+                task.cancel()
+            await asyncio.gather(*pending, return_exceptions=True)
+        for writer in self.writers:
+            writer.close()
+
+    def snapshot(self):
+        return [
+            {
+                "request": row["request"].snapshot(),
+                "response": row["response"].snapshot(),
+                "error": row["error"],
+            }
+            for row in self.connections
+        ]

tests/test_perception_tls_record_trace.py

@@ -0,0 +1,96 @@
+"""Bounded diagnostic attribution; sparse local TLS checks, no source/model workload."""
+
+# ruff: noqa: F811 -- imported shared fixtures
+import asyncio
+import hashlib
+import importlib
+from types import SimpleNamespace
+
+import pytest
+from test_perception_network_graph_adapter import adapter  # noqa: F401
+from test_perception_streaming_grpc import tls  # noqa: F401
+
+
+def record(payload=b"opaque ciphertext"):
+    return b"\x17\x03\x03" + len(payload).to_bytes(2, "big") + payload
+
+
+@pytest.mark.parametrize("cut", [1, 4, 5, 9, 21])
+def test_record_hash_identity_survives_arbitrary_tcp_chunks(adapter, cut):
+    module = importlib.import_module("pilot_tls_record_trace")
+    raw = record()
+    witness = module.RecordWitness()
+    witness.feed(raw[:cut], 10, 12)
+    witness.feed(raw[cut:] + raw, 20, 24)
+    result = witness.snapshot()
+    assert result["bytes"] == 2 * len(raw) and result["incomplete_bytes"] == 0
+    assert [r["offset"] for r in result["records"]] == [0, len(raw)]
+    assert all(r["sha256"] == hashlib.sha256(raw).hexdigest() for r in result["records"])
+    assert result["records"][0]["first_received_ns"] == "10"
+    assert result["records"][0]["received_ns"] == "20"
+    assert result["records"][1]["first_received_ns"] == "20"
+    assert "ciphertext" not in str(result)
+
+
+@pytest.mark.parametrize("fault", ["bytes", "records", "header", "record_length"])
+def test_trace_limits_fail_explicitly(adapter, fault):
+    module = importlib.import_module("pilot_tls_record_trace")
+    witness = module.RecordWitness(max_bytes=5 if fault == "bytes" else 1024, max_records=1)
+    raw = record() * 2
+    if fault == "header":
+        raw = b"\x00\x03\x03\x00\x00"
+    elif fault == "record_length":
+        raw = b"\x17\x03\x03\xff\xff"
+    with pytest.raises(ValueError, match="bounded"):
+        witness.feed(raw, 1, 2)
+
+
+def test_two_relays_match_encrypted_records_and_original_clock_calls(tmp_path, tls, adapter):
+    module = importlib.import_module("pilot_clock_route_probe")
+
+    async def check():
+        cert, key = tmp_path / "cert.pem", tmp_path / "key.pem"
+        cert.write_bytes(tls[0])
+        key.write_bytes(tls[1])
+        args = SimpleNamespace(
+            target="127.0.0.1:0",
+            bootstrap=tmp_path / "bootstrap.json",
+            certificate=cert,
+            private_key=key,
+            samples=8,
+            trace_relay=True,
+        )
+        server_report, client_report = {"samples": []}, {"samples": []}
+        task = asyncio.create_task(module.server(args, server_report))
+        try:
+            async with asyncio.timeout(2):
+                while "listen_port" not in server_report or not args.bootstrap.exists():
+                    await asyncio.sleep(0.01)
+            client_args = SimpleNamespace(**vars(args))
+            client_args.target = f"localhost:{server_report['listen_port']}"
+            await module.client(client_args, client_report)
+            await asyncio.wait_for(task, 2)
+            assert server_report["accepted"] == 8 and server_report["rejected"] == 0
+            assert len(client_report["rpc_calls"]) == len(server_report["rpc_calls"]) == 16
+            assert not args.bootstrap.exists()
+            for client, server in zip(
+                client_report["rpc_calls"], server_report["rpc_calls"], strict=True
+            ):
+                assert (client["method"], client["nonce"]) == (server["method"], server["nonce"])
+                assert int(client["invoked_ns"]) < int(server["entered_ns"])
+                assert int(server["returned_ns"]) < int(client["completed_ns"])
+            a, b = client_report["tls_records"], server_report["tls_records"]
+            assert len(a) == len(b) == 1
+            assert a[0]["error"] is b[0]["error"] is None
+            for direction in ("request", "response"):
+                x, y = a[0][direction], b[0][direction]
+                assert x["incomplete_bytes"] == y["incomplete_bytes"] == 0
+                assert [(r["offset"], r["sha256"]) for r in x["records"]] == [
+                    (r["offset"], r["sha256"]) for r in y["records"]
+                ]
+                assert len(x["records"]) >= 16
+        finally:
+            task.cancel()
+            await asyncio.gather(task, return_exceptions=True)
+
+    asyncio.run(check())