From eaad9deda183cc4f064222066bde33ca5ab90040 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Sat, 22 Aug 2026 00:03:01 +0300 Subject: [PATCH] fix(k1): stabilize live recovery and media admission --- .../src/components/MseFmp4WebSocketPlayer.tsx | 13 +- .../src/components/RerunViewport.tsx | 31 +- .../core/observation/liveCameraRecovery.ts | 4 +- .../core/observation/liveReceiverWatchdog.ts | 24 +- .../core/observation/useObservationLayout.ts | 51 +- .../src/workspaces/Workspaces.tsx | 7 +- .../test/devicePluginContracts.test.mjs | 47 ++ .../test/liveReceiverWatchdog.test.mjs | 15 +- .../test/observationSources.test.mjs | 18 +- .../rerunViewportAtomicAdmission.test.mjs | 19 + .../test/workspaceLayout.test.mjs | 63 +++ plugins/xgrids-k1/frontend/src/api.ts | 16 +- .../ActiveStreamRecoverySurface.tsx | 10 +- plugins/xgrids-k1/frontend/src/styles.css | 80 ++- .../frontend/src/useXgridsK1Runtime.ts | 2 +- scripts/build_m4_worker_shadow_artifact.py | 2 +- .../xgrids_k1/connection_supervisor.py | 16 +- src/k1link/device_plugins/xgrids_k1/facade.py | 460 +++++++++++++++- .../xgrids_k1/viewer/runtime.py | 28 +- src/k1link/web/app.py | 70 ++- tests/test_canonical_pipeline.py | 63 +++ tests/test_connection_supervisor.py | 73 ++- tests/test_plugin_runtime.py | 52 ++ tests/test_session_catalog_lifecycle.py | 27 + tests/test_web_validation_security.py | 37 ++ tests/test_xgrids_acquisition_lifecycle.py | 504 +++++++++++++++++- ...tive_acquisition_checkpoint_integration.py | 101 ++++ ..._active_acquisition_restart_rehydration.py | 6 +- tests/test_xgrids_application_session.py | 5 +- 29 files changed, 1645 insertions(+), 199 deletions(-) diff --git a/apps/control-station/src/components/MseFmp4WebSocketPlayer.tsx b/apps/control-station/src/components/MseFmp4WebSocketPlayer.tsx index ef69bb6..d75cfd2 100644 --- a/apps/control-station/src/components/MseFmp4WebSocketPlayer.tsx +++ b/apps/control-station/src/components/MseFmp4WebSocketPlayer.tsx @@ -41,7 +41,9 @@ export interface CameraStartupWatchdog { * Supervise only the disposable browser transport. One fixed first-media * deadline covers both an MSE that never opens and an open-but-silent socket. * The first non-empty fragment then starts a separate first-playable-frame - * deadline; later fragments deliberately do not extend it. + * deadline. Every subsequent fragment extends that deadline: a browser that is + * still receiving the authoritative stream must not destroy its partial MSE + * decode solely because the main thread or decoder needed longer to start. */ export function createCameraStartupWatchdog({ schedule, @@ -83,7 +85,7 @@ export function createCameraStartupWatchdog({ arm("first-media", firstMediaTimeoutMs); }, markMediaReceived() { - if (mediaReceived || playing) return; + if (playing) return; mediaReceived = true; arm("first-playable-frame", firstPlayableFrameTimeoutMs); }, @@ -189,6 +191,7 @@ export function MseFmp4WebSocketPlayer({ const leaseRetryRef = useRef(resetCameraLeaseRetryBudget(delivery.id)); const activeAuthorityRef = useRef(recoveryAuthorityIdentity); const recoveryPendingAuthorityRef = useRef(null); + const transportHealthyRef = useRef(false); const transportEpochRef = useRef(0); const transportAuthorityRef = useRef(recoveryAuthorityIdentity); const activeTransportDisposeRef = useRef<(() => void) | null>(null); @@ -243,6 +246,7 @@ export function MseFmp4WebSocketPlayer({ let receivedMedia = false; let failed = false; let startupWatchdog: CameraStartupWatchdog | null = null; + transportHealthyRef.current = false; const recovering = Boolean( activeAuthorityRef.current && recoveryPendingAuthorityRef.current === activeAuthorityRef.current, @@ -254,6 +258,7 @@ export function MseFmp4WebSocketPlayer({ if (!transportIsCurrent() || failed) return; startupWatchdog?.clear(); failed = true; + transportHealthyRef.current = false; recoveryPendingAuthorityRef.current = null; setStatus("error"); setMessage(copy); @@ -280,6 +285,7 @@ export function MseFmp4WebSocketPlayer({ } startupWatchdog?.clear(); failed = true; + transportHealthyRef.current = false; queue.length = 0; queuedBytes = 0; const retry = consumeCameraLeaseRetry(leaseRetryRef.current, delivery.id); @@ -320,6 +326,7 @@ export function MseFmp4WebSocketPlayer({ const onPlaying = () => { if (!transportIsCurrent() || failed) return; startupWatchdog?.markPlaying(); + transportHealthyRef.current = true; leaseRetryRef.current = resetCameraLeaseRetryBudget(delivery.id); recoveryPendingAuthorityRef.current = null; setStatus("playing"); @@ -464,6 +471,7 @@ export function MseFmp4WebSocketPlayer({ const disposeTransport = () => { if (disposed) return; disposed = true; + transportHealthyRef.current = false; startupWatchdog?.clear(); if (retryTimer !== undefined) window.clearTimeout(retryTimer); queue.length = 0; @@ -518,6 +526,7 @@ export function MseFmp4WebSocketPlayer({ now: Date.now(), documentVisible: document.visibilityState === "visible", networkOnline: navigator.onLine !== false, + transportHealthy: transportHealthyRef.current, }); recovery = decision.state; if (!decision.reopen) return; diff --git a/apps/control-station/src/components/RerunViewport.tsx b/apps/control-station/src/components/RerunViewport.tsx index 6e7729e..ffbe951 100644 --- a/apps/control-station/src/components/RerunViewport.tsx +++ b/apps/control-station/src/components/RerunViewport.tsx @@ -8,7 +8,7 @@ import { initialLiveReceiverRecoveryState, initialLiveReceiverWatchdogState, LIVE_RECEIVER_MAX_RECOVERY_ATTEMPTS, - LIVE_RECEIVER_OPEN_MAX_AGE_MS, + LIVE_RECEIVER_OPEN_CHECK_INTERVAL_MS, requestLiveReceiverRecovery, } from "../core/observation/liveReceiverWatchdog"; import { @@ -1091,25 +1091,6 @@ export function RerunViewport({ }; const clearRecordedAdmissionWatchdog = () => recordedOpenWatchdog?.clear(); const clearLiveRecordingOpenTimer = diagnosticLifecycle.clearAdmissionTimeout; - function refreshOpeningLiveReceiver(openForMs: number) { - if (disposed || recordingOpened) return; - diagnosticLifecycle.post({ - eventCode: "live_receiver_restart_requested", - failureStage: "recording-open-timeout", - streamId: liveStreamIdRef.current, - backendActivitySequence: liveActivitySequenceRef.current, - viewerRangeMaxNs: latestLiveRangeMaxNs, - stalledForMs: Math.round(openForMs), - recoveryAttempt: liveRecoveryRef.current.attempts || null, - }); - setStatus("loading"); - onStatusChange?.( - "loading", - "Живой визуализатор обновляет приёмник продолжающегося потока.", - ); - disposeViewer?.(); - setRetryNonce((nonce) => nonce + 1); - } const armLiveRecordingOpenTimer = () => { clearLiveRecordingOpenTimer(); diagnosticLifecycle.armAdmissionTimeout(() => { @@ -1129,17 +1110,9 @@ export function RerunViewport({ armLiveRecordingOpenTimer(); return; } - if (observed.signal === "refresh-receiver") { - // Publication is healthy, so this is presentation-only maintenance: - // refresh the aged native receiver without spending (or clearing) - // recovery debt. - recordingOpenTimedOut = true; - refreshOpeningLiveReceiver(observed.openForMs); - return; - } recordingOpenTimedOut = true; requestLiveRecovery("recording-open-timeout"); - }, LIVE_RECEIVER_OPEN_MAX_AGE_MS); + }, LIVE_RECEIVER_OPEN_CHECK_INTERVAL_MS); }; const clearLiveRecordingDiscoveryTimer = diagnosticLifecycle.clearAdmissionInterval; const clearLiveRecoveryRetryTimer = () => { diff --git a/apps/control-station/src/core/observation/liveCameraRecovery.ts b/apps/control-station/src/core/observation/liveCameraRecovery.ts index 7f3988a..f02287f 100644 --- a/apps/control-station/src/core/observation/liveCameraRecovery.ts +++ b/apps/control-station/src/core/observation/liveCameraRecovery.ts @@ -26,6 +26,7 @@ export interface CameraPlaybackRecoveryContext { now: number; documentVisible: boolean; networkOnline: boolean; + transportHealthy?: boolean; } export interface CameraPlaybackRecoveryDecision { @@ -186,7 +187,8 @@ export function reduceCameraPlaybackRecovery( } else if (event.type === "page-restore") { candidate = event.persisted; } else if (event.type === "heartbeat") { - candidate = now - current.lastObservedAt >= CAMERA_WAKE_GAP_MS; + candidate = context.transportHealthy !== true + && now - current.lastObservedAt >= CAMERA_WAKE_GAP_MS; } const outsideCooldown = current.lastReopenAt === null diff --git a/apps/control-station/src/core/observation/liveReceiverWatchdog.ts b/apps/control-station/src/core/observation/liveReceiverWatchdog.ts index 9cb045e..8ba8dbb 100644 --- a/apps/control-station/src/core/observation/liveReceiverWatchdog.ts +++ b/apps/control-station/src/core/observation/liveReceiverWatchdog.ts @@ -31,7 +31,6 @@ export interface LiveReceiverOpenWatchdogState { export type LiveReceiverOpenWatchdogSignal = | "wait-for-store" - | "refresh-receiver" | "restart-receiver"; export interface LiveReceiverOpenWatchdogResult { @@ -60,10 +59,11 @@ export interface LiveReceiverRecoveryRequest { export const LIVE_RECEIVER_STALL_THRESHOLD_MS = 5_000; export const LIVE_RECEIVER_MAX_RECOVERY_ATTEMPTS = 3; // The bridge publishes its URL only after StoreInfo, blueprint and static -// scene data have been flushed. A receiver that still has not admitted that -// store after one operator-visible four-second window is wedged, not merely -// slow; keeping it for 48 seconds made a healthy live scan look blank. -export const LIVE_RECEIVER_OPEN_MAX_AGE_MS = 4_000; +// scene data have been flushed. Poll at the original four-second boundary, +// but never discard a receiver while this exact acquisition is still proving +// fresh publication progress: doing so throws away its partial store replay and +// can keep a healthy long-running stream blank indefinitely. +export const LIVE_RECEIVER_OPEN_CHECK_INTERVAL_MS = 4_000; const LIVE_RECEIVER_RECOVERY_DELAYS_MS = [400, 1_000, 2_000, 5_000] as const; export function liveReceiverRecoveryRetryDelay(attempt: number): number { @@ -216,11 +216,10 @@ export function initialLiveReceiverOpenWatchdogState( /** * Keep one still-opening Rerun receiver alive while the backend is proving * fresh publication progress. Recreating the WASM receiver on a fixed timer - * can repeatedly discard an otherwise healthy late StoreInfo replay. Rolling - * patience is nevertheless bounded: a receiver that has not admitted a store - * by the absolute open-age limit is refreshed without consuming the recovery - * budget. A true lack of backend progress delegates to the bounded restart - * policy. Recovery debt is cleared only after viewer admission, never merely + * can repeatedly discard an otherwise healthy late StoreInfo replay. Preserve + * that receiver for as long as the backend proves fresh publication. A true + * lack of backend progress delegates to the bounded restart policy on the next + * check. Recovery debt is cleared only after viewer admission, never merely * because the backend counter advanced. */ export function advanceLiveReceiverOpenWatchdog( @@ -228,7 +227,6 @@ export function advanceLiveReceiverOpenWatchdog( recoveryState: LiveReceiverRecoveryState, backendActivitySequence: number | null, nowMs = Date.now(), - maxOpenAgeMs = LIVE_RECEIVER_OPEN_MAX_AGE_MS, ): LiveReceiverOpenWatchdogResult { const sequence = validBackendActivitySequence(backendActivitySequence); const previous = current.lastBackendActivitySequence; @@ -247,9 +245,7 @@ export function advanceLiveReceiverOpenWatchdog( return { state, recoveryState, - signal: openForMs >= maxOpenAgeMs - ? "refresh-receiver" - : "wait-for-store", + signal: "wait-for-store", openForMs, }; } diff --git a/apps/control-station/src/core/observation/useObservationLayout.ts b/apps/control-station/src/core/observation/useObservationLayout.ts index d527bfa..460a7bd 100644 --- a/apps/control-station/src/core/observation/useObservationLayout.ts +++ b/apps/control-station/src/core/observation/useObservationLayout.ts @@ -129,6 +129,26 @@ export function admitLiveDefaultPresentations( }; } +/** + * A restored workspace may hide cameras that the device plugin has not selected. + * It must still admit the exact selected delivery in a fresh browser document, + * or a replacement delivery for a camera already presented in this acquisition. + * An explicit close remains a separate acquisition-scoped fence in + * `admitLiveDefaultPresentations` and de-selects the source in the plugin first. + */ +export function restoredLayoutMayAdmitLiveDefault( + sources: readonly ObservationSourceDescriptor[], + presentedAcquisitionSources: ReadonlySet, +): boolean { + const selectedSources = sources.filter(automaticLivePresentationIdentity); + if (selectedSources.length === 0) return false; + if (presentedAcquisitionSources.size === 0) return true; + return selectedSources.some((source) => { + const lineage = livePresentationCloseFence(source); + return Boolean(lineage && presentedAcquisitionSources.has(lineage)); + }); +} + function catalogIdentity(sources: readonly ObservationSourceDescriptor[]): string { return sources .map((source) => [ @@ -170,6 +190,7 @@ export function useObservationLayout( const restoredLayoutAuthorityRef = useRef(false); const admittedLivePresentationIdentitiesRef = useRef(new Set()); const closedLiveAcquisitionSourcesRef = useRef(new Set()); + const presentedLiveAcquisitionSourcesRef = useRef(new Set()); const initializedCatalog = useRef(null); const sourceIdList = sources.map((source) => source.id).sort(); const sourceIdsIdentity = sourceIdList.join("\u0000"); @@ -182,6 +203,12 @@ export function useObservationLayout( const commitVisibleIds = useCallback((next: readonly string[]) => { const unique = [...new Set(next)]; + const visible = new Set(unique); + for (const source of sourcesRef.current) { + if (!visible.has(source.id) || !automaticLivePresentationIdentity(source)) continue; + const lineage = livePresentationCloseFence(source); + if (lineage) presentedLiveAcquisitionSourcesRef.current.add(lineage); + } visibleIdsRef.current = unique; setVisibleIds(unique); }, []); @@ -352,11 +379,18 @@ export function useObservationLayout( .filter((candidate): candidate is string => candidate !== null) .sort()); - useEffect(() => { - if (restoredLayoutAuthorityRef.current) return; + const admitAutomaticLivePresentations = useCallback(() => { + const currentSources = sourcesRef.current; + if ( + restoredLayoutAuthorityRef.current + && !restoredLayoutMayAdmitLiveDefault( + currentSources, + presentedLiveAcquisitionSourcesRef.current, + ) + ) return; const admission = admitLiveDefaultPresentations( visibleIdsRef.current, - sources, + currentSources, admittedLivePresentationIdentitiesRef.current, closedLiveAcquisitionSourcesRef.current, ); @@ -367,7 +401,11 @@ export function useObservationLayout( commitVisibleIds(admission.visibleIds); clearPresentation(admission.removedIds, false); persistLiveLayout(); - }, [clearPresentation, commitVisibleIds, persistLiveLayout, selectedDeliveryIdentity]); + }, [clearPresentation, commitVisibleIds, persistLiveLayout]); + + useEffect(() => { + admitAutomaticLivePresentations(); + }, [admitAutomaticLivePresentations, selectedDeliveryIdentity]); const markPending = useCallback((source: ObservationSourceDescriptor, pending: boolean) => { const groupId = source.activation?.groupId; @@ -545,7 +583,10 @@ export function useObservationLayout( }); restoredLayoutAuthorityRef.current = true; applyDesiredSnapshot(desiredSnapshotRef.current, "reset"); - }, [applyDesiredSnapshot]); + // The selected delivery may already have arrived before the saved layout. + // Re-run admission here so restore ordering cannot strand its camera window. + admitAutomaticLivePresentations(); + }, [admitAutomaticLivePresentations, applyDesiredSnapshot]); const visibleSourceIds = useMemo(() => new Set(visibleIds), [visibleIds]); const pendingSourceIds = useMemo(() => new Set(pendingIds), [pendingIds]); diff --git a/apps/control-station/src/workspaces/Workspaces.tsx b/apps/control-station/src/workspaces/Workspaces.tsx index e779c59..dd4d870 100644 --- a/apps/control-station/src/workspaces/Workspaces.tsx +++ b/apps/control-station/src/workspaces/Workspaces.tsx @@ -235,12 +235,7 @@ function SpatialWorkspace({ Boolean(observationLayout.maximizedFloatingSourceId); const timeline = state?.observationTimeline; const viewportRef = useRef(null); - const intentionalSourceEnd = !recordedSource && [ - "awaiting_external_stop", - "stopping", - "finalizing", - "completed", - ].includes(state?.acquisition?.state ?? ""); + const intentionalSourceEnd = !recordedSource && state?.sourceMode === "idle" && ["awaiting_external_stop", "stopping", "finalizing", "completed",].includes(state?.acquisition?.state ?? ""); const presentedViewerStatus = intentionalSourceEnd ? "idle" : rerunPresentationStatus( diff --git a/apps/control-station/test/devicePluginContracts.test.mjs b/apps/control-station/test/devicePluginContracts.test.mjs index 0e15f60..042466a 100644 --- a/apps/control-station/test/devicePluginContracts.test.mjs +++ b/apps/control-station/test/devicePluginContracts.test.mjs @@ -620,6 +620,53 @@ test("canonical K1 preparation stops before START and guards every async stage", ); }); +test("K1 control-phase polling fails closed instead of stacking timed-out state reads", async () => { + const hookSource = await readFile( + new URL( + "../../../plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts", + import.meta.url, + ), + "utf8", + ); + const controlPollingLoop = hookSource.slice( + hookSource.indexOf("async function waitForControlPhase"), + hookSource.indexOf("async function waitForPhysicalReconciliationProof"), + ); + + assert.match(hookSource, /const CONTROL_PHASE_WAIT_TIMEOUT_MS = 120_000/); + assert.match(controlPollingLoop, /xgridsK1Api\.getState\(\)/); + assert.doesNotMatch(controlPollingLoop, /ApiRequestTimeoutError/); + assert.doesNotMatch(controlPollingLoop, /catch \(error\)/); + assert.doesNotMatch( + controlPollingLoop, + /(?:openApplicationControlSession|enterApplicationWorkspace|prepareAcquisition|startAcquisition|stopAcquisition)\(/, + ); +}); + +test("K1 frontend coalesces concurrent read-only state requests without caching them", async () => { + const apiSource = await readFile( + new URL( + "../../../plugins/xgrids-k1/frontend/src/api.ts", + import.meta.url, + ), + "utf8", + ); + const singleFlight = apiSource.slice( + apiSource.indexOf("let stateReadInFlight"), + apiSource.indexOf("export const xgridsK1Api"), + ); + const getState = apiSource.slice( + apiSource.indexOf("async getState"), + apiSource.indexOf("scanBle("), + ); + + assert.match(singleFlight, /if \(stateReadInFlight\) return stateReadInFlight/); + assert.match(singleFlight, /invokeState\(xgridsK1Actions\.stateRead\)/); + assert.match(singleFlight, /stateReadInFlight === request[\s\S]*?stateReadInFlight = null/); + assert.match(getState, /return readStateSingleFlight\(\)/); + assert.doesNotMatch(singleFlight, /setTimeout|setInterval|retry/i); +}); + test("K1 control-session CAS is exact, integer-only and mapped for each mutation family", () => { const state = { application_control_session: { diff --git a/apps/control-station/test/liveReceiverWatchdog.test.mjs b/apps/control-station/test/liveReceiverWatchdog.test.mjs index d13a14f..837cf70 100644 --- a/apps/control-station/test/liveReceiverWatchdog.test.mjs +++ b/apps/control-station/test/liveReceiverWatchdog.test.mjs @@ -244,13 +244,16 @@ test("connecting Rerun authority requires an exact recovery generation lease", ( ); }); -test("opening receiver gets bounded rolling patience while backend publication advances", () => { +test("opening receiver preserves partial store replay while backend publication advances", () => { let openState = initialLiveReceiverOpenWatchdogState(0, 0); let recoveryState = initialLiveReceiverRecoveryState(); const samples = [ [134, 3_999, "wait-for-store"], - [266, 4_000, "refresh-receiver"], + [266, 4_000, "wait-for-store"], + [380, 8_000, "wait-for-store"], + [486, 12_000, "wait-for-store"], + [700, 120_000, "wait-for-store"], ]; for (const [backendActivitySequence, nowMs, expectedSignal] of samples) { const observed = advanceLiveReceiverOpenWatchdog( @@ -302,17 +305,17 @@ test("fresh backend progress preserves earlier restart debt until viewer admissi }); }); -test("aged active receiver refresh does not spend or erase restart debt", () => { +test("aged active receiver remains intact and preserves restart debt", () => { const openState = initialLiveReceiverOpenWatchdogState(486, 0); const consumedRestart = requestLiveReceiverRecovery(initialLiveReceiverRecoveryState()); const observed = advanceLiveReceiverOpenWatchdog( openState, consumedRestart.state, 900, - 4_000, + 12_000, ); - assert.equal(observed.signal, "refresh-receiver"); + assert.equal(observed.signal, "wait-for-store"); assert.deepEqual(observed.recoveryState, consumedRestart.state); - assert.equal(observed.openForMs, 4_000); + assert.equal(observed.openForMs, 12_000); }); diff --git a/apps/control-station/test/observationSources.test.mjs b/apps/control-station/test/observationSources.test.mjs index c0f39d2..cbd07bb 100644 --- a/apps/control-station/test/observationSources.test.mjs +++ b/apps/control-station/test/observationSources.test.mjs @@ -1199,7 +1199,7 @@ test("camera startup watchdog replaces an open-but-silent MSE or WebSocket", () ); }); -test("first media starts one non-sliding first-playable-frame deadline", () => { +test("continuing media extends the first-playable-frame deadline", () => { const scheduled = new Map(); const cancelled = []; const timeouts = []; @@ -1228,8 +1228,14 @@ test("first media starts one non-sliding first-playable-frame deadline", () => { assert.equal(scheduled.get(playableHandle).timeoutMs, CAMERA_FIRST_PLAYABLE_FRAME_TIMEOUT_MS); watchdog.markMediaReceived(); - assert.equal(nextHandle - 1, playableHandle, "later fragments must not extend the deadline"); - scheduled.get(playableHandle).callback(); + const extendedPlayableHandle = nextHandle - 1; + assert.notEqual(extendedPlayableHandle, playableHandle); + assert.deepEqual(cancelled, [10, playableHandle]); + assert.equal( + scheduled.get(extendedPlayableHandle).timeoutMs, + CAMERA_FIRST_PLAYABLE_FRAME_TIMEOUT_MS, + ); + scheduled.get(extendedPlayableHandle).callback(); assert.deepEqual(timeouts, ["first-playable-frame"]); assert.match( cameraStartupWatchdogRecoveryMessage("first-playable-frame"), @@ -1306,6 +1312,7 @@ test("a laptop sleep gap reopens only the exact authoritative live camera transp now: 7_000, documentVisible: true, networkOnline: true, + transportHealthy: false, }); assert.equal(wake.reopen, true); assert.equal(wake.state.lastReopenAt, 7_000); @@ -1369,15 +1376,16 @@ test("visibility, pageshow and online wake burst owns one replacement decoder", assert.equal(secondWake.state.lastReopenAt, 18_000); }); -test("healthy visible camera heartbeats do not churn its WebSocket or decoder", () => { +test("healthy camera survives a long main-thread heartbeat gap without decoder churn", () => { const authority = "camera-authority-acquisition-healthy"; let state = initialCameraPlaybackRecoveryState(authority, 1_000); - for (const now of [2_000, 3_000, 4_000, 5_000]) { + for (const now of [2_000, 3_000, 4_000, 12_000, 25_000]) { const heartbeat = reduceCameraPlaybackRecovery(state, { type: "heartbeat" }, { activeAuthorityIdentity: authority, now, documentVisible: true, networkOnline: true, + transportHealthy: true, }); assert.equal(heartbeat.reopen, false); state = heartbeat.state; diff --git a/apps/control-station/test/rerunViewportAtomicAdmission.test.mjs b/apps/control-station/test/rerunViewportAtomicAdmission.test.mjs index 91bd782..75e330c 100644 --- a/apps/control-station/test/rerunViewportAtomicAdmission.test.mjs +++ b/apps/control-station/test/rerunViewportAtomicAdmission.test.mjs @@ -295,6 +295,25 @@ test("raw replay exercises the same streaming receiver lifecycle as a live scan" ); }); +test("pending K1 STOP keeps the live Rerun source mounted until local capture ends", async () => { + const source = await readFile( + new URL("../src/workspaces/Workspaces.tsx", import.meta.url), + "utf8", + ); + assert.match( + source, + /const intentionalSourceEnd = !recordedSource &&\s*state\?\.sourceMode === "idle" && \[/, + ); + assert.match( + source, + /"awaiting_external_stop",\s*"stopping",\s*"finalizing",\s*"completed",/, + ); + assert.doesNotMatch( + source, + /const intentionalSourceEnd = !recordedSource && \[\s*"awaiting_external_stop"/, + ); +}); + test("the spatial header reports raw replay as an active source", async () => { const source = await readFile( new URL("../src/App.tsx", import.meta.url), diff --git a/apps/control-station/test/workspaceLayout.test.mjs b/apps/control-station/test/workspaceLayout.test.mjs index 1341afa..64a7bf3 100644 --- a/apps/control-station/test/workspaceLayout.test.mjs +++ b/apps/control-station/test/workspaceLayout.test.mjs @@ -16,6 +16,7 @@ let WorkspaceLayoutContractError; let admitLiveDefaultPresentations; let automaticLivePresentationIdentity; let livePresentationCloseFence; +let restoredLayoutMayAdmitLiveDefault; let observationPresentationSourceAfterLayoutApply; let visibleSourceIdsAfterRecordedCatalogActivation; @@ -39,6 +40,7 @@ before(async () => { admitLiveDefaultPresentations, automaticLivePresentationIdentity, livePresentationCloseFence, + restoredLayoutMayAdmitLiveDefault, observationPresentationSourceAfterLayoutApply, visibleSourceIdsAfterRecordedCatalogActivation, } = await server.ssrLoadModule("/src/core/observation/useObservationLayout.ts")); @@ -308,6 +310,67 @@ test("a sequential live acquisition re-arms the same camera without reopening a ]); }); +test("a restored layout admits a selected delivery after reload and only same-acquisition successors", () => { + const camera = (acquisitionId, deliveryId) => ({ + id: "k1:sensor.camera.right", + sourceId: "sensor.camera.right", + modality: "video", + availability: "streaming", + transport: "websocket", + previewUrl: null, + delivery: { + id: deliveryId, + kind: "mse-fmp4-websocket", + url: `/camera-preview/${deliveryId}`, + mediaType: 'video/mp4; codecs="avc1.641028"', + }, + activation: { + groupId: "k1:device-session-reused:camera.preview.decoder", + maxActive: 1, + selected: true, + controllable: true, + }, + binding: { + deviceId: "k1-a", + deviceSessionId: "device-session-reused", + acquisitionId, + }, + capabilities: { defaultVisible: true, overlay: true }, + }); + const original = camera("acquisition-a", "camera-preview-2"); + const successor = camera("acquisition-a", "camera-preview-3"); + const unrelated = camera("acquisition-b", "camera-preview-3"); + const presented = new Set([livePresentationCloseFence(original)]); + + assert.equal(restoredLayoutMayAdmitLiveDefault([original], new Set()), true); + assert.equal(restoredLayoutMayAdmitLiveDefault([successor], presented), true); + assert.equal(restoredLayoutMayAdmitLiveDefault([unrelated], presented), false); + assert.equal(restoredLayoutMayAdmitLiveDefault([{ + ...original, + activation: { ...original.activation, selected: false }, + }], new Set()), false); + + const admitted = admitLiveDefaultPresentations( + [], + [successor], + new Set([automaticLivePresentationIdentity(original)]), + new Set(), + ); + assert.deepEqual(admitted.visibleIds, [successor.id]); + assert.deepEqual(admitted.admittedIdentities, [ + automaticLivePresentationIdentity(successor), + ]); + + const deliberatelyClosed = admitLiveDefaultPresentations( + [], + [successor], + new Set([automaticLivePresentationIdentity(original)]), + presented, + ); + assert.deepEqual(deliberatelyClosed.visibleIds, []); + assert.deepEqual(deliberatelyClosed.admittedIdentities, []); +}); + test("workspace layout API uses the canonical endpoint and optimistic revision", async () => { const calls = []; const current = decodeObservationWorkspaceLayoutProfile(wireProfile()); diff --git a/plugins/xgrids-k1/frontend/src/api.ts b/plugins/xgrids-k1/frontend/src/api.ts index 96a55e8..6ad1963 100644 --- a/plugins/xgrids-k1/frontend/src/api.ts +++ b/plugins/xgrids-k1/frontend/src/api.ts @@ -1875,6 +1875,20 @@ function invokeState(actionId: string, input: object = {}): Promise | null = null; + +function readStateSingleFlight(): Promise { + if (stateReadInFlight) return stateReadInFlight; + + const request = invokeState(xgridsK1Actions.stateRead); + stateReadInFlight = request; + const release = () => { + if (stateReadInFlight === request) stateReadInFlight = null; + }; + void request.then(release, release); + return request; +} + export const xgridsK1Api = { async getHealth(): Promise { const payload = await requestJson( @@ -1890,7 +1904,7 @@ export const xgridsK1Api = { }, async getState(): Promise { - return invokeState(xgridsK1Actions.stateRead); + return readStateSingleFlight(); }, scanBle( diff --git a/plugins/xgrids-k1/frontend/src/components/ActiveStreamRecoverySurface.tsx b/plugins/xgrids-k1/frontend/src/components/ActiveStreamRecoverySurface.tsx index 16f1d95..01e3fcb 100644 --- a/plugins/xgrids-k1/frontend/src/components/ActiveStreamRecoverySurface.tsx +++ b/plugins/xgrids-k1/frontend/src/components/ActiveStreamRecoverySurface.tsx @@ -67,10 +67,12 @@ function RecoveryState({ {copy.showSpinner ? : null}
{copy.title} - {copy.detail} - {presentation?.progressLabel ? ( - {presentation.progressLabel} - ) : null} + + {copy.detail} + {presentation?.progressLabel ? ( + {presentation.progressLabel} + ) : null} +
); diff --git a/plugins/xgrids-k1/frontend/src/styles.css b/plugins/xgrids-k1/frontend/src/styles.css index ca037df..c0bc9da 100644 --- a/plugins/xgrids-k1/frontend/src/styles.css +++ b/plugins/xgrids-k1/frontend/src/styles.css @@ -599,6 +599,12 @@ box-sizing: border-box; gap: 0.28rem; } +.active-stream-recovery__detail { + display: grid; + min-width: 0; + gap: 0.28rem; +} + .active-stream-recovery__copy strong { color: var(--nodedc-text-primary); font-size: 0.72rem; @@ -1023,25 +1029,83 @@ box-sizing: border-box; min-width: 0; grid-template-columns: minmax(0, 1fr) auto; align-items: center; - gap: 0.65rem; + column-gap: 0.75rem; margin-top: 0; } .active-stream-recovery--compact .active-stream-recovery__state { + grid-column: 1; + grid-row: 1; min-width: 0; + align-items: center; + gap: 0.65rem; background: transparent; padding: 0; } +.active-stream-recovery--compact + .active-stream-recovery__state + > .nodedc-activity-indicator { + margin-top: 0; +} + +.active-stream-recovery--compact .active-stream-recovery__copy { + display: grid; + min-width: 0; + justify-items: center; + gap: 0.1rem; + overflow: hidden; + text-align: center; +} + +.active-stream-recovery--compact .active-stream-recovery__copy strong { + max-width: 100%; + overflow: hidden; + font-size: 0.62rem; + font-weight: 600; + line-height: 1.35; + text-overflow: ellipsis; + white-space: nowrap; +} + +.active-stream-recovery--compact .active-stream-recovery__detail { + display: flex; + min-width: 0; + max-width: 100%; + align-items: baseline; + justify-content: center; + gap: 0.4rem; + overflow: hidden; +} + +.active-stream-recovery--compact .active-stream-recovery__detail > span { + min-width: 0; + overflow: hidden; + font-size: 0.56rem; + line-height: 1.35; + text-overflow: ellipsis; + white-space: nowrap; +} + +.active-stream-recovery--compact .active-stream-recovery__copy small { + flex: 0 0 auto; + font-size: 0.54rem; + line-height: 1.35; + white-space: nowrap; +} + .active-stream-recovery--compact .active-stream-recovery__actions { - max-width: 15rem; + grid-column: 2; + grid-row: 1; + align-self: start; + max-width: none; grid-template-columns: auto; + justify-items: end; gap: 0.25rem; } .active-stream-recovery--compact .active-stream-recovery__actions p { - font-size: 0.5rem; - line-height: 1.35; + display: none; } .xgrids-k1-spatial-controls__phase { @@ -1155,14 +1219,10 @@ box-sizing: border-box; } .active-stream-recovery--compact { - grid-template-columns: minmax(0, 1fr); + grid-template-columns: minmax(0, 1fr) auto; } .active-stream-recovery--compact .active-stream-recovery__actions { - max-width: none; - } - - .active-stream-recovery--compact .active-stream-recovery__actions p { - display: none; + grid-column: 2; } } diff --git a/plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts b/plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts index f1c4b9e..a53a600 100644 --- a/plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts +++ b/plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts @@ -248,7 +248,7 @@ export function connectionActionAuthoritySnapshot( } const CONTROL_STATE_READ_INTERVAL_MS = 250; -const CONTROL_PHASE_WAIT_TIMEOUT_MS = 35_000; +const CONTROL_PHASE_WAIT_TIMEOUT_MS = 120_000; function controlPhase(state: XgridsK1State): XgridsApplicationControlPhase { return state.application_control_session?.state ?? "idle"; diff --git a/scripts/build_m4_worker_shadow_artifact.py b/scripts/build_m4_worker_shadow_artifact.py index 2c85543..8c0288c 100644 --- a/scripts/build_m4_worker_shadow_artifact.py +++ b/scripts/build_m4_worker_shadow_artifact.py @@ -25,7 +25,7 @@ WHEEL_NAME = "nodedc_mission_core-0.1.0-py3-none-any.whl" RUNNER_NAME = RUNNER.name PATCH_ID = re.compile(r"^[A-Za-z0-9._-]{1,96}$") EXPECTED_BASELINE_SHA256 = "ea10359339e6cce31b5780a2710299771cab7cc0c1c2a2b56a1621f786b31fa8" -EXPECTED_WHEEL_SHA256 = "9a60efa68eadf2267fffe3dbb89fb58d5e69e672fa3cdf7529474c9f47416acb" +EXPECTED_WHEEL_SHA256 = "411e1a040b708a118827dfd6898ef1f9911e42a86fa58b9e476688cfa046fce1" PAYLOAD_FILES = ( RUNNER_NAME, WHEEL_NAME, diff --git a/src/k1link/device_plugins/xgrids_k1/connection_supervisor.py b/src/k1link/device_plugins/xgrids_k1/connection_supervisor.py index f5e21f8..335829b 100644 --- a/src/k1link/device_plugins/xgrids_k1/connection_supervisor.py +++ b/src/k1link/device_plugins/xgrids_k1/connection_supervisor.py @@ -1868,9 +1868,10 @@ class ReadOnlyConnectionMonitor: return None previous_path = baseline.host_path intent = baseline.intent - exact_verified_control = bool( - intent is not None - and previous_path.available + if intent is None: + return None + exact_reachable_route = bool( + previous_path.available and previous_path.fingerprint is not None and previous_path.kernel_route_fingerprint is not None and baseline.device_network.state == "applied" @@ -1880,6 +1881,9 @@ class ReadOnlyConnectionMonitor: and baseline.endpoint.intent_id == intent.intent_id and baseline.endpoint.host_path_epoch == previous_path.epoch and baseline.endpoint.tcp_state == "reachable" + ) + exact_verified_control = bool( + exact_reachable_route and baseline.device_identity.state == "verified" and baseline.device_identity.intent_id == intent.intent_id and baseline.device_identity.host_path_epoch == previous_path.epoch @@ -1892,7 +1896,11 @@ class ReadOnlyConnectionMonitor: and baseline.lease.target == target and baseline.authority.control_allowed ) - if not exact_verified_control: + technical_timeout_on_exact_route = bool( + result.reason_code == "host-wifi-operation-timeout" + and exact_reachable_route + ) + if not (exact_verified_control or technical_timeout_on_exact_route): return None if ( result.kernel_route_fingerprint is None diff --git a/src/k1link/device_plugins/xgrids_k1/facade.py b/src/k1link/device_plugins/xgrids_k1/facade.py index d7a3273..c60099c 100644 --- a/src/k1link/device_plugins/xgrids_k1/facade.py +++ b/src/k1link/device_plugins/xgrids_k1/facade.py @@ -278,6 +278,13 @@ HOST_ROUTE_INSPECTION_TIMEOUT_SECONDS = 2.0 # without weakening the sub-second association cache or skipping the final # route/association continuity recheck. CONNECTION_MONITOR_ASSOCIATION_TIMEOUT_SECONDS = 3.0 +# Active-stream recovery has already frozen an exact acquisition/START owner. +# Its endpoint probe may therefore use the same short read-only association +# budget as the monitor: the exact raw route bridge above preserves continuity +# across a technical CoreWLAN timeout, while fresh DeviceInfo still has to +# verify the K1 before MQTT publication resumes. Do not inherit the generic +# 30-second helper budget on every Wi-Fi return. +ACTIVE_STREAM_RECOVERY_ASSOCIATION_TIMEOUT_SECONDS = 3.0 # Command checkpoints must not inherit either the monitor's three-second # contact budget or the host helper's generic 30-second budget. One command # may legitimately wait about three seconds for the service-lifetime @@ -303,6 +310,9 @@ CONNECTION_MONITOR_FULL_PASS_BOUND_SECONDS = ( 2 * CONNECTION_MONITOR_HOST_CONTACT_BOUND_SECONDS + CONTROL_ENDPOINT_PROBE_TIMEOUT_SECONDS ) CONNECTION_MONITOR_QUIESCE_TIMEOUT_SECONDS = 8.0 +VERIFY_POST_TRANSITION_ENDPOINT_MIN_REMAINING_SECONDS = ( + CONNECTION_MONITOR_FULL_PASS_BOUND_SECONDS + 1.0 +) CONNECTION_MONITOR_DEBOUNCEABLE_ASSOCIATION_REASONS = frozenset({"host-wifi-operation-timeout"}) LIVE_DATA_PLANE_STALL_SECONDS = 5.0 LIVE_DATA_PLANE_LOST_SECONDS = LIVE_DATA_PLANE_STALL_SECONDS * 2.0 @@ -2557,8 +2567,11 @@ class XgridsK1CompatibilityService: A privacy-limited observation has the stricter gate below: exact DeviceInfo, healthy control session, reachable lease and command authority must all still match the same intent/mode/target/epoch. The - legacy timeout bridge may retain only its exact reachable route but - cannot manufacture any missing command authority. + sole exception is an already-admitted active-stream recovery lineage: + it may retain only this exact reachable route long enough to perform a + fresh read-only DeviceInfo/MQTT rebind. That route bridge grants no + control authority and cannot send START, STOP or a network mutation. + The legacy timeout bridge has the same route-only limitation. A proven association change, a different route/interface/source, a stale or unreachable endpoint, or a different intent/target bypasses @@ -2623,7 +2636,13 @@ class XgridsK1CompatibilityService: and snapshot.lease.target == endpoint_target and snapshot.authority.control_allowed ) - if not exact_verified_control: + exact_active_stream_rebind = ( + self._exact_active_stream_recovery_route_rebind( + snapshot=snapshot, + endpoint_target=endpoint_target, + ) + ) + if not (exact_verified_control or exact_active_stream_rebind): return None return HostPathProbeResult( available=True, @@ -2640,6 +2659,68 @@ class XgridsK1CompatibilityService: kernel_route_fingerprint=path.fingerprint, ) + def _exact_active_stream_recovery_route_rebind( + self, + *, + snapshot: ConnectionSupervisorSnapshot, + endpoint_target: EndpointTarget, + ) -> bool: + """Allow one exact route-only bridge for an admitted live recovery. + + Admission of ``_ActiveStreamRecoveryLineage`` already proved the + durable original START owner. This guard rechecks its process-local + acquisition/runtime coordinates without consulting or refreshing any + device authority. The subsequent recovery path must still establish + a fresh DeviceInfo identity and MQTT control proof before it can + resume publication. + """ + + intent = snapshot.intent + with self._lock: + lineage = self._active_stream_recovery_lineage + acquisition = self._acquisition + if lineage is None or acquisition is None: + return False + exact_local_lineage = bool( + self._active_stream_recovery_state == "reconnecting" + and self._active_stream_recovery_generation + == lineage.recovery_generation + and self._snapshot_runtime_id == lineage.snapshot_runtime_id + and acquisition.acquisition_id == lineage.acquisition_id + and acquisition.device_id == lineage.device_id + and acquisition.device_session_id == lineage.device_session_id + and acquisition.state == "acquiring" + and self._acquisition_start_operation_id + in {None, lineage.start_operation_id} + and self._acquisition_session_lease is not None + and self._acquisition_out_dir is not None + and self._acquisition_out_dir.name == lineage.evidence_session_id + and self._selected_device_id == lineage.transport_ref + and self._device_id == lineage.device_id + and self._device_session_id == lineage.device_session_id + and self._connection_mode == lineage.connection_mode + and self._k1_ip == lineage.target_ipv4 + ) + if not exact_local_lineage: + return False + runtime = self.runtime.snapshot() + return bool( + runtime.get("phase") == "reconnecting" + and runtime.get("source_mode") == "live" + and runtime.get("producer_generation") + == lineage.runtime_producer_generation + and intent is not None + and intent.intent_id == lineage.intent_id + and intent.requested_mode == lineage.connection_mode + and snapshot.device_network.state == "applied" + and snapshot.device_network.intent_id == lineage.intent_id + and snapshot.device_network.transport_ref == lineage.transport_ref + and snapshot.device_network.connection_mode == lineage.connection_mode + and snapshot.device_network.target == endpoint_target + and endpoint_target + == EndpointTarget(lineage.target_ipv4, lineage.target_port) + ) + async def _monitor_host_path(self, target: EndpointTarget) -> HostPathProbeResult: if not self._connection_monitor_contact_gate.acquire(blocking=False): raise ConnectionMonitorProbeSuperseded( @@ -2715,6 +2796,57 @@ class XgridsK1CompatibilityService: ) self._connection_supervisor.record_monitor_failure("connection-monitor-start-failed") + async def _refresh_aged_verify_transport_owned( + self, + *, + application_control_session: Mapping[str, Any], + ) -> None: + """Refresh only a nearly expired Verify transport with real I/O. + + The caller owns the Verify lifecycle/network fences, so the background + monitor is deliberately superseded. If DeviceInfo or physical + reconciliation consumed most of the existing endpoint lease, perform + the same route/TCP/route observation before releasing those fences. + No cached timestamp, MQTT reconnect, BLE operation, or device command + is manufactured here. + """ + + verified_binding = application_control_session.get("verified_control") + if not isinstance(verified_binding, Mapping): + raise ConnectionVerificationError( + "Verify lost its exact control proof before transport refresh", + reason_code="connection-verify-final-binding-superseded", + ) + target = EndpointTarget( + str(verified_binding["target_ipv4"]), + int(verified_binding["target_port"]), + ) + intent_id = str(verified_binding["intent_id"]) + host_path_epoch = int(verified_binding["host_path_epoch"]) + if self._connection_supervisor.endpoint_observation_has_remaining_lease( + target=target, + intent_id=intent_id, + host_path_epoch=host_path_epoch, + minimum_remaining_seconds=( + VERIFY_POST_TRANSITION_ENDPOINT_MIN_REMAINING_SECONDS + ), + ): + return + observation = await _run_blocking_operation_without_abandonment( + self._probe_control_endpoint, + target.ipv4, + association_timeout_seconds=(CONNECTION_MONITOR_ASSOCIATION_TIMEOUT_SECONDS), + ) + if not observation.reachable: + raise ConnectionVerificationError( + "K1 endpoint не подтвердил связь после долгой read-only сверки", + reason_code="connection-verify-mqtt-unreachable", + ) + self._reconcile_connection_supervisor( + application_control_session, + self.runtime.snapshot(), + ) + def _observe_connection_transport( self, target: str, @@ -2723,9 +2855,17 @@ class XgridsK1CompatibilityService: reachable: bool | None, reason_code: str | None = None, ) -> None: - supervisor = self._connection_supervisor.snapshot() - intent = supervisor.intent endpoint = EndpointTarget(validate_private_ipv4(target), CONTROL_MQTT_PORT) + # A completed route sample is fresher than the silence fallback. Read + # the exact applied target without expiring the previous observation + # between sampling and publication; the CAS below still rejects an + # intent/target change and the sample itself decides route continuity. + supervisor = self._connection_supervisor.association_timeout_retention_candidate( + expected_target=endpoint, + ) + if supervisor is None: + return + intent = supervisor.intent if ( intent is None or supervisor.device_network.state != "applied" @@ -2734,7 +2874,14 @@ class XgridsK1CompatibilityService: or supervisor.device_network.target != endpoint ): return - epoch = self._connection_supervisor.observe_host_path(path) + epoch = self._connection_supervisor.observe_host_path_if_current( + expected_intent_id=intent.intent_id, + expected_target=endpoint, + expected_host_path_observation=supervisor.host_path, + result=path, + ) + if epoch is None: + return if path.available and reachable is not None: self._connection_supervisor.observe_endpoint( target=endpoint, @@ -3637,8 +3784,15 @@ class XgridsK1CompatibilityService: pending = self._pending_local_control_retirement if not pending: return True + control = self._application_control_session.snapshot() try: - self._retire_application_control_for_network_change() + self._retire_application_control_for_network_change( + # A failed worker may finish retiring after the first bounded + # close attempt. Retrying that host-only retirement must not + # remain permanently blocked by the terminal phase itself. + # The durable physical-command row is intentionally untouched. + allow_terminal_failure=control.get("state") == "failed", + ) except (AttributeError, RuntimeError): return False return True @@ -11616,6 +11770,21 @@ class XgridsK1CompatibilityService: and not reconciliation.observation.init_ready and not reconciliation.observation.mqtt_retained ) + ordinary_terminal_stop = bool( + record.stage == "resolved" + and record.resolution == "stop-standby-observed" + and record.publish_call_returned is True + and record.packet_id is not None + and record.qos2_completed + and record.application_response is not None + and record.application_response.success + and record.last_status is not None + and record.last_status.source == "live-control-session" + and record.last_status.session_state in {"ready", "scan_over"} + and not record.last_status.project_bound + and not record.last_status.init_ready + and not record.last_status.mqtt_retained + ) if not ( persisted is not None and status is not None @@ -11624,6 +11793,7 @@ class XgridsK1CompatibilityService: unresolved_stop or classified_terminal_stop or reconciled_ambiguous_terminal_stop + or ordinary_terminal_stop ) and record.acquisition_id == checkpoint.acquisition_id and record.identity.vendor_device_id_sha256 @@ -13107,6 +13277,75 @@ class XgridsK1CompatibilityService: return False ledger_snapshot = self._physical_command_ledger.snapshot() record = ledger_snapshot.record + ordinary_terminal_stop = bool( + ledger_snapshot.status == "resolved" + and record is not None + and record.action == "stop" + and record.stage == "resolved" + and record.resolution == "stop-standby-observed" + and record.publish_call_returned is True + and record.packet_id is not None + and record.qos2_completed + and record.application_response is not None + and record.application_response.success + and record.last_status is not None + and record.last_status.source == "live-control-session" + and record.last_status.session_state in {"ready", "scan_over"} + and not record.last_status.project_bound + and not record.last_status.init_ready + and not record.last_status.mqtt_retained + and self._checkpoint_binding_matches_physical_connection( + token.checkpoint.current_binding, + record, + ) + ) + if ordinary_terminal_stop: + assert record is not None and record.last_status is not None + store = self._active_acquisition_checkpoint + if store is None or not self._active_acquisition_checkpoint_trust_token_is_current( + token + ): + return False + try: + binding = token.checkpoint.current_binding + status_proof = self._checkpoint_status_proof( + status=record.last_status, + binding=binding, + evidence_session_id=( + token.checkpoint.current_evidence_session_id + ), + ) + physical_proof = self._checkpoint_physical_proof( + record=record, + binding=binding, + checkpoint=token.checkpoint, + ) + store.cease( + transition_id=self._active_acquisition_checkpoint_transition_id( + "restart-ordinary-stop-cease", + record.operation_id, + record.revision, + token.checkpoint_revision, + ), + expected_revision=token.checkpoint_revision, + expected_acquisition_id=token.acquisition_id, + expected_start_operation_id=token.root_start_operation_id, + physical_proof=physical_proof, + status_proof=status_proof, + ) + self._set_active_acquisition_checkpoint_reason(None) + return True + except (ActiveAcquisitionRecoveryCheckpointError, OSError, ValueError) as exc: + self._set_active_acquisition_checkpoint_reason( + str( + getattr( + exc, + "reason_code", + "restart-ordinary-stop-checkpoint-settlement-failed", + ) + ) + ) + return False reconciliation = ( record.reconciliations[-1] if record is not None and record.reconciliations @@ -14314,8 +14553,8 @@ class XgridsK1CompatibilityService: self._last_live_data_suspend_aware = time.time() self._last_live_data_session_id = session_id - physical = self._physical_command_coordinator.snapshot() - control = self._application_control_session.snapshot() + now = time.monotonic() + runtime_recovery = runtime.get("connection_recovery") with self._lock: acquisition = self._acquisition out_dir = self._acquisition_out_dir @@ -14328,6 +14567,35 @@ class XgridsK1CompatibilityService: return acquisition_id = acquisition.acquisition_id control_mode = acquisition.control_mode + lineage = (acquisition_id, out_dir.name, producer_generation) + active_recovery_lineage = self._active_stream_recovery_lineage + post_recovery_restart_candidate = bool( + control_mode == "plugin-commanded" + and active_recovery_lineage is not None + and active_recovery_lineage.acquisition_id == acquisition_id + and active_recovery_lineage.evidence_session_id == out_dir.name + and active_recovery_lineage.runtime_producer_generation + == producer_generation + and isinstance(runtime_recovery, Mapping) + and runtime_recovery.get("state") == "recovered" + ) + if not post_recovery_restart_candidate and ( + self._camera_activation_lineage == lineage + or ( + self._camera_activation_retry_lineage == lineage + and now < self._camera_activation_retry_not_before_monotonic + ) + ): + # The exact lineage already owns its one camera activation (or + # its bounded local retry delay). Repeating the durable physical + # ledger/checkpoint admission for every 10 Hz PCL blocks this + # publisher thread and evicts intervening pose frames. The + # data-plane liveness edge above remains per-frame; recovery + # deliberately bypasses this ordinary idempotence fast path. + return + + physical = self._physical_command_coordinator.snapshot() + control = self._application_control_session.snapshot() if control_mode == "plugin-commanded": start_operation_id = ( @@ -14364,21 +14632,10 @@ class XgridsK1CompatibilityService: # this local admission after the durable SCANNING proof arrives. return - lineage = (acquisition_id, out_dir.name, producer_generation) - now = time.monotonic() camera = self.camera_preview.snapshot() - with self._lock: - active_recovery_lineage = self._active_stream_recovery_lineage - runtime_recovery = runtime.get("connection_recovery") if ( - control_mode == "plugin-commanded" + post_recovery_restart_candidate and active_recovery_lineage is not None - and active_recovery_lineage.acquisition_id == acquisition_id - and active_recovery_lineage.evidence_session_id == out_dir.name - and active_recovery_lineage.runtime_producer_generation - == producer_generation - and isinstance(runtime_recovery, Mapping) - and runtime_recovery.get("state") == "recovered" and self._enqueue_post_recovery_camera_restart( active_recovery_lineage, runtime=runtime, @@ -14888,6 +15145,10 @@ class XgridsK1CompatibilityService: or control_after.get("session_generation") != control_before.get("session_generation") ) + control_stage = "post-device-info-transport-refresh" + await self._refresh_aged_verify_transport_owned( + application_control_session=control_after, + ) control_stage = "physical-reconciliation" physical_reconciliation = await self._reconcile_physical_command_after_verify_owned( verify_operation_id=verify_operation_id, @@ -14895,6 +15156,12 @@ class XgridsK1CompatibilityService: if provisional_topology is not None: control_stage = "semantic-topology-commit" self._commit_provisional_fresh_bridge_topology(provisional_topology) + control_stage = "post-reconciliation-transport-refresh" + await self._refresh_aged_verify_transport_owned( + application_control_session=dict( + self._application_control_session.snapshot() + ), + ) control_stage = "final-control-binding" ready = self._connection_supervisor.snapshot() if not ( @@ -23267,6 +23534,10 @@ class XgridsK1CompatibilityService: ) use_reconciliation = bool( reconciliation is not None + and self._physical_reconciliation_belongs_to_record( + reconciliation, + record, + ) and reconciliation.kind in { "ambiguous-outcome", @@ -24458,6 +24729,9 @@ class XgridsK1CompatibilityService: observation = await _run_blocking_operation_without_abandonment( self._probe_control_endpoint, lineage.target_ipv4, + association_timeout_seconds=( + ACTIVE_STREAM_RECOVERY_ASSOCIATION_TIMEOUT_SECONDS + ), ) if not observation.path.available or not observation.reachable: with self._lock: @@ -27846,6 +28120,15 @@ class XgridsK1CompatibilityService: self._physical_command_coordinator.snapshot() ) is None: return "not-applicable" + # A live START worker cannot be settled. Avoid touching the exact + # mark_dispatching -> publish fence on every passive state read while + # it is still initializing; the terminal branch below re-samples the + # worker under that fence before committing any no-dispatch result. + if self._application_control_session.snapshot().get("state") not in { + "failed", + "closed", + }: + return "not-applicable" if not self._k1_command_dispatch_gate.acquire(blocking=False): return "defer" try: @@ -28371,6 +28654,59 @@ class XgridsK1CompatibilityService: and durable_start_active_confirmed and durable_checkpoint_active_confirmed ) + terminal_control_failure = ( + terminal_control_proof.get("failure") + if isinstance(terminal_control_proof, Mapping) + else None + ) + terminal_control_transport = ( + terminal_control_proof.get("transport") + if isinstance(terminal_control_proof, Mapping) + else None + ) + terminal_verified_control = ( + terminal_control_proof.get("verified_control") + if isinstance(terminal_control_proof, Mapping) + else None + ) + device_reported_scan_over_without_stop = bool( + current is not None + and current.control_mode == "plugin-commanded" + and current.state in {"starting", "awaiting_external_start", "acquiring"} + and current_stop_operation_id is None + and isinstance(terminal_control_proof, Mapping) + and terminal_control_proof.get("state") == "failed" + and isinstance(terminal_control_failure, Mapping) + and terminal_control_failure.get("failed_phase") == "scanning" + and terminal_control_failure.get("modeling_command_attempted") is True + and terminal_control_failure.get("stop_command_attempted") is False + and terminal_control_failure.get("diagnostic_snapshot_unavailable") == [] + and terminal_control_failure.get("diagnostic_evidence_unavailable") == [] + and isinstance(terminal_control_transport, Mapping) + and isinstance(terminal_control_transport.get("device_status_reports"), int) + and not isinstance( + terminal_control_transport.get("device_status_reports"), + bool, + ) + and int(terminal_control_transport["device_status_reports"]) > 0 + and terminal_control_transport.get("latest_device_session_state") + == "scan_over" + and terminal_control_transport.get("latest_system_error_code") is None + and isinstance(physical_command_proof, Mapping) + and physical_command_proof.get("runtime_bound") is True + and physical_command_proof.get("reconciliation_ready") is True + and physical_command_proof.get("observed_session_state") == "scan_over" + and self._matching_start_active_confirmed( + physical_command_proof, + acquisition_id=current_acquisition_id, + start_operation_id=canonical_start_operation_id, + verified_control=( + terminal_verified_control + if isinstance(terminal_verified_control, Mapping) + else None + ), + ) + ) # Camera-only transport loss is supervised by the backend producer # watchdog and exact camera CAS. Snapshot polling is observational: it # must not move an otherwise live MQTT/PCL runtime to ``reconnecting``. @@ -28459,6 +28795,21 @@ class XgridsK1CompatibilityService: and stop_response_accepted_without_terminal_status and self._operations.deadline_reached(current_stop_operation_id) ) + terminal_unknown_stop_control = bool( + current is not None + and current.control_mode == "plugin-commanded" + and current.state == "awaiting_external_stop" + and application_control_session.get("state") == "failed" + and application_control_session.get("outcome_unknown") is True + and isinstance(application_control_session.get("failure"), Mapping) + and application_control_session["failure"].get("stop_command_attempted") is True + and self._matching_unresolved_stop_stage( + physical_command_proof, + acquisition_id=current_acquisition_id, + stop_operation_id=current_stop_operation_id, + ) + is not None + ) with self._lock: canonical_stop_standby_confirmed = bool( checkpoint_stop_completion_seen and checkpoint_stop_ceased @@ -28504,7 +28855,11 @@ class XgridsK1CompatibilityService: and stop_response_deadline_reached else None ) - if stop_response_deadline_reached: + if terminal_unknown_stop_control or stop_response_deadline_reached: + # Retire only the failed host-owned control socket. In particular, + # an UNKNOWN physical STOP keeps the capture, viewer and acquisition + # alive so an explicit read-only DeviceInfo/DeviceStatus dialogue + # can classify SCANNING or READY without a second command. self._retire_local_control_after_stop_timeout() if terminal_unknown_stop_cleanup: try: @@ -28858,6 +29213,39 @@ class XgridsK1CompatibilityService: # read-only classification and seal the old STOP as # non-replayable/none. pass + elif device_reported_scan_over_without_stop: + # This is not an inference from a quiet camera or point stream. + # The continuously DeviceInfo-bound control socket decoded a + # fresh, non-retained SCAN_OVER from this exact K1, and the + # physical coordinator independently retained the same bound + # status. No canonical STOP was requested or attempted. + # + # SCAN_OVER proves that this acquisition cannot be resumed. It + # does not prove READY and does not authorize a new START; the + # durable physical/checkpoint chain remains fenced until a later + # explicit read-only reconciliation observes the returned K1. + acquisition.transition( + "finalizing", + message_code="acquisition.recovery.device_standby_observed", + ) + camera_terminal_status = "interrupted" + camera_failure_code = "device-reported-scan-over" + recovery_standby_after_seal = True + if point_frames <= 0: + recovery_standby_start_operation_id = ( + self._acquisition_start_operation_id + ) + completion_message_code = ( + "acquisition.recovery.device_standby_observed" + ) + completion_result = { + "receiver_stopped": True, + "device_state": "scan_over", + "device_stop": "not-sent", + "automatic_command_retry": False, + "read_only_recovery": True, + "physical_reconciliation_required": True, + } elif prepared_stop_recovery_standby: acquisition.transition( "finalizing", @@ -29815,6 +30203,28 @@ class XgridsK1PluginFacade: def __init__(self, service: XgridsK1ServicePort) -> None: self.service = service + self._state_read_task: asyncio.Task[dict[str, Any]] | None = None + + def _retire_state_read_task(self, task: asyncio.Task[dict[str, Any]]) -> None: + if self._state_read_task is task: + self._state_read_task = None + with suppress(asyncio.CancelledError): + task.exception() + + async def _read_state_single_flight(self) -> dict[str, Any]: + loop = asyncio.get_running_loop() + task = self._state_read_task + if task is None or task.done() or task.get_loop() is not loop: + task = loop.create_task( + asyncio.to_thread(self.service.state), + name="xgrids-k1-state-read", + ) + self._state_read_task = task + task.add_done_callback(self._retire_state_read_task) + # An HTTP disconnect cancels only that request waiter. The shared + # read keeps running so a later poll joins it instead of stacking a + # second service.state thread behind the lifecycle gate. + return await asyncio.shield(task) async def invoke(self, invocation: RuntimeActionInvocation) -> dict[str, Any]: if invocation.plugin_id != self.plugin_id: @@ -29887,7 +30297,7 @@ class XgridsK1PluginFacade: self.service.bind_runtime_event_loop(asyncio.get_running_loop()) if action_id == ACTION_STATE_READ: EmptyRequest.model_validate(payload) - return await asyncio.to_thread(self.service.state) + return await self._read_state_single_flight() if action_id == ACTION_DISCOVERY_SCAN: scan_request = BleScanRequest.model_validate(payload) return await self.service.scan_ble(scan_request) @@ -29899,7 +30309,7 @@ class XgridsK1PluginFacade: EmptyRequest.model_validate(payload) if action_id == ACTION_DEVICE_INSPECT: return await asyncio.to_thread(self.service.inspect_device) - return await asyncio.to_thread(self.service.state) + return await self._read_state_single_flight() if action_id == ACTION_DEVICE_CALIBRATION_SNAPSHOT_READ: EmptyRequest.model_validate(payload) return await asyncio.to_thread(self.service.read_device_calibration_snapshot) @@ -29923,7 +30333,7 @@ class XgridsK1PluginFacade: return await self.service.probe_configured_endpoint(probe_request) if action_id == ACTION_APPLICATION_CONTROL_SHADOW_STATE: EmptyRequest.model_validate(payload) - return await asyncio.to_thread(self.service.state) + return await self._read_state_single_flight() if action_id == ACTION_APPLICATION_CONTROL_SHADOW_ARM: arm_request = ShadowApplicationControlArmRequest.model_validate(payload) return await asyncio.to_thread( diff --git a/src/k1link/device_plugins/xgrids_k1/viewer/runtime.py b/src/k1link/device_plugins/xgrids_k1/viewer/runtime.py index d2c737d..93f907a 100644 --- a/src/k1link/device_plugins/xgrids_k1/viewer/runtime.py +++ b/src/k1link/device_plugins/xgrids_k1/viewer/runtime.py @@ -696,6 +696,7 @@ class VisualizationRuntime: def publish_stream(bridge: RerunBridge) -> None: serve_perception_next = False + serve_non_point_cloud_next = False while ( not source_done.is_set() or not preview_empty() @@ -708,15 +709,16 @@ class VisualizationRuntime: and preview_empty() ): return - # A dedicated latest-PCL slot plus this fixed service order - # guarantees visible-scene progress even under an unbounded - # stream of perception results. Other MQTT preview traffic is - # then served before one perception result, so neither class can - # monopolize the single publisher thread. - live_point_cloud = pop_latest_live_point_cloud() - if live_point_cloud is not None: - publish_message(bridge, live_point_cloud) - continue + # The latest-PCL slot preserves visible-scene recovery, but a + # continuously full slot must not starve pose. Alternate one + # live PCL with one available non-PCL item; when no other item + # exists, the fallback below keeps PCL flowing without delay. + if not serve_non_point_cloud_next: + live_point_cloud = pop_latest_live_point_cloud() + if live_point_cloud is not None: + publish_message(bridge, live_point_cloud) + serve_non_point_cloud_next = True + continue if serve_perception_next: try: perception = self._perception_messages.get_nowait() @@ -728,6 +730,7 @@ class VisualizationRuntime: finally: self._perception_messages.task_done() serve_perception_next = False + serve_non_point_cloud_next = False continue try: message = messages.get_nowait() @@ -739,6 +742,7 @@ class VisualizationRuntime: finally: messages.task_done() serve_perception_next = True + serve_non_point_cloud_next = False continue try: perception = self._perception_messages.get_nowait() @@ -750,6 +754,12 @@ class VisualizationRuntime: finally: self._perception_messages.task_done() serve_perception_next = False + serve_non_point_cloud_next = False + continue + live_point_cloud = pop_latest_live_point_cloud() + if live_point_cloud is not None: + publish_message(bridge, live_point_cloud) + serve_non_point_cloud_next = True continue live_point_cloud_ready.wait(timeout=0.05) diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index 64faf82..85b67a4 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -355,24 +355,66 @@ def newly_finalized_recording_ids( return tuple(sorted(set(current_finalized) - known_finalized)) +def observation_archive_revision(roots: Iterable[Path]) -> tuple[tuple[object, ...], ...]: + """Return a cheap change fence for direct-child observation archives. + + The K1 writer creates and retires ``.current_session`` in the archive root, + and every session directory is a direct child of that same root. Those + operations advance the directory metadata, while growing capture/media + files do not. This lets the reconciler notice session start, completion and + removal without recursively reopening tens of thousands of immutable + evidence files every two seconds. + """ + + revisions: list[tuple[object, ...]] = [] + for configured_root in roots: + root = configured_root.expanduser().resolve(strict=False) + try: + metadata = root.lstat() + except OSError: + revisions.append((str(root), None)) + continue + revisions.append( + ( + str(root), + metadata.st_dev, + metadata.st_ino, + metadata.st_mtime_ns, + metadata.st_ctime_ns, + ) + ) + revisions.sort(key=lambda item: str(item[0])) + return tuple(revisions) + + async def _recording_preparation_reconciler() -> None: """Prepare sessions finalized during this process, never historical rows.""" known_finalized: set[str] | None = None + reconciled_archive_revision: tuple[tuple[object, ...], ...] | None = None while True: try: - await asyncio.to_thread(refresh_observation_catalog) - finalized = set(await asyncio.to_thread(finalized_replayable_recording_ids)) - newly_finalized = newly_finalized_recording_ids( - known_finalized, - finalized, + archive_revision = await asyncio.to_thread( + observation_archive_revision, + (archive.root for archive in plugin_environment.observation_archives), ) - if newly_finalized: - await asyncio.to_thread( - enqueue_replayable_recordings, - newly_finalized, + if archive_revision != reconciled_archive_revision: + await asyncio.to_thread(refresh_observation_catalog) + finalized = set(await asyncio.to_thread(finalized_replayable_recording_ids)) + newly_finalized = newly_finalized_recording_ids( + known_finalized, + finalized, ) - known_finalized = finalized + if newly_finalized: + await asyncio.to_thread( + enqueue_replayable_recordings, + newly_finalized, + ) + known_finalized = finalized + # Keep the pre-scan revision. If a writer changed the archive + # during the expensive discovery, the next two-second check + # observes the newer revision and performs one follow-up pass. + reconciled_archive_revision = archive_revision recording_reconciler_readiness.record_success() except Exception as exc: # A transient filesystem/catalog failure must not permanently @@ -496,6 +538,12 @@ _CLOSED_WEBSOCKET_SEND_ERRORS = frozenset( } ) +# This stream carries the comparatively heavy control-state snapshot, not +# camera or point-cloud media. Action requests return their own result, and the +# frontend also keeps a four-second REST fallback, so a two-second passive +# cadence preserves status recovery without starving the live media sockets. +DEVICE_PLUGIN_EVENT_POLL_INTERVAL_SECONDS = 2.0 + def _is_closed_websocket_send_error(exc: RuntimeError) -> bool: message = " ".join(str(exc).split()) @@ -528,7 +576,7 @@ async def device_plugin_events(websocket: WebSocket, plugin_id: str) -> None: if _is_closed_websocket_send_error(exc): return raise - await asyncio.sleep(0.5) + await asyncio.sleep(DEVICE_PLUGIN_EVENT_POLL_INTERVAL_SECONDS) except (PluginNotFoundError, PluginActionNotFoundError): await websocket.close(code=1008, reason="Device plugin is not available") except (PluginExecutionError, PluginRuntimeUnavailableError): diff --git a/tests/test_canonical_pipeline.py b/tests/test_canonical_pipeline.py index e4ef463..2760ac9 100644 --- a/tests/test_canonical_pipeline.py +++ b/tests/test_canonical_pipeline.py @@ -326,6 +326,7 @@ def test_live_rerun_process_and_close_failure_preserve_capture_then_recover( allow_second_bridge = threading.Event() second_bridge_ready = threading.Event() recovery_confirmed = threading.Event() + raw_capture_completed = threading.Event() raw_sequences: list[int] = [] should_stop_before_explicit_stop: list[bool] = [] confirmed_attempts: list[int] = [] @@ -389,6 +390,7 @@ def test_live_rerun_process_and_close_failure_preserve_capture_then_recover( assert second_bridge_ready.wait(timeout=2.0) raw_sequences.append(10) enqueue(_captured_live_point_cloud(10)) # type: ignore[operator] + raw_capture_completed.set() assert recovery_confirmed.wait(timeout=2.0) while not should_stop(): # type: ignore[operator] @@ -409,6 +411,7 @@ def test_live_rerun_process_and_close_failure_preserve_capture_then_recover( ) assert recovery_confirmed.wait(timeout=3.0) + assert raw_capture_completed.wait(timeout=3.0) snapshot = runtime.snapshot() assert snapshot["phase"] == "live" assert snapshot["source_ready"] is True @@ -510,6 +513,66 @@ def test_live_latest_point_cloud_survives_pose_and_perception_pressure( runtime.close() +def test_live_pose_is_published_during_continuous_point_cloud_pressure( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + producer_started = threading.Event() + producer_finished = threading.Event() + pose_published_during_pressure = threading.Event() + + class SlowPointCloudBridge(MetricRuntimeBridgeStub): + def process(self, envelope: object) -> None: + if isinstance(envelope, DecodedPointCloudView): + # Keep another PCL waiting in the latest-wins slot. The + # scheduler must still admit pose while pressure continues. + time.sleep(0.01) + elif ( + isinstance(envelope, DecodedPoseView) + and not producer_finished.is_set() + ): + pose_published_during_pressure.set() + super().process(envelope) + + def fake_capture_mqtt( + _host: str, + _out_dir: Path, + **callbacks: object, + ) -> dict[str, object]: + callbacks["on_clock_established"]() # type: ignore[operator] + callbacks["on_ready"]() # type: ignore[operator] + enqueue = callbacks["on_message_recorded"] + producer_started.set() + for index in range(300): + enqueue(_captured_live_point_cloud(index * 2 + 1)) # type: ignore[operator] + enqueue(_captured_live_pose(index * 2 + 2)) # type: ignore[operator] + time.sleep(0.001) + producer_finished.set() + + should_stop = callbacks["should_stop"] + while not should_stop(): # type: ignore[operator] + time.sleep(0.005) + return {"message_count": 600} + + monkeypatch.setattr(runtime_module, "capture_mqtt", fake_capture_mqtt) + runtime = VisualizationRuntime( + bridge_factory=lambda **kwargs: SlowPointCloudBridge(kwargs["metrics"]), # type: ignore[arg-type] + normalizer=normalize_k1_message, + ) + runtime.start_live( + "192.168.1.50", + tmp_path / "pose-fairness", + duration_seconds=None, + project_name="POSEFAIRNESS001", + ) + + assert producer_started.wait(timeout=2.0) + assert producer_finished.wait(timeout=2.0) + runtime.stop(wait_seconds=2.0) + runtime.close() + assert pose_published_during_pressure.is_set() + + @pytest.mark.parametrize("attempt", [5, 1025, 10**100]) def test_live_rerun_recovery_backoff_saturates(attempt: int) -> None: assert runtime_module._rerun_recovery_backoff_seconds(attempt) == 5.0 # noqa: SLF001 diff --git a/tests/test_connection_supervisor.py b/tests/test_connection_supervisor.py index 75e13af..02ad425 100644 --- a/tests/test_connection_supervisor.py +++ b/tests/test_connection_supervisor.py @@ -1817,7 +1817,7 @@ def test_same_path_positive_reducer_observation_resets_technical_failure_streak( asyncio.run(scenario()) -def test_unrelated_supervisor_revisions_do_not_reset_technical_failure_streak() -> None: +def test_unrelated_supervisor_revisions_preserve_exact_configured_route() -> None: async def scenario() -> None: supervisor, epoch = _configured_unverified_supervisor() path = _available_path() @@ -1836,8 +1836,12 @@ def test_unrelated_supervisor_revisions_do_not_reset_technical_failure_streak() async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: return technical_timeout + tcp_calls = 0 + async def tcp_probe(_target: EndpointTarget) -> bool: - raise AssertionError("technical failure must skip TCP") + nonlocal tcp_calls + tcp_calls += 1 + return True monitor = ReadOnlyConnectionMonitor( supervisor, @@ -1847,8 +1851,10 @@ def test_unrelated_supervisor_revisions_do_not_reset_technical_failure_streak() ) first = await monitor.poll_once() - assert first.revision == initial.revision - assert first.host_path is initial.host_path + assert first.host_path.available is True + assert first.host_path.epoch == initial.host_path.epoch + assert first.endpoint.tcp_state == "reachable" + assert first.authority.control_allowed is False assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-1", @@ -1857,11 +1863,14 @@ def test_unrelated_supervisor_revisions_do_not_reset_technical_failure_streak() ) endpoint_refresh = supervisor.snapshot() assert endpoint_refresh.revision > first.revision - assert endpoint_refresh.host_path is initial.host_path + assert endpoint_refresh.host_path.epoch == initial.host_path.epoch + assert endpoint_refresh.host_path.fingerprint == initial.host_path.fingerprint second = await monitor.poll_once() - assert second.revision == endpoint_refresh.revision - assert second.host_path is initial.host_path + assert second.host_path.available is True + assert second.host_path.epoch == initial.host_path.epoch + assert second.endpoint.tcp_state == "reachable" + assert second.authority.control_allowed is False assert supervisor.observe_endpoint( target=TARGET, intent_id="bridge-1", @@ -1870,15 +1879,17 @@ def test_unrelated_supervisor_revisions_do_not_reset_technical_failure_streak() ) confirmed = await monitor.poll_once() - assert confirmed.host_path.available is False - assert confirmed.host_path.reason_code == "host-wifi-operation-timeout" + assert confirmed.host_path.available is True + assert confirmed.host_path.epoch == initial.host_path.epoch + assert confirmed.endpoint.tcp_state == "reachable" assert confirmed.authority.control_allowed is False + assert tcp_calls == 3 await monitor.close() asyncio.run(scenario()) -def test_inflight_third_timeout_cannot_overwrite_a_concurrent_positive_observation() -> None: +def test_inflight_timeout_cannot_overwrite_a_concurrent_positive_observation() -> None: async def scenario() -> None: supervisor, initial_epoch = _configured_unverified_supervisor() path = _available_path() @@ -1892,20 +1903,20 @@ def test_inflight_third_timeout_cannot_overwrite_a_concurrent_positive_observati observation_failure_class="association-observer", kernel_route_fingerprint=path.kernel_route_fingerprint, ) - third_probe_entered = asyncio.Event() - release_third_probe = asyncio.Event() + timeout_probe_entered = asyncio.Event() + release_timeout_probe = asyncio.Event() host_calls = 0 async def host_probe(_target: EndpointTarget) -> HostPathProbeResult: nonlocal host_calls host_calls += 1 - if host_calls == 3: - third_probe_entered.set() - await release_third_probe.wait() + if host_calls == 1: + timeout_probe_entered.set() + await release_timeout_probe.wait() return technical_timeout async def tcp_probe(_target: EndpointTarget) -> bool: - raise AssertionError("technical failure must skip TCP") + return True monitor = ReadOnlyConnectionMonitor( supervisor, @@ -1913,32 +1924,20 @@ def test_inflight_third_timeout_cannot_overwrite_a_concurrent_positive_observati tcp_probe=tcp_probe, target_provider=lambda: TARGET, ) - first = await monitor.poll_once() - second = await monitor.poll_once() - assert first.authority.control_allowed is False - assert second.authority.control_allowed is False - - inflight_third = asyncio.create_task(monitor.poll_once()) - await third_probe_entered.wait() + inflight_timeout = asyncio.create_task(monitor.poll_once()) + await timeout_probe_entered.wait() refreshed_epoch = supervisor.observe_host_path(path) assert refreshed_epoch == initial_epoch external_positive = supervisor.snapshot() - release_third_probe.set() - raced_timeout = await inflight_third + release_timeout_probe.set() + raced_timeout = await inflight_timeout - assert raced_timeout.revision == external_positive.revision - assert raced_timeout.host_path is external_positive.host_path + assert raced_timeout.revision >= external_positive.revision + assert raced_timeout.host_path.available is True + assert raced_timeout.host_path.epoch == external_positive.host_path.epoch + assert raced_timeout.host_path.fingerprint == external_positive.host_path.fingerprint + assert raced_timeout.endpoint.tcp_state == "reachable" assert raced_timeout.authority.control_allowed is False - - second_after_positive = await monitor.poll_once() - assert second_after_positive.revision == external_positive.revision - assert second_after_positive.host_path is external_positive.host_path - assert second_after_positive.authority.control_allowed is False - - confirmed = await monitor.poll_once() - assert confirmed.host_path.available is False - assert confirmed.host_path.reason_code == "host-wifi-operation-timeout" - assert confirmed.authority.control_allowed is False await monitor.close() asyncio.run(scenario()) diff --git a/tests/test_plugin_runtime.py b/tests/test_plugin_runtime.py index 98fc54b..a9111be 100644 --- a/tests/test_plugin_runtime.py +++ b/tests/test_plugin_runtime.py @@ -37,6 +37,7 @@ from k1link.device_plugins.xgrids_k1.facade import ( ACTION_PHYSICAL_COMMAND_RECONCILE, ACTION_PHYSICAL_COMMAND_REOPEN_RETIRED_RECONCILIATION, ACTION_PHYSICAL_COMMAND_RETIRE_UNAVAILABLE, + ACTION_STATE_READ, ACTION_STREAM_START_LIVE, ACTION_STREAM_START_REPLAY, ACTION_STREAM_STOP, @@ -1285,6 +1286,57 @@ def test_sync_runtime_actions_run_outside_the_api_event_loop() -> None: assert service.thread_id != event_loop_thread +def test_state_reads_are_single_flight_and_survive_one_cancelled_waiter() -> None: + class BlockingStateService(FakeXgridsService): + def __init__(self) -> None: + super().__init__() + self.state_calls = 0 + self.started = threading.Event() + self.release = threading.Event() + + def state(self) -> dict[str, Any]: + self.state_calls += 1 + self.started.set() + if not self.release.wait(timeout=2): + raise AssertionError("state read was not released by the test") + return {"phase": "idle", "revision": self.state_calls} + + service = BlockingStateService() + adapter = XgridsK1PluginFacade(service) + + def invocation(invocation_id: str) -> RuntimeActionInvocation: + return RuntimeActionInvocation( + invocation_id=invocation_id, + plugin_id=adapter.plugin_id, + action_id=ACTION_STATE_READ, + requested_at=datetime.now(UTC), + parameters={}, + ) + + async def exercise() -> None: + first = asyncio.create_task(adapter.invoke(invocation("state-read-first"))) + while not service.started.is_set(): + await asyncio.sleep(0) + second = asyncio.create_task(adapter.invoke(invocation("state-read-second"))) + await asyncio.sleep(0.02) + assert service.state_calls == 1 + + first.cancel() + with pytest.raises(asyncio.CancelledError): + await first + assert service.state_calls == 1 + + service.release.set() + assert await second == {"phase": "idle", "revision": 1} + assert await adapter.invoke(invocation("state-read-fresh")) == { + "phase": "idle", + "revision": 2, + } + + asyncio.run(exercise()) + assert service.state_calls == 2 + + def test_runtime_requires_successful_handshake_before_dispatch() -> None: adapter = XgridsK1PluginFacade(FakeXgridsService()) runtime = _in_process_runtime(adapter, activate=False) diff --git a/tests/test_session_catalog_lifecycle.py b/tests/test_session_catalog_lifecycle.py index 6ca3811..8596226 100644 --- a/tests/test_session_catalog_lifecycle.py +++ b/tests/test_session_catalog_lifecycle.py @@ -200,6 +200,33 @@ def test_startup_scan_baselines_historical_sessions_without_enqueuing( manager.close() +def test_archive_revision_tracks_session_lifecycle_without_capture_churn( + tmp_path: Path, +) -> None: + sessions = tmp_path / "sessions" + sessions.mkdir() + initial = app_module.observation_archive_revision((sessions,)) + + session = sessions / "20260716T205632Z_viewer_live" + session.mkdir() + started = app_module.observation_archive_revision((sessions,)) + assert started != initial + + capture = session / "captures" / "mqtt_live" + capture.mkdir(parents=True) + (capture / "mqtt.raw.k1mqtt").write_bytes(b"growing-capture") + assert app_module.observation_archive_revision((sessions,)) == started + + marker = sessions / ".current_session" + marker.write_text(f"{session.name}\n", encoding="utf-8") + active = app_module.observation_archive_revision((sessions,)) + assert active != started + + marker.unlink() + finalized = app_module.observation_archive_revision((sessions,)) + assert finalized != active + + def test_reconciliation_skips_one_stale_session_and_prepares_later_valid_session( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/test_web_validation_security.py b/tests/test_web_validation_security.py index bac18ca..f7320cf 100644 --- a/tests/test_web_validation_security.py +++ b/tests/test_web_validation_security.py @@ -40,6 +40,24 @@ class _FailingSendWebSocket: del code, reason +class _DisconnectAfterSecondSendWebSocket: + def __init__(self) -> None: + self.accepted = False + self.send_count = 0 + + async def accept(self) -> None: + self.accepted = True + + async def send_json(self, payload: dict[str, Any]) -> None: + del payload + self.send_count += 1 + if self.send_count == 2: + raise WebSocketDisconnect(code=1001) + + async def close(self, *, code: int, reason: str) -> None: + del code, reason + + async def _post_json(path: str, payload: dict[str, Any]) -> tuple[int, str]: body = json.dumps(payload).encode() request_sent = False @@ -198,6 +216,25 @@ def test_device_plugin_events_treats_proven_transport_disconnect_as_completion( assert websocket.accepted is True +def test_device_plugin_events_keeps_heavy_state_poll_off_live_media_cadence( + monkeypatch: pytest.MonkeyPatch, +) -> None: + websocket = _DisconnectAfterSecondSendWebSocket() + sleeps: list[float] = [] + + async def record_sleep(delay: float) -> None: + sleeps.append(delay) + + monkeypatch.setattr(app_module, "plugin_dispatcher", _StateDispatcher()) + monkeypatch.setattr(app_module.asyncio, "sleep", record_sleep) + + asyncio.run(app_module.device_plugin_events(websocket, "test.plugin")) + + assert websocket.accepted is True + assert sleeps == [app_module.DEVICE_PLUGIN_EVENT_POLL_INTERVAL_SECONDS] + assert sleeps[0] == 2.0 + + def test_device_plugin_events_propagates_arbitrary_send_runtime_error( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/test_xgrids_acquisition_lifecycle.py b/tests/test_xgrids_acquisition_lifecycle.py index b2cbd41..a520276 100644 --- a/tests/test_xgrids_acquisition_lifecycle.py +++ b/tests/test_xgrids_acquisition_lifecycle.py @@ -5662,7 +5662,7 @@ def test_read_only_verify_scan_is_rejected_after_atomic_fence_before_admission( original_retire = service._retire_application_control_for_network_change # noqa: SLF001 original_apply = service._apply_read_only_device_topology # noqa: SLF001 - def retire_with_competing_scan() -> None: + def retire_with_competing_scan(*, allow_terminal_failure: bool = False) -> None: assert service._provisioning_active is True # noqa: SLF001 scan_outcomes.append( _attempt_competing_scan_from_sync_boundary( @@ -5670,7 +5670,7 @@ def test_read_only_verify_scan_is_rejected_after_atomic_fence_before_admission( operation_id="op-00000000-0000-4000-8000-000000000102", ) ) - original_retire() + original_retire(allow_terminal_failure=allow_terminal_failure) def apply_with_fence_assertion(**kwargs: Any) -> str: nonlocal admission_calls @@ -7692,6 +7692,58 @@ def test_terminal_prepared_start_without_publish_settles_and_stops_local_capture assert service._acquisition_session_lease is None # noqa: SLF001 +def test_live_prepared_start_reduction_does_not_touch_publish_fence( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + service, _ = service_with_fake_runtime(tmp_path) + acquisition_id = "acq-live-prepared" + operation_id = "op-live-prepared" + monkeypatch.setattr( + service._physical_command_coordinator, # noqa: SLF001 + "snapshot", + lambda: { + "status": "unresolved", + "record": { + "operation_id": operation_id, + "acquisition_id": acquisition_id, + "action": "start", + "stage": "prepared", + "resolution": None, + "publish_call_returned": None, + "packet_id": None, + "qos2_completed": False, + "application_response": None, + }, + }, + ) + monkeypatch.setattr( + service._application_control_session, # noqa: SLF001 + "snapshot", + lambda: {"state": "initializing", "failure": None}, + ) + + class PublishFenceMustStayFree: + def acquire(self, *args: object, **kwargs: object) -> bool: + raise AssertionError("live START reduction touched the publish fence") + + def release(self) -> None: + raise AssertionError("live START reduction released an unowned fence") + + monkeypatch.setattr( + service, + "_k1_command_dispatch_gate", + PublishFenceMustStayFree(), + ) + + outcome = service._settle_prepared_start_worker_failure( # noqa: SLF001 + acquisition_id=acquisition_id, + start_operation_id=operation_id, + ) + + assert outcome == "not-applicable" + + def _activate_real_checkpoint_for_prepared_stop_fixture( service: XgridsK1CompatibilityService, *, @@ -12273,6 +12325,7 @@ def _stop_response_without_terminal_status_fixture( tmp_path: Path, *, qos2_completed: bool = True, + application_response: bool = True, ) -> SimpleNamespace: clock_value = [datetime(2026, 8, 10, 8, 10, tzinfo=UTC)] service, runtime = service_with_fake_runtime(tmp_path) @@ -12424,20 +12477,21 @@ def _stop_response_without_terminal_status_fixture( packet_id=82, ) ledger.mark_qos2_completed(stop_operation_id, packet_id=82) - ledger.record_application_response( - stop_operation_id, - PhysicalCommandApplicationResponse( - operation_id=stop_operation_id, - action="stop", - control_session_id=physical_connection.control_session_id, - host_path_epoch=physical_connection.host_path_epoch, - producer_generation=physical_connection.producer_generation, - result_code=PHYSICAL_COMMAND_APPLICATION_SUCCESS_CODE, - success=True, - payload_sha256="a" * 64, - observed_at_utc="2026-08-10T08:10:02.000Z", - ), - ) + if application_response: + ledger.record_application_response( + stop_operation_id, + PhysicalCommandApplicationResponse( + operation_id=stop_operation_id, + action="stop", + control_session_id=physical_connection.control_session_id, + host_path_epoch=physical_connection.host_path_epoch, + producer_generation=physical_connection.producer_generation, + result_code=PHYSICAL_COMMAND_APPLICATION_SUCCESS_CODE, + success=True, + payload_sha256="a" * 64, + observed_at_utc="2026-08-10T08:10:02.000Z", + ), + ) return SimpleNamespace( service=service, runtime=runtime, @@ -12450,6 +12504,60 @@ def _stop_response_without_terminal_status_fixture( ) +def test_unknown_stop_retires_only_terminal_control_and_keeps_live_capture( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + fixture = _stop_response_without_terminal_status_fixture( + tmp_path, + qos2_completed=False, + application_response=False, + ) + service = fixture.service + control = fixture.control + close_calls = 0 + worker_retired = False + original_snapshot = control.snapshot + original_close = control.close + + def terminal_snapshot() -> dict[str, object]: + snapshot = original_snapshot() + if control.state == "failed" and not worker_retired: + snapshot["can_open"] = False + return snapshot + + def retire_failed_worker() -> None: + nonlocal close_calls, worker_retired + close_calls += 1 + worker_retired = True + original_close() + + monkeypatch.setattr(control, "snapshot", terminal_snapshot) + monkeypatch.setattr(control, "close", retire_failed_worker) + control.state = "failed" + control.outcome_unknown = True + control.failure = { + "code": "ApplicationCommandOutcomeUnknown", + "reason_code": "mqtt_response_timeout", + "stop_command_attempted": True, + "stop_publish_attempts": 1, + "safe_to_retry": False, + } + service._acquire_application_control_process_lease() # noqa: SLF001 + + recovered = service.state() + + assert close_calls == 1 + assert recovered["application_control_session"]["state"] == "idle" + assert recovered["acquisition"]["state"] == "awaiting_external_stop" + assert recovered["source_mode"] == "live" + assert recovered["physical_command"]["status"] == "unresolved" + assert recovered["physical_command"]["requires_reconciliation"] is True + assert fixture.runtime.stop_calls == 0 + assert control.stop_calls == 1 + assert service._application_control_process_lease_holders == set() # noqa: SLF001 + + @pytest.mark.parametrize("qos2_completed", [True, False]) def test_stop_success_without_terminal_status_times_out_into_local_only_recovery( tmp_path: Path, @@ -13729,7 +13837,18 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera( binding=binding, resolved=True, ) - monkeypatch.setattr(service._physical_command_coordinator, "snapshot", lambda: physical) # noqa: SLF001 + physical_snapshot_calls = 0 + + def snapshot_physical() -> dict[str, object]: + nonlocal physical_snapshot_calls + physical_snapshot_calls += 1 + return physical + + monkeypatch.setattr( + service._physical_command_coordinator, # noqa: SLF001 + "snapshot", + snapshot_physical, + ) runtime.pcl_frames = 1 frame = DecodedPointCloudView( context=ConsumerFrameContext( @@ -13818,9 +13937,11 @@ def test_confirmed_scanning_waits_for_first_authoritative_pcl_before_camera( ) # Every later authoritative PCL for the same lineage is idempotent. + physical_snapshot_calls_before = physical_snapshot_calls service._observe_published_runtime_envelope(frame, runtime.producer_generation) # noqa: SLF001 time.sleep(0.05) assert len(events) == 2 + assert physical_snapshot_calls == physical_snapshot_calls_before def test_stale_post_publish_pcl_cannot_activate_camera( @@ -15316,6 +15437,85 @@ def test_active_stream_recovery_scanning_resumes_same_physical_lineage_without_c ) +def test_active_stream_recovery_privacy_bridge_retains_exact_route_for_read_only_rebind( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + service, runtime = service_with_fake_runtime(tmp_path) + control, physical = _install_composite_active_recovery_fixture( + service, + runtime, + monkeypatch, + ) + service.state() + lineage = service._active_stream_recovery_lineage # noqa: SLF001 + assert lineage is not None + supervisor = service._connection_supervisor # noqa: SLF001 + baseline = supervisor.snapshot() + candidate = replace( + baseline, + device_identity=replace(baseline.device_identity, state="unverified"), + control_plane=replace( + baseline.control_plane, + state="lost", + session_id=None, + ), + lease=replace(baseline.lease, state="configured-unverified"), + authority=replace( + baseline.authority, + control_allowed=False, + acquisition_start_allowed=False, + ), + ) + start_projects_before = list(control.start_projects) + stop_calls_before = control.stop_calls + operation_journal_before = service._operations.snapshot() # noqa: SLF001 + physical_ledger_before = service._physical_command_ledger.snapshot() # noqa: SLF001 + network_ledger_before = service._network_mutation_ledger.snapshot() # noqa: SLF001 + monkeypatch.setattr(facade_module, "_inspect_host_path", _direct_host_path) + monkeypatch.setattr( + service._host_wifi_association_probe, # noqa: SLF001 + "observe", + lambda *_args, **_kwargs: { + "schema_version": 1, + "adapter": "CoreWLAN", + "wifi_interface": True, + "association_state": "unavailable", + "evidence_quality": "unavailable", + "continuity_proven": False, + "continuity_token": "d" * 64, + "reason_code": "association-identity-unavailable", + }, + ) + monkeypatch.setattr( + supervisor, + "association_timeout_retention_candidate", + lambda *, expected_target: ( + candidate + if expected_target + == EndpointTarget(lineage.target_ipv4, lineage.target_port) + else None + ), + ) + + sampled = service._sample_host_path(lineage.target_ipv4) # noqa: SLF001 + + assert sampled.available is True + assert sampled.reason_code is None + assert sampled.fingerprint == baseline.host_path.fingerprint + assert sampled.kernel_route_fingerprint == ( + baseline.host_path.kernel_route_fingerprint + ) + assert control.start_projects == start_projects_before + assert control.stop_calls == stop_calls_before + assert service._operations.snapshot() == operation_journal_before # noqa: SLF001 + assert ( # noqa: SLF001 + service._physical_command_ledger.snapshot() == physical_ledger_before + ) + assert service._network_mutation_ledger.snapshot() == network_ledger_before # noqa: SLF001 + assert service._physical_command_coordinator.snapshot() is physical # noqa: SLF001 + + def test_active_stream_control_adoption_gets_fresh_budget_after_slow_proof( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -15638,7 +15838,7 @@ def test_active_stream_recovery_wrong_identity_blocks_without_retry_or_camera_re monkeypatch.setattr( service, "_probe_control_endpoint", - lambda _target: facade_module._CorrelatedEndpointObservation( # noqa: SLF001 + lambda _target, **_kwargs: facade_module._CorrelatedEndpointObservation( # noqa: SLF001 path=_direct_host_path(lineage.target_ipv4), reachable=True, reason_code=None, @@ -15699,8 +15899,14 @@ def test_active_stream_recovery_owned_path_is_read_only_and_resumes_exact_lineag events.append(("monitor", True)) return True - def probe_exact_target(target_ipv4: str) -> facade_module._CorrelatedEndpointObservation: # noqa: SLF001 - events.append(("probe", target_ipv4)) + def probe_exact_target( # noqa: SLF001 + target_ipv4: str, + *, + association_timeout_seconds: float, + ) -> facade_module._CorrelatedEndpointObservation: + events.append( + ("probe", (target_ipv4, association_timeout_seconds)) + ) return facade_module._CorrelatedEndpointObservation( # noqa: SLF001 path=_direct_host_path(target_ipv4), reachable=True, @@ -15770,7 +15976,13 @@ def test_active_stream_recovery_owned_path_is_read_only_and_resumes_exact_lineag assert events == [ ("monitor", True), ("lease-acquire", "network"), - ("probe", lineage.target_ipv4), + ( + "probe", + ( + lineage.target_ipv4, + facade_module.ACTIVE_STREAM_RECOVERY_ASSOCIATION_TIMEOUT_SECONDS, + ), + ), ( "device-info-status", ( @@ -15910,6 +16122,7 @@ def test_incident_recovery_retires_epoch_one_control_and_retries_fresh_inspectio def probe_current_epoch( target_ipv4: str, + **_kwargs: object, ) -> facade_module._CorrelatedEndpointObservation: # noqa: SLF001 path = next(probe_paths) service._observe_connection_transport( # noqa: SLF001 @@ -16065,7 +16278,7 @@ def test_active_stream_recovery_waits_for_terminal_control_worker_retirement( monkeypatch.setattr( service, "_probe_control_endpoint", - lambda _target: facade_module._CorrelatedEndpointObservation( # noqa: SLF001 + lambda _target, **_kwargs: facade_module._CorrelatedEndpointObservation( # noqa: SLF001 path=_direct_host_path(lineage.target_ipv4), reachable=True, reason_code=None, @@ -16140,7 +16353,7 @@ def test_active_stream_recovery_exact_device_system_fault_is_terminal_without_co monkeypatch.setattr( service, "_probe_control_endpoint", - lambda _target: facade_module._CorrelatedEndpointObservation( # noqa: SLF001 + lambda _target, **_kwargs: facade_module._CorrelatedEndpointObservation( # noqa: SLF001 path=_direct_host_path(lineage.target_ipv4), reachable=True, reason_code=None, @@ -16223,7 +16436,7 @@ def test_active_stream_recovery_bootstrap_system_error_is_normalized_terminal_fa monkeypatch.setattr( service, "_probe_control_endpoint", - lambda _target: facade_module._CorrelatedEndpointObservation( # noqa: SLF001 + lambda _target, **_kwargs: facade_module._CorrelatedEndpointObservation( # noqa: SLF001 path=_direct_host_path(lineage.target_ipv4), reachable=True, reason_code=None, @@ -16476,6 +16689,111 @@ def test_active_stream_recovery_never_retries_invalid_fmp4_after_media_commit( assert control.stop_calls == 0 +def test_bound_scan_over_without_stop_dominates_late_empty_camera_epoch_failure( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + service, runtime = service_with_fake_runtime(tmp_path) + control, physical = _install_composite_active_recovery_fixture( + service, + runtime, + monkeypatch, + runtime_phase="live", + camera_phase="streaming", + ) + camera = service.camera_preview.snapshot() + recording = camera["recording"] + assert isinstance(recording, dict) + recording.update( + { + "active": True, + "active_epoch": 2, + "committed_media_segment_count": 4_833, + "producer_alive": False, + "completed_epochs": 2, + "last_summary": { + "codec_epoch": 2, + "status": "failed", + "media_segment_count": 0, + "failure_code": "invalid-fmp4", + }, + } + ) + camera.update( + { + "phase": "error", + "error": { + "code": "invalid-fmp4", + "message": "synthetic empty post-power-loss epoch", + }, + } + ) + physical.update( + { + "runtime_bound": True, + "reconciliation_ready": True, + "observed_session_state": "scan_over", + } + ) + control.state = "failed" + control.state_revision += 1 + control.failure = { + "reason_code": "application_acceptance_failed", + "failed_phase": "scanning", + "modeling_command_attempted": True, + "stop_command_attempted": False, + "diagnostic_snapshot_unavailable": [], + "diagnostic_evidence_unavailable": [], + "safe_to_retry": False, + } + + def scan_over_control_snapshot() -> dict[str, object]: + snapshot = FakeInteractiveControlSession.snapshot(control) + snapshot["transport"] = { + "state": "failed", + "publish_attempts": 7, + "device_status_reports": 9, + "latest_device_session_state": "scan_over", + "latest_device_project_bound": True, + "latest_device_init_ready": False, + "latest_system_error_code": None, + "automatic_retry": False, + "automatic_reconnect": False, + } + return snapshot + + monkeypatch.setattr(control, "snapshot", scan_over_control_snapshot) + monkeypatch.setattr(service, "_seal_acquisition_capture_clock", lambda: None) + camera_stop_calls: list[str] = [] + monkeypatch.setattr( + service.camera_preview, + "stop_current", + lambda: camera_stop_calls.append("stop") or {}, + ) + start_projects_before = list(control.start_projects) + stop_calls_before = control.stop_calls + + terminal = service.state() + + assert terminal["acquisition"]["state"] == "interrupted" + assert terminal["acquisition"]["message_code"] == ( + "acquisition.recovery.device_standby_observed" + ) + assert terminal["acquisition"]["result"] == { + "receiver_stopped": True, + "device_state": "scan_over", + "device_stop": "not-sent", + "automatic_command_retry": False, + "read_only_recovery": True, + "physical_reconciliation_required": True, + } + assert terminal["acquisition"]["cleanup_pending"] is False + assert camera_stop_calls == ["stop"] + assert runtime.stop_calls == 1 + assert control.start_projects == start_projects_before + assert control.stop_calls == stop_calls_before == 0 + + def test_active_stream_recovery_reopens_exact_camera_epoch_once_and_fences_stale_lineage( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -24955,6 +25273,59 @@ def test_configured_unverified_monitor_keeps_same_route_without_promoting_author assert "verify-control-device-info" in retained.allowed_actions +def test_monitor_layer_retains_exact_configured_route_across_association_lock_timeout( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + service, _ = service_with_fake_runtime(tmp_path) + binding = _seed_supervised_connection(service, with_control=False) + baseline = service._connection_supervisor.snapshot() # noqa: SLF001 + assert baseline.host_path.kernel_route_fingerprint is not None + tcp_calls: list[str] = [] + + monkeypatch.setattr( + service, + "_sample_host_path", + lambda target, **_kwargs: HostPathProbeResult( + available=False, + fingerprint=None, + interface=baseline.host_path.interface, + source_ipv4=baseline.host_path.source_ipv4, + route_class="unavailable", + reason_code="host-wifi-operation-timeout", + observation_failure_class="association-observer", + kernel_route_fingerprint=baseline.host_path.kernel_route_fingerprint, + ), + ) + monkeypatch.setattr( + facade_module, + "_probe_control_endpoint_socket", + lambda target: ( + tcp_calls.append(target) + or facade_module.TcpReachabilityProbeResult(reachable=True) + ), + ) + + async def poll_three_times() -> list[facade_module.ConnectionSupervisorSnapshot]: + return [ + await service._connection_monitor.poll_once() # noqa: SLF001 + for _ in range(3) + ] + + snapshots = asyncio.run(poll_three_times()) + + assert tcp_calls == [binding.target_ipv4] * 3 + for retained in snapshots: + assert retained.host_path.available is True + assert retained.host_path.epoch == baseline.host_path.epoch + assert retained.host_path.fingerprint == baseline.host_path.fingerprint + assert retained.endpoint.tcp_state == "reachable" + assert retained.device_identity.state == "unverified" + assert retained.authority.control_allowed is False + assert retained.authority.acquisition_start_allowed is False + assert "verify-control-device-info" in retained.allowed_actions + + def test_association_timeout_cannot_hide_a_real_kernel_route_change( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, @@ -29825,6 +30196,91 @@ def test_explicit_verify_reconciles_persisted_start_to_ready_without_device_io( assert service._application_control_process_lease_holders == {"control"} # noqa: SLF001 +def test_long_first_verify_refreshes_aged_transport_before_exposing_prestart_control( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """A long checkpoint reconciliation cannot orphan its new ready socket.""" + + service, _ = service_with_fake_runtime(tmp_path) + coordinator = _VerifyPhysicalRecoveryCoordinator( + observed_session_state="ready", + reconciliation_ready=True, + ) + _install_synthetic_verify_recovery(service, coordinator) + supervisor = service._connection_supervisor # noqa: SLF001 + monotonic_now = [100.0] + suspend_aware_now = [1_000.0] + supervisor._monotonic_clock = lambda: monotonic_now[0] # noqa: SLF001 + supervisor._suspend_aware_clock = lambda: suspend_aware_now[0] # noqa: SLF001 + before = supervisor.snapshot() + assert before.intent is not None + assert before.device_network.target is not None + host_epoch = supervisor.observe_host_path( + _association_bound_direct_host_path(before.device_network.target.ipv4) + ) + assert supervisor.observe_endpoint( + target=before.device_network.target, + intent_id=before.intent.intent_id, + host_path_epoch=host_epoch, + reachable=True, + ) + + monkeypatch.setattr(facade_module, "_inspect_host_path", _direct_host_path) + tcp_probes: list[str] = [] + + def reachable(target: str) -> bool: + tcp_probes.append(target) + return True + + monkeypatch.setattr(facade_module, "_control_endpoint_reachable", reachable) + reconcile = service._reconcile_physical_command_after_verify_owned # noqa: SLF001 + + async def reconcile_after_transport_ages( + bound_service: XgridsK1CompatibilityService, + *, + verify_operation_id: str, + allow_receiver_rehydrate: bool = True, + ) -> dict[str, Any]: + result = await reconcile( + verify_operation_id=verify_operation_id, + allow_receiver_rehydrate=allow_receiver_rehydrate, + ) + monotonic_now[0] += 20.0 + suspend_aware_now[0] += 20.0 + return result + + service._reconcile_physical_command_after_verify_owned = MethodType( # type: ignore[method-assign] # noqa: SLF001 + reconcile_after_transport_ages, + service, + ) + operation_id = "op-00000000-0000-4000-8000-000000001202" + + verified = asyncio.run( + service.verify_connection( + _retained_physical_recovery_verify_request(operation_id=operation_id) + ) + ) + + assert tcp_probes == ["192.168.1.20"] + assert verified["last_operation"]["status"] == "succeeded" + assert verified["active_connection_mode"] == "bridge" + assert verified["connection_policy"]["actions"]["start-acquisition"]["allowed"] is True + assert service._application_control_session.snapshot()["state"] == "connection-ready" # noqa: SLF001 + assert service._application_control_process_lease_holders == {"control"} # noqa: SLF001 + fresh = supervisor.snapshot() + assert fresh.intent is not None + assert fresh.device_network.target is not None + assert supervisor.endpoint_observation_has_remaining_lease( + target=fresh.device_network.target, + intent_id=fresh.intent.intent_id, + host_path_epoch=fresh.host_path.epoch, + minimum_remaining_seconds=( + facade_module.VERIFY_POST_TRANSITION_ENDPOINT_MIN_REMAINING_SECONDS + ), + ) + + def test_explicit_verify_reconciliation_does_not_require_normal_command_authority( tmp_path: Path, ) -> None: diff --git a/tests/test_xgrids_active_acquisition_checkpoint_integration.py b/tests/test_xgrids_active_acquisition_checkpoint_integration.py index 4f3dd9a..fc02a63 100644 --- a/tests/test_xgrids_active_acquisition_checkpoint_integration.py +++ b/tests/test_xgrids_active_acquisition_checkpoint_integration.py @@ -1843,6 +1843,25 @@ def test_chained_stop_facade_cessation_uses_real_ledger_ancestry( stale_reconciliation, final_record, ) + # Reproduce the append-only field shape: an otherwise valid historical + # standby reconciliation is inherited by the newer ordinary STOP. The + # checkpoint reducer must use the current STOP's own READY status, not the + # last historical reconciliation merely because it is terminal standby. + snapshot_with_stale_standby = replace( + ledger.snapshot(), + record=replace( + final_record, + reconciliations=( + *final_record.reconciliations, + stale_reconciliation, + ), + ), + ) + monkeypatch.setattr( + ledger, + "snapshot", + lambda: snapshot_with_stale_standby, + ) service._application_control_session.snapshot = lambda: { # type: ignore[method-assign] "verified_control": _verified_control(first_rebind) @@ -1862,6 +1881,88 @@ def test_chained_stop_facade_cessation_uses_real_ledger_ancestry( assert checkpoint.cessation_physical_proof.ancestor_chain == ancestry +def test_restart_ceases_active_checkpoint_after_composite_ordinary_stop( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + service = _service(tmp_path, monkeypatch) + connection = _connection( + control_session_id="control-before-ordinary-stop-restart", + host_path_epoch=1, + producer_generation=1, + ) + _prepare_and_activate_checkpoint(service, connection) + ledger = service._physical_command_ledger # noqa: SLF001 + start = ledger.snapshot().record + assert start is not None and start.operation_id == START_OPERATION_ID + identity = PhysicalCommandIdentity( + vendor_device_id_sha256=VENDOR_SHA256, + device_serial_sha256=SERIAL_SHA256, + ) + stop_operation_id = "physical-stop-ordinary-restart" + ledger.prepare( + operation_id=stop_operation_id, + parent_operation_id=START_OPERATION_ID, + acquisition_id=ACQUISITION_ID, + action="stop", + identity=identity, + connection=connection, + compatibility_profile_id=XGRIDS_K1_COMPATIBILITY_PROFILE_ID, + payload_sha256="6" * 64, + baseline_status=_status( + connection, + "scanning", + observed_at_utc="2026-08-13T12:02:00.000Z", + ), + ) + ledger.mark_dispatching(stop_operation_id) + ledger.mark_observing( + stop_operation_id, + publish_call_returned=True, + packet_id=43, + ) + ledger.mark_qos2_completed(stop_operation_id, packet_id=43) + ledger.record_application_response( + stop_operation_id, + PhysicalCommandApplicationResponse( + operation_id=stop_operation_id, + action="stop", + control_session_id=connection.control_session_id, + host_path_epoch=connection.host_path_epoch, + producer_generation=connection.producer_generation, + result_code=PHYSICAL_COMMAND_APPLICATION_SUCCESS_CODE, + success=True, + payload_sha256="7" * 64, + observed_at_utc="2026-08-13T12:02:01.000Z", + ), + ) + ledger.record_status_observation( + stop_operation_id, + _status( + connection, + "ready", + observed_at_utc="2026-08-13T12:02:02.000Z", + ), + ) + stop = ledger.resolve( + stop_operation_id, + resolution="stop-standby-observed", + ) + before_restart = ActiveAcquisitionRecoveryCheckpointStore(tmp_path).snapshot() + assert before_restart.status == "active" + + restarted = XgridsK1CompatibilityService(tmp_path) + + after_restart = ActiveAcquisitionRecoveryCheckpointStore(tmp_path).snapshot() + assert after_restart.status == "ceased" + assert after_restart.checkpoint is not None + assert after_restart.checkpoint.cessation_physical_proof is not None + assert after_restart.checkpoint.cessation_physical_proof.operation_id == ( + stop.operation_id + ) + assert restarted._active_acquisition_checkpoint_trust == "trusted" # noqa: SLF001 + + def test_restart_ready_classification_ceases_active_checkpoint_after_undispatched_stop( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/test_xgrids_active_acquisition_restart_rehydration.py b/tests/test_xgrids_active_acquisition_restart_rehydration.py index d996e08..0005d4a 100644 --- a/tests/test_xgrids_active_acquisition_restart_rehydration.py +++ b/tests/test_xgrids_active_acquisition_restart_rehydration.py @@ -887,7 +887,7 @@ def test_ambiguous_prepared_restart_ready_settles_after_dhcp_target_change( mqtt_retained=False, observed_at_utc="2026-08-13T12:20:01.000Z", ) - record = restarted._physical_command_ledger.reconcile_ambiguous( # noqa: SLF001 + restarted._physical_command_ledger.reconcile_ambiguous( # noqa: SLF001 restart_support.START_OPERATION_ID, reconciliation_id="reconciliation-ready-new-dhcp-address", resolution="physical-standby-observed", @@ -1057,7 +1057,7 @@ def test_restart_receiver_link_loss_before_first_pcl_rebinds_read_only_then_prom monkeypatch.setattr( restarted, "_probe_control_endpoint", - lambda target_ipv4: restart_support.facade_module._CorrelatedEndpointObservation( # noqa: SLF001 + lambda target_ipv4, **_kwargs: restart_support.facade_module._CorrelatedEndpointObservation( # noqa: SLF001 path=lifecycle_support._direct_host_path(target_ipv4), # noqa: SLF001 reachable=True, reason_code=None, @@ -1237,7 +1237,7 @@ def test_ambiguous_prepared_restart_second_rebind_promotes_only_after_fresh_pcl( monkeypatch.setattr( restarted, "_probe_control_endpoint", - lambda target_ipv4: restart_support.facade_module._CorrelatedEndpointObservation( # noqa: SLF001 + lambda target_ipv4, **_kwargs: restart_support.facade_module._CorrelatedEndpointObservation( # noqa: SLF001 path=lifecycle_support._direct_host_path(target_ipv4), # noqa: SLF001 reachable=True, reason_code=None, diff --git a/tests/test_xgrids_application_session.py b/tests/test_xgrids_application_session.py index 05c084e..8f296a4 100644 --- a/tests/test_xgrids_application_session.py +++ b/tests/test_xgrids_application_session.py @@ -394,7 +394,10 @@ def _wait_phase( session: InteractiveApplicationControlSession, expected: str, ) -> dict[str, object]: - deadline = time.monotonic() + 2.0 + # FakeExecutor may use its full two-second checkpoint timeout before the + # worker publishes the terminal state. Keep the observer deadline strictly + # larger so this helper does not race the transition it is asserting. + deadline = time.monotonic() + 3.0 while time.monotonic() < deadline: snapshot = session.snapshot() if snapshot["state"] == expected: