diff --git a/docs/09_OBSERVATION_SESSIONS.md b/docs/09_OBSERVATION_SESSIONS.md index eb19014..4eac88e 100644 --- a/docs/09_OBSERVATION_SESSIONS.md +++ b/docs/09_OBSERVATION_SESSIONS.md @@ -224,7 +224,7 @@ a separately attested storage root: export MISSIONCORE_DATA_DIR=/absolute/private/path/mission-core # Full K1 point-plus-camera evidence must currently remain below the checkout. export MISSIONCORE_EVIDENCE_DIR=/absolute/path/to/NODEDC_MISSION_CORE/.runtime/mission-core/evidence/sessions -# Optional operator retention quota; unset means no application byte quota. +# Optional operator override; the application default is 8 GiB. # export MISSIONCORE_RRD_CACHE_MAX_BYTES=8589934592 export MISSIONCORE_RRD_FREE_SPACE_RESERVE_BYTES=2147483648 uv run k1link serve @@ -298,10 +298,11 @@ global cross-process export gate to cap concurrent RAM, CPU and temporary-disk use. Crash leftovers from candidates, exporter temporary files and staged replay prefixes are scavenged under that lock before capacity accounting. Ready cache hits and active response leases do not wait behind that gate. The derived cache -has no application byte quota by default, so a single multi-hour RRD is not -rejected at 8 GiB. It still preserves a 2 GiB default filesystem reserve. An -operator may set `MISSIONCORE_RRD_CACHE_MAX_BYTES` to enable LRU eviction of -derived RRDs only; native evidence is never deleted. +has an 8 GiB application byte quota and a 2 GiB filesystem reserve by default. +`MISSIONCORE_RRD_CACHE_MAX_BYTES` can override the quota; LRU eviction still +affects derived RRDs only, and native evidence is never deleted. A single +derived RRD larger than the configured quota fails explicitly instead of +consuming the operator host without a bound. ## Capture-clock envelope diff --git a/docs/adr/0011-laboratory-plugin-runtime-handshake-and-transport-seam.md b/docs/adr/0011-laboratory-plugin-runtime-handshake-and-transport-seam.md index 2deb9db..fa52929 100644 --- a/docs/adr/0011-laboratory-plugin-runtime-handshake-and-transport-seam.md +++ b/docs/adr/0011-laboratory-plugin-runtime-handshake-and-transport-seam.md @@ -63,9 +63,10 @@ gateway remains a same-process data-plane route; portable stream/media IPC is a separate future gate. `GET /api/v1/device-plugin-runtimes` exposes the small lifecycle snapshots. The -main health response reports ready/total runtime counts. These values do not -claim device connectivity, sensor health, process containment or durable -supervision. +main health response includes ready/total runtime counts as one component beside +recording-cache, shared-artifact, media-tool and catalog-reconciler readiness. +The plugin values still do not claim device connectivity, sensor health, +process containment or durable supervision. ## Acceptance diff --git a/docs/adr/0022-operator-surface-ownership.md b/docs/adr/0022-operator-surface-ownership.md index b1768c8..da05e39 100644 --- a/docs/adr/0022-operator-surface-ownership.md +++ b/docs/adr/0022-operator-surface-ownership.md @@ -77,6 +77,11 @@ Contour health reads only live contracts: - `GET /api/v1/polygon/worker`; - current Mission Runtime state for active device and perception metrics. +The base health contract now separates local operational readiness from +production shared-artifact readiness and reports bounded cache, catalog +reconciler, media-tool and Map Gateway configuration states. It performs no +remote fan-out and never invents Worker 006 or device health. + Unknown or unavailable values remain empty/offline. The UI does not invent load, latency, activity or connected devices. diff --git a/docs/adr/0033-central-session-artifacts-and-bounded-host-cache.md b/docs/adr/0033-central-session-artifacts-and-bounded-host-cache.md index 2fb78b9..a8fc8af 100644 --- a/docs/adr/0033-central-session-artifacts-and-bounded-host-cache.md +++ b/docs/adr/0033-central-session-artifacts-and-bounded-host-cache.md @@ -84,6 +84,10 @@ The integrated-perception endpoint resolves results. A sealed central/local CAS hit therefore needs neither Worker 006 nor the original E10 result tree. Invalid central metadata fails closed; a genuinely absent central role can still use the existing local materialization path. +When the central store is unavailable, however, a missing local CAS role is not +treated as an absent role: base RRD export and overlay rendering both fail with +an explicit offline-cache error. This prevents one disconnected operator host +from silently producing a second derived truth. ## Persistent pins and offline use @@ -105,7 +109,15 @@ When the central root is unavailable: - a sealed local reference, manifest, and object resolve normally; - an absent local object returns an explicit offline-cache error; -- Mission Core does not silently fetch from Worker 006 or start a heavy AI run. +- Mission Core does not invoke the local RRD exporter, scan stale worker + results, render an overlay, fetch from Worker 006 or start a heavy AI run. + +`GET /api/health` exposes the effective shared/standalone mode, central-store +availability, bounded CAS and RRD-cache occupancy, media-tool availability and +the background catalog reconciler state without disclosing filesystem paths. +Operational readiness remains distinct from production shared-artifact +readiness so an offline, already pinned field host stays observable rather than +being mistaken for a fully connected host. ## RAVNOVES00 migration acceptance diff --git a/src/k1link/artifact_gateway.py b/src/k1link/artifact_gateway.py index e3ebec4..879f08c 100644 --- a/src/k1link/artifact_gateway.py +++ b/src/k1link/artifact_gateway.py @@ -16,7 +16,7 @@ from collections.abc import Iterator, Mapping, Sequence from contextlib import contextmanager, suppress from dataclasses import dataclass from pathlib import Path -from typing import Any, Final +from typing import Any, Final, Literal from uuid import uuid4 from k1link.artifacts import utc_now_iso @@ -98,6 +98,12 @@ class ArtifactCacheStatus: free_space_reserve_bytes: int +@dataclass(frozen=True, slots=True) +class ArtifactGatewayStatus: + central_status: Literal["ready", "unavailable", "invalid"] + cache: ArtifactCacheStatus + + class CentralArtifactStore: """Immutable SHA-256 objects/manifests plus atomically replaceable named refs.""" @@ -617,6 +623,22 @@ class ArtifactGateway: self.store = store self.cache = cache + def status(self) -> ArtifactGatewayStatus: + """Return a lightweight readiness snapshot without resolving an artifact.""" + + try: + self.store._require_root() + except ArtifactStoreUnavailable: + central_status: Literal["ready", "unavailable", "invalid"] = "unavailable" + except ArtifactIntegrityError: + central_status = "invalid" + else: + central_status = "ready" + return ArtifactGatewayStatus( + central_status=central_status, + cache=self.cache.status(), + ) + def publish( self, *, diff --git a/src/k1link/compute/integrated_perception.py b/src/k1link/compute/integrated_perception.py index dfe1bbe..5ec139b 100644 --- a/src/k1link/compute/integrated_perception.py +++ b/src/k1link/compute/integrated_perception.py @@ -560,8 +560,13 @@ class IntegratedPerceptionOverlayStore: role = f"integrated-overlay:{recording_id}" try: resolved = self.artifact_gateway.resolve_role("sessions", session_id, role) - except (ArtifactNotFound, ArtifactStoreUnavailable): + except ArtifactNotFound: return None + except ArtifactStoreUnavailable as exc: + raise RecordedPerceptionOverlayError( + "central integrated perception artifact is unavailable " + "and is not present in the local artifact cache" + ) from exc expected_result_id = resolved.manifest.metadata.get("integrated-result-id") if ( resolved.member.media_type != "application/vnd.rerun.rrd" diff --git a/src/k1link/sessions/recording.py b/src/k1link/sessions/recording.py index 81f01bd..67608d5 100644 --- a/src/k1link/sessions/recording.py +++ b/src/k1link/sessions/recording.py @@ -43,6 +43,7 @@ RERUN_RECORDING_MEDIA_TYPE = "application/vnd.rerun.rrd" RERUN_SESSION_TIMELINE = "session_time" SESSION_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") DEFAULT_FREE_SPACE_RESERVE_BYTES = 2 * 1024 * 1024 * 1024 +DEFAULT_CACHE_MAX_BYTES = 8 * 1024 * 1024 * 1024 RrdExporter = Callable[..., Mapping[str, object]] RecordingProgressCallback = Callable[[str, float], None] @@ -71,6 +72,15 @@ class MaterializedRecording: timeline_end_ns: int +@dataclass(frozen=True, slots=True) +class RecordingCacheStatus: + entry_count: int + total_bytes: int + cache_max_bytes: int + free_space_reserve_bytes: int + free_bytes: int + + @dataclass(frozen=True, slots=True) class _ValidatedMemoryEntry: source_identity: tuple[object, ...] @@ -165,10 +175,15 @@ class SessionRecordingMaterializer: self._exporters = dict(exporters or {}) if len(self._exporters) != len(set(self._exporters)): raise ValueError("recording exporter plugin ids must be unique") - self.cache_max_bytes = _optional_positive_configuration( + configured_cache_max_bytes = _optional_positive_configuration( cache_max_bytes, environment_name="MISSIONCORE_RRD_CACHE_MAX_BYTES", ) + self.cache_max_bytes = ( + configured_cache_max_bytes + if configured_cache_max_bytes is not None + else DEFAULT_CACHE_MAX_BYTES + ) self.free_space_reserve_bytes = _non_negative_configuration( free_space_reserve_bytes, environment_name="MISSIONCORE_RRD_FREE_SPACE_RESERVE_BYTES", @@ -199,6 +214,20 @@ class SessionRecordingMaterializer: _callable_accepts_keyword(exporter, "cancel_event") for exporter in exporters ) + def cache_status(self) -> RecordingCacheStatus: + """Return bounded cache capacity and occupancy for runtime readiness.""" + + with self._cache_guard: + total_bytes, entries = _cache_entries(self.recordings_root) + free_bytes = shutil.disk_usage(self.recordings_root).free + return RecordingCacheStatus( + entry_count=len(entries), + total_bytes=total_bytes, + cache_max_bytes=self.cache_max_bytes, + free_space_reserve_bytes=self.free_space_reserve_bytes, + free_bytes=free_bytes, + ) + def is_recording_available(self, recording: MaterializedRecording) -> bool: """Cheap no-follow check for a previously validated ready handle.""" @@ -409,8 +438,13 @@ class SessionRecordingMaterializer: session_id, "base-rrd", ) - except (ArtifactNotFound, ArtifactStoreUnavailable): + except ArtifactNotFound: return None + except ArtifactStoreUnavailable as exc: + raise RecordingMaterializationError( + "central recording is unavailable and is not present " + "in the local artifact cache" + ) from exc except ArtifactGatewayError as exc: raise RecordingMaterializationError( "central recording artifact failed validation" diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index bf588d7..4259707 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -63,6 +63,10 @@ from k1link.web.plugin_runtime import ( ) from k1link.web.polygon_api import build_polygon_router, configured_polygon_runs_root from k1link.web.runtime_diagnostics import configure_scanner_diagnostics +from k1link.web.runtime_readiness import ( + BackgroundReconcilerReadiness, + build_runtime_readiness, +) from k1link.web.session_api import build_session_router from k1link.web.system_telemetry_api import build_system_telemetry_router from k1link.web.viewer_diagnostics_api import build_viewer_diagnostics_router @@ -167,6 +171,7 @@ session_perception_epoch_store = ( else None ) map_gateway_proxy = MapGatewayProxy(MapGatewayConfiguration.from_environment()) +recording_reconciler_readiness = BackgroundReconcilerReadiness() def _prepare_recorded_media_for_launch( @@ -281,10 +286,11 @@ async def _recording_preparation_reconciler() -> None: newly_finalized, ) known_finalized = finalized - except Exception: + recording_reconciler_readiness.record_success() + except Exception as exc: # A transient filesystem/catalog failure must not permanently # disable preparation of sessions completed later in the run. - pass + recording_reconciler_readiness.record_failure(exc) await asyncio.sleep(2.0) @@ -339,18 +345,16 @@ async def request_validation_error_handler( @app.get("/api/health") def health() -> dict[str, Any]: - runtime_health = plugin_environment.runtime_health - runtimes_ready = all(item["status"] == "ready" for item in runtime_health) - return { - "ok": runtimes_ready, - "status": "ok" if runtimes_ready else "degraded", - "service": "mission-core-control-plane", - "version": __version__, - "plugin_runtimes": { - "ready": sum(item["status"] == "ready" for item in runtime_health), - "total": len(runtime_health), - }, - } + return build_runtime_readiness( + version=__version__, + plugin_runtime_health=plugin_environment.runtime_health, + recording_materializer=session_recording_materializer, + artifact_gateway=session_artifact_gateway, + map_gateway_configured=map_gateway_proxy.configured, + ffmpeg_available=_ffmpeg is not None, + ffprobe_available=_ffprobe is not None, + reconciler=recording_reconciler_readiness, + ) @app.get("/api/v1/device-plugins") diff --git a/src/k1link/web/runtime_readiness.py b/src/k1link/web/runtime_readiness.py new file mode 100644 index 0000000..6c67632 --- /dev/null +++ b/src/k1link/web/runtime_readiness.py @@ -0,0 +1,199 @@ +"""Bounded, non-secret runtime readiness for the Mission Core control plane.""" + +from __future__ import annotations + +import threading +from collections.abc import Iterable, Mapping +from datetime import UTC, datetime +from typing import Any + +from k1link.artifact_gateway import ArtifactGateway +from k1link.sessions.recording import SessionRecordingMaterializer + + +class BackgroundReconcilerReadiness: + """Track repeated catalog-reconciliation failures without exposing details.""" + + def __init__(self) -> None: + self._lock = threading.Lock() + self._completed_runs = 0 + self._consecutive_failures = 0 + self._last_success_at_utc: str | None = None + self._last_failure_at_utc: str | None = None + self._last_failure_type: str | None = None + + def record_success(self) -> None: + with self._lock: + self._completed_runs += 1 + self._consecutive_failures = 0 + self._last_success_at_utc = _utc_now_iso() + + def record_failure(self, error: BaseException) -> None: + with self._lock: + self._completed_runs += 1 + self._consecutive_failures += 1 + self._last_failure_at_utc = _utc_now_iso() + self._last_failure_type = type(error).__name__ + + def snapshot(self) -> dict[str, object]: + with self._lock: + if self._completed_runs == 0: + status = "starting" + elif self._consecutive_failures == 0: + status = "ready" + else: + status = "degraded" + return { + "status": status, + "completed_runs": self._completed_runs, + "consecutive_failures": self._consecutive_failures, + "last_success_at_utc": self._last_success_at_utc, + "last_failure_at_utc": self._last_failure_at_utc, + "last_failure_type": self._last_failure_type, + } + + +def build_runtime_readiness( + *, + version: str, + plugin_runtime_health: Iterable[Mapping[str, Any]], + recording_materializer: SessionRecordingMaterializer, + artifact_gateway: ArtifactGateway | None, + map_gateway_configured: bool, + ffmpeg_available: bool, + ffprobe_available: bool, + reconciler: BackgroundReconcilerReadiness, +) -> dict[str, Any]: + """Compose readiness from local components without network fan-out.""" + + runtime_health = tuple(plugin_runtime_health) + ready_runtime_count = sum( + item.get("status") == "ready" for item in runtime_health + ) + plugin_ready = ready_runtime_count == len(runtime_health) + + try: + recording = recording_materializer.cache_status() + recording_status = "ready" + recording_document: dict[str, object] = { + "status": recording_status, + "entries": recording.entry_count, + "bytes": recording.total_bytes, + "max_bytes": recording.cache_max_bytes, + "free_bytes": recording.free_bytes, + "free_space_reserve_bytes": recording.free_space_reserve_bytes, + } + recording_ready = ( + recording.total_bytes <= recording.cache_max_bytes + and recording.free_bytes >= recording.free_space_reserve_bytes + ) + if not recording_ready: + recording_document["status"] = "capacity-pressure" + except (OSError, RuntimeError): + recording_ready = False + recording_document = {"status": "unavailable"} + + if artifact_gateway is None: + artifact_document: dict[str, object] = { + "mode": "standalone", + "status": "not-configured", + "central_available": None, + } + shared_artifacts_ready = False + shared_artifact_degraded = False + else: + try: + artifact = artifact_gateway.status() + cache = artifact.cache + artifact_document = { + "mode": "shared", + "status": ( + "ready" + if artifact.central_status == "ready" + else ( + "offline-cache-only" + if artifact.central_status == "unavailable" + else "invalid" + ) + ), + "central_available": artifact.central_status == "ready", + "cache": { + "objects": cache.object_count, + "bytes": cache.total_bytes, + "pinned_objects": cache.pinned_object_count, + "pinned_bytes": cache.pinned_bytes, + "max_bytes": cache.cache_max_bytes, + "free_space_reserve_bytes": cache.free_space_reserve_bytes, + }, + } + shared_artifacts_ready = artifact.central_status == "ready" + shared_artifact_degraded = not shared_artifacts_ready + except (OSError, RuntimeError): + artifact_document = { + "mode": "shared", + "status": "unavailable", + "central_available": False, + } + shared_artifacts_ready = False + shared_artifact_degraded = True + + media_tools_ready = ffmpeg_available and ffprobe_available + reconciler_document = reconciler.snapshot() + failure_count = reconciler_document["consecutive_failures"] + reconciler_failures = ( + failure_count + if isinstance(failure_count, int) and not isinstance(failure_count, bool) + else 3 + ) + operational_ready = ( + plugin_ready + and recording_ready + and media_tools_ready + and reconciler_failures < 3 + ) + degraded = ( + shared_artifact_degraded + or reconciler_failures > 0 + or not operational_ready + ) + status = ( + "unavailable" + if not operational_ready + else "degraded" + if degraded + else "ok" + ) + return { + "ok": operational_ready, + "status": status, + "service": "mission-core-control-plane", + "version": version, + "readiness": { + "operational": operational_ready, + "production_shared_artifacts": ( + operational_ready and shared_artifacts_ready + ), + }, + "plugin_runtimes": { + "ready": ready_runtime_count, + "total": len(runtime_health), + }, + "components": { + "recording_cache": recording_document, + "artifact_store": artifact_document, + "background_reconciler": reconciler_document, + "media_tools": { + "status": "ready" if media_tools_ready else "unavailable", + "ffmpeg": ffmpeg_available, + "ffprobe": ffprobe_available, + }, + "map_gateway": { + "status": "configured" if map_gateway_configured else "optional", + "configured": map_gateway_configured, + }, + }, + } + + +def _utc_now_iso() -> str: + return datetime.now(UTC).isoformat().replace("+00:00", "Z") diff --git a/tests/test_e10_integrated_perception.py b/tests/test_e10_integrated_perception.py index 96d8280..61e03aa 100644 --- a/tests/test_e10_integrated_perception.py +++ b/tests/test_e10_integrated_perception.py @@ -19,12 +19,14 @@ from k1link.artifact_gateway import ( ArtifactGateway, ArtifactManifest, ArtifactMember, + ArtifactStoreUnavailable, ResolvedArtifact, ) from k1link.compute.integrated_perception import ( IntegratedPerceptionOverlayStore, _CuboidPresentationState, ) +from k1link.compute.results import RecordedPerceptionOverlayError def _write_result_descriptor( @@ -303,6 +305,54 @@ def test_integrated_overlay_uses_verified_central_cache_before_local_result_scan assert artifact.sha256 == member.sha256 +def test_integrated_overlay_does_not_render_when_central_store_is_offline( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + for name in ("jobs", "results", "lidar-packs"): + (tmp_path / name).mkdir() + + class OfflineGateway: + def resolve_role( + self, + namespace: str, + key: str, + role: str, + ) -> object: + assert (namespace, key, role) == ( + "sessions", + "session-1", + "integrated-overlay:recording-1", + ) + raise ArtifactStoreUnavailable("offline cache miss") + + store = IntegratedPerceptionOverlayStore( + jobs_root=tmp_path / "jobs", + results_root=tmp_path / "results", + lidar_packs_root=tmp_path / "lidar-packs", + cache_root=tmp_path / "cache", + ffmpeg_path=tmp_path / "ffmpeg", + artifact_gateway=OfflineGateway(), # type: ignore[arg-type] + ) + monkeypatch.setattr( + store, + "_latest", + lambda _session_id: (_ for _ in ()).throw( + AssertionError("offline CAS miss must not scan local results") + ), + ) + + with pytest.raises( + RecordedPerceptionOverlayError, + match="not present in the local artifact cache", + ): + store.materialize( + "session-1", + application_id="nodedc_mission_core_recorded", + recording_id="recording-1", + ) + + def test_e10_profile_pins_integrated_realtime_budget() -> None: _fusion, runner = _worker_modules() profile_path = ( diff --git a/tests/test_runtime_readiness.py b/tests/test_runtime_readiness.py new file mode 100644 index 0000000..a8c9b56 --- /dev/null +++ b/tests/test_runtime_readiness.py @@ -0,0 +1,112 @@ +from __future__ import annotations + +from pathlib import Path + +from k1link.artifact_gateway import ( + ArtifactGateway, + CentralArtifactStore, + LocalArtifactCache, +) +from k1link.sessions.recording import SessionRecordingMaterializer +from k1link.web.runtime_readiness import ( + BackgroundReconcilerReadiness, + build_runtime_readiness, +) + + +def _document( + tmp_path: Path, + *, + gateway: ArtifactGateway | None, + reconciler: BackgroundReconcilerReadiness, +) -> dict[str, object]: + return build_runtime_readiness( + version="test", + plugin_runtime_health=({"status": "ready"},), + recording_materializer=SessionRecordingMaterializer( + tmp_path / "data", + exporter=lambda _source, _destination: {}, + free_space_reserve_bytes=0, + ), + artifact_gateway=gateway, + map_gateway_configured=False, + ffmpeg_available=True, + ffprobe_available=True, + reconciler=reconciler, + ) + + +def test_runtime_readiness_reports_shared_store_and_bounded_cache( + tmp_path: Path, +) -> None: + central_root = tmp_path / "central" + gateway = ArtifactGateway( + CentralArtifactStore(central_root, create=True), + LocalArtifactCache( + tmp_path / "cache", + max_bytes=1024 * 1024, + free_space_reserve_bytes=0, + ), + ) + reconciler = BackgroundReconcilerReadiness() + reconciler.record_success() + + document = _document(tmp_path, gateway=gateway, reconciler=reconciler) + + assert document["ok"] is True + assert document["status"] == "ok" + assert document["readiness"] == { + "operational": True, + "production_shared_artifacts": True, + } + components = document["components"] + assert isinstance(components, dict) + assert components["recording_cache"]["max_bytes"] == 8 * 1024 * 1024 * 1024 + assert components["artifact_store"]["status"] == "ready" + assert components["map_gateway"]["status"] == "optional" + + +def test_runtime_readiness_exposes_offline_cache_only_mode( + tmp_path: Path, +) -> None: + central_root = tmp_path / "central" + gateway = ArtifactGateway( + CentralArtifactStore(central_root, create=True), + LocalArtifactCache( + tmp_path / "cache", + max_bytes=1024 * 1024, + free_space_reserve_bytes=0, + ), + ) + central_root.rename(tmp_path / "central-offline") + reconciler = BackgroundReconcilerReadiness() + reconciler.record_success() + + document = _document(tmp_path, gateway=gateway, reconciler=reconciler) + + assert document["ok"] is True + assert document["status"] == "degraded" + assert document["readiness"] == { + "operational": True, + "production_shared_artifacts": False, + } + components = document["components"] + assert isinstance(components, dict) + assert components["artifact_store"]["status"] == "offline-cache-only" + + +def test_runtime_readiness_fails_after_repeated_reconciler_errors( + tmp_path: Path, +) -> None: + reconciler = BackgroundReconcilerReadiness() + for _ in range(3): + reconciler.record_failure(OSError("private detail")) + + document = _document(tmp_path, gateway=None, reconciler=reconciler) + + assert document["ok"] is False + assert document["status"] == "unavailable" + components = document["components"] + assert isinstance(components, dict) + assert components["background_reconciler"]["last_failure_type"] == "OSError" + assert "private detail" not in str(document) diff --git a/tests/test_session_recording.py b/tests/test_session_recording.py index a77c736..b6c64ff 100644 --- a/tests/test_session_recording.py +++ b/tests/test_session_recording.py @@ -15,6 +15,7 @@ import pytest import k1link.sessions.recording as recording_module from k1link.artifact_gateway import ( ArtifactGateway, + ArtifactStoreUnavailable, CentralArtifactStore, LocalArtifactCache, ) @@ -184,7 +185,7 @@ def test_materializer_restores_exact_central_recording_without_export( assert exporter.calls == 0 -def test_default_recording_cache_has_no_application_byte_quota( +def test_default_recording_cache_has_a_bounded_application_quota( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -195,7 +196,42 @@ def test_default_recording_cache_has_no_application_byte_quota( exporter=FakeExporter(), ) - assert materializer.cache_max_bytes is None + assert materializer.cache_max_bytes == 8 * 1024 * 1024 * 1024 + + +def test_materializer_does_not_rebuild_when_central_store_is_offline( + tmp_path: Path, +) -> None: + command = _command(tmp_path / "source") + exporter = FakeExporter() + + class OfflineGateway: + def resolve_role( + self, + namespace: str, + key: str, + role: str, + ) -> object: + assert (namespace, key, role) == ( + "sessions", + command.session_id, + "base-rrd", + ) + raise ArtifactStoreUnavailable("offline cache miss") + + materializer = SessionRecordingMaterializer( + tmp_path / "private", + exporter=exporter, + artifact_gateway=OfflineGateway(), # type: ignore[arg-type] + ) + + with pytest.raises( + RecordingMaterializationError, + match="not present in the local artifact cache", + ): + materializer.materialize(command) + + assert exporter.calls == 0 def test_delete_cached_refuses_a_live_lease_then_removes_the_exact_cache(