diff --git a/docs/09_OBSERVATION_SESSIONS.md b/docs/09_OBSERVATION_SESSIONS.md index 8c9abca..84d43db 100644 --- a/docs/09_OBSERVATION_SESSIONS.md +++ b/docs/09_OBSERVATION_SESSIONS.md @@ -113,12 +113,17 @@ changing the AI result selected for its source recording. A LAB instance binds: - 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 source evidence remains the only raw sensor master. Full-session LAB replay +caches 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. A bounded +experiment instead stores a small derived RRD slice with exact start/end markers +and omits the source's unbounded recorded-video descriptor. Its integrated +perception overlay owns the matching bounded camera. This prevents Rerun from +holding the final AI frame while a longer source cloud continues to play. +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 @@ -132,7 +137,11 @@ 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. +behavior rather than presenting a fabricated gap-free run. The saved-session +duration and base RRD range must equal the E21 visual result range; attaching +the 60-second result to the complete 8:55 source replay is an invalid +presentation because latest-at image data would appear static after the E21 +window ends. 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. diff --git a/experiments/perception/LAB_E21_REPORT_2026-07-23.md b/experiments/perception/LAB_E21_REPORT_2026-07-23.md index 61c252f..374faf5 100644 --- a/experiments/perception/LAB_E21_REPORT_2026-07-23.md +++ b/experiments/perception/LAB_E21_REPORT_2026-07-23.md @@ -187,32 +187,43 @@ 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 +The original presentation `lab-e21-d0201712` is retained as historical +evidence: it attached the bounded E21 overlay to the complete 8:55 source +replay. Its 601 camera/AI frames are dynamic between 35.421857 and 95.383857 +seconds, but Rerun then holds the last image while the source cloud continues. +It is not the correct operator comparison view. + +The synchronized publication is the separate saved-session instance +`lab-e21-window2-d0201712`, displayed as +`LAB E21.3 · Real-time 1× · synchronized 60s · d0201712`. It keeps the +source result immutable, materializes a bounded 4.43 MB spatial RRD and ends +the base scene on the exact final AI timestamp. It does not replace +`RAVNOVES00`, the historical E21 presentation or the comparison baseline `lab-e19-36964643`. | Publication item | Value | | --- | --- | | LAB source session | `20260720T065719Z_viewer_live` | -| Visual result | `e10-integrated-perception-d0ecafd5e56f8ef132b8cb88f45ec275aa17470c682c6cce7c08f5597bfd7bd6` | +| Synchronized LAB session | `lab-e21-window2-d0201712` | +| Visual result | `e10-integrated-perception-5cbba0b9839ae843c05607cec8448a2b43464eb81edeb11cf7c01cfec8464c75` | | 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 | +| Replay range | 35.421857292–95.383857292 s | +| Saved-session duration | 59.962 s | +| Bounded spatial RRD | 4,429,382 bytes | +| Initial visual overlay publication | 144.54 s observed | +| Warm visual overlay retrieval | 0.046 s observed | -The 111.47-second step is viewer-container publication from already computed +The 144.54-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. +operator handoff. Warm UI openings reuse the resulting 7,882,010-byte Rerun +overlay. The synchronized base recording contains exact range markers, so the +host timeline is a 59.962-second comparison window and cannot advance into a +static post-E21 tail. ## Evidence diff --git a/src/k1link/compute/lab_instances.py b/src/k1link/compute/lab_instances.py index e979a73..10865c1 100644 --- a/src/k1link/compute/lab_instances.py +++ b/src/k1link/compute/lab_instances.py @@ -199,6 +199,8 @@ def publish_e21_lab_instance( store.data_dir, source_session_id=reference.job.session_id, lab_session_id=lab_session_id, + timeline_start_ns=round(validated.timeline_start_seconds * 1_000_000_000), + timeline_end_ns=round(validated.timeline_end_seconds * 1_000_000_000), ) binding = store.publish_lab_instance( session_id=lab_session_id, @@ -210,9 +212,13 @@ def publish_e21_lab_instance( 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"]), + duration_seconds=( + validated.timeline_end_seconds - validated.timeline_start_seconds + ), + include_recorded_media=False, provenance={ "schema_version": "missioncore.e21-lab-publication/v1", - "storage_mode": "hard-linked-source-and-bounded-derived-projection", + "storage_mode": "bounded-derived-replay-and-projection", "e21_result_id": e21_document["result_id"], "worker_result_id": worker_document["result_id"], "semantic_reference_result_id": reference.result_id, @@ -227,6 +233,8 @@ def publish_e21_lab_instance( 498, ], "visual_frame_count": frame_count, + "timeline_start_seconds": validated.timeline_start_seconds, + "timeline_end_seconds": validated.timeline_end_seconds, "source_payloads_mutated": False, }, ) diff --git a/src/k1link/sessions/lab_cache.py b/src/k1link/sessions/lab_cache.py index cd96c78..497d1dc 100644 --- a/src/k1link/sessions/lab_cache.py +++ b/src/k1link/sessions/lab_cache.py @@ -5,7 +5,11 @@ from __future__ import annotations import hashlib import json import os +import shutil import stat +import subprocess +import sys +import tempfile from pathlib import Path from typing import Any @@ -24,14 +28,35 @@ def publish_lab_replay_cache( *, source_session_id: str, lab_session_id: str, + timeline_start_ns: int | None = None, + timeline_end_ns: int | None = None, ) -> None: - """Hard-link a validated source RRD and clone its path-free sidecars. + """Publish 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. + Full-session LAB instances hard-link the immutable source bytes. Bounded + experiments materialize only their declared timeline window, so a short AI + run cannot silently hold its final frame over the rest of a long source + recording. """ + bounded = timeline_start_ns is not None or timeline_end_ns is not None + if bounded and ( + not isinstance(timeline_start_ns, int) + or isinstance(timeline_start_ns, bool) + or not isinstance(timeline_end_ns, int) + or isinstance(timeline_end_ns, bool) + or timeline_start_ns < 0 + or timeline_end_ns <= timeline_start_ns + ): + raise ValueError("LAB replay window is invalid") + window_start_ns = -1 + window_end_ns = -1 + if bounded: + assert isinstance(timeline_start_ns, int) + assert isinstance(timeline_end_ns, int) + window_start_ns = timeline_start_ns + window_end_ns = timeline_end_ns + root = data_dir.expanduser().resolve(strict=True) recordings = (root / "recordings").resolve(strict=True) source_root = recordings / source_session_id @@ -52,10 +77,32 @@ def publish_lab_replay_cache( 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") + if bounded: + _require_matching_bounded_cache( + target_rrd, + target_root / RECORDING_CACHE_SIDECAR_FILENAME, + lab_session_id=lab_session_id, + timeline_start_ns=window_start_ns, + timeline_end_ns=window_end_ns, + ) + else: + 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" + ) + elif bounded: + _publish_bounded_rrd( + source_rrd, + target_rrd, + recording_id=lab_session_id, + timeline_start_ns=window_start_ns, + timeline_end_ns=window_end_ns, + ) else: temporary = target_root / f".{RECORDING_CACHE_FILENAME}.{os.getpid()}.tmp" try: @@ -67,15 +114,196 @@ def publish_lab_replay_cache( cloned = { **document, "session_id": lab_session_id, + "recording_byte_length": target_stat.st_size, "recording_mtime_ns": target_stat.st_mtime_ns, + "recording_sha256": _sha256(target_rrd), + "timeline_start_ns": ( + window_start_ns if bounded else document["timeline_start_ns"] + ), + "timeline_end_ns": window_end_ns if bounded else document["timeline_end_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, + if not bounded: + _clone_recorded_media_sidecars( + root / "recorded-media-preparations", + source_session_id=source_session_id, + lab_session_id=lab_session_id, + ) + + +def _publish_bounded_rrd( + source_rrd: Path, + target_rrd: Path, + *, + recording_id: str, + timeline_start_ns: int, + timeline_end_ns: int, +) -> None: + rerun = _rerun_cli() + end_exclusive_ns = timeline_end_ns + 1 + split_recording_prefix = f"{recording_id}-window-" + split_recording_id = f"{split_recording_prefix}1" + with tempfile.TemporaryDirectory( + prefix=".lab-window-", + dir=target_rrd.parent, + ) as temporary_name: + temporary_root = Path(temporary_name) + split_root = temporary_root / "split" + split_root.mkdir() + _run_rerun( + rerun, + "rrd", + "split", + "--output-dir", + str(split_root), + "--timeline", + "session_time", + "--time", + str(timeline_start_ns), + "--time", + str(end_exclusive_ns), + "--recording-id", + split_recording_prefix, + str(source_rrd), + ) + split_name = ( + f"{source_rrd.stem}_{_rrd_duration(timeline_start_ns)}" + f"__{_rrd_duration(end_exclusive_ns)}.rrd" + ) + split_path = split_root / split_name + _require_regular(split_path, split_root) + marker = temporary_root / "window.rrd" + _write_window_markers( + marker, + recording_id=split_recording_id, + timeline_start_ns=timeline_start_ns, + timeline_end_ns=timeline_end_ns, + ) + merged = temporary_root / "bounded.rrd" + _run_rerun( + rerun, + "rrd", + "merge", + "--output", + str(merged), + str(split_path), + str(marker), + ) + _run_rerun(rerun, "rrd", "verify", str(merged)) + _require_regular(merged, temporary_root) + temporary_target = target_rrd.parent / f".{target_rrd.name}.{os.getpid()}.tmp" + try: + os.replace(merged, temporary_target) + os.chmod(temporary_target, 0o600) + os.replace(temporary_target, target_rrd) + finally: + temporary_target.unlink(missing_ok=True) + + +def _write_window_markers( + path: Path, + *, + recording_id: str, + timeline_start_ns: int, + timeline_end_ns: int, +) -> None: + import numpy as np + import rerun as rr + + recording = rr.RecordingStream( + "nodedc_mission_core_recorded", + recording_id=recording_id, ) + stream = rr.binary_stream(recording) + try: + for timestamp, label in ( + (timeline_start_ns, "LAB window start"), + (timeline_end_ns, "LAB window end"), + ): + recording.set_time( + "session_time", + duration=np.timedelta64(timestamp, "ns"), + ) + recording.log( + "/__mission_core/lab_window", + rr.TextDocument(label), + ) + payload = stream.read(flush=True, flush_timeout_sec=5.0) + finally: + recording.disconnect() + if payload is None or not payload.startswith(b"RRF2"): + raise SessionIntegrityError("LAB replay window markers are invalid") + path.write_bytes(payload) + os.chmod(path, 0o600) + + +def _rerun_cli() -> Path: + adjacent = Path(sys.executable).with_name("rerun") + candidate = adjacent if adjacent.is_file() else None + if candidate is None: + resolved = shutil.which("rerun") + candidate = Path(resolved) if resolved is not None else None + if candidate is None or not candidate.is_file() or not os.access(candidate, os.X_OK): + raise SessionIntegrityError("Rerun CLI is unavailable for bounded LAB replay") + return candidate.resolve() + + +def _run_rerun(executable: Path, *arguments: str) -> None: + try: + completed = subprocess.run( + [str(executable), *arguments], + check=False, + capture_output=True, + text=True, + timeout=120, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise SessionIntegrityError("bounded LAB replay command failed") from exc + if completed.returncode != 0: + detail = completed.stderr.strip()[-1000:] + raise SessionIntegrityError(f"bounded LAB replay command failed: {detail}") + + +def _rrd_duration(value_ns: int) -> str: + for suffix, unit in ( + ("s", 1_000_000_000), + ("ms", 1_000_000), + ("us", 1_000), + ("ns", 1), + ): + if value_ns < unit: + continue + whole, remainder = divmod(value_ns, unit) + if remainder == 0: + return f"{whole}{suffix}" + digits = len(str(unit)) - 1 + fraction = f"{remainder:0{digits}d}".rstrip("0") + return f"{whole}_{fraction}{suffix}" + return "0ns" + + +def _require_matching_bounded_cache( + target_rrd: Path, + target_sidecar: Path, + *, + lab_session_id: str, + timeline_start_ns: int, + timeline_end_ns: int, +) -> None: + target_root = target_rrd.parent + metadata = _require_regular(target_rrd, target_root) + _require_regular(target_sidecar, target_root) + document = _read_object(target_sidecar) + if ( + document.get("session_id") != lab_session_id + or document.get("recording_byte_length") != metadata.st_size + or document.get("recording_mtime_ns") != metadata.st_mtime_ns + or document.get("recording_sha256") != _sha256(target_rrd) + or document.get("timeline_start_ns") != timeline_start_ns + or document.get("timeline_end_ns") != timeline_end_ns + ): + raise SessionIntegrityError("LAB recording cache already contains different bytes") def _clone_recorded_media_sidecars( @@ -121,6 +349,14 @@ def _canonical_json(value: object) -> bytes: ).encode("utf-8") +def _sha256(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + def _read_object(path: Path) -> dict[str, Any]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): diff --git a/src/k1link/sessions/store.py b/src/k1link/sessions/store.py index 414ef98..9c9289d 100644 --- a/src/k1link/sessions/store.py +++ b/src/k1link/sessions/store.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import math import os import re import shutil @@ -333,6 +334,8 @@ class SessionStore: source_result_id: str | None = None, config_sha256: str | None = None, provenance: dict[str, Any] | None = None, + duration_seconds: float | None = None, + include_recorded_media: bool = True, ) -> LabSessionBinding: """Append one immutable catalog projection over an existing source session. @@ -357,6 +360,12 @@ class SessionStore: 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") + if duration_seconds is not None and ( + isinstance(duration_seconds, bool) + or not math.isfinite(duration_seconds) + or duration_seconds <= 0 + ): + raise ValueError("LAB duration must be a positive finite value") serialized_provenance = _serialize_provenance(provenance or {}) published_at = utc_now_iso() @@ -428,7 +437,11 @@ class SessionStore: source["status"], run_created_at_utc, run_created_at_utc, - source["duration_seconds"], + ( + source["duration_seconds"] + if duration_seconds is None + else duration_seconds + ), source["modalities_json"], source["replayable"], LAB_ORIGIN, @@ -449,8 +462,9 @@ class SessionStore: "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), + "FROM observation_session_artifacts WHERE session_id = ? " + "AND (? OR kind <> 'recorded-video')", + (session_id, source_session_id, include_recorded_media), ) connection.execute( "INSERT INTO observation_session_sources " @@ -458,8 +472,8 @@ class SessionStore: "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), + "WHERE session_id = ? AND (? OR modality <> 'video')", + (session_id, source_session_id, include_recorded_media), ) connection.execute( "INSERT INTO observation_lab_instances " diff --git a/tests/test_session_lab_cache.py b/tests/test_session_lab_cache.py index f525c33..87b63e6 100644 --- a/tests/test_session_lab_cache.py +++ b/tests/test_session_lab_cache.py @@ -37,6 +37,9 @@ def test_publish_lab_replay_cache_hardlinks_rrd_and_rebinds_sidecars( "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(), + "timeline": "session_time", + "timeline_start_ns": 0, + "timeline_end_ns": 1, } (source / RECORDING_CACHE_SIDECAR_FILENAME).write_text( json.dumps(source_sidecar), @@ -87,3 +90,81 @@ def test_publish_lab_replay_cache_hardlinks_rrd_and_rebinds_sidecars( lab_rrd.unlink() assert source_rrd.read_bytes() == b"RRF2immutable" + + +def test_publish_lab_replay_cache_materializes_exact_bounded_window( + tmp_path: Path, +) -> None: + import numpy as np + import rerun as rr + + data = tmp_path / "mission-core" + recordings = data / "recordings" + source = recordings / "source-session" + source.mkdir(parents=True) + source_rrd = source / RECORDING_CACHE_FILENAME + recording = rr.RecordingStream( + "nodedc_mission_core_recorded", + recording_id=source.name, + ) + stream = rr.binary_stream(recording) + try: + for index, timestamp in enumerate((0, 2_000_000_000, 4_000_000_000, 8_000_000_000)): + recording.set_time( + "session_time", + duration=np.timedelta64(timestamp, "ns"), + ) + recording.log( + "/world/points", + rr.Points3D([[float(index), 0.0, 0.0]]), + ) + payload = stream.read(flush=True, flush_timeout_sec=5.0) + finally: + recording.disconnect() + assert payload is not None + source_rrd.write_bytes(payload) + 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(), + "timeline": "session_time", + "timeline_start_ns": 0, + "timeline_end_ns": 8_000_000_000, + } + (source / RECORDING_CACHE_SIDECAR_FILENAME).write_text( + json.dumps(source_sidecar), + encoding="utf-8", + ) + + publish_lab_replay_cache( + data, + source_session_id=source.name, + lab_session_id="lab-e21-bounded", + timeline_start_ns=2_000_000_000, + timeline_end_ns=4_000_000_000, + ) + + lab = recordings / "lab-e21-bounded" + lab_rrd = lab / RECORDING_CACHE_FILENAME + assert lab_rrd.stat().st_ino != source_rrd.stat().st_ino + bounded = json.loads( + (lab / RECORDING_CACHE_SIDECAR_FILENAME).read_text(encoding="utf-8") + ) + assert bounded["timeline_start_ns"] == 2_000_000_000 + assert bounded["timeline_end_ns"] == 4_000_000_000 + assert bounded["recording_byte_length"] == lab_rrd.stat().st_size + assert bounded["recording_sha256"] == hashlib.sha256(lab_rrd.read_bytes()).hexdigest() + assert "lab_window" not in bounded + + original_stat = lab_rrd.stat() + publish_lab_replay_cache( + data, + source_session_id=source.name, + lab_session_id="lab-e21-bounded", + timeline_start_ns=2_000_000_000, + timeline_end_ns=4_000_000_000, + ) + assert lab_rrd.stat().st_ino == original_stat.st_ino + assert lab_rrd.stat().st_mtime_ns == original_stat.st_mtime_ns diff --git a/tests/test_session_store.py b/tests/test_session_store.py index e1f83af..5c64c81 100644 --- a/tests/test_session_store.py +++ b/tests/test_session_store.py @@ -849,6 +849,38 @@ def test_lab_instance_publication_is_idempotent_but_provenance_is_immutable( store.publish_lab_instance(**{**parameters, "config_sha256": "0" * 64}) +def test_bounded_lab_instance_excludes_unbounded_recorded_media( + tmp_path: Path, +) -> None: + repository = tmp_path / "repo" + sessions = repository / "sessions" + source = make_legacy_session(sessions, "20260716T205632Z_viewer_live") + make_recorded_camera_source(source) + store = SessionStore(repository, data_dir=tmp_path / "data") + store.reconcile_archive(xgrids_k1_archive_source(sessions)) + assert store.list_recorded_media(source.name) + + binding = store.publish_lab_instance( + session_id="lab-e21-window-d0201712", + source_session_id=source.name, + display_name="LAB E21.2 · bounded 60s", + lab_id="LAB E21.2", + result_kind="e21-realtime-envelope", + result_id="e10-integrated-perception-" + "1" * 64, + source_result_id="e21-realtime-envelope-" + "2" * 64, + config_sha256="3" * 64, + run_created_at_utc="2026-07-23T15:55:15.548Z", + duration_seconds=59.962, + include_recorded_media=False, + provenance={"timeline_scope": "bounded"}, + ) + + detail = store.get_session(binding.session_id) + assert detail.summary.duration_seconds == pytest.approx(59.962) + assert store.list_recorded_media(binding.session_id) == () + assert store.prepare_replay(binding.session_id).primary_artifact.path.is_file() + + def test_delete_session_rejects_a_catalog_target_outside_its_allowed_root( tmp_path: Path, ) -> None: