- {session.label}
+
+ {session.lab &&
+ !session.label.startsWith(`${session.lab.labId} ·`) ? (
+
+ {session.lab.labId}
+
+ ) : null}
+ {session.label}
+
{sessionDescription(session)}
@@ -263,10 +271,18 @@ export function ObservationSessionSelect({
title="Удалить сохранённую сессию?"
description={deleteTarget ? <>
{deleteTarget.label}
-
- Сессия, исходные данные наблюдения, подготовленная Rerun-запись и
- видеоматериалы будут удалены с этого сервера без возможности восстановления.
-
+ {deleteTarget.lab ? (
+
+ Будут удалены только LAB-запись из каталога и её подготовленный кэш.
+ Исходная запись {deleteTarget.lab.sourceSessionId} и зафиксированный
+ лабораторный результат останутся неизменными.
+
+ ) : (
+
+ Сессия, исходные данные наблюдения, подготовленная Rerun-запись и
+ видеоматериалы будут удалены с этого сервера без возможности восстановления.
+
+ )}
{sessions.error && sessions.deletingSessionId === null ? (
{sessions.error}
) : null}
diff --git a/apps/control-station/src/core/observation/sessionArchive.ts b/apps/control-station/src/core/observation/sessionArchive.ts
index 5d88940..7285465 100644
--- a/apps/control-station/src/core/observation/sessionArchive.ts
+++ b/apps/control-station/src/core/observation/sessionArchive.ts
@@ -9,6 +9,18 @@ export type ObservationSessionStatus =
| "interrupted"
| "failed";
+export interface ObservationLabInstance {
+ labId: string;
+ sourceSessionId: string;
+ resultKind: string;
+ resultId: string;
+ sourceResultId: string | null;
+ configSha256: string | null;
+ runCreatedAtUtc: string;
+ publishedAtUtc: string;
+ provenance: Readonly>;
+}
+
export interface ObservationSessionSummary {
id: string;
label: string;
@@ -19,6 +31,7 @@ export interface ObservationSessionSummary {
durationSeconds: number;
replayable: boolean;
preparation: ObservationSessionCatalogPreparation | null;
+ lab: ObservationLabInstance | null;
}
export interface ObservationSessionCatalog {
@@ -135,6 +148,18 @@ const ITEM_KEYS = new Set([
"duration_seconds",
"replayable",
"preparation",
+ "lab",
+]);
+const LAB_KEYS = new Set([
+ "lab_id",
+ "source_session_id",
+ "result_kind",
+ "result_id",
+ "source_result_id",
+ "config_sha256",
+ "run_created_at_utc",
+ "published_at_utc",
+ "provenance",
]);
const CATALOG_PREPARATION_KEYS = new Set([
"preparation_id",
@@ -344,6 +369,72 @@ function decodeItem(value: unknown, index: number): ObservationSessionSummary {
durationSeconds: value.duration_seconds,
replayable: value.replayable,
preparation: decodeCatalogPreparation(value.preparation, id),
+ lab: decodeLabInstance(value.lab, id),
+ };
+}
+
+function decodeLabInstance(
+ value: unknown,
+ sessionId: string,
+): ObservationLabInstance | null {
+ if (value === null || value === undefined) return null;
+ if (!isRecord(value)) {
+ throw new ObservationSessionContractError(
+ `LAB-привязка сессии ${sessionId} должна быть объектом или null.`,
+ );
+ }
+ assertExactKeys(value, LAB_KEYS, `LAB-привязка сессии ${sessionId}`);
+ const labId = requireString(value.lab_id, `lab(${sessionId}).lab_id`, 36);
+ if (!/^LAB [A-Z][A-Z0-9._-]{0,31}$/.test(labId)) {
+ throw new ObservationSessionContractError("Каталог содержит некорректный LAB-маркер.");
+ }
+ const sourceSessionId = requireString(
+ value.source_session_id,
+ `lab(${sessionId}).source_session_id`,
+ 128,
+ );
+ const resultKind = requireString(value.result_kind, `lab(${sessionId}).result_kind`, 128);
+ const resultId = requireString(value.result_id, `lab(${sessionId}).result_id`, 128);
+ const sourceResultId = value.source_result_id === null
+ ? null
+ : requireString(value.source_result_id, `lab(${sessionId}).source_result_id`, 128);
+ for (const [field, entry] of [
+ ["source_session_id", sourceSessionId],
+ ["result_kind", resultKind],
+ ["result_id", resultId],
+ ["source_result_id", sourceResultId],
+ ] as const) {
+ if (entry !== null && !SAFE_ID.test(entry)) {
+ throw new ObservationSessionContractError(
+ `Поле lab.${field} содержит небезопасный идентификатор.`,
+ );
+ }
+ }
+ const configSha256 = value.config_sha256 === null
+ ? null
+ : requireString(value.config_sha256, `lab(${sessionId}).config_sha256`, 64);
+ if (configSha256 !== null && !SHA256.test(configSha256)) {
+ throw new ObservationSessionContractError("LAB-конфигурация не содержит SHA-256.");
+ }
+ if (!isRecord(value.provenance)) {
+ throw new ObservationSessionContractError("LAB provenance должен быть JSON-объектом.");
+ }
+ return {
+ labId,
+ sourceSessionId,
+ resultKind,
+ resultId,
+ sourceResultId,
+ configSha256,
+ runCreatedAtUtc: requireUtcTimestamp(
+ value.run_created_at_utc,
+ `lab(${sessionId}).run_created_at_utc`,
+ ),
+ publishedAtUtc: requireUtcTimestamp(
+ value.published_at_utc,
+ `lab(${sessionId}).published_at_utc`,
+ ),
+ provenance: value.provenance,
};
}
diff --git a/apps/control-station/src/styles/observation-sessions.css b/apps/control-station/src/styles/observation-sessions.css
index 421a805..d060a45 100644
--- a/apps/control-station/src/styles/observation-sessions.css
+++ b/apps/control-station/src/styles/observation-sessions.css
@@ -201,6 +201,22 @@
white-space: nowrap;
}
+.observation-session-option__lab {
+ align-items: center;
+ background: color-mix(in srgb, var(--accent, #b8ff5a) 14%, transparent);
+ border: 1px solid color-mix(in srgb, var(--accent, #b8ff5a) 42%, transparent);
+ border-radius: 999px;
+ color: var(--accent, #b8ff5a);
+ display: inline-flex;
+ font-size: 9px;
+ font-weight: 800;
+ letter-spacing: 0.08em;
+ line-height: 1;
+ margin-right: 7px;
+ padding: 4px 6px 3px;
+ vertical-align: 1px;
+}
+
.observation-session-menu__empty {
display: grid;
min-height: 9rem;
diff --git a/docs/09_OBSERVATION_SESSIONS.md b/docs/09_OBSERVATION_SESSIONS.md
index 378dd8b..8c9abca 100644
--- a/docs/09_OBSERVATION_SESSIONS.md
+++ b/docs/09_OBSERVATION_SESSIONS.md
@@ -102,6 +102,38 @@ the obsolete browser poll; it does not cancel the process-owned conversion.
A queued job may legitimately wait behind another export and therefore uses the
overall preparation deadline rather than the active-export heartbeat timeout.
+## Immutable LAB instances
+
+An accepted experiment is published as a new saved-session identity instead of
+changing the AI result selected for its source recording. A LAB instance binds:
+
+- a stable `LAB E…` marker and opaque session id;
+- the immutable source observation session;
+- the exact result, source-result and configuration SHA-256 identities;
+- run and publication timestamps plus JSON provenance;
+- a content-addressed compute job, LiDAR pack and integrated visual result.
+
+The source evidence remains the only raw sensor master. A LAB replay cache and
+unchanged camera inputs use hard links on the same filesystem; no second copy of
+the source RRD, video or native K1 capture is created. Derived arrays are stored
+only when the run did not already persist a viewer-ready form. Deleting a LAB
+catalog entry cannot delete the referenced source session, and a source session
+with published LAB instances fails closed on deletion.
+
+The visual result is resolved by the LAB session id, so `LAB E19` and `LAB E21`
+can coexist over `RAVNOVES00` without “latest accepted result” replacing an
+earlier experiment. The operator can open either row in **Сохранённые сессии**
+and use the same **Объекты 2D / Сегментация / Кубы 3D** controls. First
+publication must also warm the result-specific Rerun overlay; later opens reuse
+that verified cache and do not rerun AI inference.
+
+E21 is a bounded special case: its worker persisted 121 semantic mask hashes,
+not duplicate mask pixels. Its visual publication materializes a mask only
+after an exact SHA-256 match against the immutable accepted E19 semantic
+reference. The seven detector frames replaced by the bounded latest-wins queue
+remain explicit empty frames. This preserves the measured near-real-time
+behavior rather than presenting a fabricated gap-free run.
+
The first preparation of a long capture can take time because every decodable
point/pose message and every valid K1 `ModelingReport` is projected into RRD.
Later openings reuse a verified, digest-bound cache. Derived RRD cache v9 is
diff --git a/experiments/perception/LAB_E21_REPORT_2026-07-23.md b/experiments/perception/LAB_E21_REPORT_2026-07-23.md
index dc9cd96..61c252f 100644
--- a/experiments/perception/LAB_E21_REPORT_2026-07-23.md
+++ b/experiments/perception/LAB_E21_REPORT_2026-07-23.md
@@ -185,6 +185,35 @@ short nominal run, an intentional overload/backpressure run, disconnect/reconnec
recovery, and then a longer soak. Model-quality work should remain separately measured
so that visual improvements cannot hide a runtime regression.
+## Immutable Mission Core publication
+
+E21 is published as the separate saved-session instance
+`lab-e21-d0201712`, displayed as
+`LAB E21 · Real-time 1× · RTX 4090 · d0201712`. It does not replace the
+result bound to `RAVNOVES00`, and it coexists with the comparison baseline
+`lab-e19-36964643`.
+
+| Publication item | Value |
+| --- | --- |
+| LAB source session | `20260720T065719Z_viewer_live` |
+| Visual result | `e10-integrated-perception-d0ecafd5e56f8ef132b8cb88f45ec275aa17470c682c6cce7c08f5597bfd7bd6` |
+| Visual frames | 601 |
+| Worker result frames | 594 |
+| Explicit latest-wins replacement frames | 12, 239–242, 433, 498 |
+| SHA-matched semantic masks | 121 / 121 |
+| Accepted cuboids materialized | 881 |
+| Raw source payloads copied | no |
+| Source RRD storage | same inode through hard link |
+| Base replay readiness | 0.03 s observed |
+| Initial visual overlay publication | 111.47 s observed |
+| Warm visual overlay retrieval | 0.04 s observed |
+
+The 111.47-second step is viewer-container publication from already computed
+results, not detector, semantic or fusion inference. It is completed before
+operator handoff. Warm UI openings reuse the resulting 7.88 MB Rerun overlay.
+The base source remains the full 8:55.718 recording; E21 AI output covers the
+accepted 59.994-second envelope beginning at source camera time 35.421857 s.
+
## Evidence
- Source report: `.runtime/compute-experiments/e21/20260723T155125Z-e21-source.json`
diff --git a/src/k1link/compute/__init__.py b/src/k1link/compute/__init__.py
index 867f575..0e073e1 100644
--- a/src/k1link/compute/__init__.py
+++ b/src/k1link/compute/__init__.py
@@ -34,6 +34,11 @@ from .jobs import (
prepare_camera_compute_job,
validate_camera_compute_job,
)
+from .lab_instances import (
+ PublishedIntegratedLabInstance,
+ publish_e21_lab_instance,
+ publish_integrated_lab_instance,
+)
from .live_perception import (
LIVE_INGRESS_SCHEMA,
LIVE_INGRESS_WIRE_SCHEMA,
@@ -109,6 +114,7 @@ __all__ = [
"LiveIngressEvent",
"LivePerceptionIngress",
"IntegratedPerceptionOverlayStore",
+ "PublishedIntegratedLabInstance",
"IntegratedPerceptionResult",
"LiveReplayQualificationResult",
"MultiratePerceptionArtifact",
@@ -137,6 +143,8 @@ __all__ = [
"validate_recorded_perception_epoch_result",
"validate_live_replay_qualification_result",
"validate_integrated_perception_result",
+ "publish_e21_lab_instance",
+ "publish_integrated_lab_instance",
"validate_multirate_perception_qualification_result",
"prepare_recorded_qualification_slice",
"DetectionFrame",
diff --git a/src/k1link/compute/lab_instances.py b/src/k1link/compute/lab_instances.py
new file mode 100644
index 0000000..e979a73
--- /dev/null
+++ b/src/k1link/compute/lab_instances.py
@@ -0,0 +1,940 @@
+"""Immutable LAB catalog projections for accepted integrated perception runs."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import secrets
+import shutil
+import stat
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+from k1link.artifacts import write_json_atomic
+from k1link.sessions import (
+ LabSessionBinding,
+ SessionIntegrityError,
+ SessionStore,
+ publish_lab_replay_cache,
+)
+
+from .integrated_perception import (
+ IntegratedPerceptionResult,
+ validate_integrated_perception_result,
+)
+from .jobs import CameraComputeJob, validate_camera_compute_job
+
+
+@dataclass(frozen=True, slots=True)
+class PublishedIntegratedLabInstance:
+ binding: LabSessionBinding
+ job: CameraComputeJob
+ result: IntegratedPerceptionResult
+
+
+def publish_integrated_lab_instance(
+ *,
+ repository_root: Path,
+ source_result_root: Path,
+ lab_session_id: str,
+ lab_id: str,
+ display_name: str,
+ provenance: dict[str, Any] | None = None,
+) -> PublishedIntegratedLabInstance:
+ """Snapshot one accepted visual result as a separately replayable LAB run.
+
+ Large immutable camera, LiDAR, array and RRD payloads are hard-linked.
+ Content-addressed manifests are rebound to the LAB session id, so the
+ existing viewer resolves one exact result instead of selecting whichever
+ source-session result happens to be newest.
+ """
+
+ root = repository_root.expanduser().resolve(strict=True)
+ jobs_root = root / ".runtime" / "compute-jobs"
+ results_root = root / ".runtime" / "compute-experiments" / "e10" / "worker-results"
+ packs_root = root / ".runtime" / "compute-experiments" / "e10" / "lidar-packs"
+ source_result_path = source_result_root.expanduser().resolve(strict=True)
+ source_document = _read_object(source_result_path / "result.json", source_result_path)
+ identity = source_document.get("identity")
+ if not isinstance(identity, dict) or not isinstance(identity.get("job_id"), str):
+ raise SessionIntegrityError("integrated LAB source has no job identity")
+ source_job_root = jobs_root / identity["job_id"]
+ source = validate_integrated_perception_result(
+ source_job_root,
+ source_result_path,
+ packs_root,
+ )
+ if not source.accepted:
+ raise SessionIntegrityError("only accepted integrated results can become LAB instances")
+
+ lab_job = _publish_lab_job(source.job, jobs_root, lab_session_id)
+ lab_pack = _publish_lab_pack(source, lab_job, packs_root, lab_session_id)
+ lab_result = _publish_lab_result(
+ source,
+ lab_job,
+ lab_pack,
+ results_root,
+ lab_session_id,
+ )
+ validated = validate_integrated_perception_result(
+ lab_job.job_root,
+ lab_result,
+ packs_root,
+ )
+ store = SessionStore(root)
+ publish_lab_replay_cache(
+ store.data_dir,
+ source_session_id=source.job.session_id,
+ lab_session_id=lab_session_id,
+ )
+ configuration = source_document["identity"].get("configuration")
+ profile_sha256 = (
+ configuration.get("profile_sha256") if isinstance(configuration, dict) else None
+ )
+ if not isinstance(profile_sha256, str) or len(profile_sha256) != 64:
+ profile_sha256 = None
+ binding = store.publish_lab_instance(
+ session_id=lab_session_id,
+ source_session_id=source.job.session_id,
+ display_name=display_name,
+ lab_id=lab_id,
+ result_kind="e10-integrated-perception",
+ result_id=validated.result_id,
+ source_result_id=source.result_id,
+ config_sha256=profile_sha256,
+ run_created_at_utc=source.created_at_utc,
+ provenance={
+ "schema_version": "missioncore.integrated-lab-publication/v1",
+ "storage_mode": "hard-linked-immutable-payloads",
+ "source_job_id": source.job.job_id,
+ "projected_job_id": lab_job.job_id,
+ "source_lidar_pack_id": source.pack_root.name,
+ "projected_lidar_pack_id": lab_pack.name,
+ **(provenance or {}),
+ },
+ )
+ return PublishedIntegratedLabInstance(binding=binding, job=lab_job, result=validated)
+
+
+def publish_e21_lab_instance(
+ *,
+ repository_root: Path,
+ e21_result_root: Path,
+ worker_result_root: Path,
+ semantic_reference_result_root: Path,
+ lab_session_id: str,
+ lab_id: str,
+ display_name: str,
+) -> PublishedIntegratedLabInstance:
+ """Publish the accepted E21 envelope as an exact 60-second visual LAB run.
+
+ E21 persisted semantic mask identities but intentionally did not duplicate
+ mask pixels. The immutable full-session reference contains those exact
+ pixels. Publication therefore materializes a bounded viewer projection only
+ after every E21 mask SHA-256 matches the corresponding reference mask.
+ Detector latest-wins replacements remain explicit empty frames.
+ """
+
+ root = repository_root.expanduser().resolve(strict=True)
+ jobs_root = root / ".runtime" / "compute-jobs"
+ results_root = root / ".runtime" / "compute-experiments" / "e10" / "worker-results"
+ packs_root = root / ".runtime" / "compute-experiments" / "e10" / "lidar-packs"
+ reference_path = semantic_reference_result_root.expanduser().resolve(strict=True)
+ reference_document = _read_object(reference_path / "result.json", reference_path)
+ reference_identity = reference_document.get("identity")
+ if not isinstance(reference_identity, dict) or not isinstance(
+ reference_identity.get("job_id"), str
+ ):
+ raise SessionIntegrityError("E21 semantic reference has no job identity")
+ reference = validate_integrated_perception_result(
+ jobs_root / reference_identity["job_id"],
+ reference_path,
+ packs_root,
+ )
+ if not reference.accepted or reference.source_start_frame_index != 0:
+ raise SessionIntegrityError("E21 semantic reference is not an accepted zero-based run")
+
+ e21_root = e21_result_root.expanduser().resolve(strict=True)
+ worker_root = worker_result_root.expanduser().resolve(strict=True)
+ e21_document, e21_report, worker_document = _validate_e21_inputs(
+ e21_root,
+ worker_root,
+ )
+ frame_count = int(e21_report["metrics"]["source"]["events_admitted"]["camera-frame"])
+ if frame_count < 2 or frame_count > reference.frame_count:
+ raise SessionIntegrityError("E21 frame count is outside the semantic reference")
+
+ lab_job = _publish_lab_job(reference.job, jobs_root, lab_session_id)
+ lab_pack = _publish_e21_pack(
+ reference,
+ lab_job,
+ packs_root,
+ lab_session_id,
+ frame_count,
+ )
+ lab_result = _publish_e21_visual_result(
+ reference=reference,
+ lab_job=lab_job,
+ lab_pack=lab_pack,
+ results_root=results_root,
+ lab_session_id=lab_session_id,
+ frame_count=frame_count,
+ e21_root=e21_root,
+ e21_document=e21_document,
+ e21_report=e21_report,
+ worker_root=worker_root,
+ worker_document=worker_document,
+ )
+ validated = validate_integrated_perception_result(
+ lab_job.job_root,
+ lab_result,
+ packs_root,
+ )
+ store = SessionStore(root)
+ publish_lab_replay_cache(
+ store.data_dir,
+ source_session_id=reference.job.session_id,
+ lab_session_id=lab_session_id,
+ )
+ binding = store.publish_lab_instance(
+ session_id=lab_session_id,
+ source_session_id=reference.job.session_id,
+ display_name=display_name,
+ lab_id=lab_id,
+ result_kind="e21-realtime-envelope",
+ result_id=validated.result_id,
+ source_result_id=str(e21_document["result_id"]),
+ config_sha256=str(e21_report["identity"]["profile_sha256"]),
+ run_created_at_utc=str(e21_report["created_at_utc"]),
+ provenance={
+ "schema_version": "missioncore.e21-lab-publication/v1",
+ "storage_mode": "hard-linked-source-and-bounded-derived-projection",
+ "e21_result_id": e21_document["result_id"],
+ "worker_result_id": worker_document["result_id"],
+ "semantic_reference_result_id": reference.result_id,
+ "semantic_mask_reconstruction": "exact-sha256-match",
+ "detector_replacement_frames": [
+ 12,
+ 239,
+ 240,
+ 241,
+ 242,
+ 433,
+ 498,
+ ],
+ "visual_frame_count": frame_count,
+ "source_payloads_mutated": False,
+ },
+ )
+ return PublishedIntegratedLabInstance(binding=binding, job=lab_job, result=validated)
+
+
+def _validate_e21_inputs(
+ e21_root: Path,
+ worker_root: Path,
+) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
+ e21_document = _read_object(e21_root / "result.json", e21_root)
+ e21_report = _read_object(e21_root / "report.json", e21_root)
+ report_descriptor = e21_document.get("report")
+ if (
+ e21_document.get("schema_version") != "missioncore.e21-realtime-envelope-result/v1"
+ or e21_document.get("result_id") != e21_root.name
+ or e21_document.get("state") != "accepted"
+ or not isinstance(report_descriptor, dict)
+ or report_descriptor.get("path") != "report.json"
+ or report_descriptor.get("byte_length") != (e21_root / "report.json").stat().st_size
+ or report_descriptor.get("sha256") != _sha256(e21_root / "report.json")
+ or e21_report.get("schema_version") != "missioncore.e21-realtime-envelope-report/v1"
+ or e21_report.get("result_id") != e21_root.name
+ or e21_report.get("state") != "accepted"
+ or not isinstance(e21_report.get("identity"), dict)
+ or not isinstance(e21_report.get("metrics"), dict)
+ ):
+ raise SessionIntegrityError("E21 accepted result identity is inconsistent")
+
+ worker_document = _read_object(worker_root / "result.json", worker_root)
+ e21_identity = e21_report["identity"]
+ if (
+ worker_document.get("schema_version") != "missioncore.e15-shadow-inference-result/v1"
+ or worker_document.get("result_id") != worker_root.name
+ or worker_document.get("result_id") != e21_identity.get("worker_result_id")
+ or _sha256(worker_root / "result.json") != e21_identity.get("worker_result_sha256")
+ or not isinstance(worker_document.get("identity"), dict)
+ or not isinstance(worker_document.get("artifacts"), list)
+ ):
+ raise SessionIntegrityError("E21 worker result identity is inconsistent")
+ required = {
+ "e15-semantic-frames": "semantic-frames.jsonl",
+ "e15-fusion-frames": "fusion-frames.jsonl",
+ "e15-world-state": "world-state.jsonl",
+ "worker-gpu-telemetry": "gpu-telemetry.jsonl",
+ }
+ descriptors = {
+ value.get("kind"): value
+ for value in worker_document["artifacts"]
+ if isinstance(value, dict) and value.get("kind") in required
+ }
+ if set(descriptors) != set(required):
+ raise SessionIntegrityError("E21 worker artifacts are incomplete")
+ for kind, name in required.items():
+ descriptor = descriptors[kind]
+ path = worker_root / name
+ if (
+ descriptor.get("path") != name
+ or descriptor.get("byte_length") != path.stat().st_size
+ or descriptor.get("sha256") != _sha256(path)
+ ):
+ raise SessionIntegrityError("E21 worker artifact identity changed")
+ return e21_document, e21_report, worker_document
+
+
+def _publish_e21_pack(
+ reference: IntegratedPerceptionResult,
+ job: CameraComputeJob,
+ packs_root: Path,
+ lab_session_id: str,
+ frame_count: int,
+) -> Path:
+ source_manifest = _read_object(reference.pack_root / "manifest.json", reference.pack_root)
+ with np.load(reference.pack_root / "lidar-pack.npz", allow_pickle=False) as arrays:
+ cloud_end = int(arrays["cloud_offsets"][frame_count])
+ payload = {
+ "frame_indices": arrays["frame_indices"][:frame_count].copy(),
+ "source_frame_indices": arrays["source_frame_indices"][:frame_count].copy(),
+ "session_seconds": arrays["session_seconds"][:frame_count].copy(),
+ "sample_available": arrays["sample_available"][:frame_count].copy(),
+ "cloud_offsets": arrays["cloud_offsets"][: frame_count + 1].copy(),
+ "cloud_points_map": arrays["cloud_points_map"][:cloud_end].copy(),
+ "pose_positions_map": arrays["pose_positions_map"][:frame_count].copy(),
+ "pose_quaternions_map_from_lidar": arrays[
+ "pose_quaternions_map_from_lidar"
+ ][:frame_count].copy(),
+ "lidar_camera_delta_ms": arrays["lidar_camera_delta_ms"][:frame_count].copy(),
+ "pose_point_delta_ms": arrays["pose_point_delta_ms"][:frame_count].copy(),
+ "intrinsic_fx_fy_cx_cy": arrays["intrinsic_fx_fy_cx_cy"].copy(),
+ "distortion_kb4": arrays["distortion_kb4"].copy(),
+ "t_camera_from_lidar": arrays["t_camera_from_lidar"].copy(),
+ }
+ identity = json.loads(json.dumps(source_manifest["identity"]))
+ identity.update(
+ {
+ "job_id": job.job_id,
+ "input_sha256": job.input_sha256,
+ "session_id": lab_session_id,
+ "frame_count": frame_count,
+ "source_start_frame_index": 0,
+ "source_end_frame_index": frame_count - 1,
+ "available_lidar_frames": int(np.count_nonzero(payload["sample_available"])),
+ "point_count": int(payload["cloud_points_map"].shape[0]),
+ "timeline_start_seconds": float(payload["session_seconds"][0]),
+ "timeline_end_seconds": float(payload["session_seconds"][-1]),
+ "visual_projection": "accepted-e21-envelope/v1",
+ }
+ )
+ identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest()
+ pack_id = f"e10-lidar-pack-{identity_sha256}"
+ destination = packs_root / pack_id
+ if destination.exists():
+ existing = _read_object(destination / "manifest.json", destination)
+ if existing.get("identity") != identity:
+ raise SessionIntegrityError("E21 LAB LiDAR pack id collides")
+ return destination
+ staging = _staging_directory(packs_root, pack_id)
+ try:
+ arrays_path = staging / "lidar-pack.npz"
+ np.savez_compressed(arrays_path, **payload)
+ os.chmod(arrays_path, 0o600)
+ write_json_atomic(
+ staging / "manifest.json",
+ {
+ **source_manifest,
+ "pack_id": pack_id,
+ "identity_sha256": identity_sha256,
+ "identity": identity,
+ "created_at_utc": str(source_manifest["created_at_utc"]),
+ "artifact": {
+ "path": arrays_path.name,
+ "media_type": "application/x-npz",
+ "byte_length": arrays_path.stat().st_size,
+ "sha256": _sha256(arrays_path),
+ },
+ },
+ )
+ _publish_directory(staging, destination)
+ finally:
+ _remove_staging(staging)
+ return destination
+
+
+def _publish_e21_visual_result(
+ *,
+ reference: IntegratedPerceptionResult,
+ lab_job: CameraComputeJob,
+ lab_pack: Path,
+ results_root: Path,
+ lab_session_id: str,
+ frame_count: int,
+ e21_root: Path,
+ e21_document: dict[str, Any],
+ e21_report: dict[str, Any],
+ worker_root: Path,
+ worker_document: dict[str, Any],
+) -> Path:
+ with np.load(reference.arrays_path, allow_pickle=False) as arrays:
+ frame_times = arrays["frame_times_ns"][:frame_count].copy()
+ reference_semantic_indices = arrays["semantic_frame_indices"]
+ selected = reference_semantic_indices < frame_count
+ reference_indices = reference_semantic_indices[selected].copy()
+ reference_masks = arrays["semantic_masks"][selected].copy()
+
+ semantic_rows = _read_jsonl(worker_root / "semantic-frames.jsonl")
+ if [row.get("frame_index") for row in semantic_rows] != reference_indices.tolist():
+ raise SessionIntegrityError("E21 semantic frame schedule changed")
+ for row, mask in zip(semantic_rows, reference_masks, strict=True):
+ if row.get("mask_sha256") != hashlib.sha256(mask.tobytes()).hexdigest():
+ raise SessionIntegrityError("E21 semantic mask does not match immutable reference")
+ row["schema_version"] = "missioncore.e10-semantic-frame/v1"
+ row["session_seconds"] = float(frame_times[int(row["frame_index"])]) / 1_000_000_000
+
+ fusion_source = {
+ int(row["source_frame_index"]): row
+ for row in _read_jsonl(worker_root / "fusion-frames.jsonl")
+ }
+ world_source = {
+ int(row["source_frame_index"]): row
+ for row in _read_jsonl(worker_root / "world-state.jsonl")
+ }
+ if set(fusion_source) != set(world_source):
+ raise SessionIntegrityError("E21 fusion and world timelines differ")
+ fusion_rows: list[dict[str, Any]] = []
+ world_rows: list[dict[str, Any]] = []
+ dropped_indices: list[int] = []
+ for index in range(frame_count):
+ session_seconds = float(frame_times[index]) / 1_000_000_000
+ fusion = fusion_source.get(index)
+ world = world_source.get(index)
+ if fusion is None or world is None:
+ dropped_indices.append(index)
+ fusion_rows.append(
+ {
+ "schema_version": "missioncore.e10-fusion-frame/v1",
+ "frame_index": index,
+ "source_frame_index": index,
+ "session_seconds": session_seconds,
+ "fusion_state": "detector-dropped-latest-wins",
+ "semantic_source_frame_index": None,
+ "semantic_status": "unavailable",
+ "objects": [],
+ }
+ )
+ world_rows.append(_dropped_world_row(index, session_seconds))
+ continue
+ normalized_fusion = json.loads(json.dumps(fusion))
+ normalized_fusion.update(
+ {
+ "schema_version": "missioncore.e10-fusion-frame/v1",
+ "frame_index": index,
+ "source_frame_index": index,
+ "session_seconds": session_seconds,
+ }
+ )
+ normalized_world = json.loads(json.dumps(world))
+ normalized_world.update(
+ {
+ "frame_index": index,
+ "source_frame_index": index,
+ "session_seconds": session_seconds,
+ }
+ )
+ fusion_rows.append(normalized_fusion)
+ world_rows.append(normalized_world)
+ expected_drops = int(
+ e21_report["metrics"]["worker"]["detector"]["queue"]["dropped_overflow"]
+ )
+ if len(dropped_indices) != expected_drops:
+ raise SessionIntegrityError("E21 detector replacement accounting changed")
+
+ box_offsets = [0]
+ centers: list[list[float]] = []
+ half_sizes: list[list[float]] = []
+ quaternions: list[list[float]] = []
+ colors: list[list[int]] = []
+ for row in fusion_rows:
+ for item in row["objects"]:
+ if not str(item.get("cuboid_status", "")).startswith("accepted-"):
+ continue
+ centers.append(item["cuboid_center_map"])
+ half_sizes.append(item["cuboid_half_size"])
+ quaternions.append(item["cuboid_quaternion_xyzw"])
+ colors.append(_cuboid_color(item))
+ box_offsets.append(len(centers))
+
+ reference_identity = _read_object(
+ reference.result_root / "result.json",
+ reference.result_root,
+ )["identity"]
+ worker_identity = worker_document["identity"]
+ configuration = {
+ "pipeline": "e21-shadow-realtime-envelope-visual-projection/v1",
+ "profile_sha256": e21_report["identity"]["profile_sha256"],
+ "profile": {
+ "source": reference_identity["configuration"]["profile"]["source"],
+ "replay": {
+ "speed": 1.0,
+ "bounded_latest_wins": True,
+ "detector_replacement_frames": dropped_indices,
+ },
+ },
+ "e21_result_id": e21_document["result_id"],
+ "worker_result_id": worker_document["result_id"],
+ "semantic_mask_materialization": {
+ "mode": "immutable-reference-exact-sha256",
+ "reference_result_id": reference.result_id,
+ "matched_masks": len(semantic_rows),
+ },
+ }
+ selection = {
+ "frame_count": frame_count,
+ "source_start_frame_index": 0,
+ "source_end_frame_index": frame_count - 1,
+ "timeline_start_seconds": float(frame_times[0]) / 1_000_000_000,
+ "timeline_end_seconds": float(frame_times[-1]) / 1_000_000_000,
+ "timeline_sha256": hashlib.sha256(frame_times.tobytes()).hexdigest(),
+ }
+ identity = {
+ "schema_version": "missioncore.e10-integrated-perception-identity/v1",
+ "job_id": lab_job.job_id,
+ "input_sha256": lab_job.input_sha256,
+ "session_id": lab_session_id,
+ "source_id": lab_job.source_id,
+ "lidar_pack_id": lab_pack.name,
+ "selection": selection,
+ "configuration": configuration,
+ "models": worker_identity["models"],
+ }
+ identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest()
+ result_id = f"e10-integrated-perception-{identity_sha256}"
+ destination = results_root / result_id
+ if destination.exists():
+ existing = _read_object(destination / "result.json", destination)
+ if existing.get("identity") != identity:
+ raise SessionIntegrityError("E21 LAB visual result id collides")
+ return destination
+
+ staging = _staging_directory(results_root, result_id)
+ try:
+ semantic_path = staging / "semantic-frames.jsonl"
+ fusion_path = staging / "fusion-frames.jsonl"
+ world_path = staging / "world-state.jsonl"
+ arrays_path = staging / "transient-perception.npz"
+ gpu_path = staging / "gpu-telemetry.jsonl"
+ report_path = staging / "run-report.json"
+ _write_jsonl(semantic_path, semantic_rows)
+ _write_jsonl(fusion_path, fusion_rows)
+ _write_jsonl(world_path, world_rows)
+ np.savez_compressed(
+ arrays_path,
+ frame_times_ns=frame_times.astype(np.int64, copy=False),
+ semantic_frame_indices=reference_indices.astype(np.int64, copy=False),
+ semantic_masks=reference_masks.astype(np.uint8, copy=False),
+ support_offsets=np.zeros(frame_count + 1, dtype=np.int64),
+ support_points=np.empty((0, 3), dtype=np.float32),
+ support_colors=np.empty((0, 3), dtype=np.uint8),
+ box_offsets=np.asarray(box_offsets, dtype=np.int64),
+ box_centers=np.asarray(centers, dtype=np.float32).reshape((-1, 3)),
+ box_half_sizes=np.asarray(half_sizes, dtype=np.float32).reshape((-1, 3)),
+ box_quaternions=np.asarray(quaternions, dtype=np.float32).reshape((-1, 4)),
+ box_colors=np.asarray(colors, dtype=np.uint8).reshape((-1, 4)),
+ )
+ shutil.copyfile(worker_root / "gpu-telemetry.jsonl", gpu_path)
+ report = {
+ "schema_version": "missioncore.e10-integrated-perception-report/v1",
+ "result_id": result_id,
+ "created_at_utc": e21_report["created_at_utc"],
+ "state": "accepted",
+ "ground_truth": False,
+ "identity": identity,
+ "acceptance": {
+ "accepted": True,
+ "navigation_or_safety_accepted": False,
+ "checks": {
+ "e21_envelope_accepted": True,
+ "semantic_hashes_exact": True,
+ "detector_replacements_explicit": True,
+ "source_payloads_immutable": True,
+ },
+ },
+ "metrics": {
+ **e21_report["metrics"],
+ "visual_projection": {
+ "frames": frame_count,
+ "semantic_masks": len(semantic_rows),
+ "detector_replacement_frames": dropped_indices,
+ "accepted_cuboids": len(centers),
+ },
+ },
+ "runtime": _read_object(worker_root / "run-report.json", worker_root).get(
+ "runtime", {}
+ ),
+ "limitations": [
+ "This is the exact accepted recorded E21 envelope, not a physical live K1 gate.",
+ "Seven latest-wins detector replacements are explicit empty visual frames.",
+ "Semantic pixels are materialized only after exact SHA-256 reference matches.",
+ "LiDAR support points remain in the immutable source scene and are not duplicated.",
+ "Navigation and safety authority remain disabled.",
+ ],
+ }
+ write_json_atomic(report_path, report)
+ artifacts = [
+ _artifact_descriptor(
+ "e10-semantic-frames",
+ semantic_path,
+ "missioncore.e10-semantic-frame/v1",
+ ),
+ _artifact_descriptor(
+ "e10-fusion-frames",
+ fusion_path,
+ "missioncore.e10-fusion-frame/v1",
+ ),
+ _artifact_descriptor(
+ "e10-world-state",
+ world_path,
+ "missioncore.live-perception-world-state/v1",
+ ),
+ _artifact_descriptor("e10-transient-perception", arrays_path, None),
+ _artifact_descriptor("worker-gpu-telemetry", gpu_path, None),
+ _artifact_descriptor(
+ "e10-run-report",
+ report_path,
+ "missioncore.e10-integrated-perception-report/v1",
+ ),
+ ]
+ write_json_atomic(
+ staging / "result.json",
+ {
+ "schema_version": "missioncore.e10-integrated-perception-result/v1",
+ "result_id": result_id,
+ "identity_sha256": identity_sha256,
+ "identity": identity,
+ "created_at_utc": e21_report["created_at_utc"],
+ "ground_truth": False,
+ "publication_scope": "recorded-integrated-realtime-qualification-only",
+ "acceptance_state": "accepted",
+ "frames_processed": frame_count,
+ "artifacts": artifacts,
+ },
+ )
+ _publish_directory(staging, destination)
+ finally:
+ _remove_staging(staging)
+ return destination
+
+
+def _dropped_world_row(frame_index: int, session_seconds: float) -> dict[str, Any]:
+ return {
+ "schema_version": "missioncore.live-perception-world-state/v1",
+ "frame_index": frame_index,
+ "source_frame_index": frame_index,
+ "session_seconds": session_seconds,
+ "fusion_state": "detector-dropped-latest-wins",
+ "object_count": 0,
+ "objects": [],
+ "delivery": {
+ "health": "degraded",
+ "semantic_status": "unavailable",
+ "result_age_ms": None,
+ "projection_status": "explicit-detector-replacement",
+ },
+ "coordinate_frames": {
+ "sensor_relative": "k1-lidar",
+ "vehicle_body": "unavailable-no-rig-to-vehicle-transform",
+ "world": "k1-map",
+ },
+ "clearance": {
+ "schema": "diagnostic-polar-obstacle-clearance/v1",
+ "state": "unavailable",
+ "frame": "k1-lidar",
+ "front_m": None,
+ "observed_sector_fraction": 0.0,
+ "sector_ranges_m": [None] * 72,
+ },
+ }
+
+
+def _cuboid_color(item: dict[str, Any]) -> list[int]:
+ key = f"e10:{item.get('association_group', 'object')}:{item.get('track_id', 0)}"
+ digest = hashlib.sha256(key.encode()).digest()
+ return [64 + digest[0] % 176, 64 + digest[1] % 176, 64 + digest[2] % 176, 88]
+
+
+def _read_jsonl(path: Path) -> list[dict[str, Any]]:
+ rows: list[dict[str, Any]] = []
+ with path.open(encoding="utf-8") as stream:
+ for line in stream:
+ value = json.loads(line)
+ if not isinstance(value, dict):
+ raise SessionIntegrityError("LAB JSONL row is not an object")
+ rows.append(value)
+ return rows
+
+
+def _write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
+ with path.open("x", encoding="utf-8") as stream:
+ for row in rows:
+ stream.write(
+ json.dumps(
+ row,
+ ensure_ascii=False,
+ sort_keys=True,
+ separators=(",", ":"),
+ allow_nan=False,
+ )
+ )
+ stream.write("\n")
+ stream.flush()
+ os.fsync(stream.fileno())
+ os.chmod(path, 0o600)
+
+
+def _artifact_descriptor(
+ kind: str,
+ path: Path,
+ schema_version: str | None,
+) -> dict[str, Any]:
+ return {
+ "kind": kind,
+ "path": path.name,
+ "schema_version": schema_version,
+ "byte_length": path.stat().st_size,
+ "sha256": _sha256(path),
+ }
+
+
+def _publish_lab_job(
+ source: CameraComputeJob,
+ jobs_root: Path,
+ lab_session_id: str,
+) -> CameraComputeJob:
+ manifest = _read_object(source.manifest_path, source.job_root)
+ input_document = json.loads(json.dumps(manifest["input"]))
+ input_document["session_id"] = lab_session_id
+ input_sha256 = hashlib.sha256(_canonical_json(input_document)).hexdigest()
+ job_id = f"recorded-camera-{input_sha256[:24]}"
+ destination = jobs_root / job_id
+ if destination.exists():
+ existing = validate_camera_compute_job(destination)
+ if existing.session_id != lab_session_id:
+ raise SessionIntegrityError("LAB compute job id collides")
+ return existing
+ staging = _staging_directory(jobs_root, job_id)
+ try:
+ _hardlink_tree(source.job_root / "input", staging / "input")
+ write_json_atomic(
+ staging / "job.json",
+ {
+ **manifest,
+ "job_id": job_id,
+ "input_sha256": input_sha256,
+ "input": input_document,
+ },
+ )
+ _publish_directory(staging, destination)
+ finally:
+ _remove_staging(staging)
+ return validate_camera_compute_job(destination)
+
+
+def _publish_lab_pack(
+ source: IntegratedPerceptionResult,
+ job: CameraComputeJob,
+ packs_root: Path,
+ lab_session_id: str,
+) -> Path:
+ source_manifest = _read_object(source.pack_root / "manifest.json", source.pack_root)
+ identity = json.loads(json.dumps(source_manifest["identity"]))
+ identity.update(
+ {
+ "job_id": job.job_id,
+ "input_sha256": job.input_sha256,
+ "session_id": lab_session_id,
+ }
+ )
+ identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest()
+ pack_id = f"e10-lidar-pack-{identity_sha256}"
+ destination = packs_root / pack_id
+ if destination.exists():
+ existing = _read_object(destination / "manifest.json", destination)
+ if existing.get("identity") != identity:
+ raise SessionIntegrityError("LAB LiDAR pack id collides")
+ return destination
+ staging = _staging_directory(packs_root, pack_id)
+ try:
+ os.link(
+ source.pack_root / "lidar-pack.npz",
+ staging / "lidar-pack.npz",
+ follow_symlinks=False,
+ )
+ write_json_atomic(
+ staging / "manifest.json",
+ {
+ **source_manifest,
+ "pack_id": pack_id,
+ "identity_sha256": identity_sha256,
+ "identity": identity,
+ },
+ )
+ _publish_directory(staging, destination)
+ finally:
+ _remove_staging(staging)
+ return destination
+
+
+def _publish_lab_result(
+ source: IntegratedPerceptionResult,
+ job: CameraComputeJob,
+ pack_root: Path,
+ results_root: Path,
+ lab_session_id: str,
+) -> Path:
+ source_document = _read_object(source.result_root / "result.json", source.result_root)
+ identity = json.loads(json.dumps(source_document["identity"]))
+ identity.update(
+ {
+ "job_id": job.job_id,
+ "input_sha256": job.input_sha256,
+ "session_id": lab_session_id,
+ "lidar_pack_id": pack_root.name,
+ }
+ )
+ identity_sha256 = hashlib.sha256(_canonical_json(identity)).hexdigest()
+ result_id = f"e10-integrated-perception-{identity_sha256}"
+ destination = results_root / result_id
+ if destination.exists():
+ existing = _read_object(destination / "result.json", destination)
+ if existing.get("identity") != identity:
+ raise SessionIntegrityError("LAB integrated result id collides")
+ return destination
+ staging = _staging_directory(results_root, result_id)
+ try:
+ descriptors: list[dict[str, Any]] = []
+ for descriptor in source_document["artifacts"]:
+ if descriptor["kind"] == "e10-run-report":
+ continue
+ source_path = source.result_root / descriptor["path"]
+ target = staging / descriptor["path"]
+ os.link(source_path, target, follow_symlinks=False)
+ descriptors.append(dict(descriptor))
+
+ report = _read_object(source.report_path, source.result_root)
+ report.update(
+ {
+ "result_id": result_id,
+ "identity": identity,
+ "lab_projection": {
+ "schema_version": "missioncore.integrated-lab-projection/v1",
+ "source_session_id": source.job.session_id,
+ "source_result_id": source.result_id,
+ "lab_session_id": lab_session_id,
+ "payloads_recomputed": False,
+ },
+ }
+ )
+ report_path = staging / "run-report.json"
+ write_json_atomic(report_path, report)
+ descriptors.append(
+ {
+ "kind": "e10-run-report",
+ "path": report_path.name,
+ "schema_version": "missioncore.e10-integrated-perception-report/v1",
+ "byte_length": report_path.stat().st_size,
+ "sha256": _sha256(report_path),
+ }
+ )
+ descriptor_order = {
+ value["kind"]: index for index, value in enumerate(source_document["artifacts"])
+ }
+ descriptors.sort(key=lambda value: descriptor_order[value["kind"]])
+ write_json_atomic(
+ staging / "result.json",
+ {
+ **source_document,
+ "result_id": result_id,
+ "identity_sha256": identity_sha256,
+ "identity": identity,
+ "artifacts": descriptors,
+ },
+ )
+ _publish_directory(staging, destination)
+ finally:
+ _remove_staging(staging)
+ return destination
+
+
+def _hardlink_tree(source: Path, destination: Path) -> None:
+ source_root = source.resolve(strict=True)
+ destination.mkdir(mode=0o700, parents=True)
+ for current, directories, files in os.walk(source_root):
+ current_path = Path(current)
+ relative = current_path.relative_to(source_root)
+ target_root = destination / relative
+ target_root.mkdir(mode=0o700, parents=True, exist_ok=True)
+ directories.sort()
+ files.sort()
+ for filename in files:
+ path = current_path / filename
+ metadata = path.lstat()
+ if stat.S_ISLNK(metadata.st_mode) or not stat.S_ISREG(metadata.st_mode):
+ raise SessionIntegrityError("LAB source tree contains a non-regular file")
+ os.link(path, target_root / filename, follow_symlinks=False)
+
+
+def _staging_directory(root: Path, identity: str) -> Path:
+ root.mkdir(mode=0o700, parents=True, exist_ok=True)
+ staging = root / f".{identity}.{secrets.token_hex(12)}.tmp"
+ staging.mkdir(mode=0o700)
+ return staging
+
+
+def _publish_directory(staging: Path, destination: Path) -> None:
+ try:
+ os.replace(staging, destination)
+ except OSError:
+ if not destination.is_dir():
+ raise
+
+
+def _remove_staging(path: Path) -> None:
+ if path.exists():
+ shutil.rmtree(path)
+
+
+def _read_object(path: Path, root: Path) -> dict[str, Any]:
+ resolved = path.resolve(strict=True)
+ if resolved.parent != root.resolve(strict=True) or resolved.is_symlink():
+ raise SessionIntegrityError("LAB manifest escapes its immutable root")
+ value = json.loads(resolved.read_text(encoding="utf-8"))
+ if not isinstance(value, dict):
+ raise SessionIntegrityError("LAB manifest is not a JSON object")
+ return value
+
+
+def _canonical_json(value: object) -> bytes:
+ return json.dumps(
+ value,
+ ensure_ascii=False,
+ sort_keys=True,
+ separators=(",", ":"),
+ allow_nan=False,
+ ).encode("utf-8")
+
+
+def _sha256(path: Path) -> str:
+ digest = hashlib.sha256()
+ with path.open("rb") as stream:
+ for block in iter(lambda: stream.read(1024 * 1024), b""):
+ digest.update(block)
+ return digest.hexdigest()
diff --git a/src/k1link/device_plugins/xgrids_k1/cli.py b/src/k1link/device_plugins/xgrids_k1/cli.py
index 8c39594..ae4833e 100644
--- a/src/k1link/device_plugins/xgrids_k1/cli.py
+++ b/src/k1link/device_plugins/xgrids_k1/cli.py
@@ -17,7 +17,11 @@ from rich.table import Table
from k1link import __version__
from k1link.artifacts import write_json_atomic
-from k1link.compute import prepare_camera_compute_job
+from k1link.compute import (
+ prepare_camera_compute_job,
+ publish_e21_lab_instance,
+ publish_integrated_lab_instance,
+)
from k1link.device_plugins.xgrids_k1.analyze import (
DEFAULT_STREAM_SUMMARY_MAX_PAYLOAD_BYTES,
MAX_STREAM_SUMMARY_PAYLOAD_BYTES,
@@ -70,12 +74,17 @@ compute_app = typer.Typer(
help="Bounded, content-addressed handoffs to external compute workers.",
no_args_is_help=True,
)
+lab_app = typer.Typer(
+ help="Append-only publication of reproducible laboratory session instances.",
+ no_args_is_help=True,
+)
app.add_typer(ble_app, name="ble")
app.add_typer(net_app, name="net")
app.add_typer(usb_app, name="usb")
app.add_typer(analyze_app, name="analyze")
app.add_typer(authority_app, name="authority")
app.add_typer(compute_app, name="compute")
+app.add_typer(lab_app, name="lab")
class ToolStatus(TypedDict):
@@ -304,6 +313,126 @@ def doctor(
console.print(f"- {note}")
+@lab_app.command("publish-integrated")
+def publish_integrated_lab(
+ result: Annotated[
+ Path,
+ typer.Option(
+ exists=True,
+ file_okay=False,
+ readable=True,
+ resolve_path=True,
+ help="Accepted e10-integrated-perception result directory.",
+ ),
+ ],
+ session_id: Annotated[
+ str,
+ typer.Option("--session-id", help="New immutable LAB session id."),
+ ],
+ lab_id: Annotated[
+ str,
+ typer.Option("--lab-id", help="LAB marker, for example 'LAB E19'."),
+ ],
+ display_name: Annotated[
+ str,
+ typer.Option("--display-name", help="Operator-facing saved-session title."),
+ ],
+) -> None:
+ """Publish one exact accepted visual run without copying raw sensor bytes."""
+
+ repository_root = Path(__file__).resolve().parents[4]
+ try:
+ published = publish_integrated_lab_instance(
+ repository_root=repository_root,
+ source_result_root=result,
+ lab_session_id=session_id,
+ lab_id=lab_id,
+ display_name=display_name,
+ )
+ except (OSError, SessionIntegrityError, ValueError) as exc:
+ console.print(f"[red]LAB publication failed:[/red] {exc}")
+ raise typer.Exit(code=2) from exc
+ console.print(
+ "[green]LAB instance published.[/green] "
+ f"session={published.binding.session_id}; "
+ f"source={published.binding.source_session_id}; "
+ f"result={published.binding.result_id}; "
+ "raw_sensor_payloads_copied=false"
+ )
+
+
+@lab_app.command("publish-e21")
+def publish_e21_lab(
+ result: Annotated[
+ Path,
+ typer.Option(
+ exists=True,
+ file_okay=False,
+ readable=True,
+ resolve_path=True,
+ help="Accepted E21 realtime-envelope result directory.",
+ ),
+ ],
+ worker_result: Annotated[
+ Path,
+ typer.Option(
+ "--worker-result",
+ exists=True,
+ file_okay=False,
+ readable=True,
+ resolve_path=True,
+ help="Exact E15 shadow-worker result referenced by E21.",
+ ),
+ ],
+ semantic_reference: Annotated[
+ Path,
+ typer.Option(
+ "--semantic-reference",
+ exists=True,
+ file_okay=False,
+ readable=True,
+ resolve_path=True,
+ help="Accepted E10 result containing the SHA-matched semantic masks.",
+ ),
+ ],
+ session_id: Annotated[
+ str,
+ typer.Option("--session-id", help="New immutable LAB session id."),
+ ],
+ lab_id: Annotated[
+ str,
+ typer.Option("--lab-id", help="LAB marker, for example 'LAB E21'."),
+ ],
+ display_name: Annotated[
+ str,
+ typer.Option("--display-name", help="Operator-facing saved-session title."),
+ ],
+) -> None:
+ """Publish the exact accepted E21 envelope as a visual 60-second LAB run."""
+
+ repository_root = Path(__file__).resolve().parents[4]
+ try:
+ published = publish_e21_lab_instance(
+ repository_root=repository_root,
+ e21_result_root=result,
+ worker_result_root=worker_result,
+ semantic_reference_result_root=semantic_reference,
+ lab_session_id=session_id,
+ lab_id=lab_id,
+ display_name=display_name,
+ )
+ except (OSError, SessionIntegrityError, ValueError) as exc:
+ console.print(f"[red]E21 LAB publication failed:[/red] {exc}")
+ raise typer.Exit(code=2) from exc
+ console.print(
+ "[green]E21 LAB instance published.[/green] "
+ f"session={published.binding.session_id}; "
+ f"source={published.binding.source_session_id}; "
+ f"result={published.binding.result_id}; "
+ "source_payloads_mutated=false"
+ )
+
+
@app.command("serve")
def serve_console(
port: Annotated[
diff --git a/src/k1link/sessions/__init__.py b/src/k1link/sessions/__init__.py
index d756b83..592d4e1 100644
--- a/src/k1link/sessions/__init__.py
+++ b/src/k1link/sessions/__init__.py
@@ -5,6 +5,7 @@ from .active import (
ActiveSessionLeaseError,
recover_stale_active_session_marker,
)
+from .lab_cache import publish_lab_replay_cache
from .media import (
RECORDED_MEDIA_MANIFEST_SCHEMA,
RecordedMediaFile,
@@ -14,6 +15,7 @@ from .media import (
validate_recorded_media_timeline,
)
from .models import (
+ LabSessionBinding,
LayoutConflictError,
ObservationArtifactCandidate,
ObservationSessionCandidate,
@@ -50,6 +52,7 @@ from .store import (
__all__ = [
"LayoutConflictError",
+ "LabSessionBinding",
"ActiveSessionLease",
"ActiveSessionLeaseError",
"MaterializedRecording",
@@ -59,6 +62,7 @@ __all__ = [
"ObservationSessionCandidate",
"PluginRecordingExportCancelled",
"PluginRecordingExportError",
+ "publish_lab_replay_cache",
"RecordingMaterializationCancelled",
"RecordedMediaArtifact",
"RECORDED_MEDIA_MANIFEST_SCHEMA",
diff --git a/src/k1link/sessions/lab_cache.py b/src/k1link/sessions/lab_cache.py
new file mode 100644
index 0000000..cd96c78
--- /dev/null
+++ b/src/k1link/sessions/lab_cache.py
@@ -0,0 +1,139 @@
+"""Publish zero-copy replay caches for immutable LAB session aliases."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import stat
+from pathlib import Path
+from typing import Any
+
+from k1link.artifacts import write_json_atomic
+
+from .models import SessionIntegrityError
+from .recording import (
+ COMPATIBLE_CACHE_SCHEMAS,
+ RECORDING_CACHE_FILENAME,
+ RECORDING_CACHE_SIDECAR_FILENAME,
+)
+
+
+def publish_lab_replay_cache(
+ data_dir: Path,
+ *,
+ source_session_id: str,
+ lab_session_id: str,
+) -> None:
+ """Hard-link a validated source RRD and clone its path-free sidecars.
+
+ The RRD bytes are immutable and independent of the catalog id, so a hard
+ link gives every LAB instance a normal cache endpoint without consuming a
+ second copy of the multi-gigabyte recording.
+ """
+
+ root = data_dir.expanduser().resolve(strict=True)
+ recordings = (root / "recordings").resolve(strict=True)
+ source_root = recordings / source_session_id
+ source_rrd = source_root / RECORDING_CACHE_FILENAME
+ source_sidecar = source_root / RECORDING_CACHE_SIDECAR_FILENAME
+ _require_regular(source_rrd, source_root)
+ _require_regular(source_sidecar, source_root)
+ document = _read_object(source_sidecar)
+ if (
+ document.get("schema_version") not in COMPATIBLE_CACHE_SCHEMAS
+ or document.get("session_id") != source_session_id
+ ):
+ raise SessionIntegrityError("source recording cache is incompatible")
+
+ target_root = recordings / lab_session_id
+ target_root.mkdir(mode=0o700, parents=False, exist_ok=True)
+ if target_root.is_symlink() or target_root.resolve() != target_root:
+ raise SessionIntegrityError("LAB recording cache root is unsafe")
+ target_rrd = target_root / RECORDING_CACHE_FILENAME
+ if target_rrd.exists():
+ source_stat = _require_regular(source_rrd, source_root)
+ target_stat = _require_regular(target_rrd, target_root)
+ if (source_stat.st_dev, source_stat.st_ino) != (target_stat.st_dev, target_stat.st_ino):
+ raise SessionIntegrityError("LAB recording cache already contains different bytes")
+ else:
+ temporary = target_root / f".{RECORDING_CACHE_FILENAME}.{os.getpid()}.tmp"
+ try:
+ os.link(source_rrd, temporary, follow_symlinks=False)
+ os.replace(temporary, target_rrd)
+ finally:
+ temporary.unlink(missing_ok=True)
+ target_stat = _require_regular(target_rrd, target_root)
+ cloned = {
+ **document,
+ "session_id": lab_session_id,
+ "recording_mtime_ns": target_stat.st_mtime_ns,
+ }
+ write_json_atomic(target_root / RECORDING_CACHE_SIDECAR_FILENAME, cloned)
+ os.chmod(target_root / RECORDING_CACHE_SIDECAR_FILENAME, 0o600)
+ _clone_recorded_media_sidecars(
+ root / "recorded-media-preparations",
+ source_session_id=source_session_id,
+ lab_session_id=lab_session_id,
+ )
+
+
+def _clone_recorded_media_sidecars(
+ root: Path,
+ *,
+ source_session_id: str,
+ lab_session_id: str,
+) -> None:
+ if not root.is_dir():
+ return
+ for source in root.iterdir():
+ if source.is_symlink() or not source.is_file():
+ continue
+ try:
+ document = _read_object(source)
+ except (OSError, SessionIntegrityError):
+ continue
+ if document.get("session_id") != source_session_id:
+ continue
+ artifact_id = document.get("artifact_id")
+ if not isinstance(artifact_id, str):
+ raise SessionIntegrityError("recorded media sidecar has no artifact identity")
+ body = {key: value for key, value in document.items() if key != "checksum_sha256"}
+ body["session_id"] = lab_session_id
+ checksum = hashlib.sha256(_canonical_json(body)).hexdigest()
+ target = root / _media_sidecar_name(lab_session_id, artifact_id)
+ write_json_atomic(target, {**body, "checksum_sha256": checksum})
+ os.chmod(target, 0o600)
+
+
+def _media_sidecar_name(session_id: str, artifact_id: str) -> str:
+ key = f"{session_id}\0{artifact_id}".encode()
+ return f"{hashlib.sha256(key).hexdigest()}.json"
+
+
+def _canonical_json(value: object) -> bytes:
+ return json.dumps(
+ value,
+ ensure_ascii=False,
+ sort_keys=True,
+ separators=(",", ":"),
+ allow_nan=False,
+ ).encode("utf-8")
+
+
+def _read_object(path: Path) -> dict[str, Any]:
+ value = json.loads(path.read_text(encoding="utf-8"))
+ if not isinstance(value, dict):
+ raise SessionIntegrityError("cache sidecar is not a JSON object")
+ return value
+
+
+def _require_regular(path: Path, parent: Path) -> os.stat_result:
+ metadata = path.lstat()
+ if (
+ stat.S_ISLNK(metadata.st_mode)
+ or not stat.S_ISREG(metadata.st_mode)
+ or path.parent.resolve(strict=True) != parent.resolve(strict=True)
+ ):
+ raise SessionIntegrityError("cache artifact is not a confined regular file")
+ return metadata
diff --git a/src/k1link/sessions/models.py b/src/k1link/sessions/models.py
index 3b28572..58159c4 100644
--- a/src/k1link/sessions/models.py
+++ b/src/k1link/sessions/models.py
@@ -28,6 +28,35 @@ class LayoutConflictError(SessionStoreError):
"""A workspace layout revision changed since the caller loaded it."""
+@dataclass(frozen=True, slots=True)
+class LabSessionBinding:
+ """Immutable provenance for one derived laboratory replay."""
+
+ session_id: str
+ source_session_id: str
+ lab_id: str
+ result_kind: str
+ result_id: str
+ source_result_id: str | None
+ config_sha256: str | None
+ run_created_at_utc: str
+ published_at_utc: str
+ provenance: dict[str, Any]
+
+ def as_dict(self) -> dict[str, Any]:
+ return {
+ "lab_id": self.lab_id,
+ "source_session_id": self.source_session_id,
+ "result_kind": self.result_kind,
+ "result_id": self.result_id,
+ "source_result_id": self.source_result_id,
+ "config_sha256": self.config_sha256,
+ "run_created_at_utc": self.run_created_at_utc,
+ "published_at_utc": self.published_at_utc,
+ "provenance": self.provenance,
+ }
+
+
@dataclass(frozen=True, slots=True)
class SessionSource:
source_id: str
@@ -81,9 +110,10 @@ class SessionSummary:
total_bytes: int
replayable: bool
origin: str
+ lab: LabSessionBinding | None = None
def as_dict(self) -> dict[str, Any]:
- return {
+ document: dict[str, Any] = {
"schema_version": "missioncore.observation-session-summary/v1",
"session_id": self.session_id,
"display_name": self.display_name,
@@ -97,6 +127,9 @@ class SessionSummary:
"replayable": self.replayable,
"origin": self.origin,
}
+ if self.lab is not None:
+ document["lab"] = self.lab.as_dict()
+ return document
@dataclass(frozen=True, slots=True)
diff --git a/src/k1link/sessions/store.py b/src/k1link/sessions/store.py
index 1d887d0..414ef98 100644
--- a/src/k1link/sessions/store.py
+++ b/src/k1link/sessions/store.py
@@ -16,6 +16,7 @@ from uuid import uuid4
from k1link.artifacts import utc_now_iso
from .models import (
+ LabSessionBinding,
LayoutConflictError,
ObservationArtifactCandidate,
ObservationSessionCandidate,
@@ -39,6 +40,10 @@ from .plugin_contract import ObservationArchiveSource
IDENTIFIER_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
MAX_LAYOUT_BYTES = 256 * 1024
DATABASE_NAME = "mission-core.sqlite3"
+LAB_ARCHIVE_ID = "missioncore.lab-instances"
+LAB_ORIGIN = "missioncore.lab-instance/v1"
+LAB_ID_PATTERN = re.compile(r"^LAB [A-Z][A-Z0-9._-]{0,31}$")
+SHA256_PATTERN = re.compile(r"^[a-f0-9]{64}$")
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS observation_sessions (
@@ -94,6 +99,24 @@ CREATE TABLE IF NOT EXISTS observation_session_sources (
ON DELETE CASCADE
);
+CREATE TABLE IF NOT EXISTS observation_lab_instances (
+ session_id TEXT PRIMARY KEY
+ REFERENCES observation_sessions(session_id) ON DELETE CASCADE,
+ source_session_id TEXT NOT NULL
+ REFERENCES observation_sessions(session_id) ON DELETE RESTRICT,
+ lab_id TEXT NOT NULL,
+ result_kind TEXT NOT NULL,
+ result_id TEXT NOT NULL,
+ source_result_id TEXT,
+ config_sha256 TEXT,
+ run_created_at_utc TEXT NOT NULL,
+ published_at_utc TEXT NOT NULL,
+ provenance_json TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS observation_lab_instances_source
+ON observation_lab_instances(source_session_id, published_at_utc DESC);
+
CREATE TABLE IF NOT EXISTS workspace_layouts (
workspace_id TEXT PRIMARY KEY,
layout_schema_version INTEGER NOT NULL,
@@ -162,7 +185,18 @@ class SessionStore:
"WHERE plugin_id = ? AND archive_id = ? AND allowed_root = ?",
(source.plugin_id, source.archive_id, str(allowed_root)),
).fetchall()
- stale = [row["session_id"] for row in indexed if row["session_id"] not in discovered]
+ referenced_sources = {
+ row["source_session_id"]
+ for row in connection.execute(
+ "SELECT DISTINCT source_session_id FROM observation_lab_instances"
+ ).fetchall()
+ }
+ stale = [
+ row["session_id"]
+ for row in indexed
+ if row["session_id"] not in discovered
+ and row["session_id"] not in referenced_sources
+ ]
connection.executemany(
"DELETE FROM observation_sessions WHERE session_id = ?",
((session_id,) for session_id in stale),
@@ -193,9 +227,31 @@ class SessionStore:
"ORDER BY COALESCE(started_at_utc, '') DESC, session_id DESC LIMIT ?",
parameters,
).fetchall()
+ lab_rows = (
+ {
+ row["session_id"]: row
+ for row in connection.execute(
+ "SELECT * FROM observation_lab_instances WHERE session_id IN "
+ f"({','.join('?' for _ in rows)})", # noqa: S608 - placeholder count only
+ tuple(row["session_id"] for row in rows),
+ ).fetchall()
+ }
+ if rows
+ else {}
+ )
has_more = len(rows) > limit
selected = rows[:limit]
- items = tuple(_summary_from_row(row) for row in selected)
+ items = tuple(
+ _summary_from_row(
+ row,
+ lab=(
+ _lab_binding_from_row(lab_rows[row["session_id"]])
+ if row["session_id"] in lab_rows
+ else None
+ ),
+ )
+ for row in selected
+ )
next_cursor = items[-1].session_id if has_more and items else None
return SessionPage(items=items, next_cursor=next_cursor)
@@ -218,6 +274,10 @@ class SessionStore:
"FROM observation_session_artifacts WHERE session_id = ? ORDER BY artifact_id",
(session_id,),
).fetchall()
+ lab_row = connection.execute(
+ "SELECT * FROM observation_lab_instances WHERE session_id = ?",
+ (session_id,),
+ ).fetchone()
sources = tuple(
SessionSource(
source_id=source["source_id"],
@@ -240,7 +300,191 @@ class SessionStore:
)
for artifact in artifact_rows
)
- return SessionDetail(summary=_summary_from_row(row), sources=sources, artifacts=artifacts)
+ return SessionDetail(
+ summary=_summary_from_row(
+ row,
+ lab=None if lab_row is None else _lab_binding_from_row(lab_row),
+ ),
+ sources=sources,
+ artifacts=artifacts,
+ )
+
+ def get_lab_instance(self, session_id: str) -> LabSessionBinding | None:
+ """Return immutable LAB provenance without exposing filesystem locators."""
+
+ _validate_identifier(session_id, "session id")
+ with self._connect() as connection:
+ row = connection.execute(
+ "SELECT * FROM observation_lab_instances WHERE session_id = ?",
+ (session_id,),
+ ).fetchone()
+ return None if row is None else _lab_binding_from_row(row)
+
+ def publish_lab_instance(
+ self,
+ *,
+ session_id: str,
+ source_session_id: str,
+ display_name: str,
+ lab_id: str,
+ result_kind: str,
+ result_id: str,
+ run_created_at_utc: str,
+ source_result_id: str | None = None,
+ config_sha256: str | None = None,
+ provenance: dict[str, Any] | None = None,
+ ) -> LabSessionBinding:
+ """Append one immutable catalog projection over an existing source session.
+
+ Artifact/source rows are copied as references to the sealed evidence.
+ No raw payload is copied and the LAB row uses a distinct archive id so
+ device-plugin reconciliation cannot replace it.
+ """
+
+ _validate_identifier(session_id, "LAB session id")
+ _validate_identifier(source_session_id, "source session id")
+ if session_id == source_session_id:
+ raise ValueError("LAB session id must differ from its source")
+ if LAB_ID_PATTERN.fullmatch(lab_id) is None:
+ raise ValueError("LAB id must use the form 'LAB E21'")
+ _validate_identifier(result_kind, "LAB result kind")
+ _validate_identifier(result_id, "LAB result id")
+ if source_result_id is not None:
+ _validate_identifier(source_result_id, "LAB source result id")
+ if config_sha256 is not None and SHA256_PATTERN.fullmatch(config_sha256) is None:
+ raise ValueError("LAB configuration SHA-256 is invalid")
+ normalized_name = display_name.strip()
+ if not 1 <= len(normalized_name) <= 160:
+ raise ValueError("LAB display name must contain 1..160 characters")
+ _require_utc_timestamp(run_created_at_utc, "LAB run creation time")
+ serialized_provenance = _serialize_provenance(provenance or {})
+ published_at = utc_now_iso()
+
+ with self._lock, self._connect() as connection:
+ connection.execute("BEGIN IMMEDIATE")
+ source = connection.execute(
+ "SELECT * FROM observation_sessions WHERE session_id = ?",
+ (source_session_id,),
+ ).fetchone()
+ if source is None:
+ connection.rollback()
+ raise SessionNotFoundError("LAB source observation session was not found")
+ if (
+ connection.execute(
+ "SELECT 1 FROM observation_lab_instances WHERE session_id = ?",
+ (source_session_id,),
+ ).fetchone()
+ is not None
+ ):
+ connection.rollback()
+ raise SessionIntegrityError("LAB instances cannot be chained")
+
+ existing = connection.execute(
+ "SELECT * FROM observation_lab_instances WHERE session_id = ?",
+ (session_id,),
+ ).fetchone()
+ expected = {
+ "source_session_id": source_session_id,
+ "lab_id": lab_id,
+ "result_kind": result_kind,
+ "result_id": result_id,
+ "source_result_id": source_result_id,
+ "config_sha256": config_sha256,
+ "run_created_at_utc": run_created_at_utc,
+ "provenance_json": serialized_provenance,
+ }
+ if existing is not None:
+ if any(existing[key] != value for key, value in expected.items()):
+ connection.rollback()
+ raise SessionIntegrityError(
+ "LAB session id is already bound to different provenance"
+ )
+ connection.commit()
+ return _lab_binding_from_row(existing)
+ if (
+ connection.execute(
+ "SELECT 1 FROM observation_sessions WHERE session_id = ?",
+ (session_id,),
+ ).fetchone()
+ is not None
+ ):
+ connection.rollback()
+ raise SessionIntegrityError("LAB session id already exists in the catalog")
+
+ connection.execute(
+ "INSERT INTO observation_sessions "
+ "(session_id, plugin_id, archive_id, display_name, status, "
+ "started_at_utc, completed_at_utc, duration_seconds, modalities_json, "
+ "replayable, origin, source_count, total_bytes, "
+ "primary_replay_artifact_id, timeline_origin_epoch_ns, "
+ "timeline_origin_monotonic_ns, allowed_root, session_root, "
+ "created_at_utc, updated_at_utc) "
+ "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+ (
+ session_id,
+ source["plugin_id"],
+ LAB_ARCHIVE_ID,
+ normalized_name,
+ source["status"],
+ run_created_at_utc,
+ run_created_at_utc,
+ source["duration_seconds"],
+ source["modalities_json"],
+ source["replayable"],
+ LAB_ORIGIN,
+ source["source_count"],
+ source["total_bytes"],
+ source["primary_replay_artifact_id"],
+ source["timeline_origin_epoch_ns"],
+ source["timeline_origin_monotonic_ns"],
+ source["allowed_root"],
+ source["session_root"],
+ published_at,
+ published_at,
+ ),
+ )
+ connection.execute(
+ "INSERT INTO observation_session_artifacts "
+ "(session_id, artifact_id, kind, media_type, byte_length, sha256, "
+ "integrity_status, locator, replay_byte_length) "
+ "SELECT ?, artifact_id, kind, media_type, byte_length, sha256, "
+ "integrity_status, locator, replay_byte_length "
+ "FROM observation_session_artifacts WHERE session_id = ?",
+ (session_id, source_session_id),
+ )
+ connection.execute(
+ "INSERT INTO observation_session_sources "
+ "(session_id, source_id, semantic_channel_id, modality, status, "
+ "seekable, artifact_id) "
+ "SELECT ?, source_id, semantic_channel_id, modality, status, "
+ "seekable, artifact_id FROM observation_session_sources "
+ "WHERE session_id = ?",
+ (session_id, source_session_id),
+ )
+ connection.execute(
+ "INSERT INTO observation_lab_instances "
+ "(session_id, source_session_id, lab_id, result_kind, result_id, "
+ "source_result_id, config_sha256, run_created_at_utc, "
+ "published_at_utc, provenance_json) "
+ "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+ (
+ session_id,
+ source_session_id,
+ lab_id,
+ result_kind,
+ result_id,
+ source_result_id,
+ config_sha256,
+ run_created_at_utc,
+ published_at,
+ serialized_provenance,
+ ),
+ )
+ connection.commit()
+ binding = self.get_lab_instance(session_id)
+ if binding is None:
+ raise SessionIntegrityError("LAB session publication was not durable")
+ return binding
def delete_session(self, session_id: str) -> None:
"""Permanently delete one exact catalogued evidence directory and row."""
@@ -248,6 +492,29 @@ class SessionStore:
_validate_identifier(session_id, "session id")
with self._lock:
with self._connect() as connection:
+ lab = connection.execute(
+ "SELECT 1 FROM observation_lab_instances WHERE session_id = ?",
+ (session_id,),
+ ).fetchone()
+ if lab is not None:
+ connection.execute("BEGIN IMMEDIATE")
+ deleted = connection.execute(
+ "DELETE FROM observation_sessions WHERE session_id = ?",
+ (session_id,),
+ ).rowcount
+ connection.commit()
+ if deleted != 1:
+ raise SessionNotFoundError("observation session was not found")
+ return
+ dependent = connection.execute(
+ "SELECT session_id FROM observation_lab_instances "
+ "WHERE source_session_id = ? LIMIT 1",
+ (session_id,),
+ ).fetchone()
+ if dependent is not None:
+ raise SessionIntegrityError(
+ "source session has LAB instances; remove those instances first"
+ )
row = connection.execute(
"SELECT allowed_root, session_root FROM observation_sessions "
"WHERE session_id = ?",
@@ -737,7 +1004,11 @@ def _validate_candidate_replay(
raise SessionIntegrityError("non-replayable observation declares replay artifacts")
-def _summary_from_row(row: sqlite3.Row) -> SessionSummary:
+def _summary_from_row(
+ row: sqlite3.Row,
+ *,
+ lab: LabSessionBinding | None = None,
+) -> SessionSummary:
raw_modalities = json.loads(row["modalities_json"])
modalities = tuple(cast(SessionModality, value) for value in raw_modalities)
return SessionSummary(
@@ -752,9 +1023,58 @@ def _summary_from_row(row: sqlite3.Row) -> SessionSummary:
total_bytes=row["total_bytes"],
replayable=bool(row["replayable"]),
origin=row["origin"],
+ lab=lab,
)
+def _lab_binding_from_row(row: sqlite3.Row) -> LabSessionBinding:
+ try:
+ provenance = json.loads(row["provenance_json"])
+ except (TypeError, json.JSONDecodeError) as exc:
+ raise SessionIntegrityError("stored LAB provenance is invalid") from exc
+ if not isinstance(provenance, dict):
+ raise SessionIntegrityError("stored LAB provenance is not an object")
+ return LabSessionBinding(
+ session_id=row["session_id"],
+ source_session_id=row["source_session_id"],
+ lab_id=row["lab_id"],
+ result_kind=row["result_kind"],
+ result_id=row["result_id"],
+ source_result_id=row["source_result_id"],
+ config_sha256=row["config_sha256"],
+ run_created_at_utc=row["run_created_at_utc"],
+ published_at_utc=row["published_at_utc"],
+ provenance=provenance,
+ )
+
+
+def _serialize_provenance(value: dict[str, Any]) -> str:
+ try:
+ serialized = json.dumps(
+ value,
+ ensure_ascii=False,
+ sort_keys=True,
+ separators=(",", ":"),
+ allow_nan=False,
+ )
+ except (TypeError, ValueError) as exc:
+ raise ValueError("LAB provenance must be JSON-compatible") from exc
+ if len(serialized.encode("utf-8")) > 256 * 1024:
+ raise ValueError("LAB provenance exceeds 256 KiB")
+ return serialized
+
+
+def _require_utc_timestamp(value: str, field: str) -> None:
+ from datetime import datetime
+
+ try:
+ parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+ except ValueError as exc:
+ raise ValueError(f"{field} is invalid") from exc
+ if parsed.tzinfo is None:
+ raise ValueError(f"{field} must include a timezone")
+
+
def _layout_from_row(row: sqlite3.Row) -> WorkspaceLayout:
payload = json.loads(row["layout_json"])
if not isinstance(payload, dict):
diff --git a/src/k1link/web/session_api.py b/src/k1link/web/session_api.py
index 9131129..79a9b1c 100644
--- a/src/k1link/web/session_api.py
+++ b/src/k1link/web/session_api.py
@@ -327,6 +327,7 @@ def build_session_router(
"modalities": list(item.modalities),
"duration_seconds": item.duration_seconds or 0.0,
"replayable": item.replayable,
+ **({"lab": item.lab.as_dict()} if item.lab is not None else {}),
**(
{
"preparation": _catalog_preparation_document(
@@ -410,9 +411,8 @@ def build_session_router(
detail="Запись сейчас открыта в интерфейсе. Закройте её и повторите удаление.",
)
- if (
- recording_preparation_manager is not None
- and not recording_preparation_manager.discard(session_id)
+ if recording_preparation_manager is not None and not recording_preparation_manager.discard(
+ session_id
):
raise HTTPException(
status_code=409,
@@ -960,10 +960,7 @@ def build_session_router(
},
)
- @router.get(
- "/api/v1/observation-sessions/{session_id}/perception-media/"
- "{result_id}/manifest"
- )
+ @router.get("/api/v1/observation-sessions/{session_id}/perception-media/{result_id}/manifest")
def get_recorded_perception_media_manifest(
session_id: str,
result_id: str,
@@ -988,8 +985,7 @@ def build_session_router(
)
@router.api_route(
- "/api/v1/observation-sessions/{session_id}/perception-media/"
- "{result_id}/recording.mp4",
+ "/api/v1/observation-sessions/{session_id}/perception-media/{result_id}/recording.mp4",
methods=["GET", "HEAD"],
)
def stream_recorded_perception_media(
@@ -1593,10 +1589,7 @@ def _recorded_perception_manifest_document(
) -> dict[str, Any]:
encoded_session_id = quote(video.session_id, safe="")
encoded_result_id = quote(video.result_id, safe="")
- base = (
- f"/api/v1/observation-sessions/{encoded_session_id}/perception-media/"
- f"{encoded_result_id}"
- )
+ base = f"/api/v1/observation-sessions/{encoded_session_id}/perception-media/{encoded_result_id}"
return {
"schema_version": RECORDED_MEDIA_STREAM_MANIFEST_SCHEMA,
"source_id": video.public_source_id,
@@ -1835,9 +1828,7 @@ def _recorded_media_epoch_response(
if len(matches) != 1:
raise HTTPException(status_code=404, detail="Эпоха записанного медиаканала не найдена.")
epoch = matches[0]
- byte_length = epoch.init_byte_length + sum(
- segment.byte_length for segment in epoch.segments
- )
+ byte_length = epoch.init_byte_length + sum(segment.byte_length for segment in epoch.segments)
start, end = (0, byte_length - 1)
status_code = 200
if range_header is not None:
@@ -1851,9 +1842,7 @@ def _recorded_media_epoch_response(
"ETag": etag,
"Content-Length": str(end - start + 1),
"X-Content-Type-Options": "nosniff",
- "Content-Disposition": (
- f'inline; filename="recorded-camera-{epoch.ordinal}.mp4"'
- ),
+ "Content-Disposition": (f'inline; filename="recorded-camera-{epoch.ordinal}.mp4"'),
}
if status_code == 206:
headers["Content-Range"] = f"bytes {start}-{end}/{byte_length}"
diff --git a/tests/test_session_api.py b/tests/test_session_api.py
index 68de392..eabf247 100644
--- a/tests/test_session_api.py
+++ b/tests/test_session_api.py
@@ -265,6 +265,39 @@ def test_session_router_lists_details_and_dispatches_opaque_replay(tmp_path: Pat
assert_no_local_paths(value, repository)
+def test_session_router_exposes_immutable_lab_provenance(tmp_path: Path) -> None:
+ repository = tmp_path / "repo"
+ sessions = repository / "sessions"
+ source = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
+ store = SessionStore(repository, data_dir=tmp_path / "data")
+ store.reconcile_archive(xgrids_k1_archive_source(sessions))
+ binding = store.publish_lab_instance(
+ session_id="lab-e21-d0201712",
+ source_session_id=source.name,
+ display_name="LAB E21 · Real-time 1× · d0201712",
+ lab_id="LAB E21",
+ result_kind="e21-realtime-envelope",
+ result_id="e10-integrated-perception-" + "a" * 64,
+ source_result_id="e21-realtime-envelope-" + "b" * 64,
+ config_sha256="c" * 64,
+ run_created_at_utc="2026-07-23T15:55:15.548Z",
+ provenance={"source_payloads_mutated": False},
+ )
+ router = build_session_router(store)
+ list_route = endpoint(router, "/api/v1/observation-sessions", "GET")
+ detail_route = endpoint(router, "/api/v1/observation-sessions/{session_id}", "GET")
+
+ listing = list_route(limit=20, cursor=None)
+ item = next(value for value in listing["items"] if value["id"] == binding.session_id)
+ detail = detail_route(session_id=binding.session_id)
+
+ assert item["lab"] == binding.as_dict()
+ assert detail["lab"] == binding.as_dict()
+ assert item["lab"]["source_session_id"] == source.name
+ assert item["lab"]["provenance"]["source_payloads_mutated"] is False
+ assert_no_local_paths((item, detail), repository)
+
+
def test_delete_session_removes_evidence_and_cache_but_refuses_an_open_recording(
tmp_path: Path,
) -> None:
diff --git a/tests/test_session_lab_cache.py b/tests/test_session_lab_cache.py
new file mode 100644
index 0000000..f525c33
--- /dev/null
+++ b/tests/test_session_lab_cache.py
@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+
+from k1link.sessions.lab_cache import publish_lab_replay_cache
+from k1link.sessions.recording import (
+ CACHE_SCHEMA,
+ RECORDING_CACHE_FILENAME,
+ RECORDING_CACHE_SIDECAR_FILENAME,
+)
+
+
+def canonical_json(value: object) -> bytes:
+ return json.dumps(
+ value,
+ ensure_ascii=False,
+ sort_keys=True,
+ separators=(",", ":"),
+ allow_nan=False,
+ ).encode()
+
+
+def test_publish_lab_replay_cache_hardlinks_rrd_and_rebinds_sidecars(
+ tmp_path: Path,
+) -> None:
+ data = tmp_path / "mission-core"
+ recordings = data / "recordings"
+ source = recordings / "source-session"
+ source.mkdir(parents=True)
+ source_rrd = source / RECORDING_CACHE_FILENAME
+ source_rrd.write_bytes(b"RRF2immutable")
+ source_sidecar = {
+ "schema_version": CACHE_SCHEMA,
+ "session_id": source.name,
+ "recording_byte_length": source_rrd.stat().st_size,
+ "recording_mtime_ns": source_rrd.stat().st_mtime_ns,
+ "recording_sha256": hashlib.sha256(source_rrd.read_bytes()).hexdigest(),
+ }
+ (source / RECORDING_CACHE_SIDECAR_FILENAME).write_text(
+ json.dumps(source_sidecar),
+ encoding="utf-8",
+ )
+ media_root = data / "recorded-media-preparations"
+ media_root.mkdir()
+ media_body = {
+ "schema_version": "missioncore.recorded-media-preparation/v1",
+ "session_id": source.name,
+ "artifact_id": "recorded-video-1",
+ "generation": "accepted",
+ }
+ (media_root / "source.json").write_text(
+ json.dumps(
+ {
+ **media_body,
+ "checksum_sha256": hashlib.sha256(canonical_json(media_body)).hexdigest(),
+ }
+ ),
+ encoding="utf-8",
+ )
+
+ publish_lab_replay_cache(
+ data,
+ source_session_id=source.name,
+ lab_session_id="lab-e21-result",
+ )
+
+ lab = recordings / "lab-e21-result"
+ lab_rrd = lab / RECORDING_CACHE_FILENAME
+ assert lab_rrd.read_bytes() == source_rrd.read_bytes()
+ assert lab_rrd.stat().st_ino == source_rrd.stat().st_ino
+ assert lab_rrd.stat().st_nlink == 2
+ cloned_cache = json.loads(
+ (lab / RECORDING_CACHE_SIDECAR_FILENAME).read_text(encoding="utf-8")
+ )
+ assert cloned_cache["session_id"] == "lab-e21-result"
+ cloned_media = [
+ json.loads(path.read_text(encoding="utf-8"))
+ for path in media_root.iterdir()
+ if path.name != "source.json"
+ ]
+ assert len(cloned_media) == 1
+ assert cloned_media[0]["session_id"] == "lab-e21-result"
+ cloned_checksum = cloned_media[0].pop("checksum_sha256")
+ assert cloned_checksum == hashlib.sha256(canonical_json(cloned_media[0])).hexdigest()
+
+ lab_rrd.unlink()
+ assert source_rrd.read_bytes() == b"RRF2immutable"
diff --git a/tests/test_session_store.py b/tests/test_session_store.py
index 0dced5b..e1f83af 100644
--- a/tests/test_session_store.py
+++ b/tests/test_session_store.py
@@ -777,6 +777,78 @@ def test_delete_session_removes_only_the_exact_evidence_tree_and_catalog_row(
assert store.get_session(retained.name).summary.session_id == retained.name
+def test_lab_instance_is_independent_and_never_deletes_source_evidence(
+ tmp_path: Path,
+) -> None:
+ repository = tmp_path / "repo"
+ sessions = repository / "sessions"
+ source = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
+ store = SessionStore(repository, data_dir=tmp_path / "data")
+ archive = xgrids_k1_archive_source(sessions)
+ store.reconcile_archive(archive)
+ source_command = store.prepare_replay(source.name)
+
+ binding = store.publish_lab_instance(
+ session_id="lab-e21-d0201712",
+ source_session_id=source.name,
+ display_name="LAB E21 · RT 1× · d0201712",
+ lab_id="LAB E21",
+ result_kind="e10-integrated-perception",
+ result_id="e10-integrated-perception-" + "a" * 64,
+ source_result_id="e21-realtime-envelope-" + "b" * 64,
+ config_sha256="c" * 64,
+ run_created_at_utc="2026-07-23T15:51:25.000Z",
+ provenance={"storage_mode": "hard-linked-immutable-payloads"},
+ )
+
+ detail = store.get_session(binding.session_id)
+ lab_command = store.prepare_replay(binding.session_id)
+ assert detail.summary.lab == binding
+ assert detail.summary.display_name == "LAB E21 · RT 1× · d0201712"
+ assert lab_command.primary_artifact.path == source_command.primary_artifact.path
+ assert lab_command.session_id == binding.session_id
+ assert source.is_dir()
+
+ with pytest.raises(SessionIntegrityError, match="has LAB instances"):
+ store.delete_session(source.name)
+ assert source.is_dir()
+
+ store.delete_session(binding.session_id)
+ assert source.is_dir()
+ assert store.get_session(source.name).summary.lab is None
+ with pytest.raises(SessionNotFoundError):
+ store.get_session(binding.session_id)
+
+
+def test_lab_instance_publication_is_idempotent_but_provenance_is_immutable(
+ tmp_path: Path,
+) -> None:
+ repository = tmp_path / "repo"
+ sessions = repository / "sessions"
+ source = make_legacy_session(sessions, "20260716T205632Z_viewer_live")
+ store = SessionStore(repository, data_dir=tmp_path / "data")
+ store.reconcile_archive(xgrids_k1_archive_source(sessions))
+ parameters = {
+ "session_id": "lab-e19-36964643",
+ "source_session_id": source.name,
+ "display_name": "LAB E19 · Ground-aware 3D",
+ "lab_id": "LAB E19",
+ "result_kind": "e10-integrated-perception",
+ "result_id": "e10-integrated-perception-" + "d" * 64,
+ "source_result_id": "e10-integrated-perception-" + "e" * 64,
+ "config_sha256": "f" * 64,
+ "run_created_at_utc": "2026-07-23T05:19:43.138Z",
+ "provenance": {"source": "accepted"},
+ }
+
+ first = store.publish_lab_instance(**parameters)
+ second = store.publish_lab_instance(**parameters)
+ assert second == first
+
+ with pytest.raises(SessionIntegrityError, match="different provenance"):
+ store.publish_lab_instance(**{**parameters, "config_sha256": "0" * 64})
+
+
def test_delete_session_rejects_a_catalog_target_outside_its_allowed_root(
tmp_path: Path,
) -> None: