refactor(platform): freeze laboratory and telemetry boundaries

src/k1link/compute/pipeline_telemetry.py

@@ -8,6 +8,7 @@ No sink is created implicitly and telemetry never grants command authority.
 
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 import re
@@ -30,6 +31,8 @@ SAFE_TOPIC_IDENTIFIER: Final = re.compile(
 )
 MAX_TEXT_LENGTH: Final = 256
 MAX_PAYLOAD_BYTES: Final = 1024 * 1024
+DEFAULT_JOURNAL_MAX_BYTES: Final = 64 * 1024 * 1024
+DEFAULT_JOURNAL_MAX_SEGMENTS: Final = 8
 STAGE_STATES: Final = frozenset({"started", "completed", "failed"})
 RUN_STATES: Final = STAGE_STATES
 _AUTHORITY: Final = {
@@ -243,10 +246,27 @@ class PipelineTelemetryEmitter:
 
 
 class JsonlPipelineTelemetrySink:
-    """Append topic-bound telemetry records for local, auditable execution evidence."""
+    """Append to a bounded, fail-closed local outbox for auditable execution evidence.
 
-    def __init__(self, path: Path) -> None:
+    Completed segments are content-addressed and never pruned implicitly.  When the
+    segment bound is reached, telemetry publication fails observably instead of
+    deleting an event that Telegraf may not have acknowledged yet.
+    """
+
+    def __init__(
+        self,
+        path: Path,
+        *,
+        max_bytes: int = DEFAULT_JOURNAL_MAX_BYTES,
+        max_segments: int = DEFAULT_JOURNAL_MAX_SEGMENTS,
+    ) -> None:
+        if max_bytes < MAX_PAYLOAD_BYTES + 4096:
+            raise PipelineTelemetryError("pipeline journal max_bytes is too small")
+        if not 1 <= max_segments <= 64:
+            raise PipelineTelemetryError("pipeline journal max_segments is invalid")
         self.path = path.expanduser().absolute()
+        self.max_bytes = max_bytes
+        self.max_segments = max_segments
         self._lock = threading.Lock()
 
     def publish(self, topic: str, payload: bytes) -> None:
@@ -261,6 +281,10 @@ class JsonlPipelineTelemetrySink:
         encoded = _canonical_json(record) + b"\n"
         with self._lock:
             self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+            if self.path.is_symlink():
+                raise PipelineTelemetryError("pipeline journal must not be a symlink")
+            if self.path.exists() and self.path.stat().st_size + len(encoded) > self.max_bytes:
+                self._rotate()
             descriptor = os.open(
                 self.path,
                 os.O_APPEND | os.O_CREAT | os.O_WRONLY,
@@ -272,6 +296,30 @@ class JsonlPipelineTelemetrySink:
             finally:
                 os.close(descriptor)
 
+    def _rotate(self) -> None:
+        if not self.path.is_file() or self.path.stat().st_size == 0:
+            return
+        segments = tuple(self.path.parent.glob(f"{self.path.stem}.*{self.path.suffix}"))
+        if len(segments) >= self.max_segments:
+            raise PipelineTelemetryError(
+                "pipeline journal segment bound reached; acknowledged segments require review"
+            )
+        digest = _file_sha256(self.path)
+        destination = self.path.with_name(
+            f"{self.path.stem}.{digest}{self.path.suffix}"
+        )
+        if destination.exists() or destination.is_symlink():
+            raise PipelineTelemetryError("pipeline journal segment identity already exists")
+        os.replace(self.path, destination)
+
+
+def _file_sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        for block in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
 
 class MqttPipelineTelemetrySink:
     """Publish through a worker-owned, already-connected Paho-compatible client."""