feat(k1): complete primary acquisition lifecycle

src/k1link/device_plugins/xgrids_k1/archive.py

@@ -6,20 +6,29 @@ import math
 import os
 import re
 import stat
+import unicodedata
 from collections.abc import Mapping
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from datetime import UTC, datetime
 from functools import lru_cache
 from pathlib import Path
 from typing import IO, Any
 
 from k1link.device_plugins.xgrids_k1.mqtt.capture import (
+    CAPTURE_CLOCK_FILENAME,
+    CAPTURE_CLOCK_ORIGIN_FILENAME,
     FRAME_HEADER,
     GROUP_COMMIT_MAX_BYTES,
     GROUP_COMMIT_MAX_MESSAGES,
     MAX_CONFIGURABLE_MESSAGE_BYTES,
     MAX_TOPIC_BYTES,
     RAW_MAGIC,
+    CaptureClockEnvelope,
+    CaptureClockOrigin,
+    CaptureFormatError,
+    is_capture_clock_filename,
+    read_capture_clock_envelope,
+    read_capture_clock_origin,
 )
 from k1link.sessions.models import (
     LegacyMediaSourceCandidate,
@@ -58,8 +67,17 @@ class LegacySessionCandidate:
     raw_byte_length: int
     replay_raw_byte_length: int
     replay_metadata_byte_length: int
+    replay_capture_clock_origin_byte_length: int
+    replay_capture_clock_byte_length: int
     raw_sha256: str | None
+    capture_clock_origin_sha256: str | None
+    capture_clock_sha256: str | None
+    capture_clock_path: Path | None
     raw_integrity_status: str
+    timeline_origin_epoch_ns: int
+    timeline_origin_monotonic_ns: int
+    timeline_completion_epoch_ns: int
+    timeline_completion_monotonic_ns: int
     media_sources: tuple[LegacyMediaSourceCandidate, ...]
 
 
@@ -72,6 +90,14 @@ class _RecoveredCapture:
     duration_seconds: float
     raw_committed_bytes: int
     metadata_committed_bytes: int
+    first_epoch_ns: int
+    last_epoch_ns: int
+    first_monotonic_ns: int
+    last_monotonic_ns: int
+    capture_clock_origin: CaptureClockOrigin | None = None
+    capture_clock: CaptureClockEnvelope | None = None
+    capture_clock_path: Path | None = None
+    capture_clock_scope: str | None = None
 
 
 def discover_legacy_viewer_sessions(root: Path) -> tuple[LegacySessionCandidate, ...]:
@@ -137,13 +163,17 @@ def _describe_session(
     message_count = (
         _non_negative_int(summary.get("message_count"))
         if completed is not None
-        else recovered.message_count if recovered is not None else 0
+        else recovered.message_count
+        if recovered is not None
+        else 0
     )
     summary_topic_counts = summary.get("topic_counts")
     topic_counts = (
         _normalized_topic_counts(summary_topic_counts)
         if completed is not None
-        else recovered.topic_counts if recovered is not None else {}
+        else recovered.topic_counts
+        if recovered is not None
+        else {}
     )
     media_sources = _discover_media_sources(session_root)
     modalities = list(_modalities(topic_counts))
@@ -151,11 +181,25 @@ def _describe_session(
         modalities.append("video")
     replayable = bool(
         not active
-        and
-        raw_magic_ok
+        and raw_magic_ok
         and raw_bytes > len(RAW_MAGIC)
         and message_count > 0
         and any(modality in {"point-cloud", "trajectory"} for modality in modalities)
+        # A v2 transport-scoped completion is provisional. When camera media
+        # exists, only the owner-sealed session envelope can advertise a
+        # combined replay. Legacy captures have no scope and retain their
+        # reviewed first/last-message fallback.
+        and not (
+            media_sources
+            and (
+                completed.capture_clock_scope
+                if completed is not None
+                else recovered.capture_clock_scope
+                if recovered is not None
+                else None
+            )
+            in {"transport", "origin"}
+        )
     )
     if completed is not None:
         status = (
@@ -171,7 +215,9 @@ def _describe_session(
         completed_at = _safe_timestamp(summary.get("completed_at_utc")) or _safe_timestamp(
             manifest.get("completed_at_utc")
         )
-        duration = _duration(summary.get("capture_elapsed_seconds"))
+        duration = _duration(summary.get("session_elapsed_seconds"))
+        if duration is None:
+            duration = _duration(summary.get("capture_elapsed_seconds"))
         declared_hash = _declared_raw_hash(summary)
         integrity_status = "verified" if declared_hash is not None else "validated-structure"
     else:
@@ -190,17 +236,62 @@ def _describe_session(
     replay_raw_bytes = (
         completed.raw_committed_bytes
         if completed is not None
-        else recovered.raw_committed_bytes if recovered is not None else 0
+        else recovered.raw_committed_bytes
+        if recovered is not None
+        else 0
     )
     replay_metadata_bytes = (
         completed.metadata_committed_bytes
         if completed is not None
-        else recovered.metadata_committed_bytes if recovered is not None else 0
+        else recovered.metadata_committed_bytes
+        if recovered is not None
+        else 0
+    )
+    timing = completed if completed is not None else recovered
+    capture_clock = None if timing is None else timing.capture_clock
+    capture_clock_origin = None if timing is None else timing.capture_clock_origin
+    replay_capture_clock_origin_bytes = (
+        capture_clock_origin.artifact_byte_length if capture_clock_origin is not None else 0
+    )
+    replay_capture_clock_bytes = (
+        capture_clock.artifact_byte_length if capture_clock is not None else 0
+    )
+    timeline_origin_epoch_ns = (
+        capture_clock.started_at_epoch_ns
+        if capture_clock is not None
+        else capture_clock_origin.started_at_epoch_ns
+        if capture_clock_origin is not None
+        else timing.first_epoch_ns
+        if timing is not None
+        else 0
+    )
+    timeline_origin_monotonic_ns = (
+        capture_clock.started_monotonic_ns
+        if capture_clock is not None
+        else capture_clock_origin.started_monotonic_ns
+        if capture_clock_origin is not None
+        else timing.first_monotonic_ns
+        if timing is not None
+        else 0
+    )
+    timeline_completion_epoch_ns = (
+        capture_clock.completed_at_epoch_ns
+        if capture_clock is not None
+        else timing.last_epoch_ns
+        if timing is not None
+        else 0
+    )
+    timeline_completion_monotonic_ns = (
+        capture_clock.completed_monotonic_ns
+        if capture_clock is not None
+        else timing.last_monotonic_ns
+        if timing is not None
+        else 0
     )
 
     return LegacySessionCandidate(
         session_id=session_root.name,
-        display_name=session_root.name,
+        display_name=_project_display_name(manifest, fallback=session_root.name),
         status=status,
         started_at_utc=started_at,
         completed_at_utc=completed_at,
@@ -214,12 +305,41 @@ def _describe_session(
         raw_byte_length=raw_bytes,
         replay_raw_byte_length=replay_raw_bytes,
         replay_metadata_byte_length=replay_metadata_bytes,
+        replay_capture_clock_origin_byte_length=replay_capture_clock_origin_bytes,
+        replay_capture_clock_byte_length=replay_capture_clock_bytes,
         raw_sha256=declared_hash,
+        capture_clock_origin_sha256=(
+            capture_clock_origin.artifact_sha256 if capture_clock_origin is not None else None
+        ),
+        capture_clock_sha256=(capture_clock.artifact_sha256 if capture_clock is not None else None),
+        capture_clock_path=None if timing is None else timing.capture_clock_path,
         raw_integrity_status=integrity_status,
+        timeline_origin_epoch_ns=timeline_origin_epoch_ns,
+        timeline_origin_monotonic_ns=timeline_origin_monotonic_ns,
+        timeline_completion_epoch_ns=timeline_completion_epoch_ns,
+        timeline_completion_monotonic_ns=timeline_completion_monotonic_ns,
         media_sources=media_sources,
     )
 
 
+def _project_display_name(manifest: Mapping[str, Any], *, fallback: str) -> str:
+    value = manifest.get("project_name")
+    if not isinstance(value, str):
+        return fallback
+    normalized = unicodedata.normalize("NFKC", value).strip()
+    try:
+        normalized.encode("utf-8", errors="strict")
+    except UnicodeEncodeError:
+        return fallback
+    if (
+        not normalized
+        or len(normalized) > 96
+        or any(unicodedata.category(character) in {"Cc", "Cs"} for character in normalized)
+    ):
+        return fallback
+    return normalized
+
+
 def _is_completed_summary(summary: dict[str, Any]) -> bool:
     return (
         "message_count" in summary
@@ -254,7 +374,7 @@ def _validate_completed_capture(
         sort_keys=True,
         separators=(",", ":"),
     )
-    return _validate_completed_capture_cached(
+    validated = _validate_completed_capture_cached(
         str(capture_root),
         str(session_root),
         str(raw_path),
@@ -262,6 +382,81 @@ def _validate_completed_capture(
         _stat_identity(metadata_stat),
         fingerprint,
     )
+    if validated is None:
+        return None
+
+    schema_version = summary.get("schema_version", 1)
+    requires_capture_clock = (
+        isinstance(schema_version, int)
+        and not isinstance(schema_version, bool)
+        and schema_version >= 2
+    )
+    capture_clock_scope = summary.get("capture_clock_scope")
+    if requires_capture_clock and capture_clock_scope not in {"transport", "session"}:
+        return None
+    artifacts = summary.get("artifacts")
+    declared_name = artifacts.get("capture_clock") if isinstance(artifacts, dict) else None
+    declared_origin_name = (
+        artifacts.get("capture_clock_origin") if isinstance(artifacts, dict) else None
+    )
+    declared_hash = _declared_capture_clock_hash(summary)
+    declared_origin_hash = _declared_capture_clock_origin_hash(summary)
+    if declared_name is None and declared_hash is None and not requires_capture_clock:
+        return validated
+    if (
+        not is_capture_clock_filename(declared_name)
+        or declared_hash is None
+        or declared_origin_name != CAPTURE_CLOCK_ORIGIN_FILENAME
+        or declared_origin_hash is None
+        or (capture_clock_scope == "transport" and declared_name != CAPTURE_CLOCK_FILENAME)
+        or (
+            capture_clock_scope == "session"
+            and isinstance(declared_name, str)
+            and declared_name != f"mqtt.timeline.session-{declared_hash}.json"
+        )
+    ):
+        return None
+    assert isinstance(declared_name, str)
+    capture_clock_path = capture_root / declared_name
+    capture_clock_origin_path = capture_root / CAPTURE_CLOCK_ORIGIN_FILENAME
+    if not _confined_file(capture_clock_path, session_root) or not _confined_file(
+        capture_clock_origin_path,
+        session_root,
+    ):
+        return None
+    try:
+        capture_clock_origin = read_capture_clock_origin(
+            capture_clock_origin_path,
+            expected_sha256=declared_origin_hash,
+        )
+        capture_clock = read_capture_clock_envelope(
+            capture_clock_path,
+            expected_sha256=declared_hash,
+        )
+    except CaptureFormatError:
+        return None
+    if not (
+        capture_clock.started_at_epoch_ns == capture_clock_origin.started_at_epoch_ns
+        and capture_clock.started_monotonic_ns == capture_clock_origin.started_monotonic_ns
+        and capture_clock.started_monotonic_ns
+        <= validated.first_monotonic_ns
+        <= validated.last_monotonic_ns
+        <= capture_clock.completed_monotonic_ns
+    ):
+        return None
+    declared_duration = _duration(summary.get("session_elapsed_seconds"))
+    if requires_capture_clock and (
+        declared_duration is None
+        or abs(declared_duration - capture_clock.duration_ns / 1_000_000_000) > 1e-9
+    ):
+        return None
+    return replace(
+        validated,
+        capture_clock_origin=capture_clock_origin,
+        capture_clock=capture_clock,
+        capture_clock_path=capture_clock_path,
+        capture_clock_scope=(capture_clock_scope if isinstance(capture_clock_scope, str) else None),
+    )
 
 
 @lru_cache(maxsize=128)
@@ -312,13 +507,29 @@ def _recover_interrupted_capture(
     session_root: Path,
     raw_path: Path,
 ) -> _RecoveredCapture | None:
-    return _scan_capture_prefix(
+    recovered = _scan_capture_prefix(
         capture_root,
         session_root,
         raw_path,
         tolerate_incomplete_metadata_tail=True,
         tolerate_raw_crash_tail=True,
     )
+    if recovered is None:
+        return None
+    origin_path = capture_root / CAPTURE_CLOCK_ORIGIN_FILENAME
+    if not _confined_file(origin_path, session_root):
+        return recovered
+    try:
+        origin = read_capture_clock_origin(origin_path)
+    except CaptureFormatError:
+        return recovered
+    if origin.started_monotonic_ns > recovered.first_monotonic_ns:
+        return recovered
+    return replace(
+        recovered,
+        capture_clock_origin=origin,
+        capture_clock_scope="origin",
+    )
 
 
 def _scan_capture_prefix(
@@ -399,8 +610,7 @@ def _scan_capture_prefix(
                 return None
             raw_committed_bytes = raw_stream.tell()
             if raw_committed_bytes != raw_size and not (
-                tolerate_raw_crash_tail
-                and _tolerable_raw_crash_tail(raw_stream, raw_size=raw_size)
+                tolerate_raw_crash_tail and _tolerable_raw_crash_tail(raw_stream, raw_size=raw_size)
             ):
                 return None
     except OSError:
@@ -423,6 +633,10 @@ def _scan_capture_prefix(
         duration_seconds=(last_monotonic_ns - first_monotonic_ns) / 1_000_000_000,
         raw_committed_bytes=raw_committed_bytes,
         metadata_committed_bytes=metadata_committed_bytes,
+        first_epoch_ns=first_epoch_ns,
+        last_epoch_ns=last_epoch_ns,
+        first_monotonic_ns=first_monotonic_ns,
+        last_monotonic_ns=last_monotonic_ns,
     )
 
 
@@ -629,6 +843,22 @@ def _declared_metadata_hash(summary: dict[str, Any]) -> str | None:
     return value if isinstance(value, str) and SHA256_PATTERN.fullmatch(value) else None
 
 
+def _declared_capture_clock_hash(summary: dict[str, Any]) -> str | None:
+    hashes = summary.get("artifact_hashes")
+    if not isinstance(hashes, dict):
+        return None
+    value = hashes.get("capture_clock_sha256")
+    return value if isinstance(value, str) and SHA256_PATTERN.fullmatch(value) else None
+
+
+def _declared_capture_clock_origin_hash(summary: dict[str, Any]) -> str | None:
+    hashes = summary.get("artifact_hashes")
+    if not isinstance(hashes, dict):
+        return None
+    value = hashes.get("capture_clock_origin_sha256")
+    return value if isinstance(value, str) and SHA256_PATTERN.fullmatch(value) else None
+
+
 def _sha256_stable(path: Path) -> str:
     try:
         current = path.lstat()
@@ -815,8 +1045,7 @@ def _validated_media_epoch(epoch: Path, expected_source_id: str) -> bool:
     except (OSError, UnicodeDecodeError, json.JSONDecodeError):
         return False
     if len(index_records) != segment_count or not all(
-        isinstance(record, dict)
-        and _non_negative_int(record.get("sequence")) > 0
+        isinstance(record, dict) and _non_negative_int(record.get("sequence")) > 0
         for record in index_records
     ):
         return False

src/k1link/device_plugins/xgrids_k1/facade.py

@@ -8,18 +8,26 @@ import json
 import secrets
 import threading
 import time
-from collections.abc import Mapping
-from contextlib import suppress
+import unicodedata
+from collections.abc import Callable, Mapping
 from datetime import UTC, datetime
+from functools import wraps
 from pathlib import Path
-from typing import Any, Literal, Protocol, cast
+from typing import Any, Concatenate, Literal, Protocol, cast
 
 from bleak.exc import BleakError
 from missioncore_plugin_sdk.v0alpha2 import (
     RuntimeActionInvocation,
     RuntimePluginDescriptor,
 )
-from pydantic import BaseModel, ConfigDict, Field, SecretStr, ValidationError
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    Field,
+    SecretStr,
+    ValidationError,
+    field_validator,
+)
 
 from k1link.artifacts import write_json_atomic
 from k1link.device_plugins.xgrids_k1.ble.scanner import scan
@@ -36,6 +44,8 @@ from k1link.device_plugins.xgrids_k1.camera import (
     build_xgrids_k1_camera_router,
 )
 from k1link.device_plugins.xgrids_k1.mqtt import validate_private_ipv4
+from k1link.device_plugins.xgrids_k1.mqtt.capture import seal_capture_clock
+from k1link.device_plugins.xgrids_k1.protocol.modeling import observe_modeling_report
 from k1link.device_plugins.xgrids_k1.protocol.normalizer import normalize_k1_message
 from k1link.device_plugins.xgrids_k1.viewer.runtime import (
     VisualizationRuntime,
@@ -74,6 +84,33 @@ ACTION_ACQUISITION_PREPARE = "acquisition.prepare"
 ACTION_ACQUISITION_START = "acquisition.start"
 ACTION_ACQUISITION_STOP = "acquisition.stop"
 ACTION_ACQUISITION_ABORT = "acquisition.abort"
+
+
+def _serialized_acquisition_access[**P, R](
+    method: Callable[Concatenate[XgridsK1CompatibilityService, P], R],
+) -> Callable[Concatenate[XgridsK1CompatibilityService, P], R]:
+    """Serialize acquisition snapshots and producer lifecycle mutations.
+
+    Runtime/camera calls may synchronously cause a state read, so the gate is
+    deliberately re-entrant for the owning thread while excluding concurrent
+    request handlers from observing a half-completed producer handoff.
+    """
+
+    @wraps(method)
+    def serialized(
+        service: XgridsK1CompatibilityService,
+        *args: P.args,
+        **kwargs: P.kwargs,
+    ) -> R:
+        with service._acquisition_lifecycle_gate:  # noqa: SLF001
+            return method(service, *args, **kwargs)
+
+    return cast(
+        "Callable[Concatenate[XgridsK1CompatibilityService, P], R]",
+        serialized,
+    )
+
+
 ACTION_ACQUISITION_STATE_READ = "acquisition.state.read"
 ACTION_STREAM_START_LIVE = "stream.start-live"
 ACTION_STREAM_START_REPLAY = "stream.start-replay"
@@ -85,12 +122,14 @@ ACTION_VIEWER_SETTINGS_UPDATE = "viewer.settings.update"
 RequestedStreamId = Literal[
     "spatial.point-cloud.live",
     "spatial.pose.live",
+    "device.modeling.live",
     "device.status.live",
     "device.heartbeat.live",
 ]
 DEFAULT_LIVE_STREAMS: tuple[RequestedStreamId, ...] = (
     "spatial.point-cloud.live",
     "spatial.pose.live",
+    "device.modeling.live",
     "device.status.live",
     "device.heartbeat.live",
 )
@@ -130,10 +169,16 @@ class ConnectRequest(StrictRequest):
 
 
 class LiveRequest(StrictRequest):
+    project_name: str = Field(min_length=1, max_length=96)
     host: str | None = Field(default=None, max_length=15)
     duration_seconds: float = Field(default=3600.0, ge=1.0, le=86_400.0)
     compatibility_attestation: CompatibilityAttestationRequest
 
+    @field_validator("project_name")
+    @classmethod
+    def validate_project_name(cls, value: str) -> str:
+        return normalize_project_name(value)
+
 
 class OperationContextRequest(StrictRequest):
     operation_id: str | None = Field(default=None, min_length=1, max_length=128)
@@ -142,12 +187,18 @@ class OperationContextRequest(StrictRequest):
 
 
 class PrepareAcquisitionRequest(OperationContextRequest):
+    project_name: str = Field(min_length=1, max_length=96)
     host: str | None = Field(default=None, max_length=15)
     duration_seconds: float = Field(default=3600.0, ge=1.0, le=86_400.0)
     requested_streams: tuple[RequestedStreamId, ...] = DEFAULT_LIVE_STREAMS
     evidence_policy: Literal["required", "best-effort", "disabled"] = "required"
     compatibility_attestation: CompatibilityAttestationRequest
 
+    @field_validator("project_name")
+    @classmethod
+    def validate_project_name(cls, value: str) -> str:
+        return normalize_project_name(value)
+
 
 class StartAcquisitionRequest(OperationContextRequest):
     acquisition_id: str | None = Field(default=None, min_length=1, max_length=128)
@@ -198,6 +249,7 @@ class XgridsK1CompatibilityService:
         self.repository_root = repository_root.resolve()
         self.evidence_root = resolve_missioncore_evidence_dir(self.repository_root)
         self._lock = threading.Lock()
+        self._acquisition_lifecycle_gate = threading.RLock()
         self._provisioning_gate = threading.Lock()
         self._provisioning_active = False
         self._fingerprint_key = secrets.token_bytes(32)
@@ -219,18 +271,23 @@ class XgridsK1CompatibilityService:
         self._operation_message: str | None = None
         self._operations = OperationJournal()
         self._acquisition: AcquisitionRecord | None = None
+        self._acquisition_project_name: str | None = None
         self._acquisition_out_dir: Path | None = None
         self._acquisition_start_operation_id: str | None = None
         self._acquisition_stop_operation_id: str | None = None
         self._acquisition_session_lease: ActiveSessionLease | None = None
         # The host-owned visual runtime receives the vendor normalizer
         # explicitly. There is no implicit K1 decoder in the visual layer.
-        self.runtime = VisualizationRuntime(normalizer=normalize_k1_message)
+        self.runtime = VisualizationRuntime(
+            normalizer=normalize_k1_message,
+            message_observer=observe_modeling_report,
+        )
         self.camera_preview = XgridsK1CameraGateway(
             self.repository_root,
             XGRIDS_K1_PLUGIN_ID,
         )
 
+    @_serialized_acquisition_access
     def state(self) -> dict[str, Any]:
         runtime = self.runtime.snapshot()
         camera_preview = self.camera_preview.snapshot()
@@ -253,6 +310,12 @@ class XgridsK1CompatibilityService:
                 else None
             )
             acquisition = self._acquisition.as_dict() if self._acquisition is not None else None
+            if acquisition is not None:
+                acquisition["project_name"] = self._acquisition_project_name
+                acquisition["cleanup_pending"] = (
+                    self._acquisition_session_lease is not None
+                    and acquisition["state"] in TERMINAL_ACQUISITION_STATES
+                )
 
         runtime_active = runtime["source_mode"] != "idle" or runtime["phase"] in {
             "starting_live",
@@ -599,6 +662,7 @@ class XgridsK1CompatibilityService:
             }
         return self.state()
 
+    @_serialized_acquisition_access
     def prepare_acquisition(self, request: PrepareAcquisitionRequest) -> dict[str, Any]:
         requested_streams = _validated_requested_streams(request)
         target = request.host or self.state()["k1_ip"]
@@ -611,6 +675,15 @@ class XgridsK1CompatibilityService:
             raise ValueError(
                 "адрес точки доступа устройства нельзя использовать как direct-LAN target"
             )
+        with self._lock:
+            if self._acquisition_session_lease is not None:
+                raise RuntimeError("предыдущая evidence-сессия ещё не остановлена и не запечатана")
+        runtime_state = self.runtime.snapshot()
+        if runtime_state.get("source_mode") != "idle":
+            raise RuntimeError(
+                "нельзя готовить acquisition во время активного live/replay источника; "
+                "сначала остановите его"
+            )
         device_id, device_session_id = self._ensure_device_context()
         request_fingerprint = self._request_fingerprint(
             ACTION_ACQUISITION_PREPARE,
@@ -619,6 +692,7 @@ class XgridsK1CompatibilityService:
                 "duration_seconds": request.duration_seconds,
                 "requested_streams": requested_streams,
                 "evidence_policy": request.evidence_policy,
+                "project_name": request.project_name,
                 "compatibility_attestation": request.compatibility_attestation.model_dump(
                     mode="json"
                 ),
@@ -649,6 +723,10 @@ class XgridsK1CompatibilityService:
                     raise RuntimeError(
                         "нельзя готовить acquisition во время настройки Wi-Fi устройства"
                     )
+                if self._acquisition_session_lease is not None:
+                    raise RuntimeError(
+                        "предыдущая evidence-сессия ещё не остановлена и не запечатана"
+                    )
                 current = self._acquisition
                 if current is not None and current.state not in TERMINAL_ACQUISITION_STATES:
                     raise RuntimeError(
@@ -674,6 +752,7 @@ class XgridsK1CompatibilityService:
                     },
                 )
                 self._acquisition = acquisition
+                self._acquisition_project_name = request.project_name
                 self._acquisition_out_dir = new_live_session_dir(self.evidence_root)
                 self._acquisition_start_operation_id = None
                 self._acquisition_stop_operation_id = None
@@ -698,6 +777,7 @@ class XgridsK1CompatibilityService:
             raise
         return self.state()
 
+    @_serialized_acquisition_access
     def start_acquisition(self, request: StartAcquisitionRequest) -> dict[str, Any]:
         acquisition = self._require_acquisition(request.acquisition_id)
         if (
@@ -731,20 +811,31 @@ class XgridsK1CompatibilityService:
             message_code="acquisition.start.arming_receiver",
         )
 
+        owns_start = False
+        cleanup_failed = False
         try:
+            runtime_state = self.runtime.snapshot()
             with self._lock:
                 if acquisition.state != "prepared":
                     raise RuntimeError(
                         f"acquisition нельзя запустить из состояния {acquisition.state}"
                     )
+                if runtime_state.get("source_mode") != "idle":
+                    raise RuntimeError(
+                        "нельзя запускать acquisition поверх активного live/replay источника"
+                    )
                 out_dir = self._acquisition_out_dir
+                project_name = self._acquisition_project_name
                 if out_dir is None:
                     raise RuntimeError("для acquisition не выделена evidence-сессия")
+                if project_name is None:
+                    raise RuntimeError("для acquisition не задано название проекта")
                 acquisition.transition(
                     "starting",
                     message_code="acquisition.start.waiting_receiver_ready",
                 )
                 self._acquisition_start_operation_id = operation.operation_id
+                owns_start = True
             lease = ActiveSessionLease.acquire(out_dir.parent, out_dir)
             with self._lock:
                 if self._acquisition_session_lease is not None:
@@ -755,35 +846,51 @@ class XgridsK1CompatibilityService:
                 acquisition.target_host,
                 out_dir,
                 duration_seconds=acquisition.duration_seconds,
+                project_name=project_name,
             )
             self._arm_camera_recording(out_dir)
         except Exception as exc:
-            with suppress(Exception):
-                self.camera_preview.stop_recording(
-                    status="failed",
-                    failure_code="acquisition-start-failed",
-                )
-            with suppress(Exception):
-                self.runtime.stop()
-            self._release_acquisition_session_lease()
-            with self._lock:
-                if acquisition.state not in TERMINAL_ACQUISITION_STATES:
-                    acquisition.transition("failed", message_code="acquisition.start.failed")
+            if owns_start:
+                with self._lock:
+                    if acquisition.state not in TERMINAL_ACQUISITION_STATES:
+                        acquisition.transition(
+                            "stopping",
+                            message_code="acquisition.start.cleanup",
+                        )
+                try:
+                    self._stop_acquisition_sources(
+                        camera_status="failed",
+                        camera_failure_code="acquisition-start-failed",
+                    )
+                except Exception as cleanup_error:
+                    cleanup_failed = True
+                    exc.add_note(
+                        f"acquisition start cleanup also failed: {type(cleanup_error).__name__}"
+                    )
+                with self._lock:
+                    if acquisition.state not in TERMINAL_ACQUISITION_STATES:
+                        acquisition.transition("failed", message_code="acquisition.start.failed")
             self._operations.transition(
                 operation.operation_id,
                 "failed",
                 stage_code="failed",
                 message_code="acquisition.start.failed",
-                error=_operation_error(exc, category="stream", side_effect_status="none"),
+                error=_operation_error(
+                    exc,
+                    category="stream" if owns_start else "conflict",
+                    side_effect_status="unknown" if cleanup_failed else "none",
+                ),
             )
             raise
         return self.state()
 
+    @_serialized_acquisition_access
     def stop_acquisition(self, request: StopAcquisitionRequest) -> dict[str, Any]:
         acquisition = self._require_acquisition(request.acquisition_id)
         with self._lock:
             acquisition_state = acquisition.state
             expected_stop_operation_id = self._acquisition_stop_operation_id
+            lease_retained = self._acquisition_session_lease is not None
 
         if request.operator_confirmed:
             if request.mode != "graceful":
@@ -794,10 +901,15 @@ class XgridsK1CompatibilityService:
                 raise ValueError(
                     "подтверждение остановки должно ссылаться на исходную stop-operation"
                 )
-        elif request.mode == "graceful" and acquisition_state not in {
-            "acquiring",
-            "awaiting_external_stop",
-        }:
+        elif (
+            request.mode == "graceful"
+            and acquisition_state
+            not in {
+                "acquiring",
+                "awaiting_external_stop",
+            }
+            and not (acquisition_state in TERMINAL_ACQUISITION_STATES and lease_retained)
+        ):
             raise ValueError(
                 "graceful stop допустим только после подтверждённого потока point cloud"
             )
@@ -843,13 +955,51 @@ class XgridsK1CompatibilityService:
         if not created and not request.operator_confirmed:
             return self.state()
         if acquisition.state in TERMINAL_ACQUISITION_STATES:
-            self._operations.transition_if_pending(
-                operation.operation_id,
-                "succeeded",
-                stage_code="already-terminal",
-                message_code="acquisition.stop.already_terminal",
-                result={"acquisition_id": acquisition.acquisition_id},
-            )
+            if lease_retained:
+                self._operations.transition_if_pending(
+                    operation.operation_id,
+                    "running",
+                    stage_code="retrying-retained-cleanup",
+                    message_code="acquisition.stop.retrying_retained_cleanup",
+                )
+                try:
+                    completed = acquisition.state == "completed"
+                    self._stop_acquisition_sources(
+                        camera_status="complete" if completed else "failed",
+                        camera_failure_code=(None if completed else "acquisition-cleanup-retry"),
+                    )
+                except Exception as exc:
+                    self._operations.transition_if_pending(
+                        operation.operation_id,
+                        "failed",
+                        stage_code="retained-cleanup-failed",
+                        message_code="acquisition.stop.retained_cleanup_failed",
+                        error=_operation_error(
+                            exc,
+                            category="stream",
+                            side_effect_status="unknown",
+                        ),
+                    )
+                    raise
+                self._operations.transition_if_pending(
+                    operation.operation_id,
+                    "succeeded",
+                    stage_code="retained-cleanup-completed",
+                    message_code="acquisition.stop.retained_cleanup_completed",
+                    result={
+                        "acquisition_id": acquisition.acquisition_id,
+                        "receiver_stopped": True,
+                        "device_stop": "unknown",
+                    },
+                )
+            else:
+                self._operations.transition_if_pending(
+                    operation.operation_id,
+                    "succeeded",
+                    stage_code="already-terminal",
+                    message_code="acquisition.stop.already_terminal",
+                    result={"acquisition_id": acquisition.acquisition_id},
+                )
             return self.state()
 
         if request.mode == "graceful" and not request.operator_confirmed:
@@ -933,6 +1083,7 @@ class XgridsK1CompatibilityService:
             raise
         return self.state()
 
+    @_serialized_acquisition_access
     def abort_acquisition(self, request: AbortAcquisitionRequest) -> dict[str, Any]:
         acquisition = self._require_acquisition(request.acquisition_id)
         request_fingerprint = self._request_fingerprint(
@@ -951,7 +1102,14 @@ class XgridsK1CompatibilityService:
         if not created:
             return self.state()
         try:
-            if acquisition.state not in TERMINAL_ACQUISITION_STATES:
+            with self._lock:
+                should_abort = acquisition.state not in TERMINAL_ACQUISITION_STATES
+                if should_abort:
+                    acquisition.transition(
+                        "stopping",
+                        message_code="acquisition.aborting",
+                    )
+            if should_abort:
                 self._stop_acquisition_sources(
                     camera_status="interrupted",
                     camera_failure_code="acquisition-aborted",
@@ -997,6 +1155,7 @@ class XgridsK1CompatibilityService:
 
     def start_live(
         self,
+        project_name: str,
         host: str | None,
         duration_seconds: float,
         compatibility_attestation: CompatibilityAttestationRequest,
@@ -1005,6 +1164,7 @@ class XgridsK1CompatibilityService:
 
         prepared = self.prepare_acquisition(
             PrepareAcquisitionRequest(
+                project_name=project_name,
                 host=host,
                 duration_seconds=duration_seconds,
                 compatibility_attestation=compatibility_attestation,
@@ -1019,28 +1179,44 @@ class XgridsK1CompatibilityService:
             StartAcquisitionRequest(acquisition_id=acquisition["acquisition_id"])
         )
 
+    @_serialized_acquisition_access
     def start_replay(self, path: str, speed: float, loop: bool) -> dict[str, Any]:
         with self._lock:
             acquisition = self._acquisition
+            lease_retained = self._acquisition_session_lease is not None
             if acquisition is not None and acquisition.state not in TERMINAL_ACQUISITION_STATES:
                 raise RuntimeError(
                     "нельзя запускать replay во время активной acquisition-сессии; "
                     "сначала остановите или отмените её"
                 )
+            if lease_retained:
+                raise RuntimeError(
+                    "нельзя запускать replay: предыдущая evidence-сессия ещё не "
+                    "остановлена и не запечатана"
+                )
         replay_path = Path(path).expanduser().resolve()
         if not replay_path.is_relative_to(self.repository_root):
             raise ValueError("файл записи должен находиться внутри репозитория")
         self.runtime.start_replay(replay_path, speed=speed, loop=loop)
         return self.state()
 
+    @_serialized_acquisition_access
     def stop(self) -> dict[str, Any]:
         """Deprecated capture-only shim; it never claims that K1 stopped scanning."""
 
         with self._lock:
             acquisition = self._acquisition
+            lease_retained = self._acquisition_session_lease is not None
         if acquisition is None or acquisition.state in TERMINAL_ACQUISITION_STATES:
-            self.camera_preview.stop_current()
-            self.runtime.stop()
+            if lease_retained:
+                completed = acquisition is not None and acquisition.state == "completed"
+                self._stop_acquisition_sources(
+                    camera_status="complete" if completed else "failed",
+                    camera_failure_code=(None if completed else "acquisition-cleanup-retry"),
+                )
+            else:
+                self.camera_preview.stop_current()
+                self.runtime.stop()
             return self.state()
         return self.stop_acquisition(
             StopAcquisitionRequest(
@@ -1049,6 +1225,7 @@ class XgridsK1CompatibilityService:
             )
         )
 
+    @_serialized_acquisition_access
     def select_camera_preview(self, request: CameraPreviewSelectRequest) -> dict[str, Any]:
         target = self._camera_target_for_session(request.device_session_id)
         with self._lock:
@@ -1064,13 +1241,28 @@ class XgridsK1CompatibilityService:
         self.camera_preview.select(request.source_id, target)
         return self.state()
 
+    @_serialized_acquisition_access
     def stop_camera_preview(self, request: CameraPreviewStopRequest) -> dict[str, Any]:
         self._camera_target_for_session(request.device_session_id)
         self.camera_preview.stop(request.generation)
         return self.state()
 
+    @_serialized_acquisition_access
     def close(self) -> None:
+        with self._lock:
+            acquisition = self._acquisition
+            close_active_acquisition = (
+                acquisition is not None and acquisition.state not in TERMINAL_ACQUISITION_STATES
+            )
+            if close_active_acquisition:
+                assert acquisition is not None
+                acquisition.transition(
+                    "stopping",
+                    message_code="acquisition.shutdown",
+                )
         camera_error: Exception | None = None
+        runtime_error: Exception | None = None
+        terminal_error: Exception | None = None
         try:
             self.camera_preview.close()
         except Exception as exc:
@@ -1078,16 +1270,44 @@ class XgridsK1CompatibilityService:
         try:
             self.runtime.close()
         except Exception as exc:
+            runtime_error = exc
             if camera_error is not None:
                 exc.add_note(
                     "camera gateway cleanup also failed: "
                     f"{type(camera_error).__name__}: {camera_error}"
                 )
-            raise
+        if runtime_error is not None:
+            terminal_error = runtime_error
+        elif camera_error is not None:
+            terminal_error = camera_error
+        else:
+            try:
+                self._seal_acquisition_capture_clock()
+            except Exception as exc:
+                terminal_error = exc
+        try:
+            if close_active_acquisition:
+                with self._lock:
+                    assert acquisition is not None
+                    if acquisition.state not in TERMINAL_ACQUISITION_STATES:
+                        acquisition.transition(
+                            "failed" if terminal_error is not None else "interrupted",
+                            message_code=(
+                                "acquisition.shutdown.failed"
+                                if terminal_error is not None
+                                else "acquisition.shutdown.interrupted"
+                            ),
+                            result={
+                                "receiver_stopped": terminal_error is None,
+                                "device_state": "unknown",
+                            },
+                        )
+                self._terminalize_acquisition_operations_on_shutdown(terminal_error)
         finally:
-            self._release_acquisition_session_lease()
-        if camera_error is not None:
-            raise camera_error
+            if terminal_error is None:
+                self._release_acquisition_session_lease()
+        if terminal_error is not None:
+            raise terminal_error
 
     def update_viewer_settings(self, request: ViewerSettingsRequest) -> dict[str, Any]:
         self.runtime.update_scene_settings(
@@ -1141,6 +1361,7 @@ class XgridsK1CompatibilityService:
         camera_failure_code: str | None,
     ) -> None:
         camera_error: Exception | None = None
+        runtime_error: Exception | None = None
         try:
             self.camera_preview.stop_recording(
                 status=camera_status,
@@ -1151,16 +1372,33 @@ class XgridsK1CompatibilityService:
         try:
             self.runtime.stop()
         except Exception as exc:
+            runtime_error = exc
             if camera_error is not None:
                 exc.add_note(
                     "camera archive cleanup also failed: "
                     f"{type(camera_error).__name__}: {camera_error}"
                 )
-            raise
+        cleanup_complete = False
+        try:
+            if runtime_error is not None:
+                raise runtime_error
+            if camera_error is not None:
+                raise camera_error
+            self._seal_acquisition_capture_clock()
+            cleanup_complete = True
         finally:
-            self._release_acquisition_session_lease()
-        if camera_error is not None:
-            raise camera_error
+            if cleanup_complete:
+                self._release_acquisition_session_lease()
+
+    def _seal_acquisition_capture_clock(self) -> None:
+        with self._lock:
+            out_dir = self._acquisition_out_dir
+        # A prepared acquisition has reserved a future path but has no producer
+        # or clock to seal. Once the live session exists, any missing or
+        # inconsistent clock is a terminal evidence error.
+        if out_dir is None or not out_dir.exists():
+            return
+        seal_capture_clock(out_dir / "captures" / "mqtt_live")
 
     def _release_acquisition_session_lease(self) -> None:
         with self._lock:
@@ -1169,6 +1407,39 @@ class XgridsK1CompatibilityService:
         if lease is not None:
             lease.release()
 
+    def _terminalize_acquisition_operations_on_shutdown(
+        self,
+        terminal_error: Exception | None,
+    ) -> None:
+        with self._lock:
+            pending = (
+                ("start", self._acquisition_start_operation_id),
+                ("stop", self._acquisition_stop_operation_id),
+            )
+        for action_kind, operation_id in pending:
+            if terminal_error is None:
+                self._operations.transition_if_pending(
+                    operation_id,
+                    "cancelled",
+                    stage_code="service-shutdown",
+                    message_code=f"acquisition.{action_kind}.service_shutdown",
+                )
+            else:
+                self._operations.transition_if_pending(
+                    operation_id,
+                    "failed",
+                    stage_code="service-shutdown-failed",
+                    message_code=f"acquisition.{action_kind}.service_shutdown_failed",
+                    error=_operation_error(
+                        terminal_error,
+                        category="stream",
+                        side_effect_status="unknown",
+                    ),
+                )
+        with self._lock:
+            self._acquisition_start_operation_id = None
+            self._acquisition_stop_operation_id = None
+
     def _set_operation(self, phase: str, message: str) -> None:
         with self._lock:
             self._operation_phase = phase
@@ -1257,6 +1528,7 @@ class XgridsK1CompatibilityService:
         camera_terminal_status: Literal["complete", "failed"] | None = None
         camera_failure_code: str | None = None
         stop_runtime_for_camera_failure = False
+        complete_after_seal = False
         with self._lock:
             acquisition = self._acquisition
             if acquisition is None or acquisition.state in TERMINAL_ACQUISITION_STATES:
@@ -1345,12 +1617,15 @@ class XgridsK1CompatibilityService:
                 camera_failure_code = "receiver-completed-without-point-data"
                 failed_operation_id = self._acquisition_start_operation_id
             elif acquisition.state == "acquiring" and source_mode == "idle":
+                # Reserve terminal reconciliation before invoking camera or
+                # filesystem callbacks. A nested/concurrent state read now
+                # observes `finalizing` and cannot seal the same session twice.
                 acquisition.transition(
-                    "completed",
-                    message_code="acquisition.receiver_completed",
-                    result={"receiver_stopped": True, "device_state": "unknown"},
+                    "finalizing",
+                    message_code="acquisition.finalizing_capture_clock",
                 )
                 camera_terminal_status = "complete"
+                complete_after_seal = True
             elif acquisition.state == "awaiting_external_stop" and source_mode == "idle":
                 acquisition.transition(
                     "failed",
@@ -1361,16 +1636,53 @@ class XgridsK1CompatibilityService:
                 camera_failure_code = "receiver-completed-before-stop-confirmation"
                 unconfirmed_stop_operation_id = self._acquisition_stop_operation_id
 
+        reconciliation_error: Exception | None = None
         if camera_terminal_status is not None:
+            terminal_error: Exception | None = None
             try:
                 self.camera_preview.stop_recording(
                     status=camera_terminal_status,
                     failure_code=camera_failure_code,
                 )
+            except Exception as exc:
+                terminal_error = exc
+            if stop_runtime_for_camera_failure:
+                try:
+                    self.runtime.stop()
+                except Exception as exc:
+                    if terminal_error is not None:
+                        exc.add_note(
+                            "camera archive cleanup also failed: "
+                            f"{type(terminal_error).__name__}: {terminal_error}"
+                        )
+                    terminal_error = exc
+            try:
+                if terminal_error is not None:
+                    raise terminal_error
+                self._seal_acquisition_capture_clock()
+                if complete_after_seal:
+                    with self._lock:
+                        if acquisition.state not in TERMINAL_ACQUISITION_STATES:
+                            acquisition.transition(
+                                "completed",
+                                message_code="acquisition.receiver_completed",
+                                result={
+                                    "receiver_stopped": True,
+                                    "device_state": "unknown",
+                                },
+                            )
+            except Exception as exc:
+                reconciliation_error = exc
+                if complete_after_seal:
+                    with self._lock:
+                        if acquisition.state not in TERMINAL_ACQUISITION_STATES:
+                            acquisition.transition(
+                                "failed",
+                                message_code="acquisition.capture_clock_failed",
+                            )
             finally:
-                self._release_acquisition_session_lease()
-        if stop_runtime_for_camera_failure:
-            self.runtime.stop()
+                if reconciliation_error is None:
+                    self._release_acquisition_session_lease()
 
         if receiver_ready_operation_id is not None:
             self._operations.transition_if_pending(
@@ -1440,6 +1752,8 @@ class XgridsK1CompatibilityService:
                 self._acquisition_stop_operation_id = None
             if self._acquisition_stop_operation_id == unconfirmed_stop_operation_id:
                 self._acquisition_stop_operation_id = None
+        if reconciliation_error is not None:
+            raise reconciliation_error
 
 
 class XgridsK1ServicePort(Protocol):
@@ -1463,6 +1777,7 @@ class XgridsK1ServicePort(Protocol):
 
     def start_live(
         self,
+        project_name: str,
         host: str | None,
         duration_seconds: float,
         compatibility_attestation: CompatibilityAttestationRequest,
@@ -1565,6 +1880,7 @@ class XgridsK1PluginFacade:
             live_request = LiveRequest.model_validate(payload)
             return await asyncio.to_thread(
                 self.service.start_live,
+                live_request.project_name,
                 live_request.host,
                 live_request.duration_seconds,
                 live_request.compatibility_attestation,
@@ -1605,6 +1921,25 @@ def _utc_now_iso() -> str:
     return datetime.now(UTC).isoformat().replace("+00:00", "Z")
 
 
+def normalize_project_name(value: str) -> str:
+    """Return a bounded display name that is never interpreted as a path."""
+
+    normalized = unicodedata.normalize("NFKC", value).strip()
+    if not normalized:
+        raise ValueError("название проекта должно содержать от 1 до 96 символов")
+    try:
+        normalized.encode("utf-8", errors="strict")
+    except UnicodeEncodeError as exc:
+        raise ValueError(
+            "название проекта не должно содержать недопустимые Unicode-символы"
+        ) from exc
+    if len(normalized) > 96:
+        raise ValueError("название проекта должно содержать не более 96 символов")
+    if any(unicodedata.category(character) in {"Cc", "Cs"} for character in normalized):
+        raise ValueError("название проекта не должно содержать управляющие символы")
+    return normalized
+
+
 def _operation_error(
     exc: Exception,
     *,
@@ -1728,12 +2063,21 @@ def _sensor_catalog(
                 "frame_id": "map",
                 "coordinate_convention": "unverified",
             },
+            {
+                "stream_id": "device.modeling.live",
+                "sensor_kind": "status",
+                "modality": "acquisition-telemetry",
+                "availability": "observed",
+                "decode_status": "profile-decoded-physical-verified",
+                "frame_id": None,
+                "coordinate_convention": None,
+            },
             {
                 "stream_id": "device.status.live",
                 "sensor_kind": "status",
                 "modality": "device-status",
                 "availability": "observed",
-                "decode_status": "raw-only",
+                "decode_status": "profile-decoded-physical-verified",
                 "frame_id": None,
                 "coordinate_convention": None,
             },

src/k1link/device_plugins/xgrids_k1/mqtt/capture.py

@@ -5,6 +5,8 @@ import ipaddress
 import json
 import math
 import os
+import re
+import stat
 import struct
 import time
 from collections.abc import Callable, Iterator
@@ -36,6 +38,13 @@ LOOP_INTERVAL_SECONDS = 0.25
 GROUP_COMMIT_INTERVAL_SECONDS = 0.5
 GROUP_COMMIT_MAX_BYTES = 4 * 1024 * 1024
 GROUP_COMMIT_MAX_MESSAGES = 32
+CAPTURE_CLOCK_FILENAME = "mqtt.timeline.json"
+CAPTURE_CLOCK_ORIGIN_FILENAME = "mqtt.timeline.origin.json"
+CAPTURE_CLOCK_ORIGIN_SCHEMA_VERSION = 1
+CAPTURE_CLOCK_SEALED_PATTERN = re.compile(r"^mqtt\.timeline\.session-[a-f0-9]{64}\.json$")
+CAPTURE_CLOCK_SCHEMA_VERSION = 1
+MAX_CAPTURE_CLOCK_BYTES = 16 * 1024
+MAX_CAPTURE_CLOCK_SPAN_NS = (1 << 53) - 1
 
 # Eight-byte file signature followed by repeated >IQ, topic UTF-8 bytes, payload bytes.
 RAW_MAGIC = b"K1MQTT\x00\x01"
@@ -60,12 +69,16 @@ StopReason = Literal[
 class ArtifactPaths(TypedDict):
     raw: str
     metadata_jsonl: str
+    capture_clock_origin: str
+    capture_clock: str
     summary: str
 
 
 class ArtifactHashes(TypedDict):
     raw_sha256: str
     metadata_jsonl_sha256: str
+    capture_clock_origin_sha256: str
+    capture_clock_sha256: str
 
 
 class RawFormat(TypedDict):
@@ -78,6 +91,7 @@ class CaptureSummary(TypedDict):
     schema_version: int
     created_at_utc: str
     completed_at_utc: str
+    capture_clock_scope: Literal["transport", "session"]
     sensitivity: str
     target_ipv4: str
     target_port: int
@@ -89,6 +103,7 @@ class CaptureSummary(TypedDict):
     subscriptions: list[str]
     requested_duration_seconds: float
     capture_elapsed_seconds: float
+    session_elapsed_seconds: float
     operation_elapsed_seconds: float
     max_message_bytes: int
     connected: bool
@@ -105,6 +120,32 @@ class CaptureSummary(TypedDict):
     artifact_hashes: ArtifactHashes
 
 
+@dataclass(frozen=True, slots=True)
+class CaptureClockEnvelope:
+    """Durable host-clock bounds shared by MQTT, camera, and derived replay."""
+
+    started_at_epoch_ns: int
+    started_monotonic_ns: int
+    completed_at_epoch_ns: int
+    completed_monotonic_ns: int
+    artifact_byte_length: int
+    artifact_sha256: str
+
+    @property
+    def duration_ns(self) -> int:
+        return self.completed_monotonic_ns - self.started_monotonic_ns
+
+
+@dataclass(frozen=True, slots=True)
+class CaptureClockOrigin:
+    """Fsynced session zero published before any camera producer can start."""
+
+    started_at_epoch_ns: int
+    started_monotonic_ns: int
+    artifact_byte_length: int
+    artifact_sha256: str
+
+
 class CaptureError(RuntimeError):
     """A one-shot capture failed after preserving all artifacts written so far."""
 
@@ -161,6 +202,8 @@ class _CaptureWriter:
         self.out_dir = out_dir.expanduser().resolve()
         self.raw_path = self.out_dir / "mqtt.raw.k1mqtt"
         self.metadata_path = self.out_dir / "mqtt.metadata.jsonl"
+        self.capture_clock_origin_path = self.out_dir / CAPTURE_CLOCK_ORIGIN_FILENAME
+        self.capture_clock_path = self.out_dir / CAPTURE_CLOCK_FILENAME
         self.summary_path = self.out_dir / "mqtt.summary.json"
         self.max_message_bytes = max_message_bytes
         self.message_count = 0
@@ -172,10 +215,18 @@ class _CaptureWriter:
         self._pending_metadata: list[str] = []
         self._pending_raw_bytes = 0
         self._last_commit_monotonic = time.monotonic()
+        self._started_at_epoch_ns: int | None = None
+        self._started_monotonic_ns: int | None = None
 
     def open(self) -> None:
         self.out_dir.mkdir(parents=True, exist_ok=True)
-        artifact_paths = (self.raw_path, self.metadata_path, self.summary_path)
+        artifact_paths = (
+            self.raw_path,
+            self.metadata_path,
+            self.capture_clock_origin_path,
+            self.capture_clock_path,
+            self.summary_path,
+        )
         existing = [path.name for path in artifact_paths if path.exists()]
         if existing:
             names = ", ".join(existing)
@@ -186,6 +237,19 @@ class _CaptureWriter:
             self._raw.write(RAW_MAGIC)
             self._metadata = _open_text_exclusive(self.metadata_path)
             _fsync_directory(self.out_dir)
+            # This is the earliest durable point from which the capture can
+            # accept evidence. The camera is armed only after this writer is
+            # running, so both modalities share this exact monotonic zero.
+            self._started_at_epoch_ns = time.time_ns()
+            self._started_monotonic_ns = time.monotonic_ns()
+            _write_json_atomic_new(
+                self.capture_clock_origin_path,
+                {
+                    "schema_version": CAPTURE_CLOCK_ORIGIN_SCHEMA_VERSION,
+                    "started_at_epoch_ns": self._started_at_epoch_ns,
+                    "started_monotonic_ns": self._started_monotonic_ns,
+                },
+            )
         except BaseException:
             with suppress(OSError):
                 self.close()
@@ -262,8 +326,7 @@ class _CaptureWriter:
         if (
             len(self._pending_metadata) >= GROUP_COMMIT_MAX_MESSAGES
             or self._pending_raw_bytes >= GROUP_COMMIT_MAX_BYTES
-            or time.monotonic() - self._last_commit_monotonic
-            >= GROUP_COMMIT_INTERVAL_SECONDS
+            or time.monotonic() - self._last_commit_monotonic >= GROUP_COMMIT_INTERVAL_SECONDS
         ):
             self._commit_pending()
         return CapturedMqttMessage(
@@ -304,6 +367,31 @@ class _CaptureWriter:
         if first_error is not None:
             raise first_error
 
+    def finalize_capture_clock(self) -> CaptureClockEnvelope:
+        """Publish the exact capture envelope after every producer is sealed."""
+
+        started_at_epoch_ns = self._started_at_epoch_ns
+        started_monotonic_ns = self._started_monotonic_ns
+        if started_at_epoch_ns is None or started_monotonic_ns is None:
+            raise RuntimeError("capture writer has no start clock")
+        origin = read_capture_clock_origin(self.capture_clock_origin_path)
+        if (
+            origin.started_at_epoch_ns != started_at_epoch_ns
+            or origin.started_monotonic_ns != started_monotonic_ns
+        ):
+            raise CaptureFormatError("capture clock origin changed before finalization")
+        completed_at_epoch_ns = time.time_ns()
+        completed_monotonic_ns = time.monotonic_ns()
+        document = {
+            "schema_version": CAPTURE_CLOCK_SCHEMA_VERSION,
+            "started_at_epoch_ns": started_at_epoch_ns,
+            "started_monotonic_ns": started_monotonic_ns,
+            "completed_at_epoch_ns": completed_at_epoch_ns,
+            "completed_monotonic_ns": completed_monotonic_ns,
+        }
+        _write_json_atomic_new(self.capture_clock_path, document)
+        return read_capture_clock_envelope(self.capture_clock_path)
+
     def maybe_commit(self, now_monotonic: float | None = None) -> None:
         if not self._pending_metadata:
             return
@@ -348,9 +436,7 @@ class _CaptureWriter:
         metadata.write(payload)
         metadata.flush()
         os.fsync(metadata.fileno())
-        self._last_commit_monotonic = (
-            time.monotonic() if now_monotonic is None else now_monotonic
-        )
+        self._last_commit_monotonic = time.monotonic() if now_monotonic is None else now_monotonic
 
 
 def validate_private_ipv4(value: str) -> str:
@@ -437,6 +523,231 @@ def iter_capture_frames(
             )
 
 
+def read_capture_clock_envelope(
+    path: Path,
+    *,
+    expected_sha256: str | None = None,
+) -> CaptureClockEnvelope:
+    """Read one bounded, no-follow capture clock artifact and prove stability."""
+
+    candidate = path.expanduser().absolute()
+    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(candidate, flags)
+    except OSError as exc:
+        raise CaptureFormatError("capture clock artifact is unavailable") from exc
+    try:
+        before = os.fstat(descriptor)
+        if not stat.S_ISREG(before.st_mode) or not 1 <= before.st_size <= MAX_CAPTURE_CLOCK_BYTES:
+            raise CaptureFormatError("capture clock artifact is not a bounded regular file")
+        payload = bytearray()
+        while len(payload) <= MAX_CAPTURE_CLOCK_BYTES:
+            chunk = os.read(descriptor, min(4096, MAX_CAPTURE_CLOCK_BYTES + 1 - len(payload)))
+            if not chunk:
+                break
+            payload.extend(chunk)
+        after = os.fstat(descriptor)
+        try:
+            current = os.lstat(candidate)
+        except OSError as exc:
+            raise CaptureFormatError("capture clock artifact changed during validation") from exc
+        if (
+            len(payload) != before.st_size
+            or _file_identity(before) != _file_identity(after)
+            or stat.S_ISLNK(current.st_mode)
+            or (current.st_dev, current.st_ino) != (before.st_dev, before.st_ino)
+        ):
+            raise CaptureFormatError("capture clock artifact changed during validation")
+    finally:
+        os.close(descriptor)
+
+    digest = hashlib.sha256(payload).hexdigest()
+    if expected_sha256 is not None and digest != expected_sha256:
+        raise CaptureFormatError("capture clock digest does not match its summary")
+    try:
+        value = json.loads(payload.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise CaptureFormatError("capture clock artifact is not valid JSON") from exc
+    expected_keys = {
+        "schema_version",
+        "started_at_epoch_ns",
+        "started_monotonic_ns",
+        "completed_at_epoch_ns",
+        "completed_monotonic_ns",
+    }
+    if not isinstance(value, dict) or set(value) != expected_keys:
+        raise CaptureFormatError("capture clock artifact has an unsupported shape")
+    if value["schema_version"] != CAPTURE_CLOCK_SCHEMA_VERSION:
+        raise CaptureFormatError("capture clock schema is unsupported")
+    for key in expected_keys - {"schema_version"}:
+        item = value[key]
+        if not isinstance(item, int) or isinstance(item, bool) or item < 0:
+            raise CaptureFormatError("capture clock contains an invalid timestamp")
+    started_at_epoch_ns = value["started_at_epoch_ns"]
+    started_monotonic_ns = value["started_monotonic_ns"]
+    completed_at_epoch_ns = value["completed_at_epoch_ns"]
+    completed_monotonic_ns = value["completed_monotonic_ns"]
+    duration_ns = completed_monotonic_ns - started_monotonic_ns
+    if duration_ns < 0 or duration_ns > MAX_CAPTURE_CLOCK_SPAN_NS:
+        raise CaptureFormatError("capture clock monotonic span is outside bounds")
+    return CaptureClockEnvelope(
+        started_at_epoch_ns=started_at_epoch_ns,
+        started_monotonic_ns=started_monotonic_ns,
+        completed_at_epoch_ns=completed_at_epoch_ns,
+        completed_monotonic_ns=completed_monotonic_ns,
+        artifact_byte_length=len(payload),
+        artifact_sha256=digest,
+    )
+
+
+def read_capture_clock_origin(
+    path: Path,
+    *,
+    expected_sha256: str | None = None,
+) -> CaptureClockOrigin:
+    candidate = path.expanduser().absolute()
+    try:
+        payload = _read_stable_payload(candidate, MAX_CAPTURE_CLOCK_BYTES)
+    except OSError as exc:
+        raise CaptureFormatError("capture clock origin is unavailable") from exc
+    digest = hashlib.sha256(payload).hexdigest()
+    if expected_sha256 is not None and digest != expected_sha256:
+        raise CaptureFormatError("capture clock origin digest does not match its summary")
+    try:
+        value = json.loads(payload.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise CaptureFormatError("capture clock origin is not valid JSON") from exc
+    expected_keys = {
+        "schema_version",
+        "started_at_epoch_ns",
+        "started_monotonic_ns",
+    }
+    if not isinstance(value, dict) or set(value) != expected_keys:
+        raise CaptureFormatError("capture clock origin has an unsupported shape")
+    if value["schema_version"] != CAPTURE_CLOCK_ORIGIN_SCHEMA_VERSION:
+        raise CaptureFormatError("capture clock origin schema is unsupported")
+    epoch_ns = value["started_at_epoch_ns"]
+    monotonic_ns = value["started_monotonic_ns"]
+    if (
+        not isinstance(epoch_ns, int)
+        or isinstance(epoch_ns, bool)
+        or epoch_ns < 0
+        or not isinstance(monotonic_ns, int)
+        or isinstance(monotonic_ns, bool)
+        or monotonic_ns < 0
+    ):
+        raise CaptureFormatError("capture clock origin contains an invalid timestamp")
+    return CaptureClockOrigin(
+        started_at_epoch_ns=epoch_ns,
+        started_monotonic_ns=monotonic_ns,
+        artifact_byte_length=len(payload),
+        artifact_sha256=digest,
+    )
+
+
+def is_capture_clock_filename(value: object) -> bool:
+    return isinstance(value, str) and (
+        value == CAPTURE_CLOCK_FILENAME or CAPTURE_CLOCK_SEALED_PATTERN.fullmatch(value) is not None
+    )
+
+
+def seal_capture_clock(capture_root: Path) -> CaptureClockEnvelope:
+    """Extend the provisional MQTT clock after every session producer is sealed.
+
+    The acquisition owner must call this after camera shutdown and runtime
+    shutdown, but before releasing its active-session lease. A failure raises
+    ``CaptureError``. The content-addressed clock is published first and the
+    summary pointer switches second, so a crash leaves the old pointer valid
+    and a later owner retry can finish sealing.
+    """
+
+    try:
+        root = capture_root.expanduser().resolve(strict=True)
+    except OSError as exc:
+        raise CaptureError("capture clock root is unavailable") from exc
+    if not root.is_dir():
+        raise CaptureError("capture clock root is not a directory")
+    summary_path = root / "mqtt.summary.json"
+    try:
+        summary = _read_bounded_json_object(summary_path, MAX_CAPTURE_CLOCK_BYTES * 8)
+        artifacts = summary.get("artifacts")
+        hashes = summary.get("artifact_hashes")
+        declared_name = artifacts.get("capture_clock") if isinstance(artifacts, dict) else None
+        declared_origin_name = (
+            artifacts.get("capture_clock_origin") if isinstance(artifacts, dict) else None
+        )
+        if (
+            summary.get("schema_version") != 2
+            or not isinstance(artifacts, dict)
+            or not is_capture_clock_filename(declared_name)
+            or declared_origin_name != CAPTURE_CLOCK_ORIGIN_FILENAME
+            or not isinstance(hashes, dict)
+            or not isinstance(hashes.get("capture_clock_origin_sha256"), str)
+            or not isinstance(hashes.get("capture_clock_sha256"), str)
+        ):
+            raise CaptureFormatError("capture summary has no supported clock contract")
+        assert isinstance(declared_name, str)
+        origin = read_capture_clock_origin(
+            root / CAPTURE_CLOCK_ORIGIN_FILENAME,
+            expected_sha256=hashes["capture_clock_origin_sha256"],
+        )
+        clock_path = root / declared_name
+        current = read_capture_clock_envelope(
+            clock_path,
+            expected_sha256=hashes["capture_clock_sha256"],
+        )
+        if (
+            current.started_at_epoch_ns != origin.started_at_epoch_ns
+            or current.started_monotonic_ns != origin.started_monotonic_ns
+        ):
+            raise CaptureFormatError("capture clock does not match its durable origin")
+        scope = summary.get("capture_clock_scope")
+        if scope == "session":
+            if CAPTURE_CLOCK_SEALED_PATTERN.fullmatch(declared_name) is None:
+                raise CaptureFormatError("sealed capture summary points to a provisional clock")
+            if declared_name != f"mqtt.timeline.session-{current.artifact_sha256}.json":
+                raise CaptureFormatError(
+                    "sealed capture clock filename does not match its content digest"
+                )
+            return current
+        if scope != "transport" or declared_name != CAPTURE_CLOCK_FILENAME:
+            raise CaptureFormatError("provisional capture summary has an invalid clock pointer")
+        completed_at_epoch_ns = time.time_ns()
+        completed_monotonic_ns = time.monotonic_ns()
+        if completed_monotonic_ns < current.completed_monotonic_ns:
+            raise CaptureFormatError("session completion precedes MQTT completion")
+        document = {
+            "schema_version": CAPTURE_CLOCK_SCHEMA_VERSION,
+            "started_at_epoch_ns": current.started_at_epoch_ns,
+            "started_monotonic_ns": current.started_monotonic_ns,
+            "completed_at_epoch_ns": completed_at_epoch_ns,
+            "completed_monotonic_ns": completed_monotonic_ns,
+        }
+        sealed_digest = hashlib.sha256(_serialize_json(document)).hexdigest()
+        sealed_name = f"mqtt.timeline.session-{sealed_digest}.json"
+        sealed_path = root / sealed_name
+        # A retry can converge on an already durable content-addressed
+        # artifact. It is still fully revalidated below.
+        with suppress(FileExistsError):
+            _write_json_atomic_new(sealed_path, document)
+        sealed = read_capture_clock_envelope(
+            sealed_path,
+            expected_sha256=sealed_digest,
+        )
+        artifacts["capture_clock"] = sealed_name
+        hashes["capture_clock_sha256"] = sealed.artifact_sha256
+        summary["capture_clock_scope"] = "session"
+        summary["session_elapsed_seconds"] = round(
+            sealed.duration_ns / 1_000_000_000,
+            9,
+        )
+        summary["completed_at_utc"] = utc_now_iso()
+        _write_json_atomic_replace(summary_path, summary)
+        return sealed
+    except (OSError, CaptureFormatError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise CaptureError(f"could not seal session capture clock: {exc}") from exc
+
+
 def capture_mqtt(
     host: str,
     out_dir: Path,
@@ -444,6 +755,7 @@ def capture_mqtt(
     port: int = 1883,
     duration_seconds: float = 60.0,
     max_message_bytes: int = DEFAULT_MAX_MESSAGE_BYTES,
+    on_clock_established: Callable[[], None] | None = None,
     on_ready: Callable[[], None] | None = None,
     on_message_recorded: Callable[[CapturedMqttMessage], None] | None = None,
     should_stop: Callable[[], bool] | None = None,
@@ -471,7 +783,14 @@ def capture_mqtt(
         )
     )
     writer = _CaptureWriter(out_dir, max_message_bytes)
-    writer.open()
+    try:
+        writer.open()
+        if on_clock_established is not None:
+            on_clock_established()
+    except BaseException:
+        with suppress(OSError):
+            writer.close()
+        raise
     state = _CaptureState()
     created_at_utc = utc_now_iso()
     operation_started = time.monotonic()
@@ -615,7 +934,11 @@ def capture_mqtt(
             if state.error is None:
                 fail("capture_error", f"artifact close failed: {type(exc).__name__}: {exc}")
 
-    operation_completed = time.monotonic()
+    try:
+        capture_clock = writer.finalize_capture_clock()
+    except (OSError, RuntimeError, CaptureFormatError) as exc:
+        raise CaptureError(f"could not publish capture clock: {type(exc).__name__}: {exc}") from exc
+    operation_completed = capture_clock.completed_monotonic_ns / 1_000_000_000
     capture_elapsed = 0.0 if capture_started is None else operation_completed - capture_started
 
     summary = _build_summary(
@@ -627,6 +950,7 @@ def capture_mqtt(
         operation_elapsed=operation_completed - operation_started,
         max_message_bytes=max_message_bytes,
         created_at_utc=created_at_utc,
+        capture_clock=capture_clock,
         state=state,
     )
     try:
@@ -651,12 +975,15 @@ def _build_summary(
     operation_elapsed: float,
     max_message_bytes: int,
     created_at_utc: str,
+    capture_clock: CaptureClockEnvelope,
     state: _CaptureState,
 ) -> CaptureSummary:
+    capture_clock_origin = read_capture_clock_origin(writer.capture_clock_origin_path)
     return {
-        "schema_version": 1,
+        "schema_version": 2,
         "created_at_utc": created_at_utc,
         "completed_at_utc": utc_now_iso(),
+        "capture_clock_scope": "transport",
         "sensitivity": "contains raw K1 MQTT payloads and local addressing; do not commit",
         "target_ipv4": target_ipv4,
         "target_port": port,
@@ -668,6 +995,7 @@ def _build_summary(
         "subscriptions": list(REPORT_TOPICS),
         "requested_duration_seconds": duration_seconds,
         "capture_elapsed_seconds": round(capture_elapsed, 6),
+        "session_elapsed_seconds": round(capture_clock.duration_ns / 1_000_000_000, 9),
         "operation_elapsed_seconds": round(operation_elapsed, 6),
         "max_message_bytes": max_message_bytes,
         "connected": state.connected,
@@ -687,11 +1015,15 @@ def _build_summary(
         "artifacts": {
             "raw": writer.raw_path.name,
             "metadata_jsonl": writer.metadata_path.name,
+            "capture_clock_origin": writer.capture_clock_origin_path.name,
+            "capture_clock": writer.capture_clock_path.name,
             "summary": writer.summary_path.name,
         },
         "artifact_hashes": {
             "raw_sha256": _sha256_file(writer.raw_path),
             "metadata_jsonl_sha256": _sha256_file(writer.metadata_path),
+            "capture_clock_origin_sha256": capture_clock_origin.artifact_sha256,
+            "capture_clock_sha256": capture_clock.artifact_sha256,
         },
     }
 
@@ -740,6 +1072,105 @@ def _write_summary_exclusive(path: Path, summary: CaptureSummary) -> None:
     _fsync_directory(path.parent)
 
 
+def _write_json_atomic_new(path: Path, value: object) -> None:
+    """Publish a new JSON artifact atomically without replacing evidence."""
+
+    serialized = _serialize_json(value)
+    temporary = path.with_name(f".{path.name}.{os.getpid()}.{time.monotonic_ns()}.tmp")
+    descriptor = os.open(
+        temporary,
+        os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0),
+        0o600,
+    )
+    try:
+        view = memoryview(serialized)
+        while view:
+            written = os.write(descriptor, view)
+            view = view[written:]
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+    try:
+        # Hard-link publication is atomic and fails if another artifact has
+        # appeared at the destination; unlike replace it never overwrites.
+        os.link(temporary, path, follow_symlinks=False)
+        temporary.unlink()
+        _fsync_directory(path.parent)
+    except BaseException:
+        temporary.unlink(missing_ok=True)
+        raise
+
+
+def _write_json_atomic_replace(path: Path, value: object) -> None:
+    serialized = _serialize_json(value)
+    temporary = path.with_name(f".{path.name}.{os.getpid()}.{time.monotonic_ns()}.tmp")
+    descriptor = os.open(
+        temporary,
+        os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_CLOEXEC", 0),
+        0o600,
+    )
+    try:
+        view = memoryview(serialized)
+        while view:
+            written = os.write(descriptor, view)
+            view = view[written:]
+        os.fsync(descriptor)
+    finally:
+        os.close(descriptor)
+    try:
+        os.replace(temporary, path)
+        _fsync_directory(path.parent)
+    except BaseException:
+        temporary.unlink(missing_ok=True)
+        raise
+
+
+def _read_bounded_json_object(path: Path, max_bytes: int) -> dict[str, object]:
+    payload = _read_stable_payload(path, max_bytes)
+    value = json.loads(payload.decode("utf-8"))
+    if not isinstance(value, dict):
+        raise CaptureFormatError("JSON artifact is not an object")
+    return value
+
+
+def _read_stable_payload(path: Path, max_bytes: int) -> bytes:
+    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+    descriptor = os.open(path, flags)
+    try:
+        before = os.fstat(descriptor)
+        if not stat.S_ISREG(before.st_mode) or not 1 <= before.st_size <= max_bytes:
+            raise CaptureFormatError("artifact is not a bounded regular file")
+        chunks: list[bytes] = []
+        remaining = before.st_size
+        while remaining:
+            chunk = os.read(descriptor, min(4096, remaining))
+            if not chunk:
+                break
+            chunks.append(chunk)
+            remaining -= len(chunk)
+        payload = b"".join(chunks)
+        after = os.fstat(descriptor)
+        current = os.lstat(path)
+        if (
+            len(payload) != before.st_size
+            or _file_identity(before) != _file_identity(after)
+            or stat.S_ISLNK(current.st_mode)
+            or (current.st_dev, current.st_ino) != (before.st_dev, before.st_ino)
+        ):
+            raise CaptureFormatError("artifact changed during validation")
+        return payload
+    finally:
+        os.close(descriptor)
+
+
+def _file_identity(value: os.stat_result) -> tuple[int, int, int, int, int]:
+    return (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, value.st_ctime_ns)
+
+
+def _serialize_json(value: object) -> bytes:
+    return (json.dumps(value, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8")
+
+
 def _fsync_directory(path: Path) -> None:
     descriptor = os.open(path, os.O_RDONLY)
     try:

src/k1link/device_plugins/xgrids_k1/observation.py

@@ -2,10 +2,9 @@
 
 from __future__ import annotations
 
-import json
-import os
 import stat
 import threading
+from collections.abc import Mapping
 from pathlib import Path
 
 from k1link.device_plugins.xgrids_k1.archive import (
@@ -34,7 +33,6 @@ from k1link.sessions.store import resolve_missioncore_evidence_dir
 from k1link.web.camera_archive import recover_incomplete_camera_archives
 
 XGRIDS_K1_PLUGIN_ID = "nodedc.device.xgrids-lixelkity-k1"
-MAX_TIMELINE_ORIGIN_LINE_BYTES = 64 * 1024
 
 
 def build_xgrids_k1_observation(repository_root: Path) -> ObservationRuntimeContribution:
@@ -95,8 +93,12 @@ def _to_host_candidate(candidate: LegacySessionCandidate) -> ObservationSessionC
     raw_bytes = candidate.raw_byte_length
     replay_raw_bytes = candidate.replay_raw_byte_length
     replay_metadata_bytes = candidate.replay_metadata_byte_length
+    replay_capture_clock_origin_bytes = candidate.replay_capture_clock_origin_byte_length
+    replay_capture_clock_bytes = candidate.replay_capture_clock_byte_length
     replayable = candidate.replayable
     metadata_path = raw_path.with_name("mqtt.metadata.jsonl")
+    capture_clock_origin_path = raw_path.with_name("mqtt.timeline.origin.json")
+    capture_clock_path = candidate.capture_clock_path
 
     artifacts = [
         ObservationArtifactCandidate(
@@ -123,6 +125,34 @@ def _to_host_candidate(candidate: LegacySessionCandidate) -> ObservationSessionC
                 integrity_status=candidate.raw_integrity_status,
             )
         )
+    if replay_capture_clock_bytes > 0:
+        if capture_clock_path is None:
+            raise SessionIntegrityError("K1 capture clock locator is unavailable")
+        artifacts.append(
+            ObservationArtifactCandidate(
+                artifact_id="raw-transport-clock",
+                kind="raw-transport-clock",
+                media_type="application/vnd.nodedc.capture-clock+json",
+                locator=capture_clock_path,
+                byte_length=replay_capture_clock_bytes,
+                replay_byte_length=replay_capture_clock_bytes if replayable else 0,
+                sha256=candidate.capture_clock_sha256,
+                integrity_status="verified",
+            )
+        )
+    if replay_capture_clock_origin_bytes > 0:
+        artifacts.append(
+            ObservationArtifactCandidate(
+                artifact_id="raw-transport-clock-origin",
+                kind="raw-transport-clock-origin",
+                media_type="application/vnd.nodedc.capture-clock-origin+json",
+                locator=capture_clock_origin_path,
+                byte_length=replay_capture_clock_origin_bytes,
+                replay_byte_length=(replay_capture_clock_origin_bytes if replayable else 0),
+                sha256=candidate.capture_clock_origin_sha256,
+                integrity_status="verified",
+            )
+        )
 
     media_sources = candidate.media_sources
     artifacts.extend(
@@ -175,7 +205,6 @@ def _to_host_candidate(candidate: LegacySessionCandidate) -> ObservationSessionC
         for media in media_sources
     )
 
-    timeline_origin = _timeline_origin(metadata_path) if replayable else None
     return ObservationSessionCandidate(
         session_id=session_id,
         display_name=candidate.display_name,
@@ -189,8 +218,10 @@ def _to_host_candidate(candidate: LegacySessionCandidate) -> ObservationSessionC
         allowed_root=candidate.allowed_root,
         session_root=candidate.session_root,
         primary_replay_artifact_id="raw-transport-primary" if replayable else None,
-        timeline_origin_epoch_ns=None if timeline_origin is None else timeline_origin[0],
-        timeline_origin_monotonic_ns=None if timeline_origin is None else timeline_origin[1],
+        timeline_origin_epoch_ns=(candidate.timeline_origin_epoch_ns if replayable else None),
+        timeline_origin_monotonic_ns=(
+            candidate.timeline_origin_monotonic_ns if replayable else None
+        ),
         sources=tuple(sources),
         artifacts=tuple(artifacts),
     )
@@ -206,42 +237,11 @@ def _regular_file_size(path: Path) -> int:
     return value.st_size
 
 
-def _timeline_origin(path: Path) -> tuple[int, int]:
-    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
-    try:
-        descriptor = os.open(path, flags)
-        try:
-            payload = os.read(descriptor, MAX_TIMELINE_ORIGIN_LINE_BYTES + 1)
-        finally:
-            os.close(descriptor)
-        first_line = payload.splitlines(keepends=True)[0]
-        if len(first_line) > MAX_TIMELINE_ORIGIN_LINE_BYTES or not first_line.endswith(
-            (b"\n", b"\r")
-        ):
-            raise ValueError
-        document = json.loads(first_line)
-        epoch_ns = document.get("received_at_epoch_ns")
-        monotonic_ns = document.get("received_monotonic_ns")
-        if (
-            document.get("record_type") != "message"
-            or document.get("sequence") != 1
-            or not _non_negative_int(epoch_ns)
-            or not _non_negative_int(monotonic_ns)
-        ):
-            raise ValueError
-        return int(epoch_ns), int(monotonic_ns)
-    except (IndexError, OSError, UnicodeDecodeError, json.JSONDecodeError, ValueError) as exc:
-        raise SessionIntegrityError("K1 transport timeline origin is invalid") from exc
-
-
-def _non_negative_int(value: object) -> bool:
-    return isinstance(value, int) and not isinstance(value, bool) and value >= 0
-
-
 def _export_recording(
     source: Path,
     destination: Path,
     *,
+    artifacts: Mapping[str, Path] | None = None,
     cancel_event: threading.Event | None = None,
     activity_callback: object | None = None,
 ) -> dict[str, object]:
@@ -250,6 +250,12 @@ def _export_recording(
             export_k1mqtt_to_rrd(
                 source,
                 destination,
+                capture_clock_path=(
+                    None if artifacts is None else artifacts.get("raw-transport-clock")
+                ),
+                capture_clock_origin_path=(
+                    None if artifacts is None else artifacts.get("raw-transport-clock-origin")
+                ),
                 cancel_event=cancel_event,
                 activity_callback=activity_callback if callable(activity_callback) else None,
             )

src/k1link/device_plugins/xgrids_k1/protocol/modeling.py

@@ -0,0 +1,168 @@
+from __future__ import annotations
+
+import math
+import struct
+from dataclasses import dataclass
+from typing import Protocol
+
+from k1link.device_plugins.xgrids_k1.protocol.protobuf_wire import (
+    ProtobufWireError,
+    ProtoField,
+    iter_fields,
+)
+from k1link.viewer.metrics import BridgeMetrics
+
+MODELING_REPORT_TOPIC = "lixel/application/report/modeling"
+# Retained physical FW 3.0.2 captures independently correlate ScanTime with
+# host-monotonic duration at two ticks per second (one tick = 0.5 s). Keeping
+# the profile-scoped scale named avoids silently treating a vendor counter as
+# SI seconds.
+SCAN_TIME_TICKS_PER_SECOND = 2
+
+
+class ModelingReportDecodeError(ValueError):
+    """A K1 ModelingReport violated the profile-scoped protobuf contract."""
+
+
+@dataclass(frozen=True, slots=True)
+class ModelingTelemetry:
+    """Profile-scoped acquisition telemetry reported by the K1 itself."""
+
+    move_distance_meters: float
+    move_speed_meters_per_second: float
+    scan_time_ticks: int
+    pgo_progress: int
+
+    @property
+    def elapsed_seconds(self) -> float:
+        return self.scan_time_ticks / SCAN_TIME_TICKS_PER_SECOND
+
+
+class ModelingTransportMessage(Protocol):
+    @property
+    def topic(self) -> str: ...
+
+    @property
+    def payload(self) -> bytes: ...
+
+
+def is_modeling_report_topic(topic: str) -> bool:
+    return topic == MODELING_REPORT_TOPIC
+
+
+def observe_modeling_report(
+    message: ModelingTransportMessage,
+    metrics: BridgeMetrics,
+) -> bool:
+    """Consume one K1 ModelingReport before the visual preview queue."""
+
+    if not is_modeling_report_topic(message.topic):
+        return False
+    metrics.received(len(message.payload))
+    try:
+        telemetry = decode_modeling_report(message.payload)
+    except ModelingReportDecodeError:
+        metrics.modeling_decode_error()
+    else:
+        metrics.acquisition_telemetry(
+            elapsed_seconds=telemetry.elapsed_seconds,
+            route_distance_meters=telemetry.move_distance_meters,
+            speed_meters_per_second=telemetry.move_speed_meters_per_second,
+            pgo_progress=telemetry.pgo_progress,
+        )
+    return True
+
+
+def decode_modeling_report(
+    payload: bytes, *, max_payload_bytes: int = 64 * 1024
+) -> ModelingTelemetry:
+    """Decode the verified FW 3.0.2 ``ModelingReport`` status message.
+
+    The nested schema recovered from the reviewed LixelGO build is:
+    ``1 Header, 2 int32 PgoProgress, 3 ScanStatus`` and ScanStatus contains
+    ``1 float MoveDistance, 2 float MoveSpeed, 3 int64 ScanTime``. Header data
+    is intentionally not retained here because it can contain device identity
+    and an OpenAPI credential.
+    """
+
+    if len(payload) > max_payload_bytes:
+        raise ModelingReportDecodeError("ModelingReport exceeds configured limit")
+
+    pgo_progress = 0
+    scan_status: bytes | None = None
+    seen: set[int] = set()
+    try:
+        for field in iter_fields(payload, max_fields=64):
+            if field.number == 2:
+                _mark_once(seen, 2, "modeling.pgo_progress")
+                pgo_progress = _nonnegative_int(
+                    field,
+                    "modeling.pgo_progress",
+                    max_value=0x7FFF_FFFF,
+                )
+            elif field.number == 3:
+                _mark_once(seen, 3, "modeling.scan_status")
+                scan_status = _bytes(field, "modeling.scan_status")
+    except ProtobufWireError as exc:
+        raise ModelingReportDecodeError(f"invalid ModelingReport: {exc}") from exc
+
+    if scan_status is None:
+        raise ModelingReportDecodeError("ModelingReport has no scan_status")
+
+    distance = 0.0
+    speed = 0.0
+    scan_time = 0
+    seen.clear()
+    try:
+        for field in iter_fields(scan_status, max_fields=16):
+            if field.number == 1:
+                _mark_once(seen, 1, "scan_status.move_distance")
+                distance = _nonnegative_float32(field, "scan_status.move_distance")
+            elif field.number == 2:
+                _mark_once(seen, 2, "scan_status.move_speed")
+                speed = _nonnegative_float32(field, "scan_status.move_speed")
+            elif field.number == 3:
+                _mark_once(seen, 3, "scan_status.scan_time")
+                scan_time = _nonnegative_int(
+                    field,
+                    "scan_status.scan_time",
+                    max_value=0x7FFF_FFFF_FFFF_FFFF,
+                )
+    except ProtobufWireError as exc:
+        raise ModelingReportDecodeError(f"invalid ModelingReport.scan_status: {exc}") from exc
+
+    return ModelingTelemetry(
+        move_distance_meters=distance,
+        move_speed_meters_per_second=speed,
+        scan_time_ticks=scan_time,
+        pgo_progress=pgo_progress,
+    )
+
+
+def _bytes(field: ProtoField, name: str) -> bytes:
+    if field.wire_type != 2 or not isinstance(field.value, bytes):
+        raise ModelingReportDecodeError(f"{name} has the wrong protobuf wire type")
+    return field.value
+
+
+def _nonnegative_int(field: ProtoField, name: str, *, max_value: int) -> int:
+    if field.wire_type != 0 or not isinstance(field.value, int):
+        raise ModelingReportDecodeError(f"{name} has the wrong protobuf wire type")
+    if field.value > max_value:
+        raise ModelingReportDecodeError(f"{name} is outside its recovered signed range")
+    return field.value
+
+
+def _mark_once(seen: set[int], number: int, name: str) -> None:
+    if number in seen:
+        raise ModelingReportDecodeError(f"{name} is duplicated")
+    seen.add(number)
+
+
+def _nonnegative_float32(field: ProtoField, name: str) -> float:
+    if field.wire_type != 5 or not isinstance(field.value, bytes):
+        raise ModelingReportDecodeError(f"{name} has the wrong protobuf wire type")
+    value = float(struct.unpack("<f", field.value)[0])
+    if not math.isfinite(value) or value < 0:
+        raise ModelingReportDecodeError(f"{name} must be finite and nonnegative")
+    return value

src/k1link/device_plugins/xgrids_k1/protocol/modeling_control.py

@@ -0,0 +1,568 @@
+from __future__ import annotations
+
+import hmac
+from dataclasses import dataclass, field
+from enum import IntEnum
+
+from k1link.device_plugins.xgrids_k1.protocol.protobuf_wire import (
+    ProtobufWireError,
+    ProtoField,
+    iter_fields,
+)
+
+# The values below are scoped to the reviewed LixelGO/K1 protocol profile.  This
+# module deliberately has no MQTT dependency: it can build and validate bytes,
+# but it cannot send them to equipment.
+MAX_CONTROL_PAYLOAD_BYTES = 64 * 1024
+MAX_HEADER_BYTES = 4 * 1024
+MAX_TEXT_BYTES = 4 * 1024
+MODELING_SESSION_SUFFIX = ":ModelingRequest"
+
+OPENAPI_RESULT_BASE = 302_252_032
+OPENAPI_SUCCESS = OPENAPI_RESULT_BASE + 1
+MODELING_STATE_BASE = 302_252_032
+
+
+class ModelingProtocolError(ValueError):
+    """A bounded K1 modeling-control payload violated the recovered schema."""
+
+
+class ModelingEncodeError(ModelingProtocolError):
+    """A command could not be encoded without guessing vendor-owned input."""
+
+
+class ModelingResponseCorrelationError(ModelingProtocolError):
+    """A response did not match the exact command identity and action."""
+
+
+class ModelingCommandRejected(ModelingProtocolError):
+    """The device returned a non-success result for a correlated command."""
+
+    def __init__(self, code: int, description: str) -> None:
+        # Device descriptions are retained for a caller that needs structured
+        # diagnostics, but are not interpolated into the exception string.
+        super().__init__(f"K1 modeling command rejected with result code {code}")
+        self.code = code
+        self.description = description
+
+
+class ModelingAction(IntEnum):
+    START = 1
+    STOP = 2
+
+
+class RecordMode(IntEnum):
+    CALCULATE_ONLY = 0
+    RECORD_ONLY = 1
+    RECORD_AND_CALCULATE = 2
+
+
+class ScanMode(IntEnum):
+    POINT_CLOUD = 0
+    LCC = 1
+    PORTRAIT = 2
+
+
+class MountType(IntEnum):
+    HANDHELD = 0
+    VEHICLE = 1
+    UAV = 2
+    BACKPACK = 3
+
+
+class SessionState(IntEnum):
+    OTHER_STATUS = 0
+    READY = 300
+    SCAN_STARTING = 301
+    SCANNING = 302
+    SCAN_STOPPING = 303
+    SCAN_OVER = 304
+    DISK_ERROR = 305
+    SAVE_ERROR = 306
+    ALGORITHM_ERROR = 307
+    NO_CONTINUE_ERROR = 308
+    NO_SPACE_ERROR = 309
+    CAMERA_ERROR = 310
+    CONTINUE_SUCCESS = 311
+    CONTINUE_FAIL = 312
+    USB_DISK = 313
+    SD_SPACE_NOT_ENOUGH = 314
+    MEMORY_NOT_ENOUGH = 315
+    LIDAR_DATA_ERROR = 316
+    MAPPING_ERROR = 317
+
+    @property
+    def is_fault(self) -> bool:
+        return self in _FAULT_SESSION_STATES
+
+
+_FAULT_SESSION_STATES = frozenset(
+    {
+        SessionState.DISK_ERROR,
+        SessionState.SAVE_ERROR,
+        SessionState.ALGORITHM_ERROR,
+        SessionState.NO_CONTINUE_ERROR,
+        SessionState.NO_SPACE_ERROR,
+        SessionState.CAMERA_ERROR,
+        SessionState.CONTINUE_FAIL,
+        SessionState.SD_SPACE_NOT_ENOUGH,
+        SessionState.MEMORY_NOT_ENOUGH,
+        SessionState.LIDAR_DATA_ERROR,
+        SessionState.MAPPING_ERROR,
+    }
+)
+
+
+@dataclass(frozen=True, slots=True)
+class CommandHeaderIdentity:
+    """Exact recovered modeling identity; only the credential stays caller-provided."""
+
+    device_id: str = field(repr=False)
+    openapi_key: str = field(repr=False)
+
+    def __post_init__(self) -> None:
+        _validate_required_ascii_identity(self.device_id, "device_id", MAX_TEXT_BYTES)
+        _validate_required_ascii_identity(self.openapi_key, "openapi_key", MAX_TEXT_BYTES)
+        if len(self.session_id.encode("ascii")) > MAX_TEXT_BYTES:
+            raise ModelingEncodeError("derived modeling session_id exceeds configured limit")
+
+    @property
+    def session_id(self) -> str:
+        """Return the literal relation proven in the retained wire evidence."""
+
+        return f"{self.device_id}{MODELING_SESSION_SUFFIX}"
+
+    def matches(self, observed: ObservedHeader) -> bool:
+        """Compare every recovered identity field without exposing its value."""
+
+        if (
+            observed.device_id is None
+            or observed.session_id is None
+            or observed.openapi_key is None
+        ):
+            return False
+        return all(
+            hmac.compare_digest(expected.encode("ascii"), actual.encode("utf-8"))
+            for expected, actual in (
+                (self.device_id, observed.device_id),
+                (self.session_id, observed.session_id),
+                (self.openapi_key, observed.openapi_key),
+            )
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ObservedHeader:
+    """Identity subset decoded from an inbound header; repr stays redacted."""
+
+    device_id: str | None = field(default=None, repr=False)
+    session_id: str | None = field(default=None, repr=False)
+    openapi_key: str | None = field(default=None, repr=False)
+
+
+@dataclass(frozen=True, slots=True)
+class EncodedModelingCommand:
+    """An inert command envelope.  No transport operation exists in this type."""
+
+    action: ModelingAction
+    header: CommandHeaderIdentity = field(repr=False)
+    payload: bytes = field(repr=False)
+
+
+@dataclass(frozen=True, slots=True)
+class ModelingResponseError:
+    code: int
+    description: str = field(repr=False)
+
+
+@dataclass(frozen=True, slots=True)
+class ModelingResponse:
+    header: ObservedHeader = field(repr=False)
+    action: ModelingAction
+    error: ModelingResponseError
+
+
+@dataclass(frozen=True, slots=True)
+class DeviceStatusReport:
+    """Recovered acquisition subset of ``DeviceStatusReport``.
+
+    Vendor identity and project strings are intentionally redacted from repr.
+    Nested system/RTK messages are bounded by the outer parser but are not
+    interpreted by this modeling state machine.
+    """
+
+    header: ObservedHeader | None = field(repr=False)
+    modeling_state_code: int
+    session_state: SessionState | None
+    device_sn: str | None = field(repr=False)
+    project_id: str | None = field(repr=False)
+    time_zone: str | None = field(repr=False)
+    init_ready: bool
+    system_status_present: bool
+    rtk_status_present: bool
+
+
+def encode_modeling_start(
+    header: CommandHeaderIdentity,
+    *,
+    project_name: str,
+    record_mode: RecordMode,
+    scan_mode: ScanMode,
+    mount_type: MountType,
+    pre_project_id: str | None = None,
+) -> EncodedModelingCommand:
+    """Encode the recovered ``ModelingRequest`` start shape.
+
+    All modes are mandatory so that a future caller cannot inherit an implicit
+    vendor default.  Proto3 zero-valued enums are canonically omitted, matching
+    the reviewed application serializer.
+    """
+
+    _expect_exact_enum(record_mode, RecordMode, "record_mode")
+    _expect_exact_enum(scan_mode, ScanMode, "scan_mode")
+    _expect_exact_enum(mount_type, MountType, "mount_type")
+    _validate_required_text(project_name, "project_name", MAX_TEXT_BYTES)
+    if pre_project_id is not None:
+        _validate_required_text(pre_project_id, "pre_project_id", MAX_TEXT_BYTES)
+
+    fields = [
+        _bytes_field(1, _encode_command_header(header)),
+        _varint_field(2, ModelingAction.START),
+        _text_field(3, project_name),
+    ]
+    if record_mode:
+        fields.append(_varint_field(4, record_mode))
+    if scan_mode:
+        fields.append(_varint_field(5, scan_mode))
+    if mount_type:
+        fields.append(_varint_field(6, mount_type))
+    if pre_project_id is not None:
+        fields.append(_text_field(7, pre_project_id))
+
+    payload = b"".join(fields)
+    _ensure_encoded_bound(payload)
+    return EncodedModelingCommand(ModelingAction.START, header, payload)
+
+
+def encode_modeling_stop(header: CommandHeaderIdentity) -> EncodedModelingCommand:
+    """Encode the recovered stop shape: header plus action, and nothing else."""
+
+    payload = b"".join(
+        (
+            _bytes_field(1, _encode_command_header(header)),
+            _varint_field(2, ModelingAction.STOP),
+        )
+    )
+    _ensure_encoded_bound(payload)
+    return EncodedModelingCommand(ModelingAction.STOP, header, payload)
+
+
+def decode_modeling_response(
+    payload: bytes, *, max_payload_bytes: int = MAX_CONTROL_PAYLOAD_BYTES
+) -> ModelingResponse:
+    """Decode a bounded response without treating its description as success."""
+
+    _check_payload_bound(payload, max_payload_bytes, "ModelingResponse")
+    header: ObservedHeader | None = None
+    action: ModelingAction | None = None
+    response_error: ModelingResponseError | None = None
+    seen: set[int] = set()
+    try:
+        for proto_field in iter_fields(payload, max_fields=64):
+            if proto_field.number == 1:
+                _mark_once(seen, 1, "response.header")
+                header = _decode_observed_header(
+                    _bytes_value(proto_field, "response.header"), require_identity=True
+                )
+            elif proto_field.number == 2:
+                _mark_once(seen, 2, "response.action")
+                raw_action = _uint_value(proto_field, "response.action", max_value=0xFFFF_FFFF)
+                try:
+                    action = ModelingAction(raw_action)
+                except ValueError as exc:
+                    raise ModelingProtocolError("response.action is not start or stop") from exc
+            elif proto_field.number == 15:
+                _mark_once(seen, 15, "response.error")
+                response_error = _decode_response_error(_bytes_value(proto_field, "response.error"))
+    except ProtobufWireError as exc:
+        raise ModelingProtocolError(f"invalid ModelingResponse: {exc}") from exc
+
+    if header is None:
+        raise ModelingProtocolError("ModelingResponse has no header")
+    if action is None:
+        raise ModelingProtocolError("ModelingResponse has no action")
+    if response_error is None:
+        raise ModelingProtocolError("ModelingResponse has no error result")
+    return ModelingResponse(header, action, response_error)
+
+
+def correlate_modeling_response(
+    payload: bytes,
+    expected: EncodedModelingCommand,
+    *,
+    max_payload_bytes: int = MAX_CONTROL_PAYLOAD_BYTES,
+) -> ModelingResponse:
+    """Require identity, action, and the physically observed success code."""
+
+    response = decode_modeling_response(payload, max_payload_bytes=max_payload_bytes)
+    if not expected.header.matches(response.header):
+        raise ModelingResponseCorrelationError("response header identity mismatch")
+    if response.action is not expected.action:
+        raise ModelingResponseCorrelationError("response action mismatch")
+    if response.error.code != OPENAPI_SUCCESS:
+        raise ModelingCommandRejected(response.error.code, response.error.description)
+    return response
+
+
+def decode_device_status_report(
+    payload: bytes, *, max_payload_bytes: int = MAX_CONTROL_PAYLOAD_BYTES
+) -> DeviceStatusReport:
+    """Decode the fields that bound the K1 acquisition lifecycle."""
+
+    _check_payload_bound(payload, max_payload_bytes, "DeviceStatusReport")
+    header: ObservedHeader | None = None
+    modeling_state_code: int | None = None
+    device_sn: str | None = None
+    project_id: str | None = None
+    time_zone: str | None = None
+    init_ready = False
+    system_status_present = False
+    rtk_status_present = False
+    seen: set[int] = set()
+    try:
+        for proto_field in iter_fields(payload, max_fields=64):
+            if proto_field.number == 1:
+                _mark_once(seen, 1, "device_status.header")
+                header = _decode_observed_header(
+                    _bytes_value(proto_field, "device_status.header"), require_identity=False
+                )
+            elif proto_field.number == 2:
+                _mark_once(seen, 2, "device_status.modeling_state")
+                modeling_state_code = _uint_value(
+                    proto_field, "device_status.modeling_state", max_value=0xFFFF_FFFF
+                )
+            elif proto_field.number == 3:
+                _mark_once(seen, 3, "device_status.device_sn")
+                device_sn = _text_value(proto_field, "device_status.device_sn")
+            elif proto_field.number == 4:
+                _mark_once(seen, 4, "device_status.project_id")
+                project_id = _text_value(proto_field, "device_status.project_id")
+            elif proto_field.number == 5:
+                _mark_once(seen, 5, "device_status.time_zone")
+                time_zone = _text_value(proto_field, "device_status.time_zone")
+            elif proto_field.number == 6:
+                _mark_once(seen, 6, "device_status.init_ready")
+                raw_ready = _uint_value(proto_field, "device_status.init_ready", max_value=1)
+                init_ready = bool(raw_ready)
+            elif proto_field.number == 7:
+                _mark_once(seen, 7, "device_status.system_status")
+                _bytes_value(proto_field, "device_status.system_status")
+                system_status_present = True
+            elif proto_field.number == 8:
+                _mark_once(seen, 8, "device_status.rtk_status")
+                _bytes_value(proto_field, "device_status.rtk_status")
+                rtk_status_present = True
+    except ProtobufWireError as exc:
+        raise ModelingProtocolError(f"invalid DeviceStatusReport: {exc}") from exc
+
+    if modeling_state_code is None:
+        raise ModelingProtocolError("DeviceStatusReport has no modeling_state")
+    return DeviceStatusReport(
+        header=header,
+        modeling_state_code=modeling_state_code,
+        session_state=session_state_from_code(modeling_state_code),
+        device_sn=device_sn,
+        project_id=project_id,
+        time_zone=time_zone,
+        init_ready=init_ready,
+        system_status_present=system_status_present,
+        rtk_status_present=rtk_status_present,
+    )
+
+
+def session_state_from_code(code: int) -> SessionState | None:
+    """Map the captured base-offset status code, preserving unknown values."""
+
+    if code < MODELING_STATE_BASE:
+        return None
+    try:
+        return SessionState(code - MODELING_STATE_BASE)
+    except ValueError:
+        return None
+
+
+def _encode_command_header(header: CommandHeaderIdentity) -> bytes:
+    # seq/stamp/scaler were default-valued and omitted in the reviewed command
+    # requests.  The three identities must be supplied explicitly by a caller.
+    payload = b"".join(
+        (
+            _text_field(4, header.device_id),
+            _text_field(5, header.session_id),
+            _text_field(6, header.openapi_key),
+        )
+    )
+    if len(payload) > MAX_HEADER_BYTES:
+        raise ModelingEncodeError("encoded command header exceeds configured limit")
+    return payload
+
+
+def _decode_observed_header(payload: bytes, *, require_identity: bool) -> ObservedHeader:
+    if len(payload) > MAX_HEADER_BYTES:
+        raise ModelingProtocolError("header exceeds configured limit")
+    identities: dict[int, str] = {}
+    seen: set[int] = set()
+    try:
+        for proto_field in iter_fields(payload, max_fields=64):
+            if proto_field.number in {4, 5, 6}:
+                _mark_once(seen, proto_field.number, "header identity")
+                identities[proto_field.number] = _identity_text_value(
+                    proto_field,
+                    "header identity",
+                )
+    except ProtobufWireError as exc:
+        raise ModelingProtocolError(f"invalid header: {exc}") from exc
+
+    header = ObservedHeader(
+        device_id=identities.get(4),
+        session_id=identities.get(5),
+        openapi_key=identities.get(6),
+    )
+    if require_identity and (
+        header.device_id is None or header.session_id is None or header.openapi_key is None
+    ):
+        raise ModelingProtocolError("response header identity is incomplete")
+    return header
+
+
+def _decode_response_error(payload: bytes) -> ModelingResponseError:
+    if len(payload) > MAX_HEADER_BYTES:
+        raise ModelingProtocolError("response error exceeds configured limit")
+    code: int | None = None
+    description = ""
+    seen: set[int] = set()
+    try:
+        for proto_field in iter_fields(payload, max_fields=16):
+            if proto_field.number == 1:
+                _mark_once(seen, 1, "response.error.code")
+                code = _uint_value(proto_field, "response.error.code", max_value=0xFFFF_FFFF)
+            elif proto_field.number == 2:
+                _mark_once(seen, 2, "response.error.description")
+                description = _text_value(proto_field, "response.error.description")
+    except ProtobufWireError as exc:
+        raise ModelingProtocolError(f"invalid response error: {exc}") from exc
+    if code is None:
+        raise ModelingProtocolError("response error has no code")
+    return ModelingResponseError(code, description)
+
+
+def _expect_exact_enum(value: object, enum_type: type[IntEnum], name: str) -> None:
+    if type(value) is not enum_type:
+        raise ModelingEncodeError(f"{name} must be an explicit {enum_type.__name__}")
+
+
+def _validate_required_text(value: object, name: str, max_bytes: int) -> None:
+    if not isinstance(value, str) or not value:
+        raise ModelingEncodeError(f"{name} must be a non-empty string")
+    try:
+        encoded = value.encode("utf-8")
+    except UnicodeEncodeError as exc:
+        raise ModelingEncodeError(f"{name} is not valid UTF-8 text") from exc
+    if len(encoded) > max_bytes:
+        raise ModelingEncodeError(f"{name} exceeds configured limit")
+    if any(ord(character) < 0x20 or ord(character) == 0x7F for character in value):
+        raise ModelingEncodeError(f"{name} contains a control character")
+
+
+def _validate_required_ascii_identity(value: object, name: str, max_bytes: int) -> None:
+    if not isinstance(value, str) or not value:
+        raise ModelingEncodeError(f"{name} must be a non-empty string")
+    try:
+        encoded = value.encode("ascii")
+    except UnicodeEncodeError as exc:
+        raise ModelingEncodeError(f"{name} must use printable ASCII") from exc
+    if len(encoded) > max_bytes:
+        raise ModelingEncodeError(f"{name} exceeds configured limit")
+    if any(byte <= 0x20 or byte > 0x7E for byte in encoded):
+        raise ModelingEncodeError(f"{name} must use printable ASCII without spaces")
+
+
+def _mark_once(seen: set[int], number: int, name: str) -> None:
+    if number in seen:
+        raise ModelingProtocolError(f"{name} is duplicated")
+    seen.add(number)
+
+
+def _check_payload_bound(payload: bytes, max_payload_bytes: int, name: str) -> None:
+    if max_payload_bytes < 1:
+        raise ValueError("max_payload_bytes must be positive")
+    if len(payload) > max_payload_bytes:
+        raise ModelingProtocolError(f"{name} exceeds configured limit")
+
+
+def _ensure_encoded_bound(payload: bytes) -> None:
+    if len(payload) > MAX_CONTROL_PAYLOAD_BYTES:
+        raise ModelingEncodeError("encoded ModelingRequest exceeds configured limit")
+
+
+def _uint_value(proto_field: ProtoField, name: str, *, max_value: int) -> int:
+    if proto_field.wire_type != 0 or not isinstance(proto_field.value, int):
+        raise ModelingProtocolError(f"{name} has the wrong protobuf wire type")
+    if proto_field.value > max_value:
+        raise ModelingProtocolError(f"{name} exceeds its recovered integer range")
+    return proto_field.value
+
+
+def _bytes_value(proto_field: ProtoField, name: str) -> bytes:
+    if proto_field.wire_type != 2 or not isinstance(proto_field.value, bytes):
+        raise ModelingProtocolError(f"{name} has the wrong protobuf wire type")
+    return proto_field.value
+
+
+def _text_value(proto_field: ProtoField, name: str) -> str:
+    raw = _bytes_value(proto_field, name)
+    if len(raw) > MAX_TEXT_BYTES:
+        raise ModelingProtocolError(f"{name} exceeds configured limit")
+    try:
+        return raw.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise ModelingProtocolError(f"{name} is not valid UTF-8") from exc
+
+
+def _identity_text_value(proto_field: ProtoField, name: str) -> str:
+    value = _text_value(proto_field, name)
+    try:
+        encoded = value.encode("ascii")
+    except UnicodeEncodeError as exc:
+        raise ModelingProtocolError(f"{name} must use printable ASCII") from exc
+    if not encoded or any(byte <= 0x20 or byte > 0x7E for byte in encoded):
+        raise ModelingProtocolError(f"{name} must use printable ASCII without spaces")
+    return value
+
+
+def _varint(value: int) -> bytes:
+    if value < 0 or value > 0xFFFF_FFFF_FFFF_FFFF:
+        raise ModelingEncodeError("protobuf varint is outside uint64")
+    encoded = bytearray()
+    while value > 0x7F:
+        encoded.append((value & 0x7F) | 0x80)
+        value >>= 7
+    encoded.append(value)
+    return bytes(encoded)
+
+
+def _key(number: int, wire_type: int) -> bytes:
+    if number < 1:
+        raise ModelingEncodeError("protobuf field number must be positive")
+    return _varint((number << 3) | wire_type)
+
+
+def _varint_field(number: int, value: int) -> bytes:
+    return _key(number, 0) + _varint(int(value))
+
+
+def _bytes_field(number: int, value: bytes) -> bytes:
+    return _key(number, 2) + _varint(len(value)) + value
+
+
+def _text_field(number: int, value: str) -> bytes:
+    return _bytes_field(number, value.encode("utf-8"))

src/k1link/device_plugins/xgrids_k1/protocol/modeling_state.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import StrEnum
+
+from k1link.device_plugins.xgrids_k1.protocol.modeling_control import (
+    DeviceStatusReport,
+    SessionState,
+)
+
+
+class AcquisitionPhase(StrEnum):
+    UNOBSERVED = "unobserved"
+    READY = "ready"
+    CALIBRATING = "calibrating"
+    SCANNING = "scanning"
+    STOPPING = "stopping"
+    SCAN_OVER_UNVERIFIED = "scan_over_unverified"
+    FAULT = "fault"
+    UNKNOWN = "unknown"
+
+
+class SaveEvidence(StrEnum):
+    """Evidence levels seen in status reports; none means durable completion."""
+
+    NONE = "none"
+    SCAN_OVER_OBSERVED = "scan_over_observed"
+    SCAN_OVER_THEN_READY_OBSERVED = "scan_over_then_ready_observed"
+    FAULT_BEFORE_DURABLE_CONFIRMATION = "fault_before_durable_confirmation"
+
+
+@dataclass(frozen=True, slots=True)
+class AcquisitionSnapshot:
+    phase: AcquisitionPhase
+    session_state: SessionState | None
+    modeling_state_code: int | None
+    init_ready: bool
+    project_bound: bool
+    save_evidence: SaveEvidence
+
+    @property
+    def durable_save_complete(self) -> bool:
+        """Stay false until the physical save gate and artifacts are proven."""
+
+        return False
+
+
+class DeviceAcquisitionStateMachine:
+    """Observe K1 lifecycle reports without asserting an unproved save result.
+
+    The reviewed physical runs prove Ready -> ScanStarting -> Scanning and
+    Scanning -> ScanStopping.  ``ScanOver`` exists in the recovered enum, but a
+    physical stop run has not yet demonstrated the complete ScanOver/Ready plus
+    stable-artifact sequence.  Consequently every snapshot is fail-closed for
+    durable save completion.
+    """
+
+    def __init__(self) -> None:
+        self._scan_over_observed = False
+        self._ready_after_scan_over_observed = False
+        self._snapshot = AcquisitionSnapshot(
+            phase=AcquisitionPhase.UNOBSERVED,
+            session_state=None,
+            modeling_state_code=None,
+            init_ready=False,
+            project_bound=False,
+            save_evidence=SaveEvidence.NONE,
+        )
+
+    @property
+    def snapshot(self) -> AcquisitionSnapshot:
+        return self._snapshot
+
+    def observe(self, report: DeviceStatusReport) -> AcquisitionSnapshot:
+        state = report.session_state
+
+        # A new start invalidates evidence retained from a previous acquisition.
+        if state is SessionState.SCAN_STARTING or (
+            state is SessionState.SCANNING
+            and self._snapshot.session_state in {None, SessionState.READY, SessionState.SCAN_OVER}
+        ):
+            self._scan_over_observed = False
+            self._ready_after_scan_over_observed = False
+
+        if state is SessionState.SCAN_OVER:
+            self._scan_over_observed = True
+        elif state is SessionState.READY and self._scan_over_observed:
+            self._ready_after_scan_over_observed = True
+
+        phase = _phase_for(state)
+        if state is not None and state.is_fault:
+            evidence = SaveEvidence.FAULT_BEFORE_DURABLE_CONFIRMATION
+        elif self._ready_after_scan_over_observed:
+            evidence = SaveEvidence.SCAN_OVER_THEN_READY_OBSERVED
+        elif self._scan_over_observed:
+            evidence = SaveEvidence.SCAN_OVER_OBSERVED
+        else:
+            evidence = SaveEvidence.NONE
+
+        self._snapshot = AcquisitionSnapshot(
+            phase=phase,
+            session_state=state,
+            modeling_state_code=report.modeling_state_code,
+            init_ready=report.init_ready,
+            project_bound=bool(report.project_id),
+            save_evidence=evidence,
+        )
+        return self._snapshot
+
+
+def _phase_for(state: SessionState | None) -> AcquisitionPhase:
+    if state is None:
+        return AcquisitionPhase.UNKNOWN
+    if state is SessionState.READY:
+        return AcquisitionPhase.READY
+    if state is SessionState.SCAN_STARTING:
+        return AcquisitionPhase.CALIBRATING
+    if state is SessionState.SCANNING:
+        return AcquisitionPhase.SCANNING
+    if state is SessionState.SCAN_STOPPING:
+        return AcquisitionPhase.STOPPING
+    if state is SessionState.SCAN_OVER:
+        return AcquisitionPhase.SCAN_OVER_UNVERIFIED
+    if state.is_fault:
+        return AcquisitionPhase.FAULT
+    return AcquisitionPhase.UNKNOWN

src/k1link/device_plugins/xgrids_k1/rrd_export.py

@@ -16,6 +16,21 @@ import rerun as rr
 from rerun import blueprint as rrb
 
 from k1link.data_plane import DecodedPointCloudView, DecodedPoseView, NormalizationError
+from k1link.device_plugins.xgrids_k1.mqtt.capture import (
+    CAPTURE_CLOCK_FILENAME,
+    CAPTURE_CLOCK_ORIGIN_FILENAME,
+    CaptureClockEnvelope,
+    CaptureClockOrigin,
+    CaptureFormatError,
+    read_capture_clock_envelope,
+    read_capture_clock_origin,
+)
+from k1link.device_plugins.xgrids_k1.protocol.modeling import (
+    ModelingReportDecodeError,
+    ModelingTelemetry,
+    decode_modeling_report,
+    is_modeling_report_topic,
+)
 from k1link.device_plugins.xgrids_k1.protocol.normalizer import normalize_k1_message
 from k1link.device_plugins.xgrids_k1.viewer.replay import (
     ReplayFormatError,
@@ -45,6 +60,7 @@ JS_MAX_SAFE_INTEGER = (1 << 53) - 1
 RECORDED_SPATIAL_VIEW_ID = UUID("5f5f11d5-3b0a-4a81-887b-2be767cba1c0")
 RECORDED_ROOT_CONTAINER_ID = UUID("b02f2aca-8471-4dcb-b786-53df5a320fc8")
 RECORDED_POINTS_VISUALIZER_ID = UUID("ca037ec0-8761-4417-86ee-846fa2875303")
+RECORDED_METRICS_VIEW_ID = UUID("f973fc11-0867-4732-ad3c-97008621fab7")
 
 
 class RrdExportSummary(TypedDict):
@@ -58,6 +74,7 @@ class RrdExportSummary(TypedDict):
     decoded_messages: int
     point_frames: int
     pose_frames: int
+    modeling_reports: int
     ignored_messages: int
     points: int
     trajectory_poses: int
@@ -86,6 +103,7 @@ class _ExportCounters:
     source_messages: int = 0
     point_frames: int = 0
     pose_frames: int = 0
+    modeling_reports: int = 0
     ignored_messages: int = 0
     points: int = 0
     first_decoded_time_ns: int | None = None
@@ -166,15 +184,18 @@ def export_k1mqtt_to_rrd(
     input_path: Path,
     output_path: Path,
     *,
+    capture_clock_path: Path | None = None,
+    capture_clock_origin_path: Path | None = None,
     cancel_event: threading.Event | None = None,
     activity_callback: Callable[[], None] | None = None,
 ) -> RrdExportSummary:
     """Losslessly project every decodable K1 data-plane frame into one RRD.
 
     The raw capture remains the source of record. The derived RRD uses a
-    recording-local duration timeline whose zero is the first raw message's
-    receive-monotonic timestamp. It never traverses the bounded live-preview
-    queue, so export throughput cannot drop point or pose frames.
+    recording-local duration timeline whose zero is the durable capture-clock
+    origin for v2 recordings (or the first raw message for legacy captures).
+    It never traverses the bounded live-preview queue, so export throughput
+    cannot drop point or pose frames.
 
     The destination is replaced only after the temporary RRD has been closed,
     flushed and fsynced. Any decode, timing, sink or rename failure therefore
@@ -202,8 +223,30 @@ def export_k1mqtt_to_rrd(
 
     counters = _ExportCounters()
     trajectory = _TrajectoryBuffer.empty()
-    session_origin_ns: int | None = None
+    capture_clock = _optional_capture_clock(source, capture_clock_path)
+    capture_clock_origin = _optional_capture_clock_origin(
+        source,
+        capture_clock_origin_path,
+    )
+    if (
+        capture_clock is not None
+        and capture_clock_origin is not None
+        and (
+            capture_clock.started_at_epoch_ns != capture_clock_origin.started_at_epoch_ns
+            or capture_clock.started_monotonic_ns != capture_clock_origin.started_monotonic_ns
+        )
+    ):
+        raise RrdExportError("native capture clock does not match its durable origin")
+    session_origin_ns: int | None = (
+        capture_clock.started_monotonic_ns
+        if capture_clock is not None
+        else capture_clock_origin.started_monotonic_ns
+        if capture_clock_origin is not None
+        else None
+    )
     previous_monotonic_ns: int | None = None
+    last_source_time_ns: int | None = None
+    last_source_capture_ns: int | None = None
 
     try:
         recording = rr.RecordingStream(APPLICATION_ID, recording_id=recording_id)
@@ -230,12 +273,37 @@ def export_k1mqtt_to_rrd(
                 )
             if session_origin_ns is None:
                 session_origin_ns = monotonic_ns
+            if monotonic_ns < session_origin_ns or (
+                capture_clock is not None and monotonic_ns > capture_clock.completed_monotonic_ns
+            ):
+                raise RrdExportError(
+                    f"native capture message {message.sequence} is outside its clock envelope"
+                )
             session_time_ns = monotonic_ns - session_origin_ns
             if session_time_ns > JS_MAX_SAFE_INTEGER:
                 raise RrdExportError(
                     "session duration exceeds the exact JavaScript nanosecond range"
                 )
             previous_monotonic_ns = monotonic_ns
+            last_source_time_ns = session_time_ns
+            last_source_capture_ns = message.received_at_epoch_ns
+
+            if is_modeling_report_topic(message.topic):
+                try:
+                    telemetry = decode_modeling_report(message.payload)
+                except ModelingReportDecodeError as exc:
+                    raise RrdExportError(
+                        f"known K1 modeling report {message.sequence} failed validation"
+                    ) from exc
+                _set_frame_time(
+                    recording,
+                    message.sequence,
+                    session_time_ns,
+                    message.received_at_epoch_ns,
+                )
+                _log_modeling_telemetry(recording, telemetry)
+                counters.modeling_reports += 1
+                continue
 
             try:
                 decoded = normalize_k1_message(
@@ -279,6 +347,23 @@ def export_k1mqtt_to_rrd(
             raise RrdExportError("native capture contains no decodable point or pose frames")
         assert counters.first_decoded_time_ns is not None
         assert counters.last_decoded_time_ns is not None
+        if capture_clock is None:
+            assert last_source_time_ns is not None
+            assert last_source_capture_ns is not None
+            timeline_end_ns = last_source_time_ns
+            timeline_end_epoch_ns = last_source_capture_ns
+        else:
+            timeline_end_ns = capture_clock.duration_ns
+            timeline_end_epoch_ns = capture_clock.completed_at_epoch_ns
+        # Materialize the durable capture completion independently of any
+        # opaque MQTT packet. This is the exact envelope end and remains a
+        # real RRD row for the strict browser buffering gate.
+        _set_terminal_time(
+            recording,
+            timeline_end_ns,
+            timeline_end_epoch_ns,
+        )
+        _log_session_end(recording)
 
         _raise_if_cancelled(cancel_event)
         recording.flush(timeout_sec=30.0)
@@ -306,18 +391,17 @@ def export_k1mqtt_to_rrd(
             decoded_messages=counters.decoded_messages,
             point_frames=counters.point_frames,
             pose_frames=counters.pose_frames,
+            modeling_reports=counters.modeling_reports,
             ignored_messages=counters.ignored_messages,
             points=counters.points,
             trajectory_poses=len(trajectory.positions),
             trajectory_updates=trajectory.updates,
             session_origin_monotonic_ns=session_origin_ns,
             timeline_start_ns=0,
-            # Playback completeness is defined by data actually written to
-            # the RRD. K1 status/heartbeat packets may continue long after the
-            # final point or pose frame; advertising that raw tail as the RRD
-            # end makes a strict browser buffering gate wait forever.
-            timeline_end_ns=counters.last_decoded_time_ns,
-            timeline_span_ns=counters.last_decoded_time_ns,
+            # Both endpoints are real RRD rows. This keeps camera media inside
+            # one seekable session without advertising an unmaterialized tail.
+            timeline_end_ns=timeline_end_ns,
+            timeline_span_ns=timeline_end_ns,
             first_decoded_time_ns=counters.first_decoded_time_ns,
             last_decoded_time_ns=counters.last_decoded_time_ns,
             source_sha256=source_sha256,
@@ -356,6 +440,58 @@ def _validate_paths(source: Path, destination: Path) -> None:
         raise RrdExportError("RRD export accepts only native K1MQTT captures")
 
 
+def _optional_capture_clock(
+    source: Path,
+    explicit_path: Path | None,
+) -> CaptureClockEnvelope | None:
+    """Load the exact v2 capture envelope, retaining legacy-message fallback."""
+
+    path = explicit_path if explicit_path is not None else source.with_name(CAPTURE_CLOCK_FILENAME)
+    path = _confined_clock_artifact(source, path)
+    try:
+        path.lstat()
+    except FileNotFoundError:
+        return None
+    except OSError as exc:
+        raise RrdExportError("native capture clock is unavailable") from exc
+    try:
+        return read_capture_clock_envelope(path)
+    except CaptureFormatError as exc:
+        raise RrdExportError(f"native capture clock is invalid: {exc}") from exc
+
+
+def _optional_capture_clock_origin(
+    source: Path,
+    explicit_path: Path | None,
+) -> CaptureClockOrigin | None:
+    path = (
+        explicit_path
+        if explicit_path is not None
+        else source.with_name(CAPTURE_CLOCK_ORIGIN_FILENAME)
+    )
+    path = _confined_clock_artifact(source, path)
+    try:
+        path.lstat()
+    except FileNotFoundError:
+        return None
+    except OSError as exc:
+        raise RrdExportError("native capture clock origin is unavailable") from exc
+    try:
+        return read_capture_clock_origin(path)
+    except CaptureFormatError as exc:
+        raise RrdExportError(f"native capture clock origin is invalid: {exc}") from exc
+
+
+def _confined_clock_artifact(source: Path, path: Path) -> Path:
+    candidate = path.expanduser().absolute()
+    try:
+        if candidate.parent.resolve(strict=True) != source.parent.resolve(strict=True):
+            raise RrdExportError("native capture clock escapes its source directory")
+    except OSError as exc:
+        raise RrdExportError("native capture clock directory is unavailable") from exc
+    return candidate
+
+
 def _recorded_blueprint(
     settings: RerunSceneSettings,
     *,
@@ -384,9 +520,7 @@ def _recorded_blueprint(
         # each Points3D row. A uniform custom color is the one color mode
         # that can be replaced safely by a singleton blueprint override.
         colors=(
-            [_parse_hex_color(settings.custom_color)]
-            if settings.palette == "custom"
-            else None
+            [_parse_hex_color(settings.custom_color)] if settings.palette == "custom" else None
         ),
     ).visualizer()
     point_visualizer.id = RECORDED_POINTS_VISUALIZER_ID
@@ -417,7 +551,12 @@ def _recorded_blueprint(
         time_ranges=time_ranges,
     )
     spatial_view.id = RECORDED_SPATIAL_VIEW_ID
-    root_container = rrb.Tabs(spatial_view)
+    metrics_view = rrb.TimeSeriesView(
+        origin="/metrics/device",
+        name="Маршрут и время",
+    )
+    metrics_view.id = RECORDED_METRICS_VIEW_ID
+    root_container = rrb.Tabs(spatial_view, metrics_view, active_tab=0)
     root_container.id = RECORDED_ROOT_CONTAINER_ID
 
     if include_initial_playback_state:
@@ -490,6 +629,21 @@ def _log_static_scene(recording: rr.RecordingStream) -> None:
         rr.TransformAxes3D(axis_length=0.45, show_frame=False),
         static=True,
     )
+    recording.log(
+        "/metrics/device/route_distance_meters",
+        rr.SeriesLines(names=["Дистанция, м"], colors=[[67, 191, 255, 255]]),
+        static=True,
+    )
+    recording.log(
+        "/metrics/device/speed_meters_per_second",
+        rr.SeriesLines(names=["Скорость, м/с"], colors=[[255, 193, 92, 255]]),
+        static=True,
+    )
+    recording.log(
+        "/metrics/device/scan_elapsed_seconds",
+        rr.SeriesLines(names=["Время сканирования, с"], colors=[[126, 224, 160, 255]]),
+        static=True,
+    )
 
 
 def _log_session_origin(recording: rr.RecordingStream) -> None:
@@ -505,6 +659,31 @@ def _log_session_origin(recording: rr.RecordingStream) -> None:
     )
 
 
+def _log_session_end(recording: rr.RecordingStream) -> None:
+    recording.log(
+        "/__mission_core/session_end",
+        rr.AnyValues(session_end=True),
+    )
+
+
+def _log_modeling_telemetry(
+    recording: rr.RecordingStream,
+    telemetry: ModelingTelemetry,
+) -> None:
+    recording.log(
+        "/metrics/device/route_distance_meters",
+        rr.Scalars([telemetry.move_distance_meters]),
+    )
+    recording.log(
+        "/metrics/device/speed_meters_per_second",
+        rr.Scalars([telemetry.move_speed_meters_per_second]),
+    )
+    recording.log(
+        "/metrics/device/scan_elapsed_seconds",
+        rr.Scalars([telemetry.elapsed_seconds]),
+    )
+
+
 def _set_frame_time(
     recording: rr.RecordingStream,
     sequence: int,
@@ -522,6 +701,22 @@ def _set_frame_time(
     recording.set_time("message_sequence", sequence=sequence)
 
 
+def _set_terminal_time(
+    recording: rr.RecordingStream,
+    session_time_ns: int,
+    capture_time_ns: int,
+) -> None:
+    recording.set_time(
+        SESSION_TIMELINE,
+        duration=np.timedelta64(session_time_ns, "ns"),
+    )
+    recording.set_time(
+        CAPTURE_TIMELINE,
+        timestamp=np.datetime64(capture_time_ns, "ns"),
+    )
+    recording.disable_timeline("message_sequence")
+
+
 def _log_points(
     recording: rr.RecordingStream,
     frame: DecodedPointCloudView,

src/k1link/device_plugins/xgrids_k1/viewer/runtime.py

@@ -5,6 +5,7 @@ import queue
 import threading
 import time
 from collections.abc import Callable
+from contextlib import suppress
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Literal, Protocol, TypedDict
@@ -25,6 +26,7 @@ RuntimePhase = Literal[
     "stopping",
     "error",
 ]
+LIVE_CAPTURE_CLOCK_READY_TIMEOUT_SECONDS = 20.0
 SourceMode = Literal["idle", "live", "replay"]
 StateCallback = Callable[[], None]
 BridgeFactory = Callable[..., RerunBridge]
@@ -43,6 +45,10 @@ class CanonicalNormalizer(Protocol):
     ) -> DecodedDataPlaneView | None: ...
 
 
+class RawMessageObserver(Protocol):
+    def __call__(self, message: StreamMessage, metrics: BridgeMetrics) -> bool: ...
+
+
 class RuntimeSnapshot(TypedDict):
     phase: RuntimePhase
     message: str
@@ -65,6 +71,7 @@ class VisualizationRuntime:
         grpc_port: int = DEFAULT_GRPC_PORT,
         bridge_factory: BridgeFactory | None = None,
         normalizer: CanonicalNormalizer,
+        message_observer: RawMessageObserver | None = None,
     ) -> None:
         self._lock = threading.Lock()
         self._on_state_change = on_state_change
@@ -80,6 +87,7 @@ class VisualizationRuntime:
         self._grpc_port = grpc_port
         self._bridge_factory = bridge_factory or RerunBridge
         self._normalizer = normalizer
+        self._message_observer = message_observer
         self._bridge: RerunBridge | None = None
         self._closed = False
         self._scene_settings = RerunSceneSettings()
@@ -133,9 +141,11 @@ class VisualizationRuntime:
         out_dir: Path,
         *,
         duration_seconds: float = 3600.0,
+        project_name: str,
     ) -> None:
         if not math.isfinite(duration_seconds) or duration_seconds <= 0:
             raise ValueError("длительность приёма должна быть больше нуля")
+        clock_established = threading.Event()
         self._start(
             source_mode="live",
             phase="starting_live",
@@ -144,8 +154,21 @@ class VisualizationRuntime:
                 host,
                 out_dir.expanduser().resolve(),
                 duration_seconds=duration_seconds,
+                project_name=project_name,
+                clock_established=clock_established,
             ),
         )
+        deadline = time.monotonic() + LIVE_CAPTURE_CLOCK_READY_TIMEOUT_SECONDS
+        while not clock_established.wait(timeout=0.05):
+            snapshot = self.snapshot()
+            with self._lock:
+                thread_alive = self._thread is not None and self._thread.is_alive()
+            if snapshot["phase"] == "error" or not thread_alive:
+                raise RuntimeError("live capture завершился до фиксации общего session clock")
+            if time.monotonic() >= deadline:
+                with suppress(RuntimeError):
+                    self.stop()
+                raise RuntimeError("live capture clock не был готов за отведённое время")
 
     def stop(self, *, wait_seconds: float = 5.0) -> None:
         notify_only = False
@@ -194,6 +217,11 @@ class VisualizationRuntime:
             if not thread_alive:
                 self._bridge = None
                 self._rerun_grpc_url = None
+        if thread_alive:
+            self._notify()
+            raise RuntimeError(
+                "поток не завершился за отведённое время; runtime и evidence lease сохранены"
+            )
         if bridge is not None:
             try:
                 bridge.close()
@@ -278,8 +306,16 @@ class VisualizationRuntime:
 
         self._run_pipeline(produce, running_phase="replay")
 
-    def _run_live(self, host: str, out_dir: Path, *, duration_seconds: float) -> None:
-        _write_live_session_preamble(out_dir, host, duration_seconds)
+    def _run_live(
+        self,
+        host: str,
+        out_dir: Path,
+        *,
+        duration_seconds: float,
+        project_name: str,
+        clock_established: threading.Event,
+    ) -> None:
+        _write_live_session_preamble(out_dir, host, duration_seconds, project_name)
 
         def produce(put: Callable[[StreamMessage], None]) -> str:
             def on_message(message: CapturedMqttMessage) -> None:
@@ -298,6 +334,7 @@ class VisualizationRuntime:
                 host,
                 out_dir / "captures" / "mqtt_live",
                 duration_seconds=duration_seconds,
+                on_clock_established=clock_established.set,
                 on_ready=lambda: self._set_running(
                     "live",
                     "Приём запущен. Теперь дважды нажмите физическую кнопку устройства.",
@@ -325,6 +362,14 @@ class VisualizationRuntime:
         publisher_error: list[BaseException] = []
 
         def enqueue(message: StreamMessage) -> None:
+            # A plugin may consume non-visual status before the bounded preview
+            # FIFO. This keeps control telemetry from evicting visual frames
+            # without embedding a vendor decoder in the visual runtime.
+            if self._message_observer is not None and self._message_observer(
+                message, self._metrics
+            ):
+                self._notify()
+                return
             try:
                 messages.put_nowait(message)
                 return
@@ -525,7 +570,12 @@ def new_live_session_dir(sessions_root: Path) -> Path:
     return candidate
 
 
-def _write_live_session_preamble(out_dir: Path, host: str, duration_seconds: float) -> None:
+def _write_live_session_preamble(
+    out_dir: Path,
+    host: str,
+    duration_seconds: float,
+    project_name: str,
+) -> None:
     out_dir.mkdir(parents=True, exist_ok=False)
     started_at_utc = utc_now_iso()
     write_json_atomic(
@@ -537,6 +587,9 @@ def _write_live_session_preamble(out_dir: Path, host: str, duration_seconds: flo
             "operation": "k1_live_mqtt_to_rerun",
             "target": "owner-controlled K1 at redacted RFC1918 address",
             "requested_duration_seconds": duration_seconds,
+            # Operator-provided display metadata only. It is never used as a
+            # path component and is normalized/validated at the plugin API.
+            "project_name": project_name,
             "raw_capture": "captures/mqtt_live/mqtt.raw.k1mqtt",
             "credential_storage": "none",
         },