From 85035fa07b20135cf4f5a439ccc3613a43e04cc9 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Sat, 22 Aug 2026 13:09:24 +0300 Subject: [PATCH] fix(k1): harden live handoff and camera recovery --- apps/control-station/src/App.tsx | 14 +- .../src/components/MseFmp4WebSocketPlayer.tsx | 132 ++++++++- .../src/components/RerunViewport.tsx | 12 +- .../core/observation/liveViewerDiagnostics.ts | 93 ++++-- .../core/observation/viewerSettingsTarget.ts | 19 ++ .../test/devicePluginContracts.test.mjs | 52 ++++ .../test/k1SupervisorPresentation.test.mjs | 23 ++ .../test/liveReceiverWatchdog.test.mjs | 15 + .../test/liveViewerDiagnostics.test.mjs | 97 ++++++- .../test/observationSources.test.mjs | 21 ++ .../rerunViewportAtomicAdmission.test.mjs | 8 + .../test/viewerSettingsTarget.test.mjs | 88 ++++++ .../frontend/src/useXgridsK1Runtime.ts | 110 +++++++- .../active_acquisition_recovery_checkpoint.py | 138 +++++++-- src/k1link/device_plugins/xgrids_k1/camera.py | 55 +++- src/k1link/device_plugins/xgrids_k1/facade.py | 156 +++++++--- ...etwork_provisioning_idempotency_journal.py | 4 +- src/k1link/web/runtime_diagnostics.py | 12 + src/k1link/web/viewer_diagnostics_api.py | 57 +++- tests/test_rerun_bridge.py | 9 +- tests/test_viewer_diagnostics_api.py | 78 ++++- tests/test_xgrids_acquisition_lifecycle.py | 267 ++++++++++++++++++ ...tive_acquisition_checkpoint_integration.py | 53 ++-- ..._active_acquisition_recovery_checkpoint.py | 68 ++++- tests/test_xgrids_camera_gateway.py | 10 + .../test_xgrids_connection_scenario_reset.py | 57 ++++ 26 files changed, 1478 insertions(+), 170 deletions(-) create mode 100644 apps/control-station/src/core/observation/viewerSettingsTarget.ts create mode 100644 apps/control-station/test/viewerSettingsTarget.test.mjs diff --git a/apps/control-station/src/App.tsx b/apps/control-station/src/App.tsx index 1fe2766..b4ff40e 100644 --- a/apps/control-station/src/App.tsx +++ b/apps/control-station/src/App.tsx @@ -50,6 +50,7 @@ import type { } from "./core/observation/sessionArchive"; import { useRecordedSessionAdmission } from "./core/observation/useRecordedSessionAdmission"; import { useWorkspaceLayoutProfile } from "./core/observation/useWorkspaceLayoutProfile"; +import { viewerSettingsTargetIdentity } from "./core/observation/viewerSettingsTarget"; import { resolvePolygonRunRoute } from "./core/polygon/runArchive"; import { OBSERVATION_WORKSPACE_ID, @@ -302,14 +303,7 @@ export default function App() { }, }; }, [recordedReplay, replayPresented, replaySources, runtime.state]); - const viewerSettingsTargetIdentity = [ - runtime.state?.activeDevice?.pluginId, - runtime.state?.activeDevice?.modelId, - runtime.state?.activeDevice?.instanceId, - runtime.state?.deviceSession?.sessionId, - runtime.state?.deviceSession?.deviceId, - runtime.state?.acquisition?.acquisitionId, - ].filter(Boolean).join(":") || "local-runtime"; + const viewerSettingsTarget = viewerSettingsTargetIdentity(runtime.state); const observationLayout = useObservationLayout( activeObservationSources, replayPresented ? undefined : runtime.setObservationSourceActive, @@ -389,7 +383,7 @@ export default function App() { useEffect(() => { const profile = workspaceLayoutProfile.profile; const applicationKey = profile - ? `${profile.revision}:${viewerSettingsTargetIdentity}` + ? `${profile.revision}:${viewerSettingsTarget}` : null; if ( !profile || @@ -410,7 +404,7 @@ export default function App() { }, [ runtime.backendStatus, runtime.updateViewerSettings, - viewerSettingsTargetIdentity, + viewerSettingsTarget, workspaceLayoutProfile.profile, ]); diff --git a/apps/control-station/src/components/MseFmp4WebSocketPlayer.tsx b/apps/control-station/src/components/MseFmp4WebSocketPlayer.tsx index d75cfd2..b9865b0 100644 --- a/apps/control-station/src/components/MseFmp4WebSocketPlayer.tsx +++ b/apps/control-station/src/components/MseFmp4WebSocketPlayer.tsx @@ -9,7 +9,14 @@ import { reduceCameraPlaybackRecovery, type CameraPlaybackRecoveryEvent, } from "../core/observation/liveCameraRecovery"; -import { subscribeToLiveViewerBuildFence } from "../core/observation/liveViewerDiagnostics"; +import { + createLiveViewerDiagnosticLifecycle, + createLiveViewerInstanceId, + createLiveViewerLineage, + subscribeToLiveViewerBuildFence, + type LiveViewerDiagnosticLifecycle, + type LiveViewerFailureStage, +} from "../core/observation/liveViewerDiagnostics"; type PlayerStatus = "connecting" | "buffering" | "playing" | "error"; @@ -168,6 +175,15 @@ export function cameraTransportCloseRecoveryMessage(code: number): string { return "Browser-preview прерван; восстанавливаем текущую камеру."; } +export function cameraPlaybackRecoveryFailureStage( + event: CameraPlaybackRecoveryEvent, +): LiveViewerFailureStage { + if (event.type === "heartbeat") return "camera-heartbeat-reopen"; + if (event.type === "network-online") return "camera-network-online-reopen"; + if (event.type === "page-restore") return "camera-page-restore-reopen"; + return "camera-visibility-reopen"; +} + function websocketUrl(path: string): string { const url = new URL(path, window.location.href); if (url.protocol === "http:") url.protocol = "ws:"; @@ -195,6 +211,9 @@ export function MseFmp4WebSocketPlayer({ const transportEpochRef = useRef(0); const transportAuthorityRef = useRef(recoveryAuthorityIdentity); const activeTransportDisposeRef = useRef<(() => void) | null>(null); + const activeDiagnosticRef = useRef(null); + const diagnosticInstanceIdRef = useRef(null); + const diagnosticGenerationRef = useRef(0); const uiBuildStaleRef = useRef(false); const [attempt, setAttempt] = useState(0); const [uiBuildStale, setUiBuildStale] = useState(false); @@ -232,6 +251,16 @@ export function MseFmp4WebSocketPlayer({ let disposed = false; const transportEpoch = transportEpochRef.current + 1; transportEpochRef.current = transportEpoch; + diagnosticInstanceIdRef.current ??= createLiveViewerInstanceId(); + diagnosticGenerationRef.current += 1; + const diagnosticLifecycle = createLiveViewerDiagnosticLifecycle({ + lineage: createLiveViewerLineage( + diagnosticInstanceIdRef.current, + diagnosticGenerationRef.current, + ), + }); + activeDiagnosticRef.current = diagnosticLifecycle; + diagnosticLifecycle.verifyBuild(); const transportIsCurrent = () => cameraTransportCallbackIsCurrent( transportEpochRef.current, transportEpoch, @@ -245,6 +274,7 @@ export function MseFmp4WebSocketPlayer({ let retryTimer: number | undefined; let receivedMedia = false; let failed = false; + let playingReported = false; let startupWatchdog: CameraStartupWatchdog | null = null; transportHealthyRef.current = false; const recovering = Boolean( @@ -271,7 +301,16 @@ export function MseFmp4WebSocketPlayer({ } }; - const retryTransport = (copy: string) => { + const retryTransport = ( + copy: string, + failureStage: LiveViewerFailureStage, + websocketCloseCode?: number, + appendFailure?: { + cameraAppendErrorName: string; + cameraMediaSourceState: "closed" | "open" | "ended"; + cameraVideoErrorCode: number | null; + }, + ) => { if (!transportIsCurrent() || failed) return; if (!cameraTransportRecoveryIsCurrent( activeAuthorityRef.current, @@ -286,10 +325,21 @@ export function MseFmp4WebSocketPlayer({ startupWatchdog?.clear(); failed = true; transportHealthyRef.current = false; - queue.length = 0; - queuedBytes = 0; const retry = consumeCameraLeaseRetry(leaseRetryRef.current, delivery.id); leaseRetryRef.current = retry.budget; + diagnosticLifecycle.post({ + eventCode: "live_camera_transport_restart_requested", + failureStage, + streamId: delivery.id, + cameraQueueBytes: queuedBytes, + cameraQueueSegments: queue.length, + cameraRetryCount: retry.budget.count, + websocketCloseCode, + transportEpoch, + ...appendFailure, + }); + queue.length = 0; + queuedBytes = 0; recoveryPendingAuthorityRef.current = recoveryAuthorityIdentity; setStatus("connecting"); setMessage(copy); @@ -318,7 +368,12 @@ export function MseFmp4WebSocketPlayer({ schedule: (callback, timeoutMs) => window.setTimeout(callback, timeoutMs), cancel: (handle) => window.clearTimeout(handle), onTimeout: (stage) => { - retryTransport(cameraStartupWatchdogRecoveryMessage(stage)); + retryTransport( + cameraStartupWatchdogRecoveryMessage(stage), + stage === "first-media" + ? "camera-first-media-timeout" + : "camera-first-playable-timeout", + ); }, }); } @@ -331,6 +386,16 @@ export function MseFmp4WebSocketPlayer({ recoveryPendingAuthorityRef.current = null; setStatus("playing"); setMessage(""); + if (!playingReported) { + playingReported = true; + diagnosticLifecycle.post({ + eventCode: "live_camera_transport_playing", + streamId: delivery.id, + cameraQueueBytes: queuedBytes, + cameraQueueSegments: queue.length, + transportEpoch, + }); + } }; video.addEventListener("playing", onPlaying); @@ -349,9 +414,24 @@ export function MseFmp4WebSocketPlayer({ try { sourceBuffer.appendBuffer(chunk); } catch (error) { - retryTransport(error instanceof DOMException && error.name === "QuotaExceededError" - ? "Live-буфер переполнен; восстанавливаем канал без накопленной задержки." - : "Не удалось добавить видеосегмент; восстанавливаем browser-preview."); + const quotaExceeded = error instanceof DOMException + && error.name === "QuotaExceededError"; + retryTransport( + quotaExceeded + ? "Live-буфер переполнен; восстанавливаем канал без накопленной задержки." + : "Не удалось добавить видеосегмент; восстанавливаем browser-preview.", + quotaExceeded ? "camera-append-quota" : "camera-append-error", + undefined, + { + cameraAppendErrorName: error instanceof DOMException + ? error.name + : error instanceof Error + ? error.name + : "UnknownError", + cameraMediaSourceState: mediaSource.readyState, + cameraVideoErrorCode: video.error?.code ?? null, + }, + ); } }; @@ -361,7 +441,10 @@ export function MseFmp4WebSocketPlayer({ // Never drop arbitrary fMP4 fragments: the following samples may depend // on them. A bounded reset is safer and keeps live latency deterministic. if (!cameraPendingQueueCanAccept(queuedBytes, queue.length, chunk.byteLength)) { - retryTransport("Видеодекодер отстал от эфира; восстанавливаем live-буфер."); + retryTransport( + "Видеодекодер отстал от эфира; восстанавливаем live-буфер.", + "camera-queue-capacity", + ); return; } queue.push(chunk); @@ -427,7 +510,10 @@ export function MseFmp4WebSocketPlayer({ appendNext(); }; onBufferError = () => { - retryTransport("MSE сбросил видеосегмент; восстанавливаем decoder."); + retryTransport( + "MSE сбросил видеосегмент; восстанавливаем decoder.", + "camera-source-buffer-error", + ); }; sourceBuffer.addEventListener("updateend", onBufferUpdateEnd); sourceBuffer.addEventListener("error", onBufferError); @@ -453,16 +539,26 @@ export function MseFmp4WebSocketPlayer({ enqueue(event.data); } else if (event.data instanceof Blob) { void event.data.arrayBuffer().then(enqueue).catch(() => { - retryTransport("Получен повреждённый видеосегмент; восстанавливаем канал."); + retryTransport( + "Получен повреждённый видеосегмент; восстанавливаем канал.", + "camera-fragment-decode", + ); }); } }); socket.addEventListener("error", () => { - retryTransport("Связь с локальным video adapter потеряна; переподключаемся."); + retryTransport( + "Связь с локальным video adapter потеряна; переподключаемся.", + "camera-websocket-error", + ); }); socket.addEventListener("close", (event) => { if (!transportIsCurrent() || failed) return; - retryTransport(cameraTransportCloseRecoveryMessage(event.code)); + retryTransport( + cameraTransportCloseRecoveryMessage(event.code), + "camera-websocket-close", + event.code, + ); }); }; @@ -504,6 +600,10 @@ export function MseFmp4WebSocketPlayer({ if (activeTransportDisposeRef.current === disposeTransport) { activeTransportDisposeRef.current = null; } + if (activeDiagnosticRef.current === diagnosticLifecycle) { + activeDiagnosticRef.current = null; + } + diagnosticLifecycle.dispose(); disposeTransport(); }; }, [attempt, recoveryAuthorityIdentity, transportIdentity, uiBuildStale]); @@ -530,6 +630,12 @@ export function MseFmp4WebSocketPlayer({ }); recovery = decision.state; if (!decision.reopen) return; + activeDiagnosticRef.current?.post({ + eventCode: "live_camera_transport_restart_requested", + failureStage: cameraPlaybackRecoveryFailureStage(event), + streamId: delivery.id, + transportEpoch: transportEpochRef.current, + }); leaseRetryRef.current = resetCameraLeaseRetryBudget(delivery.id); recoveryPendingAuthorityRef.current = recoveryAuthorityIdentity; setStatus("connecting"); diff --git a/apps/control-station/src/components/RerunViewport.tsx b/apps/control-station/src/components/RerunViewport.tsx index ffbe951..4c2a898 100644 --- a/apps/control-station/src/components/RerunViewport.tsx +++ b/apps/control-station/src/components/RerunViewport.tsx @@ -1063,9 +1063,10 @@ export function RerunViewport({ let liveRecoveryRetryTimer: number | undefined; let recordingOpened = false; let recordingOpenTimedOut = false; - let liveOpenWatchdog = initialLiveReceiverOpenWatchdogState( - liveActivitySequenceRef.current, - ); + // A new browser receiver has not observed any backend publication yet. + // Starting from the already-consumed runtime sequence makes the first + // four-second check look stalled and destroys a healthy partial replay. + let liveOpenWatchdog = initialLiveReceiverOpenWatchdogState(null); let viewerStartResolved = false; let latestLiveRangeMaxNs: number | null = null; let liveWatchdog = initialLiveReceiverWatchdogState(); @@ -1718,10 +1719,7 @@ export function RerunViewport({ if (!isRecordedSource) { // Measure native store admission from the resolved viewer start, // not from dynamic module import or React effect setup time. - liveOpenWatchdog = initialLiveReceiverOpenWatchdogState( - liveActivitySequenceRef.current, - Date.now(), - ); + liveOpenWatchdog = initialLiveReceiverOpenWatchdogState(null, Date.now()); discoverActiveLiveRecording(); if (!recordingOpened) { diagnosticLifecycle.armAdmissionInterval( diff --git a/apps/control-station/src/core/observation/liveViewerDiagnostics.ts b/apps/control-station/src/core/observation/liveViewerDiagnostics.ts index df50b7a..b88062e 100644 --- a/apps/control-station/src/core/observation/liveViewerDiagnostics.ts +++ b/apps/control-station/src/core/observation/liveViewerDiagnostics.ts @@ -4,13 +4,28 @@ export type LiveViewerDiagnosticEventCode = | "live_receiver_recovered" | "live_receiver_recovery_exhausted" | "live_receiver_active_store_admitted" - | "live_receiver_error"; + | "live_receiver_error" + | "live_camera_transport_restart_requested" + | "live_camera_transport_playing"; export type LiveViewerFailureStage = | "recording-open-timeout" | "viewer-start" | "module-load" - | "receiver-stalled"; + | "receiver-stalled" + | "camera-first-media-timeout" + | "camera-first-playable-timeout" + | "camera-queue-capacity" + | "camera-append-quota" + | "camera-append-error" + | "camera-source-buffer-error" + | "camera-fragment-decode" + | "camera-websocket-error" + | "camera-websocket-close" + | "camera-heartbeat-reopen" + | "camera-visibility-reopen" + | "camera-network-online-reopen" + | "camera-page-restore-reopen"; export interface LiveViewerDiagnostic { eventCode: LiveViewerDiagnosticEventCode; @@ -20,6 +35,14 @@ export interface LiveViewerDiagnostic { viewerRangeMaxNs?: number | null; stalledForMs?: number | null; recoveryAttempt?: number | null; + cameraQueueBytes?: number | null; + cameraQueueSegments?: number | null; + cameraRetryCount?: number | null; + websocketCloseCode?: number | null; + transportEpoch?: number | null; + cameraAppendErrorName?: string | null; + cameraMediaSourceState?: "closed" | "open" | "ended" | null; + cameraVideoErrorCode?: number | null; } export interface LiveViewerLineage { @@ -69,7 +92,6 @@ const DEVELOPMENT_UI_BUILD_ID = "development"; const SAFE_STREAM_ID = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/; const HASHED_UI_BUILD_ID = /^\/assets\/[A-Za-z0-9._/-]+-[A-Za-z0-9_-]{8,}\.js$/; const UI_BUILD_CHECK_INTERVAL_MILLISECONDS = 15_000; -const UI_BUILD_RELOAD_DELAY_MILLISECONDS = 50; let documentInstanceId: string | null = null; let sharedUiBuildCoordinator: UiBuildStaleCoordinator | null = null; @@ -172,32 +194,45 @@ export function liveViewerDiagnosticBody( ...(safeInteger(event.recoveryAttempt) === undefined ? {} : { recovery_attempt: safeInteger(event.recoveryAttempt) }), + ...(safeInteger(event.cameraQueueBytes) === undefined + ? {} + : { camera_queue_bytes: safeInteger(event.cameraQueueBytes) }), + ...(safeInteger(event.cameraQueueSegments) === undefined + ? {} + : { camera_queue_segments: safeInteger(event.cameraQueueSegments) }), + ...(safeInteger(event.cameraRetryCount) === undefined + ? {} + : { camera_retry_count: safeInteger(event.cameraRetryCount) }), + ...(safeInteger(event.websocketCloseCode) === undefined + ? {} + : { websocket_close_code: safeInteger(event.websocketCloseCode) }), + ...(safeInteger(event.transportEpoch) === undefined + ? {} + : { transport_epoch: safeInteger(event.transportEpoch) }), + ...(event.cameraAppendErrorName + ? { camera_append_error_name: event.cameraAppendErrorName } + : {}), + ...(event.cameraMediaSourceState + ? { camera_media_source_state: event.cameraMediaSourceState } + : {}), + ...(safeInteger(event.cameraVideoErrorCode) === undefined + ? {} + : { camera_video_error_code: safeInteger(event.cameraVideoErrorCode) }), }; } -export function createUiBuildStaleCoordinator({ - scheduleReload, - reload, -}: { - scheduleReload: (callback: () => void, delayMilliseconds: number) => void; - reload: () => void; -}): UiBuildStaleCoordinator { - const listeners = new Set<(event: StaleUiBuild) => void>(); +export function createUiBuildStaleCoordinator(): UiBuildStaleCoordinator { let staleEvent: StaleUiBuild | null = null; - let reloadScheduled = false; return { - subscribe(listener) { - listeners.add(listener); - if (staleEvent) listener(staleEvent); - return () => listeners.delete(listener); + subscribe(_listener) { + // Build drift is diagnostic-only. The loaded document and its local + // Rerun/camera transports remain authoritative until normal unmount or + // their existing bounded recovery replaces them. + return () => undefined; }, report(event) { if (event.loadedUiBuildId === event.expectedUiBuildId || staleEvent) return; staleEvent = event; - for (const listener of [...listeners]) listener(event); - if (reloadScheduled) return; - reloadScheduled = true; - scheduleReload(reload, UI_BUILD_RELOAD_DELAY_MILLISECONDS); }, stale() { return staleEvent !== null; @@ -206,12 +241,7 @@ export function createUiBuildStaleCoordinator({ } function browserUiBuildCoordinator(): UiBuildStaleCoordinator { - sharedUiBuildCoordinator ??= createUiBuildStaleCoordinator({ - scheduleReload: (callback, delayMilliseconds) => { - window.setTimeout(callback, delayMilliseconds); - }, - reload: () => window.location.reload(), - }); + sharedUiBuildCoordinator ??= createUiBuildStaleCoordinator(); return sharedUiBuildCoordinator; } @@ -254,10 +284,17 @@ export function verifyLiveViewerClientBuild( lineage: Pick, signal?: AbortSignal, ): void { - if (signal?.aborted || lineage.uiBuildId === DEVELOPMENT_UI_BUILD_ID) return; + if ( + signal?.aborted || + lineage.uiBuildId === DEVELOPMENT_UI_BUILD_ID || + browserUiBuildCoordinator().stale() + ) return; void fetch("/api/v1/viewer/client-contract", { method: "GET", - headers: { Accept: "application/json" }, + headers: { + Accept: "application/json", + "X-MissionCore-UI-Build": lineage.uiBuildId, + }, cache: "no-store", signal, }).then(async (response) => { diff --git a/apps/control-station/src/core/observation/viewerSettingsTarget.ts b/apps/control-station/src/core/observation/viewerSettingsTarget.ts new file mode 100644 index 0000000..f334043 --- /dev/null +++ b/apps/control-station/src/core/observation/viewerSettingsTarget.ts @@ -0,0 +1,19 @@ +import type { MissionRuntimeState } from "../runtime/contracts"; + +export function viewerSettingsTargetIdentity( + state: MissionRuntimeState | null, +): string { + const deviceSessionId = state?.deviceSession?.sessionId?.trim(); + const sessionDeviceId = state?.deviceSession?.deviceId?.trim(); + if (deviceSessionId || sessionDeviceId) { + return ["device-session", deviceSessionId, sessionDeviceId] + .filter(Boolean) + .join(":"); + } + + return [ + state?.activeDevice?.pluginId, + state?.activeDevice?.modelId, + state?.activeDevice?.instanceId, + ].filter(Boolean).join(":") || "local-runtime"; +} diff --git a/apps/control-station/test/devicePluginContracts.test.mjs b/apps/control-station/test/devicePluginContracts.test.mjs index 042466a..136c824 100644 --- a/apps/control-station/test/devicePluginContracts.test.mjs +++ b/apps/control-station/test/devicePluginContracts.test.mjs @@ -20,6 +20,7 @@ let operatorIntentGeneration; let configuration; let compatibility; let networkProvisionFailureMessage; +let awaitNetworkProvisionSettlementAfterLostResponse; let discoveryScanFailureMessage; let connectionVerificationFailureMessage; let operationById; @@ -99,6 +100,7 @@ before(async () => { )); ({ networkProvisionFailureMessage, + awaitNetworkProvisionSettlementAfterLostResponse, discoveryScanFailureMessage, connectionVerificationFailureMessage, operationById, @@ -1043,6 +1045,56 @@ test("K1 API uses bounded no-retry timeout classes per operation family", async assert.doesNotMatch(apiSource, /automaticRetry\s*=\s*true/); }); +test("lost Quick response follows the admitted Apply by journal reads only", async () => { + const idempotencyKey = "network-provision:quick-lost-response"; + let nowMs = 1_000; + let reads = 0; + const accepted = []; + const operation = { + operation_id: "op-quick-lost-response", + action: "network.provision", + status: "running", + idempotency_key: idempotencyKey, + deadline_at: new Date(5_000).toISOString(), + }; + const initialState = { + snapshot_runtime_id: "runtime-quick-lost-response", + operations: [operation], + }; + const terminalState = { + ...initialState, + operations: [{ + ...operation, + status: "succeeded", + result: { phase: "network_applied" }, + }], + }; + + const settled = await awaitNetworkProvisionSettlementAfterLostResponse( + initialState, + idempotencyKey, + async () => { + reads += 1; + if (reads === 1) { + throw new ApiError("transient localhost handoff", 0, true); + } + return terminalState; + }, + (state) => accepted.push(state), + () => {}, + { + now: () => nowMs, + wait: async (delayMs) => { + nowMs += delayMs; + }, + }, + ); + + assert.equal(reads, 2); + assert.deepEqual(accepted, [terminalState]); + assert.equal(settled.operations[0].status, "succeeded"); +}); + test("one explicit K1 operator action supplies the backend physical acceptance", () => { assert.deepEqual(physicalCommandConfirmation.operatorActionPhysicalAcceptance(), { operator_present: true, diff --git a/apps/control-station/test/k1SupervisorPresentation.test.mjs b/apps/control-station/test/k1SupervisorPresentation.test.mjs index 8723657..d23cb45 100644 --- a/apps/control-station/test/k1SupervisorPresentation.test.mjs +++ b/apps/control-station/test/k1SupervisorPresentation.test.mjs @@ -72,6 +72,7 @@ let transportRefEquivalenceKey; let shouldRevealProvisioningNetworkStep; let retiredPhysicalReopenAuthority; let SnapshotRuntimeActionArbiter; +let runtimeActionResponseAlreadyAccepted; let connectionActionAuthoritySnapshot; let exactAppliedNetworkIntentCompleted; let selectMonotonicXgridsState; @@ -106,6 +107,7 @@ before(async () => { SnapshotRuntimeActionArbiter, connectionActionAuthoritySnapshot, exactAppliedNetworkIntentCompleted, + runtimeActionResponseAlreadyAccepted, } = await server.ssrLoadModule( "@xgrids-k1/frontend/useXgridsK1Runtime.ts", )); @@ -2026,6 +2028,27 @@ test("K1 runtime replacement retires A and isolates a new B action from late set assert.equal(pendingAction, null); }); +test("successful action response may be superseded only by the same runtime revision", () => { + const response = { + snapshot_runtime_id: "runtime-a", + snapshot_revision: 41, + }; + + assert.equal(runtimeActionResponseAlreadyAccepted(response, { + snapshot_runtime_id: "runtime-a", + snapshot_revision: 42, + }), true); + assert.equal(runtimeActionResponseAlreadyAccepted(response, { + snapshot_runtime_id: "runtime-a", + snapshot_revision: 40, + }), false); + assert.equal(runtimeActionResponseAlreadyAccepted(response, { + snapshot_runtime_id: "runtime-b", + snapshot_revision: 99, + }), false); + assert.equal(runtimeActionResponseAlreadyAccepted(response, null), false); +}); + test("explicit scenario reset B supersedes pending callback A in one runtime", () => { const arbiter = new SnapshotRuntimeActionArbiter(); const actionA = arbiter.begin(12); diff --git a/apps/control-station/test/liveReceiverWatchdog.test.mjs b/apps/control-station/test/liveReceiverWatchdog.test.mjs index 837cf70..eb0756e 100644 --- a/apps/control-station/test/liveReceiverWatchdog.test.mjs +++ b/apps/control-station/test/liveReceiverWatchdog.test.mjs @@ -273,6 +273,21 @@ test("opening receiver preserves partial store replay while backend publication } }); +test("new receiver grants its first confirmed backend publication a full open window", () => { + const openState = initialLiveReceiverOpenWatchdogState(null, 0); + const recoveryState = initialLiveReceiverRecoveryState(); + const firstObservedPublication = advanceLiveReceiverOpenWatchdog( + openState, + recoveryState, + 1, + 4_000, + ); + + assert.equal(firstObservedPublication.signal, "wait-for-store"); + assert.equal(firstObservedPublication.state.lastBackendActivitySequence, 1); + assert.deepEqual(firstObservedPublication.recoveryState, recoveryState); +}); + test("unchanged opening sequence delegates to bounded receiver restart", () => { const openState = initialLiveReceiverOpenWatchdogState(486); const recoveryState = initialLiveReceiverRecoveryState(); diff --git a/apps/control-station/test/liveViewerDiagnostics.test.mjs b/apps/control-station/test/liveViewerDiagnostics.test.mjs index 0447951..326525d 100644 --- a/apps/control-station/test/liveViewerDiagnostics.test.mjs +++ b/apps/control-station/test/liveViewerDiagnostics.test.mjs @@ -9,6 +9,7 @@ let createUiBuildStaleCoordinator; let liveViewerDiagnosticBody; let server; let uiBuildIdFromModuleScripts; +let verifyLiveViewerClientBuild; before(async () => { server = await createServer({ @@ -22,6 +23,7 @@ before(async () => { createUiBuildStaleCoordinator, liveViewerDiagnosticBody, uiBuildIdFromModuleScripts, + verifyLiveViewerClientBuild, } = await server.ssrLoadModule("/src/core/observation/liveViewerDiagnostics.ts")); }); @@ -129,9 +131,9 @@ test("two mounted viewers keep timer and diagnostic lineage isolated", () => { assert.equal(posts[0].eventLineage.lifecycleGeneration, 7); }); -test("stale-build and unmount fence callbacks before one reload", () => { +test("stale build drift neither fences transports nor reloads the document", () => { const clock = createFakeScheduler(); - const order = []; + const sideEffects = []; const posts = []; const lifecycle = createLiveViewerDiagnosticLifecycle({ lineage: lineage("00000000-0000-4000-8000-000000000031"), @@ -142,15 +144,9 @@ test("stale-build and unmount fence callbacks before one reload", () => { lifecycle.armAdmissionTimeout(() => { lifecycle.post({ eventCode: "live_receiver_error" }); }, 12_000); - const coordinator = createUiBuildStaleCoordinator({ - scheduleReload: (callback, delay) => { - order.push(`scheduled:${delay}`); - clock.scheduler.setTimeout(callback, delay); - }, - reload: () => order.push("reload"), - }); + const coordinator = createUiBuildStaleCoordinator(); coordinator.subscribe(() => { - order.push("local-transports-closed"); + sideEffects.push("local-transports-closed"); lifecycle.dispose(); }); @@ -165,9 +161,47 @@ test("stale-build and unmount fence callbacks before one reload", () => { clock.advance(60_000); lifecycle.post({ eventCode: "live_receiver_error" }); - assert.deepEqual(order, ["local-transports-closed", "scheduled:50", "reload"]); - assert.deepEqual(posts, []); - assert.equal(lifecycle.active(), false); + assert.deepEqual(sideEffects, []); + assert.deepEqual(posts, [ + { eventCode: "live_receiver_error" }, + { eventCode: "live_receiver_error" }, + ]); + assert.equal(lifecycle.active(), true); + assert.equal(coordinator.stale(), true); +}); + +test("build drift is reported once with the exact loaded build", async () => { + const originalFetch = globalThis.fetch; + const calls = []; + globalThis.fetch = async (url, options) => { + calls.push({ url, options }); + return new Response(JSON.stringify({ + schema_version: "missioncore.live-viewer-client-contract/v1", + status: "ready", + ui_build_id: "/assets/index-ijklmnop.js", + diagnostic_schema_version: "missioncore.live-viewer-diagnostic/v2", + }), { + status: 200, + headers: { "X-MissionCore-UI-Build": "/assets/index-ijklmnop.js" }, + }); + }; + + try { + const eventLineage = lineage("00000000-0000-4000-8000-000000000032"); + verifyLiveViewerClientBuild(eventLineage); + await new Promise((resolve) => setTimeout(resolve, 0)); + verifyLiveViewerClientBuild(eventLineage); + await new Promise((resolve) => setTimeout(resolve, 0)); + } finally { + globalThis.fetch = originalFetch; + } + + assert.equal(calls.length, 1); + assert.equal(calls[0].url, "/api/v1/viewer/client-contract"); + assert.equal( + calls[0].options.headers["X-MissionCore-UI-Build"], + "/assets/index-abcdefgh.js", + ); }); test("last unsubscribe fences an already queued build verification callback", () => { @@ -213,4 +247,41 @@ test("diagnostic body and build id retain exact document/viewer/build lineage", ), "/assets/index-dT7dN-y4.js", ); + + assert.deepEqual( + liveViewerDiagnosticBody( + { + eventCode: "live_camera_transport_restart_requested", + failureStage: "camera-queue-capacity", + streamId: "camera-preview-2", + cameraQueueBytes: 12_000_000, + cameraQueueSegments: 96, + cameraRetryCount: 3, + websocketCloseCode: 4_008, + transportEpoch: 7, + cameraAppendErrorName: "InvalidStateError", + cameraMediaSourceState: "open", + cameraVideoErrorCode: 3, + }, + eventLineage, + ), + { + schema_version: "missioncore.live-viewer-diagnostic/v2", + event_code: "live_camera_transport_restart_requested", + ui_build_id: "/assets/index-abcdefgh.js", + document_instance_id: "00000000-0000-4000-8000-000000000001", + viewer_instance_id: "00000000-0000-4000-8000-000000000041", + lifecycle_generation: 9, + failure_stage: "camera-queue-capacity", + stream_id: "camera-preview-2", + camera_queue_bytes: 12_000_000, + camera_queue_segments: 96, + camera_retry_count: 3, + websocket_close_code: 4_008, + transport_epoch: 7, + camera_append_error_name: "InvalidStateError", + camera_media_source_state: "open", + camera_video_error_code: 3, + }, + ); }); diff --git a/apps/control-station/test/observationSources.test.mjs b/apps/control-station/test/observationSources.test.mjs index cbd07bb..a9cc814 100644 --- a/apps/control-station/test/observationSources.test.mjs +++ b/apps/control-station/test/observationSources.test.mjs @@ -16,6 +16,7 @@ let cameraTransportRecoveryIsCurrent; let cameraTransportCanOpen; let cameraPendingQueueCanAccept; let cameraTransportCloseRecoveryMessage; +let cameraPlaybackRecoveryFailureStage; let createCameraStartupWatchdog; let cameraStartupWatchdogRecoveryMessage; let CAMERA_FIRST_MEDIA_TIMEOUT_MS; @@ -69,6 +70,7 @@ before(async () => { cameraTransportCanOpen, cameraPendingQueueCanAccept, cameraTransportCloseRecoveryMessage, + cameraPlaybackRecoveryFailureStage, createCameraStartupWatchdog, cameraStartupWatchdogRecoveryMessage, CAMERA_FIRST_MEDIA_TIMEOUT_MS, @@ -1297,6 +1299,25 @@ test("server slow-reader close is an automatic browser-only camera recovery", () assert.match(cameraTransportCloseRecoveryMessage(1_011), /восстанавливаем/); }); +test("camera transport recovery events keep exact diagnostic causes", () => { + assert.equal( + cameraPlaybackRecoveryFailureStage({ type: "heartbeat" }), + "camera-heartbeat-reopen", + ); + assert.equal( + cameraPlaybackRecoveryFailureStage({ type: "document-visible" }), + "camera-visibility-reopen", + ); + assert.equal( + cameraPlaybackRecoveryFailureStage({ type: "network-online" }), + "camera-network-online-reopen", + ); + assert.equal( + cameraPlaybackRecoveryFailureStage({ type: "page-restore", persisted: true }), + "camera-page-restore-reopen", + ); +}); + test("a laptop sleep gap reopens only the exact authoritative live camera transport", () => { const source = xgridsK1ObservationSources( cameraStreamingState("sensor.camera.right"), diff --git a/apps/control-station/test/rerunViewportAtomicAdmission.test.mjs b/apps/control-station/test/rerunViewportAtomicAdmission.test.mjs index 75e330c..f939861 100644 --- a/apps/control-station/test/rerunViewportAtomicAdmission.test.mjs +++ b/apps/control-station/test/rerunViewportAtomicAdmission.test.mjs @@ -237,6 +237,14 @@ test("recorded RRD bytes are never split across LogChannel.send_rrd calls", asyn source, /if \(readyToRender && !readyPublished\)[\s\S]*eventCode: "live_receiver_active_store_admitted"[\s\S]*diagnosticLifecycle\.markAdmitted\(\)/, ); + assert.match( + source, + /liveOpenWatchdog = initialLiveReceiverOpenWatchdogState\(null, Date\.now\(\)\)/, + ); + assert.doesNotMatch( + source, + /initialLiveReceiverOpenWatchdogState\(\s*liveActivitySequenceRef\.current/, + ); assert.doesNotMatch( source, /\], \[followLive, liveRecoveryAuthorityIdentity, liveStreamId, sourceUrl\]\);/, diff --git a/apps/control-station/test/viewerSettingsTarget.test.mjs b/apps/control-station/test/viewerSettingsTarget.test.mjs new file mode 100644 index 0000000..88c780c --- /dev/null +++ b/apps/control-station/test/viewerSettingsTarget.test.mjs @@ -0,0 +1,88 @@ +import assert from "node:assert/strict"; +import { after, before, test } from "node:test"; + +import { createServer } from "vite"; + +let server; +let viewerSettingsTargetIdentity; + +before(async () => { + server = await createServer({ + appType: "custom", + logLevel: "silent", + server: { middlewareMode: true }, + }); + ({ viewerSettingsTargetIdentity } = await server.ssrLoadModule( + "/src/core/observation/viewerSettingsTarget.ts", + )); +}); + +after(async () => { + await server?.close(); +}); + +function runtimeState({ acquisitionId, sessionId = "session-a", deviceId = "device-a" }) { + return { + phase: "connected", + sourceMode: "idle", + activeDevice: { + pluginId: "xgrids-k1", + modelId: "lixel-k1", + instanceId: "instance-a", + displayName: "K1", + }, + deviceSession: { sessionId, deviceId }, + acquisition: acquisitionId + ? { + acquisitionId, + deviceId, + deviceSessionId: sessionId, + compatibilityProfileId: "profile-a", + controlMode: "live", + state: "streaming", + stateRevision: 1, + operatorInstructions: [], + } + : null, + }; +} + +test("new acquisition on the same control session keeps the viewer settings target stable", () => { + const beforeStart = viewerSettingsTargetIdentity(runtimeState({})); + const firstRun = viewerSettingsTargetIdentity(runtimeState({ acquisitionId: "acquisition-a" })); + const secondRun = viewerSettingsTargetIdentity(runtimeState({ acquisitionId: "acquisition-b" })); + + assert.equal(firstRun, beforeStart); + assert.equal(secondRun, beforeStart); +}); + +test("temporary loss of control authority does not reapply settings within one device session", () => { + const authoritative = runtimeState({ acquisitionId: "acquisition-a" }); + const authorityUnavailable = { + ...authoritative, + activeDevice: null, + phase: "starting", + }; + + assert.equal( + viewerSettingsTargetIdentity(authorityUnavailable), + viewerSettingsTargetIdentity(authoritative), + ); +}); + +test("a new control session or device produces a new viewer settings target", () => { + const original = viewerSettingsTargetIdentity(runtimeState({ acquisitionId: "acquisition-a" })); + + assert.notEqual( + viewerSettingsTargetIdentity(runtimeState({ sessionId: "session-b" })), + original, + ); + assert.notEqual( + viewerSettingsTargetIdentity(runtimeState({ deviceId: "device-b" })), + original, + ); +}); + +test("an unconfigured runtime uses the local target", () => { + assert.equal(viewerSettingsTargetIdentity(null), "local-runtime"); +}); diff --git a/plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts b/plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts index a53a600..1b8bd38 100644 --- a/plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts +++ b/plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts @@ -135,6 +135,23 @@ export class SnapshotRuntimeActionArbiter { } } +export function runtimeActionResponseAlreadyAccepted( + responseState: XgridsK1State, + acceptedState: XgridsK1State | null, +): boolean { + const responseRuntimeId = responseState.snapshot_runtime_id?.trim() ?? ""; + const acceptedRuntimeId = acceptedState?.snapshot_runtime_id?.trim() ?? ""; + const responseRevision = responseState.snapshot_revision; + const acceptedRevision = acceptedState?.snapshot_revision; + return Boolean( + responseRuntimeId + && acceptedRuntimeId === responseRuntimeId + && Number.isSafeInteger(responseRevision) + && Number.isSafeInteger(acceptedRevision) + && (acceptedRevision as number) >= (responseRevision as number) + ); +} + export type AcquisitionPreparationDraft = Omit< PrepareAcquisitionRequest, | "operation_id" @@ -248,6 +265,87 @@ export function connectionActionAuthoritySnapshot( } const CONTROL_STATE_READ_INTERVAL_MS = 250; +const NETWORK_PROVISION_SETTLEMENT_FALLBACK_MS = 30_000; +const NETWORK_PROVISION_SETTLEMENT_MAX_MS = 300_000; +const NETWORK_PROVISION_SETTLEMENT_GRACE_MS = 1_000; + +function networkProvisionSettlementDeadline( + operation: XgridsOperation, + startedAtMs: number, +): number { + const operationDeadlineMs = operation.deadline_at + ? Date.parse(operation.deadline_at) + : Number.NaN; + const requestedDeadlineMs = Number.isFinite(operationDeadlineMs) + && operationDeadlineMs > startedAtMs + ? operationDeadlineMs + NETWORK_PROVISION_SETTLEMENT_GRACE_MS + : startedAtMs + NETWORK_PROVISION_SETTLEMENT_FALLBACK_MS; + return Math.min( + requestedDeadlineMs, + startedAtMs + NETWORK_PROVISION_SETTLEMENT_MAX_MS, + ); +} + +/** + * Follow one already-admitted Apply after a browser response is interrupted by + * the host Wi-Fi handoff. This loop only reads the local journal; it never + * retries the HTTP mutation, BLE write, CoreWLAN association or control open. + */ +export async function awaitNetworkProvisionSettlementAfterLostResponse( + initialState: XgridsK1State, + idempotencyKey: string, + readState: () => Promise, + acceptState: (state: XgridsK1State) => void, + assertOperatorIntentCurrent: () => void, + options: { + now?: () => number; + wait?: (delayMs: number) => Promise; + } = {}, +): Promise { + const now = options.now ?? Date.now; + const wait = options.wait ?? ((delayMs: number) => new Promise((resolve) => { + globalThis.setTimeout(resolve, delayMs); + })); + let state = initialState; + let operation = operationByIdempotencyKey( + state, + "network.provision", + idempotencyKey, + ); + if (!operation || !["accepted", "running"].includes(operation.status)) { + return state; + } + const deadlineMs = networkProvisionSettlementDeadline(operation, now()); + + while (["accepted", "running"].includes(operation.status)) { + assertOperatorIntentCurrent(); + const remainingMs = deadlineMs - now(); + if (remainingMs <= 0) return state; + await wait(Math.min(CONTROL_STATE_READ_INTERVAL_MS, remainingMs)); + assertOperatorIntentCurrent(); + try { + const nextState = await readState(); + assertOperatorIntentCurrent(); + acceptState(nextState); + state = nextState; + } catch (readError) { + // A host Wi-Fi transition can briefly abort even a localhost fetch. + // Preserve the admitted operation and perform only the next bounded + // journal read; the original Apply is never reissued. + if (!(readError instanceof ApiError) || !readError.transportUnavailable) { + throw readError; + } + continue; + } + operation = operationByIdempotencyKey( + state, + "network.provision", + idempotencyKey, + ); + if (!operation) return state; + } + return state; +} const CONTROL_PHASE_WAIT_TIMEOUT_MS = 120_000; function controlPhase(state: XgridsK1State): XgridsApplicationControlPhase { @@ -977,7 +1075,10 @@ export function useXgridsK1Runtime(enabled: boolean) { !operatorIntents.current.isRuntimeCurrent(runtimeToken) || !runtimeActionArbiter.current.isCurrent(actionToken) ) return false; - if (!acceptState(nextState)) return false; + if ( + !acceptState(nextState) + && !runtimeActionResponseAlreadyAccepted(nextState, latestState.current) + ) return false; if (!runtimeActionArbiter.current.isCurrent(actionToken)) return false; return true; } catch (operationError) { @@ -1355,6 +1456,13 @@ export function useXgridsK1Runtime(enabled: boolean) { throw connectError; } acceptState(failedState); + failedState = await awaitNetworkProvisionSettlementAfterLostResponse( + failedState, + request.idempotency_key, + () => xgridsK1Api.getState(), + acceptState, + assertOperatorIntentCurrent, + ); const failedOperation = operationByIdempotencyKey( failedState, "network.provision", diff --git a/src/k1link/device_plugins/xgrids_k1/active_acquisition_recovery_checkpoint.py b/src/k1link/device_plugins/xgrids_k1/active_acquisition_recovery_checkpoint.py index 64c9d70..d05325b 100644 --- a/src/k1link/device_plugins/xgrids_k1/active_acquisition_recovery_checkpoint.py +++ b/src/k1link/device_plugins/xgrids_k1/active_acquisition_recovery_checkpoint.py @@ -1873,6 +1873,7 @@ class ActiveAcquisitionRecoveryCheckpointStore: validation_checkpoint, status_proof=cessation_status_proof, physical_proof=cessation_physical_proof, + allow_reconciled_transport_change=True, ) _require_active_reconciled_standby_shape( current, @@ -2251,9 +2252,9 @@ class ActiveAcquisitionRecoveryCheckpointStore: """Cease a proven START that became standby before restart PCL. This is deliberately distinct from ``cease_prepared_reconciled``: - READY/SCAN_OVER proves that the old successful START is no longer - active, so there is no transient ACTIVE checkpoint, capture promotion, - first-PCL receipt, or invented STOP edge. + READY/SCAN_OVER proves that the previously observed active START is no + longer active, so there is no transient ACTIVE checkpoint, capture + promotion, first-PCL receipt, or invented STOP edge. """ _validate_mutation_request( @@ -2351,7 +2352,11 @@ class ActiveAcquisitionRecoveryCheckpointStore: self._clock(), floor=current.updated_at_utc, ) - assert origin_proof.original_project_id_sha256 is not None + active_project_id_sha256 = ( + origin_proof.original_project_id_sha256 + or origin_proof.reconciled_active_project_id_sha256 + ) + assert active_project_id_sha256 is not None candidate = replace( current, revision=revision, @@ -2360,9 +2365,7 @@ class ActiveAcquisitionRecoveryCheckpointStore: physical_lineage_head_revision=( cessation_physical_proof.ledger_revision ), - active_project_id_sha256=( - origin_proof.original_project_id_sha256 - ), + active_project_id_sha256=active_project_id_sha256, current_evidence_session_id=( cessation_status_proof.evidence_session_id ), @@ -2614,6 +2617,7 @@ def _require_binding_matches_checkpoint_values( compatibility_profile_id: str, binding: ActiveAcquisitionRecoveryTransportBinding, allow_target_ipv4_change: bool = False, + allow_connection_mode_change: bool = False, ) -> None: if ( binding.logical_device_id != identity.logical_device_id @@ -2621,7 +2625,10 @@ def _require_binding_matches_checkpoint_values( or binding.device_serial_sha256 != identity.device_serial_sha256 or binding.compatibility_profile_id != compatibility_profile_id or binding.transport_ref != connection.transport_ref - or binding.connection_mode != connection.connection_mode + or ( + not allow_connection_mode_change + and binding.connection_mode != connection.connection_mode + ) or ( not allow_target_ipv4_change and binding.target_ipv4 != connection.target_ipv4 @@ -2684,6 +2691,7 @@ def _require_physical_lineage_base( proof: ActiveAcquisitionRecoveryPhysicalLineageProof, *, allow_target_ipv4_change: bool = False, + allow_connection_mode_change: bool = False, ) -> None: if ( proof.acquisition_id != checkpoint.acquisition_id @@ -2706,6 +2714,7 @@ def _require_physical_lineage_base( and proof.binding == checkpoint.current_binding ) ), + allow_connection_mode_change=allow_connection_mode_change, ) @@ -2994,22 +3003,40 @@ def _require_prepared_resolved_start_standby( physical_proof=cessation_physical_proof, ) baseline = origin_proof.baseline_status_proof - if not ( + exact_origin = bool( origin_proof.origin_kind == "composite-resolved" - and origin_proof.operation_id == checkpoint.original_start_operation_id - and origin_proof.acquisition_id == checkpoint.acquisition_id - and origin_proof.payload_sha256 == checkpoint.start_payload_sha256 and origin_proof.original_attempt_stage == "resolved" and origin_proof.original_attempt_resolution == "start-active-observed" and origin_proof.original_project_id_sha256 is not None + and origin_proof.reconciled_active_project_id_sha256 + in {None, origin_proof.original_project_id_sha256} and origin_proof.project_evidence_strength == "exact-vendor-project-id" + ) + ambiguous_origin = bool( + origin_proof.origin_kind == "ambiguous-reconciled" + and origin_proof.original_attempt_stage in {"dispatching", "observing"} + and origin_proof.original_attempt_resolution is None + and origin_proof.original_project_id_sha256 is None + and origin_proof.reconciled_active_project_id_sha256 is not None + and origin_proof.project_evidence_strength + == "edge-correlated-vendor-project-id" + ) + settlement_project_id_sha256 = ( + origin_proof.original_project_id_sha256 + or origin_proof.reconciled_active_project_id_sha256 + ) + if not ( + (exact_origin or ambiguous_origin) + and origin_proof.operation_id == checkpoint.original_start_operation_id + and origin_proof.acquisition_id == checkpoint.acquisition_id + and origin_proof.payload_sha256 == checkpoint.start_payload_sha256 and origin_proof.automatic_replay_allowed is False and origin_proof.ledger_revision == cessation_physical_proof.ledger_revision and origin_proof.physical_proof_id == cessation_physical_proof.proof_id and origin_proof.reconciliation_id == reconciliation_id and origin_proof.original_attempt_sha256 == reconciliation_original_attempt_sha256 - and origin_proof.original_project_id_sha256 + and settlement_project_id_sha256 == reconciliation_original_project_id_sha256 and baseline.binding == checkpoint.prepared_binding and baseline.evidence_session_id @@ -3029,7 +3056,7 @@ def _require_prepared_resolved_start_standby( raise ActiveAcquisitionRecoveryCheckpointTransitionError( "restart standby settlement requires new runtime, control and evidence sessions" ) - if not ( + exact_terminal_lineage = bool( cessation_physical_proof.operation_id == checkpoint.original_start_operation_id and cessation_physical_proof.action == "start" @@ -3044,9 +3071,30 @@ def _require_prepared_resolved_start_standby( == "physical-standby-observed" and cessation_physical_proof.composite_complete and cessation_physical_proof.stop_fence == "none" + ) + ambiguous_terminal_lineage = bool( + cessation_physical_proof.operation_id + == checkpoint.original_start_operation_id + and cessation_physical_proof.action == "start" + and cessation_physical_proof.resolution + == "physical-standby-observed" + and cessation_physical_proof.payload_sha256 + == checkpoint.start_payload_sha256 + and cessation_physical_proof.original_start_payload_sha256 + == checkpoint.start_payload_sha256 + and cessation_physical_proof.reconciliation_kind + == "resolved-active-cessation" + and cessation_physical_proof.reconciliation_resolution + == "physical-standby-observed" + and not cessation_physical_proof.composite_complete + and cessation_physical_proof.stop_fence == "none" + ) + if not ( + (exact_origin and exact_terminal_lineage) + or (ambiguous_origin and ambiguous_terminal_lineage) ): raise ActiveAcquisitionRecoveryCheckpointTransitionError( - "resolved START standby settlement lacks exact terminal lineage" + "reconciled START standby settlement lacks exact terminal lineage" ) if _validated_timestamp( cessation_status_proof.observed_at_utc, @@ -3215,11 +3263,13 @@ def _require_active_cessation( *, status_proof: ActiveAcquisitionRecoveryStatusProof, physical_proof: ActiveAcquisitionRecoveryPhysicalLineageProof, + allow_reconciled_transport_change: bool = False, ) -> None: _require_active_cessation_shape( checkpoint, status_proof=status_proof, physical_proof=physical_proof, + allow_reconciled_transport_change=allow_reconciled_transport_change, ) if ( checkpoint.physical_lineage_head_revision is None @@ -3236,12 +3286,23 @@ def _require_active_cessation_shape( *, status_proof: ActiveAcquisitionRecoveryStatusProof, physical_proof: ActiveAcquisitionRecoveryPhysicalLineageProof, + allow_reconciled_transport_change: bool = False, ) -> None: if status_proof.session_state not in {"ready", "scan_over"}: raise ActiveAcquisitionRecoveryCheckpointTransitionError( "active cessation requires fresh READY/SCAN_OVER" ) - _require_status_binding(checkpoint, status_proof) + if allow_reconciled_transport_change: + _require_binding_matches_checkpoint_values( + identity=checkpoint.identity, + connection=checkpoint.connection, + compatibility_profile_id=checkpoint.compatibility_profile_id, + binding=status_proof.binding, + allow_target_ipv4_change=True, + allow_connection_mode_change=True, + ) + else: + _require_status_binding(checkpoint, status_proof) _require_matching_observed_proofs( status_proof=status_proof, physical_proof=physical_proof, @@ -3342,8 +3403,24 @@ def _require_active_reconciled_standby_shape( raise ActiveAcquisitionRecoveryCheckpointTransitionError( "restart standby settlement requires fresh READY/SCAN_OVER" ) - _require_status_binding(checkpoint, cessation_status_proof) - _require_physical_lineage_base(checkpoint, cessation_physical_proof) + # This is a terminal, read-only settlement of the same pinned K1. The + # scanner may become observable through Bridge after a Quick Connect STOP + # became ambiguous (or vice versa), so the verified route may change + # without granting START, STOP, replay, or network-mutation authority. + _require_binding_matches_checkpoint_values( + identity=checkpoint.identity, + connection=checkpoint.connection, + compatibility_profile_id=checkpoint.compatibility_profile_id, + binding=cessation_status_proof.binding, + allow_target_ipv4_change=True, + allow_connection_mode_change=True, + ) + _require_physical_lineage_base( + checkpoint, + cessation_physical_proof, + allow_target_ipv4_change=True, + allow_connection_mode_change=True, + ) _require_matching_observed_proofs( status_proof=cessation_status_proof, physical_proof=cessation_physical_proof, @@ -4642,6 +4719,12 @@ def _validate_checkpoint_semantics( checkpoint.state != "prepared" and checkpoint.last_gap_recovered_at_utc is not None ), + allow_connection_mode_change=( + checkpoint.state == "ceased" + and bool(checkpoint.transition_receipts) + and checkpoint.transition_receipts[-1].kind + == "cease-active-reconciled-standby" + ), ) if checkpoint.last_gap_failed_binding is not None: _require_binding_matches_checkpoint_values( @@ -4944,7 +5027,16 @@ def _validate_checkpoint_semantics( cessation_physical = checkpoint.cessation_physical_proof if checkpoint.physical_lineage_head_revision != cessation_physical.ledger_revision: raise ValueError("ceased physical lineage head does not match cessation proof") - _require_physical_lineage_base(checkpoint, cessation_physical) + _require_physical_lineage_base( + checkpoint, + cessation_physical, + allow_target_ipv4_change=( + receipts[-1].kind == "cease-active-reconciled-standby" + ), + allow_connection_mode_change=( + receipts[-1].kind == "cease-active-reconciled-standby" + ), + ) if resolved_start_reconciliation_id is not None: if not ( checkpoint.prepared_resolution_proof == cessation_physical @@ -4983,7 +5075,10 @@ def _validate_checkpoint_semantics( checkpoint.current_evidence_session_id != checkpoint.cessation_status_proof.evidence_session_id or checkpoint.active_project_id_sha256 - != origin_proof.original_project_id_sha256 + != ( + origin_proof.original_project_id_sha256 + or origin_proof.reconciled_active_project_id_sha256 + ) or receipts[-1].kind != "cease-prepared-resolved-start-standby" ): @@ -5048,6 +5143,9 @@ def _validate_checkpoint_semantics( checkpoint, status_proof=status, physical_proof=cessation_physical, + allow_reconciled_transport_change=( + receipts[-1].kind == "cease-active-reconciled-standby" + ), ) if checkpoint.current_evidence_session_id != status.evidence_session_id: raise ValueError("active cessation evidence session is inconsistent") diff --git a/src/k1link/device_plugins/xgrids_k1/camera.py b/src/k1link/device_plugins/xgrids_k1/camera.py index 9a9f4c8..09ae30b 100644 --- a/src/k1link/device_plugins/xgrids_k1/camera.py +++ b/src/k1link/device_plugins/xgrids_k1/camera.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +import logging import os import queue import signal @@ -27,6 +28,8 @@ from k1link.web.camera_archive import ( CameraSourceId = Literal["sensor.camera.left", "sensor.camera.right"] +logger = logging.getLogger(__name__) + CAMERA_SOURCE_PATHS: Final[dict[CameraSourceId, str]] = { "sensor.camera.left": "/live/chn_left_main", "sensor.camera.right": "/live/chn_right_main", @@ -48,7 +51,11 @@ MAX_CAMERA_PREVIEW_QUEUED_SEGMENTS: Final = 64 # media fragment. This remains a strict per-reader bound and matches the # frontend's reviewed 12 MiB receive envelope. MAX_CAMERA_PREVIEW_QUEUED_BYTES: Final = 12 * 1024 * 1024 -MAX_CAMERA_PREVIEW_QUEUE_AGE_SECONDS: Final = 3.0 +# The browser can pause WebSocket consumption for just over three seconds while +# the live 3D workspace commits a large reactive update. The byte and segment +# caps above remain the hard memory/latency fence; this age fence only avoids +# discarding an otherwise healthy preview at that measured UI pause boundary. +MAX_CAMERA_PREVIEW_QUEUE_AGE_SECONDS: Final = 6.0 MAX_CAMERA_PREVIEW_CONSUMERS: Final = 8 CAMERA_PREVIEW_SEND_TIMEOUT_SECONDS: Final = 3.0 CAMERA_DRAIN_TIMEOUT_SECONDS: Final = 5.0 @@ -143,6 +150,7 @@ class _CameraPreviewSegmentQueue: self._queued_bytes = 0 self._closed = False self._clock = clock + self._last_rejection_reason: str | None = None @property def queued_bytes(self) -> int: @@ -158,14 +166,19 @@ class _CameraPreviewSegmentQueue: payload_size = len(segment[1]) with self._condition: if self._closed: + self._last_rejection_reason = "queue-closed" return False now = self._clock() if self._segments and now - self._segments[0][0] > MAX_CAMERA_PREVIEW_QUEUE_AGE_SECONDS: + self._last_rejection_reason = "queue-age" return False if len(self._segments) >= MAX_CAMERA_PREVIEW_QUEUED_SEGMENTS: + self._last_rejection_reason = "queue-segments" return False if self._queued_bytes + payload_size > MAX_CAMERA_PREVIEW_QUEUED_BYTES: + self._last_rejection_reason = "queue-bytes" return False + self._last_rejection_reason = None self._segments.append((now, segment)) self._queued_bytes += payload_size self._condition.notify() @@ -188,6 +201,23 @@ class _CameraPreviewSegmentQueue: return segment return None + def diagnostic_snapshot(self) -> dict[str, int | str | None]: + """Return bounded observer facts without changing delivery state.""" + + with self._condition: + now = self._clock() + oldest_age_ms = ( + max(0, int((now - self._segments[0][0]) * 1_000)) + if self._segments + else 0 + ) + return { + "queued_bytes": self._queued_bytes, + "queued_segments": len(self._segments), + "oldest_age_ms": oldest_age_ms, + "rejection_reason": self._last_rejection_reason, + } + def close(self) -> None: with self._condition: self._segments.clear() @@ -1269,11 +1299,34 @@ class XgridsK1CameraGateway: *, failure_code: str = "consumer-too-slow", ) -> None: + queue_diagnostic = delivery.segments.diagnostic_snapshot() with self._lock: if self._producer is producer and producer.deliveries.get(id(delivery)) is delivery: producer.deliveries.pop(id(delivery), None) delivery.failure_code = failure_code self._revision += 1 + logger.warning( + "K1 camera preview browser lease retired: generation=%s " + "failure=%s rejection=%s queued_segments=%s queued_bytes=%s " + "oldest_age_ms=%s", + delivery.generation, + failure_code, + queue_diagnostic["rejection_reason"], + queue_diagnostic["queued_segments"], + queue_diagnostic["queued_bytes"], + queue_diagnostic["oldest_age_ms"], + extra={ + "event_code": "k1_camera_preview_delivery_retired", + "failure_code": failure_code, + "camera_generation": delivery.generation, + "camera_queue_rejection_reason": queue_diagnostic["rejection_reason"], + "camera_queue_segments": queue_diagnostic["queued_segments"], + "camera_queue_bytes": queue_diagnostic["queued_bytes"], + "camera_queue_oldest_age_ms": queue_diagnostic["oldest_age_ms"], + "device_write_performed": False, + "automatic_retry": False, + }, + ) _close_segment_queue(delivery) def _mark_producer_failure( diff --git a/src/k1link/device_plugins/xgrids_k1/facade.py b/src/k1link/device_plugins/xgrids_k1/facade.py index c60099c..2399bb7 100644 --- a/src/k1link/device_plugins/xgrids_k1/facade.py +++ b/src/k1link/device_plugins/xgrids_k1/facade.py @@ -10019,7 +10019,9 @@ class XgridsK1CompatibilityService: "retryable": terminal.retryable, "safe_to_retry": terminal.safe_to_retry, "side_effect_status": ( - "none" if terminal.side_effect_status == "none" else "confirmed" + terminal.side_effect_status + if terminal.side_effect_status in {"none", "unknown"} + else "confirmed" ), "durable_replay": True, }, @@ -10748,6 +10750,7 @@ class XgridsK1CompatibilityService: quick_connect_host_profile_id(selected_device_name) if quick_connect else None ) operation_stage = "device-ap-activation" if quick_connect else "ble-provisioning-write" + quick_host_association: Mapping[str, Any] | None = None # Resolve the durable identity expectation while the previous # control session is still intact. A corrupt/unavailable pin store @@ -11018,6 +11021,7 @@ class XgridsK1CompatibilityService: self._duplicate_network_process_fence_descriptor ), ) + quick_host_association = association except HostWifiProfileError as exc: write_json_atomic( session_dir / "host-wifi-association.redacted.json", @@ -11249,8 +11253,12 @@ class XgridsK1CompatibilityService: # decide whether that separately authorized host mutation is # necessary at all. host_route_class: str | None = None - host_association: Mapping[str, Any] | None = None - host_wifi_association_outcome: str | None = None + host_association: Mapping[str, Any] | None = quick_host_association + host_wifi_association_outcome: str | None = ( + str(quick_host_association.get("outcome") or "unknown") + if quick_host_association is not None + else None + ) if request.connection_mode == "bridge" and request.allow_host_wifi_switch: host_route_class = _host_route_class(ipv4) if ( @@ -11758,11 +11766,6 @@ class XgridsK1CompatibilityService: and reconciliation.original_attempt.action == "stop" and reconciliation.original_attempt.stage == "observing" and reconciliation.original_attempt.resolution is None - and reconciliation.original_attempt.publish_call_returned is True - and reconciliation.original_attempt.packet_id is not None - and reconciliation.original_attempt.qos2_completed - and reconciliation.original_attempt.application_response is not None - and reconciliation.original_attempt.application_response.success and reconciliation.observation.source == "explicit-read-only-reconciliation" and reconciliation.observation.session_state in {"ready", "scan_over"} @@ -12932,25 +12935,35 @@ class XgridsK1CompatibilityService: raise ActiveAcquisitionRecoveryCheckpointError( "ambiguous START origin has the wrong physical classification" ) - if reconciliation.kind == "resolved-active-rebind": - prior_active_project_id_sha256 = next( - ( - item.observation.project_id_sha256 - for item in reversed(record.reconciliations[:-1]) - if item.original_attempt_sha256 - == reconciliation.original_attempt_sha256 - and item.resolution == "physical-active-observed" - ), - None, + prior_active_project_id_sha256 = next( + ( + item.observation.project_id_sha256 + for item in reversed(record.reconciliations[:-1]) + if item.original_attempt_sha256 + == reconciliation.original_attempt_sha256 + and item.resolution == "physical-active-observed" + and item.observation.session_state == "scanning" + and item.observation.project_bound + and item.observation.init_ready + and not item.observation.mqtt_retained + ), + None, + ) + if reconciliation.kind == "resolved-active-rebind" and ( + prior_active_project_id_sha256 is None + or reconciliation.observation.project_id_sha256 + != prior_active_project_id_sha256 + ): + raise ActiveAcquisitionRecoveryCheckpointError( + "ambiguous START rebind changed or lacks its edge-correlated project" + ) + if ( + reconciliation.kind == "resolved-active-cessation" + and prior_active_project_id_sha256 is None + ): + raise ActiveAcquisitionRecoveryCheckpointError( + "ambiguous START cessation lacks its prior active project proof" ) - if ( - prior_active_project_id_sha256 is None - or reconciliation.observation.project_id_sha256 - != prior_active_project_id_sha256 - ): - raise ActiveAcquisitionRecoveryCheckpointError( - "ambiguous START rebind changed or lacks its edge-correlated project" - ) origin_kind = "ambiguous-reconciled" project_strength = "edge-correlated-vendor-project-id" original_project_id_sha256 = None @@ -12980,6 +12993,8 @@ class XgridsK1CompatibilityService: reconciled_active_project_id_sha256=( reconciliation.observation.project_id_sha256 if active_observation + else prior_active_project_id_sha256 + if ambiguous_origin else None ), ) @@ -13151,7 +13166,7 @@ class XgridsK1CompatibilityService: else status.observed_at_utc ) if checkpoint.state == "active": - store.cease_active_reconciled_standby( + settled_checkpoint = store.cease_active_reconciled_standby( transition_id=self._active_acquisition_checkpoint_transition_id( "cease-active-reconciled-standby", checkpoint.acquisition_id, @@ -13167,6 +13182,25 @@ class XgridsK1CompatibilityService: cessation_status_proof=status_proof, cessation_physical_proof=physical_proof, ) + logger.info( + "active acquisition checkpoint closed from verified standby", + extra={ + "event_code": ( + "active_acquisition_checkpoint_reconciled_standby_settled" + ), + "acquisition_id": checkpoint.acquisition_id, + "checkpoint_revision_before": checkpoint.revision, + "checkpoint_revision_after": settled_checkpoint.revision, + "connection_mode_before": ( + checkpoint.current_binding.connection_mode + ), + "connection_mode_after": binding.connection_mode, + "reconciliation_id": reconciliation_id, + "observed_session_state": status.session_state, + "device_write_performed": False, + "automatic_retry": False, + }, + ) elif checkpoint.state == "prepared" and ( reconciliation.kind == "resolved-active-cessation" ): @@ -13176,14 +13210,13 @@ class XgridsK1CompatibilityService: reconciliation=reconciliation, physical_proof=physical_proof, ) - original_project_id_sha256 = ( - reconciliation.original_attempt.last_status.project_id_sha256 - if reconciliation.original_attempt.last_status is not None - else None + settlement_project_id_sha256 = ( + origin_proof.original_project_id_sha256 + or origin_proof.reconciled_active_project_id_sha256 ) - if original_project_id_sha256 is None: + if settlement_project_id_sha256 is None: raise ActiveAcquisitionRecoveryCheckpointError( - "resolved START standby lacks its original vendor project" + "reconciled START standby lacks its proven vendor project" ) store.cease_prepared_resolved_start_standby( transition_id=self._active_acquisition_checkpoint_transition_id( @@ -13206,7 +13239,7 @@ class XgridsK1CompatibilityService: reconciliation.original_attempt_sha256 ), reconciliation_original_project_id_sha256=( - original_project_id_sha256 + settlement_project_id_sha256 ), ) else: @@ -13361,7 +13394,7 @@ class XgridsK1CompatibilityService: ) and reconciliation.kind == "prepared-stop-classification" ) - dispatched_then_observed_standby = bool( + ambiguous_stop_observed_standby = bool( record is not None and record.resolution == "physical-standby-observed" and reconciliation is not None @@ -13371,11 +13404,6 @@ class XgridsK1CompatibilityService: and reconciliation.original_attempt.action == "stop" and reconciliation.original_attempt.stage == "observing" and reconciliation.original_attempt.resolution is None - and reconciliation.original_attempt.publish_call_returned is True - and reconciliation.original_attempt.packet_id is not None - and reconciliation.original_attempt.qos2_completed - and reconciliation.original_attempt.application_response is not None - and reconciliation.original_attempt.application_response.success ) if not ( ledger_snapshot.status == "resolved" @@ -13383,7 +13411,7 @@ class XgridsK1CompatibilityService: and record.action == "stop" and record.stage == "resolved" and reconciliation is not None - and (definitely_undispatched or dispatched_then_observed_standby) + and (definitely_undispatched or ambiguous_stop_observed_standby) and reconciliation.resolution == "physical-standby-observed" and reconciliation.observation.source == "explicit-read-only-reconciliation" @@ -15573,13 +15601,47 @@ class XgridsK1CompatibilityService: else: restart_outcome = None if allow_receiver_rehydrate: - restart_outcome = ( - await self._rehydrate_active_acquisition_after_restart( - token=checkpoint_trust_token, - reconciliation_id=reconciliation_id, - reconciled_record=reconciled_record, + try: + restart_outcome = ( + await self._rehydrate_active_acquisition_after_restart( + token=checkpoint_trust_token, + reconciliation_id=reconciliation_id, + reconciled_record=reconciled_record, + ) + ) + except ActiveAcquisitionRecoveryCheckpointError as exc: + # Receiver resurrection is optional after an exact + # read-only SCANNING proof. A stale/corrupt local + # checkpoint must revoke that broader authority, but + # it must not discard the narrower, already durable + # STOP-only authority derived from DeviceInfo plus a + # fresh non-retained DeviceStatus. + reason_code = str( + getattr( + exc, + "reason_code", + ( + "active-acquisition-recovery-" + "checkpoint-rehydrate-failed" + ), + ) + ) + self._mark_active_acquisition_checkpoint_untrusted( + trust="unavailable", + reason_code=reason_code, + ) + logger.warning( + "K1 checkpoint rejected receiver rehydration; " + "continuing with proven STOP-only control", + extra={ + "event_code": ( + "k1_restart_rehydration_checkpoint_rejected" + ), + "reason_code": reason_code, + "device_write_performed": False, + "automatic_retry": False, + }, ) - ) if restart_outcome is None: checkpoint_lineage = ( self._restart_stop_only_checkpoint_lineage( diff --git a/src/k1link/device_plugins/xgrids_k1/network_provisioning_idempotency_journal.py b/src/k1link/device_plugins/xgrids_k1/network_provisioning_idempotency_journal.py index 38789ae..c3323e8 100644 --- a/src/k1link/device_plugins/xgrids_k1/network_provisioning_idempotency_journal.py +++ b/src/k1link/device_plugins/xgrids_k1/network_provisioning_idempotency_journal.py @@ -37,11 +37,11 @@ NetworkProvisioningIdempotencyStage = Literal["prepared", "unresolved", "termina NetworkProvisioningIdempotencyDisposition = Literal["admitted", "terminal-replay"] NetworkProvisioningIdempotencyStatus = Literal["empty", "ready", "blocked", "corrupt"] NetworkProvisioningTerminalOutcome = Literal["succeeded", "failed", "cancelled"] -NetworkProvisioningSideEffectStatus = Literal["none", "applied", "reconciled"] +NetworkProvisioningSideEffectStatus = Literal["none", "applied", "reconciled", "unknown"] _STAGES = frozenset({"prepared", "unresolved", "terminal"}) _OUTCOMES = frozenset({"succeeded", "failed", "cancelled"}) -_SIDE_EFFECT_STATUSES = frozenset({"none", "applied", "reconciled"}) +_SIDE_EFFECT_STATUSES = frozenset({"none", "applied", "reconciled", "unknown"}) _SAFE_ACTION = re.compile(r"^[a-z][a-z0-9._-]{0,95}$") _SAFE_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:+-]{0,159}$") _SAFE_CODE = re.compile(r"^[a-z][a-z0-9._-]{0,127}$") diff --git a/src/k1link/web/runtime_diagnostics.py b/src/k1link/web/runtime_diagnostics.py index 5abb8ba..2640066 100644 --- a/src/k1link/web/runtime_diagnostics.py +++ b/src/k1link/web/runtime_diagnostics.py @@ -56,6 +56,7 @@ _EXTRA_FIELDS: Final = ( "stale_session_retired", "stream_id", "ui_build_id", + "expected_ui_build_id", "document_instance_id", "viewer_instance_id", "lifecycle_generation", @@ -63,6 +64,17 @@ _EXTRA_FIELDS: Final = ( "viewer_range_max_ns", "stalled_for_ms", "recovery_attempt", + "camera_queue_bytes", + "camera_queue_segments", + "camera_queue_oldest_age_ms", + "camera_queue_rejection_reason", + "camera_retry_count", + "camera_generation", + "websocket_close_code", + "transport_epoch", + "camera_append_error_name", + "camera_media_source_state", + "camera_video_error_code", "preferred_port", "selected_port", ) diff --git a/src/k1link/web/viewer_diagnostics_api.py b/src/k1link/web/viewer_diagnostics_api.py index 75b5ff8..7b1ecd6 100644 --- a/src/k1link/web/viewer_diagnostics_api.py +++ b/src/k1link/web/viewer_diagnostics_api.py @@ -1,10 +1,11 @@ from __future__ import annotations import logging +import re from collections.abc import Callable from typing import Literal -from fastapi import APIRouter, Response +from fastapi import APIRouter, Request, Response from fastapi.responses import JSONResponse from pydantic import BaseModel, ConfigDict, Field @@ -17,12 +18,27 @@ LiveViewerEventCode = Literal[ "live_receiver_recovery_exhausted", "live_receiver_active_store_admitted", "live_receiver_error", + "live_camera_transport_restart_requested", + "live_camera_transport_playing", ] LiveViewerFailureStage = Literal[ "recording-open-timeout", "viewer-start", "module-load", "receiver-stalled", + "camera-first-media-timeout", + "camera-first-playable-timeout", + "camera-queue-capacity", + "camera-append-quota", + "camera-append-error", + "camera-source-buffer-error", + "camera-fragment-decode", + "camera-websocket-error", + "camera-websocket-close", + "camera-heartbeat-reopen", + "camera-visibility-reopen", + "camera-network-online-reopen", + "camera-page-restore-reopen", ] LIVE_VIEWER_DIAGNOSTIC_SCHEMA = "missioncore.live-viewer-diagnostic/v2" @@ -61,6 +77,17 @@ class LiveViewerDiagnosticEvent(BaseModel): viewer_range_max_ns: int | None = Field(default=None, ge=0) stalled_for_ms: int | None = Field(default=None, ge=0, le=600_000) recovery_attempt: int | None = Field(default=None, ge=1, le=3) + camera_queue_bytes: int | None = Field(default=None, ge=0, le=16_777_216) + camera_queue_segments: int | None = Field(default=None, ge=0, le=128) + camera_retry_count: int | None = Field(default=None, ge=1, le=4) + websocket_close_code: int | None = Field(default=None, ge=0, le=4_999) + transport_epoch: int | None = Field(default=None, ge=1) + camera_append_error_name: str | None = Field( + default=None, + pattern=r"^[A-Za-z][A-Za-z0-9]{0,63}$", + ) + camera_media_source_state: Literal["closed", "open", "ended"] | None = None + camera_video_error_code: int | None = Field(default=None, ge=0, le=4) def build_viewer_diagnostics_router( @@ -70,7 +97,7 @@ def build_viewer_diagnostics_router( router = APIRouter(prefix="/api/v1/viewer", tags=["viewer"]) @router.get("/client-contract") - def get_live_viewer_client_contract() -> JSONResponse: + def get_live_viewer_client_contract(request: Request) -> JSONResponse: expected = expected_ui_build_id() if expected is None: return JSONResponse( @@ -81,6 +108,24 @@ def build_viewer_diagnostics_router( }, headers={"Cache-Control": "no-store"}, ) + loaded = request.headers.get(UI_BUILD_HEADER) + if ( + loaded not in {None, "development", expected} + and re.fullmatch(_UI_BUILD_ID_PATTERN, loaded) is not None + ): + logger.info( + "Mission Core UI build drift observed; automatic page reload suppressed: " + "loaded=%s expected=%s", + loaded, + expected, + extra={ + "event_code": "ui_build_drift_reload_suppressed", + "ui_build_id": loaded, + "expected_ui_build_id": expected, + "device_write_performed": False, + "automatic_retry": False, + }, + ) return JSONResponse( content={ "schema_version": LIVE_VIEWER_CLIENT_CONTRACT_SCHEMA, @@ -142,6 +187,14 @@ def build_viewer_diagnostics_router( "viewer_range_max_ns": event.viewer_range_max_ns, "stalled_for_ms": event.stalled_for_ms, "recovery_attempt": event.recovery_attempt, + "camera_queue_bytes": event.camera_queue_bytes, + "camera_queue_segments": event.camera_queue_segments, + "camera_retry_count": event.camera_retry_count, + "websocket_close_code": event.websocket_close_code, + "transport_epoch": event.transport_epoch, + "camera_append_error_name": event.camera_append_error_name, + "camera_media_source_state": event.camera_media_source_state, + "camera_video_error_code": event.camera_video_error_code, }, ) return Response(status_code=204, headers={UI_BUILD_HEADER: expected}) diff --git a/tests/test_rerun_bridge.py b/tests/test_rerun_bridge.py index 8f92aef..4491680 100644 --- a/tests/test_rerun_bridge.py +++ b/tests/test_rerun_bridge.py @@ -20,6 +20,7 @@ from k1link.device_plugins.xgrids_k1.protocol.normalizer import normalize_k1_mes from k1link.device_plugins.xgrids_k1.viewer.messages import StreamMessage from k1link.device_plugins.xgrids_k1.viewer.runtime import VisualizationRuntime from k1link.viewer.rerun_bridge import ( + LIVE_GRPC_BUFFER_LIMIT, RerunBridge, RerunSceneSettings, _live_time_panel, @@ -37,8 +38,10 @@ class FakeRecording: self.blueprints: list[object] = [] self.disconnected = False self.flush_count = 0 + self.serve_grpc_options: dict[str, object] | None = None - def serve_grpc(self, **_: object) -> str: + def serve_grpc(self, **options: object) -> str: + self.serve_grpc_options = options return "rerun+http://127.0.0.1:9876/proxy" def log(self, path: str, entity: object, *, static: bool = False) -> None: @@ -299,6 +302,10 @@ def test_legacy_points_and_pose_are_logged_to_rerun( "stream_time", } assert recording.flush_count == 1 + assert recording.serve_grpc_options is not None + assert recording.serve_grpc_options["server_memory_limit"] == "32MiB" + assert recording.serve_grpc_options["newest_first"] is False + assert LIVE_GRPC_BUFFER_LIMIT == "32MiB" bridge.close() assert recording.disconnected is True diff --git a/tests/test_viewer_diagnostics_api.py b/tests/test_viewer_diagnostics_api.py index b381a6a..c77cd2a 100644 --- a/tests/test_viewer_diagnostics_api.py +++ b/tests/test_viewer_diagnostics_api.py @@ -11,6 +11,7 @@ import pytest from fastapi import APIRouter from fastapi.routing import APIRoute from pydantic import ValidationError +from starlette.requests import Request from k1link.web.runtime_diagnostics import ( SCANNER_LOGGER_NAME, @@ -34,6 +35,13 @@ def _endpoint(router: APIRouter, path: str, method: str) -> Callable[..., Any]: raise AssertionError(f"{method} {path} route is missing") +def _request(*, ui_build_id: str | None = None) -> Request: + headers = [] + if ui_build_id is not None: + headers.append((b"x-missioncore-ui-build", ui_build_id.encode("ascii"))) + return Request({"type": "http", "method": "GET", "path": "/", "headers": headers}) + + def test_private_scanner_diagnostics_are_durable_structured_and_bounded( tmp_path: Path, ) -> None: @@ -76,6 +84,11 @@ def test_private_scanner_diagnostics_are_durable_structured_and_bounded( "device_write_performed": False, "preferred_port": 9876, "selected_port": 9877, + "camera_queue_bytes": 12_000_000, + "camera_queue_segments": 96, + "camera_retry_count": 3, + "websocket_close_code": 4_008, + "transport_epoch": 7, "unapproved_secret_field": "must-not-be-written", }, ) @@ -118,6 +131,11 @@ def test_private_scanner_diagnostics_are_durable_structured_and_bounded( assert document["device_write_performed"] is False assert document["preferred_port"] == 9876 assert document["selected_port"] == 9877 + assert document["camera_queue_bytes"] == 12_000_000 + assert document["camera_queue_segments"] == 96 + assert document["camera_retry_count"] == 3 + assert document["websocket_close_code"] == 4_008 + assert document["transport_epoch"] == 7 assert "unapproved_secret_field" not in document parent = logging.getLogger(SCANNER_LOGGER_NAME) @@ -180,6 +198,40 @@ def test_live_viewer_diagnostic_endpoint_accepts_only_bounded_events( ) assert fallback.failure_stage is None + camera_restart = LiveViewerDiagnosticEvent( + schema_version="missioncore.live-viewer-diagnostic/v2", + event_code="live_camera_transport_restart_requested", + ui_build_id=expected_build, + document_instance_id="00000000-0000-4000-8000-000000000001", + viewer_instance_id="00000000-0000-4000-8000-000000000003", + lifecycle_generation=2, + failure_stage="camera-queue-capacity", + stream_id="camera-preview-2", + camera_queue_bytes=12_000_000, + camera_queue_segments=96, + camera_retry_count=3, + websocket_close_code=4_008, + transport_epoch=7, + camera_append_error_name="InvalidStateError", + camera_media_source_state="open", + camera_video_error_code=3, + ) + with caplog.at_level( + logging.INFO, + logger="k1link.device_plugins.xgrids_k1.viewer_receiver", + ): + camera_response = endpoint(camera_restart) + assert camera_response.status_code == 204 + assert caplog.records[-1].failure_stage == "camera-queue-capacity" + assert caplog.records[-1].camera_queue_bytes == 12_000_000 + assert caplog.records[-1].camera_queue_segments == 96 + assert caplog.records[-1].camera_retry_count == 3 + assert caplog.records[-1].websocket_close_code == 4_008 + assert caplog.records[-1].transport_epoch == 7 + assert caplog.records[-1].camera_append_error_name == "InvalidStateError" + assert caplog.records[-1].camera_media_source_state == "open" + assert caplog.records[-1].camera_video_error_code == 3 + def test_live_viewer_diagnostic_rejects_stale_build_before_logging( caplog: pytest.LogCaptureFixture, @@ -212,7 +264,7 @@ def test_live_viewer_client_contract_is_no_store_and_exact_build() -> None: router = build_viewer_diagnostics_router(expected_ui_build_id=lambda: expected_build) endpoint = _endpoint(router, "/api/v1/viewer/client-contract", "GET") - response = endpoint() + response = endpoint(_request(ui_build_id=expected_build)) assert response.status_code == 200 assert response.headers["cache-control"] == "no-store" @@ -225,11 +277,33 @@ def test_live_viewer_client_contract_is_no_store_and_exact_build() -> None: } +def test_live_viewer_client_contract_logs_suppressed_stale_build_reload( + caplog: pytest.LogCaptureFixture, +) -> None: + loaded_build = "/assets/index-abcdefgh.js" + expected_build = "/assets/index-ijklmnop.js" + router = build_viewer_diagnostics_router(expected_ui_build_id=lambda: expected_build) + endpoint = _endpoint(router, "/api/v1/viewer/client-contract", "GET") + + with caplog.at_level( + logging.INFO, + logger="k1link.device_plugins.xgrids_k1.viewer_receiver", + ): + response = endpoint(_request(ui_build_id=loaded_build)) + + assert response.status_code == 200 + assert caplog.records[-1].event_code == "ui_build_drift_reload_suppressed" + assert caplog.records[-1].ui_build_id == loaded_build + assert caplog.records[-1].expected_ui_build_id == expected_build + assert caplog.records[-1].device_write_performed is False + assert caplog.records[-1].automatic_retry is False + + def test_live_viewer_client_contract_no_dist_is_retryable_without_reload_header() -> None: router = build_viewer_diagnostics_router(expected_ui_build_id=lambda: None) endpoint = _endpoint(router, "/api/v1/viewer/client-contract", "GET") - response = endpoint() + response = endpoint(_request()) assert response.status_code == 503 assert response.headers["cache-control"] == "no-store" diff --git a/tests/test_xgrids_acquisition_lifecycle.py b/tests/test_xgrids_acquisition_lifecycle.py index a520276..93cda05 100644 --- a/tests/test_xgrids_acquisition_lifecycle.py +++ b/tests/test_xgrids_acquisition_lifecycle.py @@ -19833,6 +19833,8 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host( assert state["connection_mode"] == "quick-connect" assert state["k1_ip"] == "192.168.56.1" assert state["compatibility"]["attestation"]["topology"] == "device-ap" + assert state["last_operation"]["result"]["host_wifi_association_performed"] is True + assert state["last_operation"]["result"]["host_wifi_association_outcome"] == "associated" quick_sessions = sorted(service.evidence_root.glob("*viewer_k1_ap_association*")) assert len(quick_sessions) == 1 assert not (quick_sessions[0] / "provisioning.sensitive.json").exists() @@ -19843,6 +19845,8 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host( assert '"credentials_resolved_by_plugin": true' in redacted_manifest assert "credential_provider_id" in redacted_manifest assert "device_ap_activation_profile_id" in redacted_manifest + assert '"host_wifi_association_performed": true' in redacted_manifest + assert '"host_wifi_association_outcome": "associated"' in redacted_manifest assert (quick_sessions[0] / "ap-activation.redacted.json").exists() assert ( service._camera_target_for_session( # noqa: SLF001 @@ -29487,6 +29491,114 @@ _RECOVERY_ACQUISITION_ID = "acquisition-persisted-active-k1" _RECOVERY_START_OPERATION_ID = "physical-start-persisted-active-k1" +def _persist_ambiguous_start_with_prepared_checkpoint( + service: XgridsK1CompatibilityService, +) -> None: + """Persist the exact pre-crash shape: PREPARED token + ambiguous START.""" + + identity = PhysicalCommandIdentity( + vendor_device_id_sha256=_RECOVERY_VENDOR_HASH, + device_serial_sha256=_RECOVERY_SERIAL_HASH, + ) + connection = PhysicalCommandConnectionBinding( + intent_id="ambiguous-start-intent", + transport_ref="test-ble-transport", + connection_mode="bridge", + target_ipv4="192.168.68.52", + target_port=facade_module.CONTROL_MQTT_PORT, + host_path_epoch=1, + control_session_id="ambiguous-start-control", + producer_generation=1, + ) + observed_at_utc = datetime.now(UTC).isoformat(timespec="milliseconds").replace( + "+00:00", + "Z", + ) + baseline = PhysicalCommandStatusEvidence( + source="live-control-session", + vendor_device_id_sha256=_RECOVERY_VENDOR_HASH, + device_serial_sha256=_RECOVERY_SERIAL_HASH, + control_session_id=connection.control_session_id, + host_path_epoch=connection.host_path_epoch, + producer_generation=connection.producer_generation, + session_state="ready", + session_state_code=300, + project_bound=False, + project_id_sha256=None, + init_ready=False, + status_message_sha256="1" * 64, + mqtt_retained=False, + observed_at_utc=observed_at_utc, + ) + operation_id = "physical-start-persisted-ambiguous-k1" + acquisition_id = "acquisition-persisted-ambiguous-k1" + payload_sha256 = "2" * 64 + ledger = service._physical_command_ledger # noqa: SLF001 + ledger.prepare( + operation_id=operation_id, + parent_operation_id=None, + acquisition_id=acquisition_id, + action="start", + identity=identity, + connection=connection, + compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID, + payload_sha256=payload_sha256, + baseline_status=baseline, + ) + store = service._active_acquisition_checkpoint # noqa: SLF001 + assert store is not None + store.prepare( + transition_id="prepare-persisted-ambiguous-start", + predecessor_revision=0, + acquisition_id=acquisition_id, + original_start_operation_id=operation_id, + start_payload_sha256=payload_sha256, + identity=ActiveAcquisitionRecoveryIdentity( + logical_device_id="known-k1", + vendor_device_id_sha256=_RECOVERY_VENDOR_HASH, + device_serial_sha256=_RECOVERY_SERIAL_HASH, + ), + connection=ActiveAcquisitionRecoveryConnection( + transport_ref=connection.transport_ref, + connection_mode=connection.connection_mode, + target_ipv4=connection.target_ipv4, + target_port=connection.target_port, + ), + compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID, + project_name="AMBIGUOUS_RECOVERY", + project_name_wire_sha256=active_acquisition_project_name_sha256( + "AMBIGUOUS_RECOVERY" + ), + original_evidence_session_id="evidence-before-crash", + duration_seconds=None, + requested_streams=("spatial.point-cloud.live", "camera.rgb.live"), + evidence_policy="required", + mount_type="handheld", + gnss_mode="none", + prepared_binding=ActiveAcquisitionRecoveryTransportBinding( + runtime_instance_id="runtime-before-crash", + intent_id=connection.intent_id, + transport_ref=connection.transport_ref, + connection_mode=connection.connection_mode, + target_ipv4=connection.target_ipv4, + target_port=connection.target_port, + host_path_epoch=connection.host_path_epoch, + control_session_id=connection.control_session_id, + producer_generation=connection.producer_generation, + logical_device_id="known-k1", + compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID, + vendor_device_id_sha256=_RECOVERY_VENDOR_HASH, + device_serial_sha256=_RECOVERY_SERIAL_HASH, + ), + ) + ledger.mark_dispatching(operation_id) + ledger.mark_observing( + operation_id, + publish_call_returned=True, + packet_id=41, + ) + + def _persist_resolved_active_start_for_restart( service: XgridsK1CompatibilityService, ) -> None: @@ -30448,6 +30560,54 @@ def test_explicit_verify_reconciles_scanning_as_active_without_unblocking_mode_c } +def test_explicit_verify_scanning_checkpoint_rehydrate_failure_keeps_stop_only( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A rejected local checkpoint cannot erase proven physical STOP authority.""" + + service, runtime = service_with_fake_runtime(tmp_path) + coordinator = _VerifyPhysicalRecoveryCoordinator( + observed_session_state="scanning", + reconciliation_ready=True, + ) + _install_synthetic_verify_recovery(service, coordinator) + rehydrate_calls: list[str] = [] + + async def reject_receiver_rehydration(**_: object) -> None: + rehydrate_calls.append("rejected") + raise facade_module.ActiveAcquisitionRecoveryCheckpointError( + "synthetic stale prepared checkpoint" + ) + + monkeypatch.setattr( + service, + "_rehydrate_active_acquisition_after_restart", + reject_receiver_rehydration, + ) + operation_id = "op-00000000-0000-4000-8000-000000001217" + + state = asyncio.run( + service.verify_connection( + _retained_physical_recovery_verify_request(operation_id=operation_id) + ) + ) + + assert rehydrate_calls == ["rejected"] + assert state["last_operation"]["status"] == "succeeded" + assert state["physical_command"]["record"]["resolution"] == ( + "physical-active-observed" + ) + assert state["acquisition"]["acquisition_id"] == "acq-persisted-start" + assert state["acquisition"]["state"] == "failed" + assert state["acquisition"]["result"]["recovery_only"] is True + assert state["application_control_session"]["state"] == "scanning" + assert state["application_control_session"]["can_stop"] is True + assert state["connection_policy"]["actions"]["stop-acquisition"]["allowed"] is True + assert runtime.start_calls == [] + assert runtime.stop_calls == 0 + + def test_explicit_verify_scanning_cleans_terminal_camera_residual_for_stop_only_shell( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, @@ -30934,6 +31094,113 @@ def test_active_reconciliation_survives_post_commit_adoption_failure_and_next_re ]["mode_selection"] +def test_prepared_ambiguous_start_active_then_ready_settles_checkpoint_without_command( + tmp_path: Path, +) -> None: + """Fresh SCANNING then READY closes the pre-crash token without replay.""" + + service, runtime = service_with_fake_runtime(tmp_path) + _persist_ambiguous_start_with_prepared_checkpoint(service) + coordinator = service._physical_command_coordinator # noqa: SLF001 + observed_base = datetime.now(UTC) + timedelta(seconds=1) + + def bind_fresh_status( + *, + suffix: str, + generation: int, + session_state: str, + ) -> None: + observed_at_utc = ( + observed_base + timedelta(seconds=generation) + ).isoformat(timespec="milliseconds").replace("+00:00", "Z") + coordinator.application_response( + ApplicationMqttResponseEvidence( + operation_key=f"bootstrap:{suffix}:DeviceInfoRequest", + response_topic="lixel/application/response/device_info", + payload_sha256=hashlib.sha256( + f"device-info:{suffix}".encode() + ).hexdigest(), + modeling_action=None, + result_code=None, + success=None, + observed_at_utc=observed_at_utc, + ) + ) + coordinator.bind_control_session( + PhysicalCommandRuntimeBinding( + vendor_device_id_sha256=_RECOVERY_VENDOR_HASH, + device_serial_sha256=_RECOVERY_SERIAL_HASH, + compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID, + intent_id=f"recovery-{suffix}-intent", + transport_ref="test-ble-transport", + connection_mode="bridge", + target_ipv4="192.168.68.52", + target_port=facade_module.CONTROL_MQTT_PORT, + host_path_epoch=generation, + control_session_id=f"recovery-{suffix}-control", + producer_generation=generation, + ) + ) + scanning = session_state == "scanning" + coordinator.device_status( + ApplicationMqttDeviceStatusEvidence( + vendor_device_id_sha256=_RECOVERY_VENDOR_HASH, + device_serial_sha256=_RECOVERY_SERIAL_HASH, + session_state=session_state, # type: ignore[arg-type] + session_state_code=MODELING_STATE_BASE + (302 if scanning else 300), + project_bound=scanning, + project_id_sha256=_RECOVERY_PROJECT_HASH if scanning else None, + init_ready=scanning, + status_message_sha256=hashlib.sha256( + f"status:{suffix}:{session_state}".encode() + ).hexdigest(), + mqtt_retained=False, + observed_at_utc=observed_at_utc, + ) + ) + + bind_fresh_status(suffix="active", generation=2, session_state="scanning") + active_record = coordinator.reconcile_unresolved( + reconciliation_id="reconcile-ambiguous-active" + ) + assert active_record["resolution"] == "physical-active-observed" + token = service._validate_active_acquisition_checkpoint_lineage() # noqa: SLF001 + assert token is not None and token.checkpoint_state == "prepared" + + bind_fresh_status(suffix="ready", generation=3, session_state="ready") + standby_record = coordinator.reconcile_resolved_active( + reconciliation_id="reconcile-ambiguous-ready" + ) + assert standby_record["resolution"] == "physical-standby-observed" + assert standby_record["reconciliations"][-1]["kind"] == ( + "resolved-active-cessation" + ) + + settled = service._settle_restart_checkpoint_after_verified_standby( # noqa: SLF001 + token=token, + reconciliation_id="reconcile-ambiguous-ready", + reconciled_record=standby_record, + ) + + assert settled is True + store = service._active_acquisition_checkpoint # noqa: SLF001 + assert store is not None + checkpoint = store.snapshot().checkpoint + assert checkpoint is not None and checkpoint.state == "ceased" + assert checkpoint.active_project_id_sha256 == _RECOVERY_PROJECT_HASH + assert checkpoint.activated_at_utc is None + assert checkpoint.first_published_pcl_proof is None + assert checkpoint.reconciled_start_origin_proof is not None + assert checkpoint.reconciled_start_origin_proof.origin_kind == ( + "ambiguous-reconciled" + ) + assert checkpoint.reconciled_start_origin_proof.reconciled_active_project_id_sha256 == ( + _RECOVERY_PROJECT_HASH + ) + assert runtime.start_calls == [] + assert runtime.stop_calls == 0 + + @pytest.mark.parametrize( ("proof_kind", "observed_session_state"), [ diff --git a/tests/test_xgrids_active_acquisition_checkpoint_integration.py b/tests/test_xgrids_active_acquisition_checkpoint_integration.py index fc02a63..154b31f 100644 --- a/tests/test_xgrids_active_acquisition_checkpoint_integration.py +++ b/tests/test_xgrids_active_acquisition_checkpoint_integration.py @@ -126,12 +126,14 @@ def _connection( control_session_id: str, host_path_epoch: int, producer_generation: int, + connection_mode: str = "bridge", + target_ipv4: str = "192.168.68.52", ) -> PhysicalCommandConnectionBinding: return PhysicalCommandConnectionBinding( intent_id="intent-checkpoint-integration", transport_ref="transport-checkpoint-integration", - connection_mode="bridge", - target_ipv4="192.168.68.52", + connection_mode=connection_mode, # type: ignore[arg-type] + target_ipv4=target_ipv4, target_port=1883, host_path_epoch=host_path_epoch, control_session_id=control_session_id, @@ -460,10 +462,22 @@ def test_repeated_reset_reopen_fresh_ready_ceases_old_active_checkpoint( (False, True), ids=("process-restart", "same-process-new-control-epoch"), ) +@pytest.mark.parametrize( + "acknowledged_stop", + (True, False), + ids=("acknowledged-stop", "ambiguous-stop"), +) +@pytest.mark.parametrize( + "cross_mode", + (False, True), + ids=("same-mode", "quick-to-bridge"), +) def test_ready_after_dispatched_stop_ceases_active_checkpoint( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, same_process: bool, + acknowledged_stop: bool, + cross_mode: bool, ) -> None: """Fresh READY settles a dispatched STOP across either recovery boundary.""" @@ -472,6 +486,8 @@ def test_ready_after_dispatched_stop_ceases_active_checkpoint( control_session_id="checkpoint-dispatched-stop-original-control", host_path_epoch=1, producer_generation=1, + connection_mode="quick-connect" if cross_mode else "bridge", + target_ipv4="192.168.56.1" if cross_mode else "192.168.68.52", ) _prepare_and_activate_checkpoint(service, original) ledger = service._physical_command_ledger # noqa: SLF001 @@ -500,21 +516,22 @@ def test_ready_after_dispatched_stop_ceases_active_checkpoint( publish_call_returned=True, packet_id=42, ) - ledger.mark_qos2_completed(FIRST_STOP_OPERATION_ID, packet_id=42) - ledger.record_application_response( - FIRST_STOP_OPERATION_ID, - PhysicalCommandApplicationResponse( - operation_id=FIRST_STOP_OPERATION_ID, - action="stop", - control_session_id=original.control_session_id, - host_path_epoch=original.host_path_epoch, - producer_generation=original.producer_generation, - result_code=PHYSICAL_COMMAND_APPLICATION_SUCCESS_CODE, - success=True, - payload_sha256="3" * 64, - observed_at_utc="2026-08-13T12:01:01.000Z", - ), - ) + if acknowledged_stop: + ledger.mark_qos2_completed(FIRST_STOP_OPERATION_ID, packet_id=42) + ledger.record_application_response( + FIRST_STOP_OPERATION_ID, + PhysicalCommandApplicationResponse( + operation_id=FIRST_STOP_OPERATION_ID, + action="stop", + control_session_id=original.control_session_id, + host_path_epoch=original.host_path_epoch, + producer_generation=original.producer_generation, + result_code=PHYSICAL_COMMAND_APPLICATION_SUCCESS_CODE, + success=True, + payload_sha256="3" * 64, + observed_at_utc="2026-08-13T12:01:01.000Z", + ), + ) if not same_process: service._snapshot_runtime_id = ( # noqa: SLF001 "snapshot-runtime-checkpoint-dispatched-stop-successor" @@ -527,6 +544,8 @@ def test_ready_after_dispatched_stop_ceases_active_checkpoint( # behind an ACTIVE checkpoint, as observed in the live Bridge flow. host_path_epoch=(original.host_path_epoch if same_process else 2), producer_generation=2, + connection_mode="bridge", + target_ipv4="192.168.68.52", ) coordinator = service._physical_command_coordinator # noqa: SLF001 coordinator.application_response( diff --git a/tests/test_xgrids_active_acquisition_recovery_checkpoint.py b/tests/test_xgrids_active_acquisition_recovery_checkpoint.py index be3a949..2eb7e4a 100644 --- a/tests/test_xgrids_active_acquisition_recovery_checkpoint.py +++ b/tests/test_xgrids_active_acquisition_recovery_checkpoint.py @@ -537,6 +537,7 @@ def _cease_prepared_resolved_start_standby_kwargs( prepared_binding: ActiveAcquisitionRecoveryTransportBinding, *, terminal_state: str, + origin_kind: str = "composite-resolved", ) -> dict[str, Any]: binding = _reconciled_binding() status = _status( @@ -545,16 +546,21 @@ def _cease_prepared_resolved_start_standby_kwargs( evidence_session_id="evidence-restarted", observed_at="2026-08-13T12:00:03.000Z", ) + composite = origin_kind == "composite-resolved" physical = ActiveAcquisitionRecoveryPhysicalLineageProof( ledger_schema_version=ACTIVE_ACQUISITION_RECOVERY_PHYSICAL_LEDGER_SCHEMA, ledger_revision=21, - proof_id="physical-composite-resolved-standby", + proof_id=f"physical-{origin_kind}-standby", operation_id=START_OPERATION_ID, original_start_operation_id=START_OPERATION_ID, parent_operation_id=None, acquisition_id=ACQUISITION_ID, action="start", - resolution="start-active-observed", + resolution=( + "start-active-observed" + if composite + else "physical-standby-observed" + ), payload_sha256=START_PAYLOAD_SHA256, original_start_payload_sha256=START_PAYLOAD_SHA256, reconciliation_kind="resolved-active-cessation", @@ -562,18 +568,22 @@ def _cease_prepared_resolved_start_standby_kwargs( status_message_sha256=status.status_message_sha256, observed_session_state=status.session_state, binding=binding, - composite_complete=True, + composite_complete=composite, edge_terminal=True, late_start_excluded=True, stop_fence="none", observed_at_utc=status.observed_at_utc, ) origin = _origin( - origin_kind="composite-resolved", + origin_kind=origin_kind, prepared_binding=prepared_binding, physical_proof=physical, ) - assert origin.original_project_id_sha256 is not None + settlement_project_id_sha256 = ( + origin.original_project_id_sha256 + or origin.reconciled_active_project_id_sha256 + ) + assert settlement_project_id_sha256 is not None return { "transition_id": ( f"transition-cease-prepared-resolved-start-{terminal_state}" @@ -590,7 +600,7 @@ def _cease_prepared_resolved_start_standby_kwargs( origin.original_attempt_sha256 ), "reconciliation_original_project_id_sha256": ( - origin.original_project_id_sha256 + settlement_project_id_sha256 ), } @@ -981,6 +991,52 @@ def test_cease_prepared_resolved_start_standby_is_terminal_without_activation( assert restarted.snapshot().checkpoint == ceased +@pytest.mark.parametrize("terminal_state", ("ready", "scan_over")) +def test_cease_prepared_ambiguous_active_start_standby_is_terminal_without_activation( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + terminal_state: str, +) -> None: + """Two read-only proofs close an ambiguous START without inventing STOP/PCL.""" + + prepared_binding = _binding() + store = _store(tmp_path, monkeypatch) + prepared = _prepare(store, prepared_binding) + kwargs = _cease_prepared_resolved_start_standby_kwargs( + prepared_binding, + terminal_state=terminal_state, + origin_kind="ambiguous-reconciled", + ) + + ceased = store.cease_prepared_resolved_start_standby(**kwargs) + + assert ceased.state == "ceased" + assert ceased.revision == prepared.revision + 1 + assert ceased.active_project_id_sha256 == ACTIVE_PROJECT_ID_SHA256 + assert ceased.activated_at_utc is None + assert ceased.activation_status_proof is None + assert ceased.activation_physical_proof is None + assert ceased.current_active_status_proof is None + assert ceased.current_active_physical_proof is None + assert ceased.first_published_pcl_proof is None + assert ceased.reconciled_start_origin_proof is not None + assert ceased.reconciled_start_origin_proof.origin_kind == ( + "ambiguous-reconciled" + ) + assert ceased.reconciled_start_origin_proof.original_project_id_sha256 is None + assert ceased.reconciled_start_origin_proof.reconciled_active_project_id_sha256 == ( + ACTIVE_PROJECT_ID_SHA256 + ) + assert ceased.cessation_status_proof == kwargs["cessation_status_proof"] + assert ceased.cessation_physical_proof == kwargs["cessation_physical_proof"] + assert ( + ActiveAcquisitionRecoveryCheckpointStore(tmp_path / "repository") + .snapshot() + .checkpoint + == ceased + ) + + def test_cease_prepared_resolved_start_standby_rejects_inexact_restart_proofs( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/test_xgrids_camera_gateway.py b/tests/test_xgrids_camera_gateway.py index 137c7fd..53ffa13 100644 --- a/tests/test_xgrids_camera_gateway.py +++ b/tests/test_xgrids_camera_gateway.py @@ -801,6 +801,16 @@ def test_preview_queue_enforces_exact_fragment_byte_and_age_bounds() -> None: assert age_bounded.offer(("media", b"too-old")) is False +def test_preview_queue_survives_measured_ui_pause() -> None: + now = [100.0] + preview_queue = _CameraPreviewSegmentQueue(clock=lambda: now[0]) + + assert preview_queue.offer(("media", b"first")) is True + now[0] += 3.2 + + assert preview_queue.offer(("media", b"next")) is True + + def test_camera_router_closes_lagging_preview_with_explicit_private_code( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/test_xgrids_connection_scenario_reset.py b/tests/test_xgrids_connection_scenario_reset.py index 527a697..281490a 100644 --- a/tests/test_xgrids_connection_scenario_reset.py +++ b/tests/test_xgrids_connection_scenario_reset.py @@ -470,6 +470,63 @@ def test_reset_attempt_cutoff_uses_operation_identity_not_local_stage_sequence( assert projected["status"] == "failed" +def test_quick_connect_unknown_write_can_reset_to_bridge_without_device_restart( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + service = _service(monkeypatch, tmp_path) + operation_id = "op-00000000-0000-4000-8000-000000000103" + journal = service._require_network_provisioning_idempotency_journal() # noqa: SLF001 + admitted = journal.begin( + idempotency_key="network-provision:quick-to-bridge-reset", + action=facade_module.ACTION_NETWORK_PROVISION, + operation_id=operation_id, + request_binding_sha256=hashlib.sha256(b"quick-to-bridge-reset").hexdigest(), + ) + journal.mark_unresolved( + operation_id, + expected_revision=admitted.record.revision, + ) + prepared = service._network_mutation_ledger.prepare( # noqa: SLF001 + operation_id=operation_id, + transport_ref="RESET-QUICK-K1-UUID", + intended_mode="quick-connect", + write_mode="with_response", + baseline_status=NetworkStatusEvidence( + mode="WIFI_AP", + ipv4="192.168.56.1", + status_code=1, + reserved=0, + ), + ) + service._network_mutation_ledger.mark_dispatching( # noqa: SLF001 + operation_id, + expected_revision=prepared.revision, + ) + + reset = service.select_connection_mode( + _reset( + mode="bridge", + revision=0, + reset_id="op-reset-quick-to-bridge-without-device-restart-01", + ) + ) + + assert reset["desired_connection_mode"] == "bridge" + assert reset["desired_connection_mode_revision"] == 1 + assert reset["connection_scenario_reset"]["network_write_performed"] is False + network_record = service._network_mutation_ledger.snapshot().record # noqa: SLF001 + assert network_record is not None + assert network_record.resolution == "superseded" + terminal_record = journal.snapshot().records[-1] + assert terminal_record.operation_id == operation_id + assert terminal_record.stage == "terminal" + assert terminal_record.terminal is not None + assert terminal_record.terminal.outcome == "cancelled" + assert terminal_record.terminal.side_effect_status == "unknown" + assert terminal_record.terminal.safe_to_retry is False + + def test_scenario_reset_preserves_physical_audit_and_unrelated_plugin_state( monkeypatch: pytest.MonkeyPatch, tmp_path: Path,