diff --git a/apps/control-station/src/App.tsx b/apps/control-station/src/App.tsx index 7b94d5e..abdd0f7 100644 --- a/apps/control-station/src/App.tsx +++ b/apps/control-station/src/App.tsx @@ -199,6 +199,7 @@ export default function App() { const livePerceptionLayersRef = useRef(defaultLivePerceptionLayers); const livePerceptionRevisionRef = useRef(0); const replayActiveRef = useRef(false); + const selectedRecordedSessionIdRef = useRef(null); const sceneSettingsCommitterActiveRef = useRef(true); const sceneSettingsCommitterRef = useRef | null>(null); const polygonDatasetRouteOpenedRef = useRef(false); @@ -536,6 +537,7 @@ export default function App() { // produced a validated launch descriptor. Keep the old scene mounted // before this point; now perform one controlled receiver teardown before // accepting the already-ready archive. + selectedRecordedSessionIdRef.current = null; setReplayTransitioning(true); setRecordedReplay(null); setRecordedReplayLabel(null); @@ -553,6 +555,7 @@ export default function App() { if (sourceSwitchBlockedRef.current) { throw new Error(SPATIAL_SOURCE_SWITCH_BLOCKED_REASON); } + selectedRecordedSessionIdRef.current = session.id; setRecordedReplay(launch); setRecordedReplayLabel(session.label); setSourceUrl(launch.sourceUrl); @@ -564,11 +567,28 @@ export default function App() { if (outcome !== "accepted") setReplayTransitioning(false); }, []); + const releaseRecordedReplayForDelete = useCallback(async (sessionId: string) => { + if (selectedRecordedSessionIdRef.current !== sessionId) return; + // Deleting the selected archive is a source transition. Unmount Rerun and + // recorded media first so no admission request can outlive the evidence. + selectedRecordedSessionIdRef.current = null; + setReplayTransitioning(true); + setRecordedReplay(null); + setRecordedReplayLabel(null); + setSourceUrl(""); + setSourceDraft(""); + await new Promise((resolve) => { + window.requestAnimationFrame(() => window.setTimeout(resolve, 0)); + }); + if (selectedRecordedSessionIdRef.current === null) setReplayTransitioning(false); + }, []); + const activateAutomaticSpatialSource = useCallback(() => { // A plugin-owned live/file-replay start must release any host-selected // archive or manual URL before the new acquisition becomes non-terminal. // This transition is an internal start boundary, not an operator source // switch, so it intentionally does not consult the acquisition guard. + selectedRecordedSessionIdRef.current = null; setReplayTransitioning(false); setRecordedReplay(null); setRecordedReplayLabel(null); @@ -786,6 +806,7 @@ export default function App() { onReplayAccepted={acceptRecordedReplay} onReplaySettled={(_session, outcome) => settleRecordedReplaySwitch(outcome)} + onDeleteBegin={releaseRecordedReplayForDelete} /> ) : activeDefinition.kind === "datasets" ? ( Offline evaluation @@ -838,6 +859,7 @@ export default function App() { onReplayAccepted: acceptRecordedReplay, onReplaySettled: (_session, outcome) => settleRecordedReplaySwitch(outcome), + onDeleteBegin: releaseRecordedReplayForDelete, }} navigation={{ openView, @@ -886,6 +908,7 @@ export default function App() { title={sourceSwitchBlockedReason ?? undefined} onClick={() => { if (sourceSwitchBlockedRef.current) return; + selectedRecordedSessionIdRef.current = null; setSourceDraft(""); setSourceUrl(""); setRecordedReplay(null); @@ -901,6 +924,7 @@ export default function App() { title={sourceSwitchBlockedReason ?? undefined} onClick={() => { if (sourceSwitchBlockedRef.current) return; + selectedRecordedSessionIdRef.current = null; setSourceUrl(sourceDraft.trim()); setRecordedReplay(null); setRecordedReplayLabel(null); diff --git a/apps/control-station/src/components/ObservationSessionSelect.tsx b/apps/control-station/src/components/ObservationSessionSelect.tsx index f030c71..d37f1a6 100644 --- a/apps/control-station/src/components/ObservationSessionSelect.tsx +++ b/apps/control-station/src/components/ObservationSessionSelect.tsx @@ -105,6 +105,7 @@ export interface ObservationSessionReplayCallbacks { session: ObservationSessionSummary, outcome: ObservationReplayOutcome, ) => void | Promise; + onDeleteBegin?: (sessionId: string) => void | Promise; } export function ObservationSessionSelect({ @@ -114,6 +115,7 @@ export function ObservationSessionSelect({ onReplayBegin, onReplayAccepted, onReplaySettled, + onDeleteBegin, }: ObservationSessionReplayCallbacks & { limit?: number; disabled?: boolean; @@ -127,6 +129,7 @@ export function ObservationSessionSelect({ onReplayBegin, onReplayAccepted, onReplaySettled, + onDeleteBegin, }); const triggerCopy = sessions.replayProgress ? progressCopy(sessions.replayProgress.phase) @@ -313,6 +316,7 @@ export function ObservationSessionArchive({ onReplayBegin, onReplayAccepted, onReplaySettled, + onDeleteBegin, }: ObservationSessionReplayCallbacks & { limit?: number; labsOnly?: boolean; @@ -327,6 +331,7 @@ export function ObservationSessionArchive({ onReplayBegin, onReplayAccepted, onReplaySettled, + onDeleteBegin, }); const items = sessions.items; diff --git a/apps/control-station/src/core/observation/useObservationSessions.ts b/apps/control-station/src/core/observation/useObservationSessions.ts index 3b9a532..729790d 100644 --- a/apps/control-station/src/core/observation/useObservationSessions.ts +++ b/apps/control-station/src/core/observation/useObservationSessions.ts @@ -52,6 +52,20 @@ export interface ObservationReplayCoordinator { cancel: () => void; } +export async function deleteObservationSessionAfterTeardown( + sessionId: string, + { + onDeleteBegin, + deleteSession = deleteObservationSession, + }: { + onDeleteBegin?: (sessionId: string) => void | Promise; + deleteSession?: (sessionId: string) => Promise; + } = {}, +): Promise { + await onDeleteBegin?.(sessionId); + await deleteSession(sessionId); +} + export interface ObservationPreparationPollingOptions { signal: AbortSignal; fetcher?: ObservationSessionFetch; @@ -375,6 +389,7 @@ export function useObservationSessions({ onReplayBegin, onReplayAccepted, onReplaySettled, + onDeleteBegin, }: { limit?: number; scope?: ObservationSessionScope; @@ -392,6 +407,8 @@ export function useObservationSessions({ session: ObservationSessionSummary, outcome: ObservationReplayOutcome, ) => void | Promise; + /** Releases an active viewer before its evidence and derived cache are deleted. */ + onDeleteBegin?: (sessionId: string) => void | Promise; } = {}): ObservationSessionsController { const [items, setItems] = useState([]); const [state, setState] = useState("idle"); @@ -606,7 +623,7 @@ export function useObservationSessions({ setDeletingSessionId(sessionId); setError(null); try { - await deleteObservationSession(sessionId); + await deleteObservationSessionAfterTeardown(sessionId, { onDeleteBegin }); if (!mounted.current) return false; setItems((current) => current.filter((item) => item.id !== sessionId)); setFailedSessionId((current) => current === sessionId ? null : current); @@ -617,7 +634,7 @@ export function useObservationSessions({ } finally { if (mounted.current) setDeletingSessionId(null); } - }, [deletingSessionId, replayingSessionId]); + }, [deletingSessionId, onDeleteBegin, replayingSessionId]); return { items, diff --git a/apps/control-station/src/workspaces/laboratory/LaboratoryArchiveWorkspace.tsx b/apps/control-station/src/workspaces/laboratory/LaboratoryArchiveWorkspace.tsx index f27d9c3..df78554 100644 --- a/apps/control-station/src/workspaces/laboratory/LaboratoryArchiveWorkspace.tsx +++ b/apps/control-station/src/workspaces/laboratory/LaboratoryArchiveWorkspace.tsx @@ -583,6 +583,7 @@ export function LaboratoryArchiveWorkspace(props: LaboratoryWorkspaceProps) { onReplayBegin: props.sessionArchive.onReplayBegin, onReplayAccepted: props.sessionArchive.onReplayAccepted, onReplaySettled: props.sessionArchive.onReplaySettled, + onDeleteBegin: props.sessionArchive.onDeleteBegin, }); const publishedWorks = useMemo( () => sessions.items.filter((session) => ( diff --git a/apps/control-station/test/observationSessions.test.mjs b/apps/control-station/test/observationSessions.test.mjs index caa4239..df6ff83 100644 --- a/apps/control-station/test/observationSessions.test.mjs +++ b/apps/control-station/test/observationSessions.test.mjs @@ -17,6 +17,7 @@ let ObservationSessionApiError; let ObservationSessionContractError; let decodeObservationRecordedMediaManifest; let createObservationReplayCoordinator; +let deleteObservationSessionAfterTeardown; let resolveObservationSessionReplay; let waitForObservationReplayPreparation; let storeObservationReplayPreparation; @@ -45,6 +46,7 @@ before(async () => { } = await server.ssrLoadModule("/src/core/observation/sessionArchive.ts")); ({ createObservationReplayCoordinator, + deleteObservationSessionAfterTeardown, resolveObservationSessionReplay, waitForObservationReplayPreparation, storeObservationReplayPreparation, @@ -437,6 +439,56 @@ test("delete API uses one opaque same-origin target and requires an empty 204", ); }); +test("session deletion releases the selected presentation before the DELETE request", async () => { + const events = []; + await deleteObservationSessionAfterTeardown("session-20260716T205632Z", { + onDeleteBegin: async (sessionId) => { + events.push(`release:${sessionId}`); + }, + deleteSession: async (sessionId) => { + events.push(`delete:${sessionId}`); + }, + }); + assert.deepEqual(events, [ + "release:session-20260716T205632Z", + "delete:session-20260716T205632Z", + ]); + + let deleteCalled = false; + await assert.rejects( + deleteObservationSessionAfterTeardown("session-20260716T205632Z", { + onDeleteBegin: async () => { + throw new Error("viewer teardown failed"); + }, + deleteSession: async () => { + deleteCalled = true; + }, + }), + /viewer teardown failed/, + ); + assert.equal(deleteCalled, false); +}); + +test("App atomically clears the selected Rerun source before deleting its session", async () => { + const appSource = await readFile(new URL("../src/App.tsx", import.meta.url), "utf8"); + assert.match( + appSource, + /selectedRecordedSessionIdRef\.current = session\.id[\s\S]*setRecordedReplay\(launch\)/, + ); + assert.match( + appSource, + /releaseRecordedReplayForDelete[\s\S]*selectedRecordedSessionIdRef\.current !== sessionId[\s\S]*selectedRecordedSessionIdRef\.current = null[\s\S]*setRecordedReplay\(null\)[\s\S]*setSourceUrl\(""\)/, + ); + assert.match( + appSource, + /onDeleteBegin=\{releaseRecordedReplayForDelete\}/, + ); + assert.match( + appSource, + /onDeleteBegin:\s*releaseRecordedReplayForDelete/, + ); +}); + test("replay API accepts only a same-origin seekable recording descriptor", async () => { const calls = []; const launch = await replayObservationSession("session-20260716T205632Z", { diff --git a/src/k1link/sessions/preparation.py b/src/k1link/sessions/preparation.py index 0ae0dce..b3cf68a 100644 --- a/src/k1link/sessions/preparation.py +++ b/src/k1link/sessions/preparation.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging import os import queue import threading @@ -20,6 +21,8 @@ from .recording import ( SessionRecordingMaterializer, ) +LOGGER = logging.getLogger(__name__) + PreparationState = Literal[ "queued", "validating", @@ -582,6 +585,12 @@ class SessionRecordingPreparationManager: except Exception: # The public status intentionally does not expose paths, # broker payloads or exporter internals. + LOGGER.exception( + "recording preparation materialization failed: " + "session_id=%s preparation_id=%s", + job.command.session_id, + job.preparation_id, + ) with self._guard: if job.cancel_event.is_set(): self._transition_locked(job, "cancelled", job.progress) @@ -616,6 +625,12 @@ class SessionRecordingPreparationManager: recording_end_seconds=(recording.timeline_end_ns / 1_000_000_000), ) except Exception: + LOGGER.exception( + "recording preparation finalization failed: " + "session_id=%s preparation_id=%s", + job.command.session_id, + job.preparation_id, + ) with self._guard: if job.cancel_event.is_set(): self._transition_locked(job, "cancelled", job.progress) diff --git a/src/k1link/web/session_api.py b/src/k1link/web/session_api.py index 1650ecf..e5b765b 100644 --- a/src/k1link/web/session_api.py +++ b/src/k1link/web/session_api.py @@ -372,8 +372,10 @@ def build_session_router( try: store.get_session(session_id) recorded_artifacts = store.list_recorded_media(session_id) - except SessionNotFoundError as exc: - raise HTTPException(status_code=404, detail=str(exc)) from exc + except SessionNotFoundError: + # DELETE is idempotent. Multiple mounted catalogs or another + # operator tab may have already removed the exact session. + return Response(status_code=204) except ValueError as exc: raise HTTPException( status_code=422, @@ -427,8 +429,10 @@ def build_session_router( (artifact.artifact_id for artifact in recorded_artifacts), ) await run_in_threadpool(store.delete_session, session_id) - except SessionNotFoundError as exc: - raise HTTPException(status_code=404, detail=str(exc)) from exc + except SessionNotFoundError: + # A concurrent idempotent delete won the race after both callers + # resolved the catalog row. + return Response(status_code=204) except (OSError, SessionIntegrityError) as exc: raise HTTPException( status_code=409, @@ -469,7 +473,12 @@ def build_session_router( recording_preparation_manager.restore_published, command, ) - if snapshot is None: + if snapshot is None or snapshot.state in {"failed", "cancelled"}: + # A terminal preparation describes one attempt, not the + # durable session. A fresh POST /replay is explicit retry + # intent and must replace that attempt. This also recovers + # the camera-finalization race where the RRD is already + # published but the media archive seals moments later. snapshot = recording_preparation_manager.enqueue( command, retry_failed=True, diff --git a/tests/test_session_api.py b/tests/test_session_api.py index 12ea5a8..ae3ab14 100644 --- a/tests/test_session_api.py +++ b/tests/test_session_api.py @@ -22,6 +22,7 @@ from k1link.device_plugins.xgrids_k1.mqtt.capture import FRAME_HEADER, RAW_MAGIC from k1link.sessions import ( MaterializedRecording, RecordedMediaInspector, + RecordedMediaManifest, ReplayCommand, SessionIntegrityError, SessionNotFoundError, @@ -362,8 +363,10 @@ def test_delete_session_removes_evidence_and_cache_but_refuses_an_open_recording release() response = asyncio.run(delete_route(session_id=session.name)) + repeated_response = asyncio.run(delete_route(session_id=session.name)) assert response.status_code == 204 + assert repeated_response.status_code == 204 assert not session.exists() assert not recording.path.parent.exists() with pytest.raises(SessionNotFoundError): @@ -763,6 +766,90 @@ def test_cold_replay_returns_quick_202_then_status_returns_ready_launch( manager.close() +def test_replay_post_retries_terminal_camera_finalization_failure( + tmp_path: Path, +) -> None: + repository = tmp_path / "repo" + sessions = repository / "sessions" + session = make_legacy_session(sessions, "20260716T205632Z_viewer_live") + store = SessionStore(repository, data_dir=tmp_path / "data") + store.reconcile_archive(xgrids_k1_archive_source(sessions)) + payload = b"recording-ready-before-camera-archive" + export_calls = 0 + finalization_calls = 0 + + def export_recording(source: Path, destination: Path) -> dict[str, object]: + nonlocal export_calls + export_calls += 1 + destination.write_bytes(payload) + return { + "source_sha256": hashlib.sha256(source.read_bytes()).hexdigest(), + "rrd_sha256": hashlib.sha256(payload).hexdigest(), + "rrd_bytes": len(payload), + "timeline": "session_time", + "timeline_start_ns": 0, + "timeline_end_ns": 2_500_000_000, + } + + def prepare_recorded_media( + _command: ReplayCommand, + _recording: MaterializedRecording, + ) -> tuple[RecordedMediaManifest, ...]: + nonlocal finalization_calls + finalization_calls += 1 + if finalization_calls == 1: + raise SessionIntegrityError("camera archive is still sealing") + return () + + materializer = SessionRecordingMaterializer( + store.data_dir, + exporter=export_recording, + ) + manager = SessionRecordingPreparationManager( + materializer, + ready_preparer=prepare_recorded_media, + ) + router = build_session_router( + store, + recording_materializer=materializer, + recording_preparation_manager=manager, + ) + replay_route = endpoint( + router, + "/api/v1/observation-sessions/{session_id}/replay", + "POST", + ) + try: + first = asyncio.run(replay_route(session_id=session.name, request=None)) + assert first.status_code == 202 + deadline = time.monotonic() + 2 + failed = manager.status(session.name) + while failed is None or failed.state != "failed": + assert time.monotonic() < deadline + time.sleep(0.005) + failed = manager.status(session.name) + failed_preparation_id = failed.preparation_id + + retry = asyncio.run(replay_route(session_id=session.name, request=None)) + assert retry.status_code == 202 + retry_document = json.loads(retry.body) + assert retry_document["preparation"]["preparation_id"] != failed_preparation_id + + deadline = time.monotonic() + 2 + ready = manager.status(session.name) + while ready is None or ready.state != "ready": + assert time.monotonic() < deadline + time.sleep(0.005) + ready = manager.status(session.name) + + launch = asyncio.run(replay_route(session_id=session.name, request=None)) + assert launch["launch"]["session_id"] == session.name + assert export_calls == 1 + assert finalization_calls == 2 + finally: + manager.close() + + def test_catalog_read_never_prepares_a_cold_historical_session(tmp_path: Path) -> None: repository = tmp_path / "repo" sessions = repository / "sessions"