From 2a97cf28c01143f17850a85b35a8589fd22ed155 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 29 Jul 2026 01:55:12 +0300 Subject: [PATCH] fix(k1): stabilize repeated acquisition and live viewer recovery --- .../src/components/RerunViewport.tsx | 206 ++++- .../core/observation/liveReceiverWatchdog.ts | 156 ++++ .../core/observation/liveViewerDiagnostics.ts | 61 ++ .../src/core/runtime/contracts.ts | 1 + .../src/workspaces/Workspaces.tsx | 3 +- .../test/devicePluginContracts.test.mjs | 5 + .../test/liveReceiverWatchdog.test.mjs | 94 ++ .../rerunViewportAtomicAdmission.test.mjs | 14 +- plugins/xgrids-k1/frontend/src/api.ts | 44 + .../xgrids-k1/frontend/src/runtimeContext.tsx | 8 + .../frontend/src/useXgridsK1Runtime.ts | 6 +- src/k1link/compute/live_perception.py | 20 +- .../xgrids_k1/ble/wifi_provisioning.py | 15 +- src/k1link/device_plugins/xgrids_k1/facade.py | 852 +++++++++++++++--- .../xgrids_k1/protocol/application_mqtt.py | 52 +- .../xgrids_k1/protocol/application_session.py | 201 ++++- .../xgrids_k1/viewer/runtime.py | 51 +- src/k1link/viewer/rerun_bridge.py | 48 +- src/k1link/web/app.py | 6 + src/k1link/web/runtime_diagnostics.py | 97 ++ src/k1link/web/viewer_diagnostics_api.py | 66 ++ tests/test_live_perception.py | 25 + tests/test_rerun_bridge.py | 56 +- tests/test_viewer_diagnostics_api.py | 142 +++ tests/test_wifi_provisioning.py | 57 ++ tests/test_xgrids_acquisition_lifecycle.py | 535 ++++++++++- tests/test_xgrids_application_mqtt.py | 33 + tests/test_xgrids_application_session.py | 208 +++++ 28 files changed, 2776 insertions(+), 286 deletions(-) create mode 100644 apps/control-station/src/core/observation/liveReceiverWatchdog.ts create mode 100644 apps/control-station/src/core/observation/liveViewerDiagnostics.ts create mode 100644 apps/control-station/test/liveReceiverWatchdog.test.mjs create mode 100644 src/k1link/web/runtime_diagnostics.py create mode 100644 src/k1link/web/viewer_diagnostics_api.py create mode 100644 tests/test_viewer_diagnostics_api.py diff --git a/apps/control-station/src/components/RerunViewport.tsx b/apps/control-station/src/components/RerunViewport.tsx index b55ed71..5a198ed 100644 --- a/apps/control-station/src/components/RerunViewport.tsx +++ b/apps/control-station/src/components/RerunViewport.tsx @@ -1,6 +1,15 @@ import { useEffect, useRef, useState } from "react"; import type { SceneSettings } from "../sceneSettings"; +import { + advanceLiveReceiverWatchdog, + initialLiveReceiverRecoveryState, + initialLiveReceiverWatchdogState, + LIVE_RECEIVER_MAX_RECOVERY_ATTEMPTS, + requestLiveReceiverRecovery, +} from "../core/observation/liveReceiverWatchdog"; +import { postLiveViewerDiagnostic } from "../core/observation/liveViewerDiagnostics"; +import type { LiveViewerFailureStage } from "../core/observation/liveViewerDiagnostics"; import type { RecordedAdmissionPhase } from "../core/observation/recordedSessionAdmission"; export type RerunViewportStatus = "idle" | "loading" | "ready" | "error"; @@ -66,6 +75,8 @@ export interface RerunViewportProps { sourceUrl: string; recordedArtifact?: RecordedRrdArtifactDescriptor | null; followLive?: boolean; + liveActivitySequence?: number | null; + liveStreamId?: string | null; autoplayWhenReady?: boolean; presentationGate?: RecordedAdmissionPhase; expectedTimelineStartSeconds?: number; @@ -796,6 +807,8 @@ export function RerunViewport({ sourceUrl, recordedArtifact = null, followLive = false, + liveActivitySequence = null, + liveStreamId = null, autoplayWhenReady = false, presentationGate = "ready", expectedTimelineStartSeconds, @@ -824,6 +837,11 @@ export function RerunViewport({ const [status, setStatus] = useState(sourceUrl ? "loading" : "idle"); const [recordingBufferProgress, setRecordingBufferProgress] = useState(null); const [retryNonce, setRetryNonce] = useState(0); + const liveActivitySequenceRef = useRef(liveActivitySequence); + liveActivitySequenceRef.current = liveActivitySequence; + const liveStreamIdRef = useRef(liveStreamId); + liveStreamIdRef.current = liveStreamId; + const liveRecoveryRef = useRef(initialLiveReceiverRecoveryState()); const blueprintChannelRef = useRef(null); const perceptionChannelRef = useRef(null); const loadedPerceptionChannelRef = useRef(null); @@ -849,6 +867,10 @@ export function RerunViewport({ recordedArtifact !== null, ); + useEffect(() => { + liveRecoveryRef.current = initialLiveReceiverRecoveryState(); + }, [followLive, liveStreamId, sourceUrl]); + useEffect(() => { const normalizedSource = sourceUrl.trim(); const host = hostRef.current; @@ -889,6 +911,7 @@ export function RerunViewport({ let disposed = false; let disposeViewer: (() => void) | undefined; let recordingOpenTimer: number | undefined; + let liveRecordingDiscoveryTimer: number | undefined; let recordedOpenWatchdog: { arm: () => void; clear: () => void; @@ -898,6 +921,8 @@ export function RerunViewport({ let recordingOpened = false; let recordingOpenTimedOut = false; let viewerStartResolved = false; + let latestLiveRangeMaxNs: number | null = null; + let liveWatchdog = initialLiveReceiverWatchdogState(); // The backend serves this exact digest-bound URL or rejects it with 412. // Presentation still waits for Rerun to decode the complete declared range. const artifactVerified = true; @@ -925,9 +950,16 @@ export function RerunViewport({ recordingOpenTimer = undefined; } }; + const clearLiveRecordingDiscoveryTimer = () => { + if (liveRecordingDiscoveryTimer !== undefined) { + window.clearInterval(liveRecordingDiscoveryTimer); + liveRecordingDiscoveryTimer = undefined; + } + }; const clearRecordingTimers = () => { clearRecordedAdmissionWatchdog(); clearLiveRecordingOpenTimer(); + clearLiveRecordingDiscoveryTimer(); }; const clearPlaybackRangeTimer = () => { if (playbackRangeTimer === undefined) return; @@ -950,11 +982,118 @@ export function RerunViewport({ requestFrame: (callback) => window.requestAnimationFrame(callback), cancelFrame: (handle) => window.cancelAnimationFrame(handle), }); - const reportError = (message: string) => { + const reportError = (message: string, failureStage?: LiveViewerFailureStage) => { if (disposed) return; + if (followLive) { + postLiveViewerDiagnostic({ + eventCode: "live_receiver_error", + failureStage, + streamId: liveStreamIdRef.current, + backendActivitySequence: liveActivitySequenceRef.current, + viewerRangeMaxNs: latestLiveRangeMaxNs, + recoveryAttempt: liveRecoveryRef.current.attempts || null, + }); + } setStatus("error"); onStatusChange?.("error", message); }; + const requestLiveRecovery = ( + failureStage: LiveViewerFailureStage, + stalledForMs = 0, + emitErrorEvent = true, + ) => { + if (!followLive || disposed) return false; + if (emitErrorEvent) { + postLiveViewerDiagnostic({ + eventCode: "live_receiver_error", + failureStage, + streamId: liveStreamIdRef.current, + backendActivitySequence: liveActivitySequenceRef.current, + viewerRangeMaxNs: latestLiveRangeMaxNs, + stalledForMs: Math.round(stalledForMs) || null, + recoveryAttempt: liveRecoveryRef.current.attempts || null, + }); + } + const recovery = requestLiveReceiverRecovery(liveRecoveryRef.current); + liveRecoveryRef.current = recovery.state; + if (recovery.signal === "exhausted") { + postLiveViewerDiagnostic({ + eventCode: "live_receiver_recovery_exhausted", + failureStage, + streamId: liveStreamIdRef.current, + backendActivitySequence: liveActivitySequenceRef.current, + viewerRangeMaxNs: latestLiveRangeMaxNs, + stalledForMs: Math.round(stalledForMs) || null, + recoveryAttempt: LIVE_RECEIVER_MAX_RECOVERY_ATTEMPTS, + }); + setStatus("error"); + onStatusChange?.( + "error", + "Живой поток продолжает поступать, но визуализатор не восстановился после трёх переподключений.", + ); + return true; + } + postLiveViewerDiagnostic({ + eventCode: "live_receiver_restart_requested", + failureStage, + streamId: liveStreamIdRef.current, + backendActivitySequence: liveActivitySequenceRef.current, + viewerRangeMaxNs: latestLiveRangeMaxNs, + stalledForMs: Math.round(stalledForMs) || null, + recoveryAttempt: recovery.attempt, + }); + setStatus("loading"); + onStatusChange?.( + "loading", + "Живой визуализатор переподключается к продолжающемуся потоку.", + ); + disposeViewer?.(); + setRetryNonce((nonce) => nonce + 1); + return true; + }; + const observeLiveReceiver = (viewerRangeMaxNs: number | null) => { + if (!followLive || disposed) return; + latestLiveRangeMaxNs = viewerRangeMaxNs; + const receiverOpenedAfterRecovery = + liveRecoveryRef.current.awaitingRecovery && + liveWatchdog.lastViewerRangeMaxNs === null && + viewerRangeMaxNs !== null; + const observed = advanceLiveReceiverWatchdog(liveWatchdog, { + nowMs: Date.now(), + backendActivitySequence: liveActivitySequenceRef.current, + viewerRangeMaxNs, + }); + liveWatchdog = observed.state; + if ( + (receiverOpenedAfterRecovery || observed.signal === "receiver-advanced") && + liveRecoveryRef.current.awaitingRecovery + ) { + postLiveViewerDiagnostic({ + eventCode: "live_receiver_recovered", + streamId: liveStreamIdRef.current, + backendActivitySequence: liveActivitySequenceRef.current, + viewerRangeMaxNs, + recoveryAttempt: liveRecoveryRef.current.attempts, + }); + liveRecoveryRef.current = initialLiveReceiverRecoveryState(); + return; + } + if (observed.signal !== "stalled") return; + + postLiveViewerDiagnostic({ + eventCode: "live_receiver_stalled", + failureStage: "receiver-stalled", + streamId: liveStreamIdRef.current, + backendActivitySequence: liveActivitySequenceRef.current, + viewerRangeMaxNs, + stalledForMs: Math.round(observed.stalledForMs), + recoveryAttempt: Math.min( + liveRecoveryRef.current.attempts + 1, + LIVE_RECEIVER_MAX_RECOVERY_ATTEMPTS, + ), + }); + requestLiveRecovery("receiver-stalled", observed.stalledForMs, false); + }; host.replaceChildren(); appliedPointColorKeyRef.current = null; @@ -1016,7 +1155,7 @@ export function RerunViewport({ viewer.stop(); } catch { // A partially initialized WASM handle can already be gone after a - // startup failure. The host still has to be cleared below. + // startup failure. } host.replaceChildren(); }; @@ -1034,14 +1173,20 @@ export function RerunViewport({ }); } - unsubscribers.push( - viewer.on("recording_open", (event) => { + const admitRecording = (event: { + application_id: string; + recording_id: string; + }) => { if ( disposed || recordingOpened || (isRecordedSource && event.application_id !== "nodedc_mission_core_recorded") ) return; recordingOpened = true; + if (!isRecordedSource) { + clearLiveRecordingOpenTimer(); + clearLiveRecordingDiscoveryTimer(); + } if ( recordedBlueprintUrl && event.application_id === "nodedc_mission_core_recorded" && @@ -1060,7 +1205,6 @@ export function RerunViewport({ setPerceptionChannelRevision((revision) => revision + 1); } } - if (!isRecordedSource) clearLiveRecordingOpenTimer(); // Durable Mission Core recordings use a zero-based session clock. // Keeping wall-clock capture_time as a secondary RRD timeline is // useful for audit, but the operator scrubber must remain a small, @@ -1144,6 +1288,7 @@ export function RerunViewport({ } catch { return; } + observeLiveReceiver(rangeNs?.max ?? null); const recordedBuffer = recordedPlaybackBufferState( rangeNs, expectedTimelineEndSeconds, @@ -1225,8 +1370,31 @@ export function RerunViewport({ clearPlaybackRangeTimer(); publishBufferedState(); playbackRangeTimer = window.setInterval(publishBufferedState, 500); - }), + }; + unsubscribers.push( + viewer.on("recording_open", admitRecording), ); + const discoverActiveLiveRecording = () => { + if (isRecordedSource || disposed || recordingOpened || !viewer.ready) return; + try { + const recordingId = viewer.get_active_recording_id(); + if (!recordingId) return; + // Rerun 0.34.1 may ingest an SDK gRPC store without forwarding its + // recording_open event to the JavaScript wrapper. The active store + // is the authoritative fallback and avoids hiding a ready canvas. + postLiveViewerDiagnostic({ + eventCode: "live_receiver_active_store_admitted", + streamId: liveStreamIdRef.current, + backendActivitySequence: liveActivitySequenceRef.current, + }); + admitRecording({ + application_id: "nodedc_mission_core_spatial", + recording_id: recordingId, + }); + } catch { + // The native receiver is still opening; the bounded timer retries. + } + }; unsubscribers.push( viewer.on("time_update", (event) => { @@ -1316,25 +1484,34 @@ export function RerunViewport({ setPerceptionChannelRevision((revision) => revision + 1); } - if (!recordingOpened && !isRecordedSource) { - recordingOpenTimer = window.setTimeout(() => { - recordingOpenTimedOut = true; - disposeViewer?.(); - reportError("Визуализатор запущен, но поток записи не открылся."); - }, 12_000); + if (!isRecordedSource) { + discoverActiveLiveRecording(); + if (!recordingOpened) { + liveRecordingDiscoveryTimer = window.setInterval( + discoverActiveLiveRecording, + 100, + ); + recordingOpenTimer = window.setTimeout(() => { + recordingOpenTimedOut = true; + requestLiveRecovery("recording-open-timeout"); + }, 12_000); + } } } catch { disposeViewer(); if (!recordingOpenTimedOut && !disposed) { - reportError(isRecordedSource + const message = isRecordedSource ? "Запись не прошла атомарную проверку целостности." - : "Не удалось запустить встроенный визуализатор."); + : "Не удалось запустить встроенный визуализатор."; + if (!isRecordedSource && requestLiveRecovery("viewer-start")) return; + reportError(message); } } }) .catch(() => { if (disposed) return; disposeViewer?.(); + if (requestLiveRecovery("module-load")) return; reportError("Не удалось загрузить модуль визуализатора."); }); @@ -1355,6 +1532,7 @@ export function RerunViewport({ expectedTimelineStartSeconds, followLive, initialPlaybackStartSeconds, + liveStreamId, onPlaybackChange, onPlaybackControllerChange, onSelectionChange, diff --git a/apps/control-station/src/core/observation/liveReceiverWatchdog.ts b/apps/control-station/src/core/observation/liveReceiverWatchdog.ts new file mode 100644 index 0000000..dc4c9da --- /dev/null +++ b/apps/control-station/src/core/observation/liveReceiverWatchdog.ts @@ -0,0 +1,156 @@ +export interface LiveReceiverWatchdogState { + lastBackendActivitySequence: number | null; + lastViewerRangeMaxNs: number | null; + staticRangeBackendAdvances: number; + stalledSinceMs: number | null; + stallReported: boolean; +} + +export type LiveReceiverWatchdogSignal = "none" | "receiver-advanced" | "stalled"; + +export interface LiveReceiverWatchdogResult { + state: LiveReceiverWatchdogState; + signal: LiveReceiverWatchdogSignal; + stalledForMs: number; +} + +export interface LiveReceiverRecoveryState { + attempts: number; + awaitingRecovery: boolean; +} + +export type LiveReceiverRecoverySignal = "retry" | "exhausted"; + +export interface LiveReceiverRecoveryResult { + state: LiveReceiverRecoveryState; + signal: LiveReceiverRecoverySignal; + attempt: number; +} + +export const LIVE_RECEIVER_STALL_THRESHOLD_MS = 5_000; +export const LIVE_RECEIVER_MAX_RECOVERY_ATTEMPTS = 3; + +export function initialLiveReceiverWatchdogState(): LiveReceiverWatchdogState { + return { + lastBackendActivitySequence: null, + lastViewerRangeMaxNs: null, + staticRangeBackendAdvances: 0, + stalledSinceMs: null, + stallReported: false, + }; +} + +export function initialLiveReceiverRecoveryState(): LiveReceiverRecoveryState { + return { + attempts: 0, + awaitingRecovery: false, + }; +} + +/** + * Bound viewer-only restarts independently from scanner and acquisition + * lifecycle. The caller may dispose and recreate the browser receiver, but + * must never issue START/STOP or reconnect the physical device. + */ +export function requestLiveReceiverRecovery( + current: LiveReceiverRecoveryState, + maxAttempts = LIVE_RECEIVER_MAX_RECOVERY_ATTEMPTS, +): LiveReceiverRecoveryResult { + if (current.attempts >= maxAttempts) { + return { + state: { + attempts: current.attempts, + awaitingRecovery: false, + }, + signal: "exhausted", + attempt: current.attempts, + }; + } + const attempt = current.attempts + 1; + return { + state: { + attempts: attempt, + awaitingRecovery: true, + }, + signal: "retry", + attempt, + }; +} + +/** + * Detect a disposable receiver stall only when the backend frame counter keeps + * advancing while Rerun's visible time range does not. Scanner commands are + * deliberately absent from this state machine. + */ +export function advanceLiveReceiverWatchdog( + current: LiveReceiverWatchdogState, + sample: { + nowMs: number; + backendActivitySequence: number | null; + viewerRangeMaxNs: number | null; + }, + thresholdMs = LIVE_RECEIVER_STALL_THRESHOLD_MS, +): LiveReceiverWatchdogResult { + const backendSequence = Number.isSafeInteger(sample.backendActivitySequence) && + (sample.backendActivitySequence ?? -1) >= 0 + ? sample.backendActivitySequence + : null; + const viewerRange = Number.isFinite(sample.viewerRangeMaxNs) && + (sample.viewerRangeMaxNs ?? -1) >= 0 + ? sample.viewerRangeMaxNs + : null; + const state = { ...current }; + + if (state.lastBackendActivitySequence === null && state.lastViewerRangeMaxNs === null) { + state.lastBackendActivitySequence = backendSequence; + state.lastViewerRangeMaxNs = viewerRange; + return { state, signal: "none", stalledForMs: 0 }; + } + + const backendAdvanced = backendSequence !== null && + (state.lastBackendActivitySequence === null || + backendSequence > state.lastBackendActivitySequence); + const receiverAdvanced = viewerRange !== null && + (state.lastViewerRangeMaxNs === null || viewerRange > state.lastViewerRangeMaxNs); + + if (backendSequence !== null) { + state.lastBackendActivitySequence = Math.max( + backendSequence, + state.lastBackendActivitySequence ?? 0, + ); + } + if (viewerRange !== null) { + state.lastViewerRangeMaxNs = Math.max(viewerRange, state.lastViewerRangeMaxNs ?? 0); + } + + if (receiverAdvanced) { + const hadStaticBackendProgress = state.stalledSinceMs !== null; + state.staticRangeBackendAdvances = 0; + state.stalledSinceMs = null; + state.stallReported = false; + return { + state, + signal: hadStaticBackendProgress ? "receiver-advanced" : "none", + stalledForMs: 0, + }; + } + + if (backendAdvanced) { + state.staticRangeBackendAdvances += 1; + if (state.staticRangeBackendAdvances >= 2 && state.stalledSinceMs === null) { + state.stalledSinceMs = sample.nowMs; + } + } + const stalledForMs = state.stalledSinceMs === null + ? 0 + : Math.max(0, sample.nowMs - state.stalledSinceMs); + if ( + !state.stallReported && + state.staticRangeBackendAdvances >= 2 && + stalledForMs >= thresholdMs + ) { + state.stallReported = true; + return { state, signal: "stalled", stalledForMs }; + } + return { state, signal: "none", stalledForMs }; +} diff --git a/apps/control-station/src/core/observation/liveViewerDiagnostics.ts b/apps/control-station/src/core/observation/liveViewerDiagnostics.ts new file mode 100644 index 0000000..0424bb9 --- /dev/null +++ b/apps/control-station/src/core/observation/liveViewerDiagnostics.ts @@ -0,0 +1,61 @@ +export type LiveViewerDiagnosticEventCode = + | "live_receiver_stalled" + | "live_receiver_restart_requested" + | "live_receiver_recovered" + | "live_receiver_recovery_exhausted" + | "live_receiver_active_store_admitted" + | "live_receiver_error"; + +export type LiveViewerFailureStage = + | "recording-open-timeout" + | "viewer-start" + | "module-load" + | "receiver-stalled"; + +export interface LiveViewerDiagnostic { + eventCode: LiveViewerDiagnosticEventCode; + failureStage?: LiveViewerFailureStage | null; + streamId?: string | null; + backendActivitySequence?: number | null; + viewerRangeMaxNs?: number | null; + stalledForMs?: number | null; + recoveryAttempt?: number | null; +} + +const SAFE_STREAM_ID = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/; + +function safeInteger(value: number | null | undefined): number | undefined { + return typeof value === "number" && Number.isSafeInteger(value) && value >= 0 + ? value + : undefined; +} + +export function postLiveViewerDiagnostic(event: LiveViewerDiagnostic): void { + const streamId = event.streamId?.trim(); + const body = { + schema_version: "missioncore.live-viewer-diagnostic/v1", + event_code: event.eventCode, + ...(event.failureStage ? { failure_stage: event.failureStage } : {}), + ...(streamId && SAFE_STREAM_ID.test(streamId) ? { stream_id: streamId } : {}), + ...(safeInteger(event.backendActivitySequence) === undefined + ? {} + : { backend_activity_sequence: safeInteger(event.backendActivitySequence) }), + ...(safeInteger(event.viewerRangeMaxNs) === undefined + ? {} + : { viewer_range_max_ns: safeInteger(event.viewerRangeMaxNs) }), + ...(safeInteger(event.stalledForMs) === undefined + ? {} + : { stalled_for_ms: safeInteger(event.stalledForMs) }), + ...(safeInteger(event.recoveryAttempt) === undefined + ? {} + : { recovery_attempt: safeInteger(event.recoveryAttempt) }), + }; + void fetch("/api/v1/viewer/live-diagnostics", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + keepalive: true, + }).catch(() => { + // Diagnostics must never interfere with the live receiver recovery path. + }); +} diff --git a/apps/control-station/src/core/runtime/contracts.ts b/apps/control-station/src/core/runtime/contracts.ts index e0f7ec3..cdf6deb 100644 --- a/apps/control-station/src/core/runtime/contracts.ts +++ b/apps/control-station/src/core/runtime/contracts.ts @@ -28,6 +28,7 @@ export interface ViewerSettings { } export interface StreamMetrics { + publishedFrameCount?: number | null; latencyMs?: number | null; frameRateHz?: number | null; pointCount?: number | null; diff --git a/apps/control-station/src/workspaces/Workspaces.tsx b/apps/control-station/src/workspaces/Workspaces.tsx index ba6ffb8..41a98e9 100644 --- a/apps/control-station/src/workspaces/Workspaces.tsx +++ b/apps/control-station/src/workspaces/Workspaces.tsx @@ -499,7 +499,8 @@ function SpatialWorkspace({ { assert.equal(lifecycle.sourceStatusLabel(waiting), "Ожидание реальных данных"); assert.equal(lifecycle.isConfirmedLiveState(acquiring), true); assert.equal(lifecycle.confirmedRuntimeSourceMode(acquiring), "live"); + assert.equal(lifecycle.liveStartPlan(acquiring), "already-running"); + assert.equal( + lifecycle.controlSessionEntryPlan("scanning", false, false), + "continue", + ); }); test("prepared acquisition resumes without another prepare and remains recoverable", () => { diff --git a/apps/control-station/test/liveReceiverWatchdog.test.mjs b/apps/control-station/test/liveReceiverWatchdog.test.mjs new file mode 100644 index 0000000..e0cf274 --- /dev/null +++ b/apps/control-station/test/liveReceiverWatchdog.test.mjs @@ -0,0 +1,94 @@ +import assert from "node:assert/strict"; +import { after, before, test } from "node:test"; + +import { createServer } from "vite"; + +let server; +let advanceLiveReceiverWatchdog; +let initialLiveReceiverWatchdogState; +let initialLiveReceiverRecoveryState; +let requestLiveReceiverRecovery; + +before(async () => { + server = await createServer({ + appType: "custom", + logLevel: "silent", + server: { middlewareMode: true }, + }); + ({ + advanceLiveReceiverWatchdog, + initialLiveReceiverRecoveryState, + initialLiveReceiverWatchdogState, + requestLiveReceiverRecovery, + } = await server.ssrLoadModule("/src/core/observation/liveReceiverWatchdog.ts")); +}); + +after(async () => { + await server?.close(); +}); + +test("watchdog ignores healthy live progress", () => { + let state = initialLiveReceiverWatchdogState(); + for (let index = 0; index < 20; index += 1) { + const observed = advanceLiveReceiverWatchdog(state, { + nowMs: index * 500, + backendActivitySequence: index * 2, + viewerRangeMaxNs: index * 1_000_000_000, + }); + state = observed.state; + assert.notEqual(observed.signal, "stalled"); + } +}); + +test("watchdog detects a frozen receiver while backend publication advances", () => { + let state = initialLiveReceiverWatchdogState(); + let signal = "none"; + let stalledForMs = 0; + for (let index = 0; index <= 13; index += 1) { + const observed = advanceLiveReceiverWatchdog(state, { + nowMs: index * 500, + backendActivitySequence: 100 + index, + viewerRangeMaxNs: 42_000_000_000, + }); + state = observed.state; + signal = observed.signal === "stalled" ? observed.signal : signal; + stalledForMs = Math.max(stalledForMs, observed.stalledForMs); + } + assert.equal(signal, "stalled"); + assert.ok(stalledForMs >= 5_000); +}); + +test("one backend increment at stream finalization does not create a false stall", () => { + let state = initialLiveReceiverWatchdogState(); + state = advanceLiveReceiverWatchdog(state, { + nowMs: 0, + backendActivitySequence: 10, + viewerRangeMaxNs: 20_000_000_000, + }).state; + state = advanceLiveReceiverWatchdog(state, { + nowMs: 500, + backendActivitySequence: 11, + viewerRangeMaxNs: 20_000_000_000, + }).state; + const observed = advanceLiveReceiverWatchdog(state, { + nowMs: 10_000, + backendActivitySequence: 11, + viewerRangeMaxNs: 20_000_000_000, + }); + assert.equal(observed.signal, "none"); +}); + +test("startup failures request only three bounded viewer restarts", () => { + let state = initialLiveReceiverRecoveryState(); + for (let attempt = 1; attempt <= 3; attempt += 1) { + const recovery = requestLiveReceiverRecovery(state); + assert.equal(recovery.signal, "retry"); + assert.equal(recovery.attempt, attempt); + assert.equal(recovery.state.awaitingRecovery, true); + state = recovery.state; + } + const exhausted = requestLiveReceiverRecovery(state); + assert.equal(exhausted.signal, "exhausted"); + assert.equal(exhausted.attempt, 3); + assert.equal(exhausted.state.awaitingRecovery, false); +}); diff --git a/apps/control-station/test/rerunViewportAtomicAdmission.test.mjs b/apps/control-station/test/rerunViewportAtomicAdmission.test.mjs index 7464e78..807625e 100644 --- a/apps/control-station/test/rerunViewportAtomicAdmission.test.mjs +++ b/apps/control-station/test/rerunViewportAtomicAdmission.test.mjs @@ -129,7 +129,11 @@ test("recorded RRD bytes are never split across LogChannel.send_rrd calls", asyn assert.match(source, /viewer\.start\(\s*rerunViewerInitialSource\(resolvedSource\)/s); assert.match( source, - /recordingOpened = true;[\s\S]*if \(!isRecordedSource\) clearLiveRecordingOpenTimer\(\);/, + /recordingOpened = true;[\s\S]*clearLiveRecordingOpenTimer\(\);[\s\S]*clearLiveRecordingDiscoveryTimer\(\);/, + ); + assert.match( + source, + /viewer\.get_active_recording_id\(\)[\s\S]*eventCode: "live_receiver_active_store_admitted"[\s\S]*admitRecording\(\{[\s\S]*application_id: "nodedc_mission_core_spatial"/, ); assert.match( source, @@ -141,6 +145,14 @@ test("recorded RRD bytes are never split across LogChannel.send_rrd calls", asyn ); }); +test("raw replay exercises the same streaming receiver lifecycle as a live scan", async () => { + const source = await readFile( + new URL("../src/workspaces/Workspaces.tsx", import.meta.url), + "utf8", + ); + assert.match(source, /followLive=\{!recordedReplay && streamActive\}/); +}); + test("the complete vendor canvas host is hidden during partial and failed admission", async () => { const css = await readFile(new URL("../src/styles/spatial.css", import.meta.url), "utf8"); assert.match( diff --git a/plugins/xgrids-k1/frontend/src/api.ts b/plugins/xgrids-k1/frontend/src/api.ts index 5905d48..4b98baf 100644 --- a/plugins/xgrids-k1/frontend/src/api.ts +++ b/plugins/xgrids-k1/frontend/src/api.ts @@ -58,6 +58,30 @@ export interface XgridsDeviceSession { connectivity?: "unknown" | "offline" | "connecting" | "connected" | "degraded"; } +export interface XgridsConnectionVerification { + status?: + | "not-probed" + | "configured" + | "live-address-observed" + | "reachable" + | "recovered" + | "unreachable"; + lease_state?: "disconnected" | "configured" | "reachable"; + lease_generation?: number; + endpoint_validation?: string | null; + network_reachability?: + | "unknown" + | "not-probed" + | "reachable" + | "degraded" + | "unreachable"; + address_changed?: boolean; + previous_address_present?: boolean; + write_performed?: boolean; + observed_at?: string | null; + reason_code?: string | null; +} + export interface XgridsCompatibilityState { profile_id?: string | null; decision?: "compatible" | "limited" | "unknown" | "incompatible"; @@ -167,6 +191,23 @@ export interface XgridsApplicationControlSession { expected?: Record; observed?: Record; } | null; + status_reconciliation?: { + device_session_state?: string; + device_project_bound?: boolean; + system_error_code?: number | null; + decision?: "safe-explicit-prestart-retry"; + automatic_retry?: false; + } | null; + network_change_admissible?: boolean; + network_change_reconciliation?: { + device_session_state?: "scan_stopping"; + device_project_bound?: true; + system_error_code?: null; + stop_complete?: true; + standby_confirmed?: false; + decision?: "explicit-network-change-only-after-acknowledged-stop"; + automatic_retry?: false; + } | null; safe_to_retry?: boolean; } | null; dialogue?: Record | null; @@ -220,6 +261,8 @@ export interface XgridsOperation { } export interface XgridsK1Metrics { + pcl_frames?: number | null; + pose_frames?: number | null; mqtt_to_decode_ms?: number | null; decode_ms?: number | null; publish_ms?: number | null; @@ -308,6 +351,7 @@ export interface XgridsK1State { application_control_session?: XgridsApplicationControlSession | null; device_ref?: XgridsDeviceRef | null; device_session?: XgridsDeviceSession | null; + connection_verification?: XgridsConnectionVerification | null; acquisition?: XgridsAcquisition | null; operations?: XgridsOperation[]; last_operation?: XgridsOperation | null; diff --git a/plugins/xgrids-k1/frontend/src/runtimeContext.tsx b/plugins/xgrids-k1/frontend/src/runtimeContext.tsx index 514246b..9c3322f 100644 --- a/plugins/xgrids-k1/frontend/src/runtimeContext.tsx +++ b/plugins/xgrids-k1/frontend/src/runtimeContext.tsx @@ -104,6 +104,14 @@ function normalizeState( viewerSettings: state.viewer_settings, sourceMode: confirmedRuntimeSourceMode(state), metrics: { + publishedFrameCount: ( + Number.isSafeInteger(metrics?.pcl_frames) && + Number.isSafeInteger(metrics?.pose_frames) && + (metrics?.pcl_frames ?? -1) >= 0 && + (metrics?.pose_frames ?? -1) >= 0 + ) + ? (metrics?.pcl_frames ?? 0) + (metrics?.pose_frames ?? 0) + : null, latencyMs: pipelineLatency(metrics), frameRateHz: finiteMetric(metrics?.frame_rate ?? metrics?.frame_rate_hz), pointCount: finiteMetric(metrics?.point_count), diff --git a/plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts b/plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts index c3b0d4f..420d42c 100644 --- a/plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts +++ b/plugins/xgrids-k1/frontend/src/useXgridsK1Runtime.ts @@ -70,6 +70,8 @@ function controlFailure(state: XgridsK1State): ApiError { "Сканер не подтвердил канонические MQTT-подписки.", mqtt_response_timeout: "Ожидаемый ответ сканера не пришёл до безопасной границы ожидания.", + mqtt_network_loop_failed: + "Локальный MQTT-клиент потерял управляющее соединение со сканером.", response_identity_decode_failed: "Ответ сканера не удалось безопасно разобрать и привязать к операции.", modeling_response_decode_failed: @@ -124,7 +126,9 @@ function controlFailure(state: XgridsK1State): ApiError { ? "Команды START и STOP не отправлялись." : "Факт отправки START или STOP диагностически не подтверждён; повтор запрещён."; const retryStatus = failure?.safe_to_retry - ? "Новая попытка возможна только отдельным нажатием оператора." + ? failure.status_reconciliation?.decision === "safe-explicit-prestart-retry" + ? "Живой статус READY без связанного проекта подтверждён; новая попытка возможна только отдельным нажатием оператора." + : "Новая попытка возможна только отдельным нажатием оператора." : "Повтор заблокирован до ручной проверки состояния."; const exchanges = typeof failure?.publish_attempts === "number" ? `MQTT-публикаций до остановки: ${failure.publish_attempts}.` diff --git a/src/k1link/compute/live_perception.py b/src/k1link/compute/live_perception.py index 825327d..d942484 100644 --- a/src/k1link/compute/live_perception.py +++ b/src/k1link/compute/live_perception.py @@ -396,6 +396,11 @@ class LivePerceptionIngress: if self._session_id == session_id: return raise RuntimeError("another live perception session is active") + # Queued items belong to one acquisition. A worker may be absent + # while a session ends, so clear bounded leftovers before the next + # session can become visible to that worker. + for queue in self._queues.values(): + queue.items.clear() self._session_id = session_id self._active = True self._publish_locked( @@ -681,9 +686,7 @@ class LiveSensorSynchronizer: or retention_seconds <= 0 ): raise ValueError("live sensor synchronizer bounds are invalid") - self._maximum_lidar_camera_delta_ns = round( - maximum_lidar_camera_delta_ms * 1_000_000 - ) + self._maximum_lidar_camera_delta_ns = round(maximum_lidar_camera_delta_ms * 1_000_000) self._maximum_pose_point_delta_ns = round(maximum_pose_point_delta_ms * 1_000_000) self._capacity = capacity_per_modality self._retention_ns = round(retention_seconds * 1_000_000_000) @@ -801,8 +804,7 @@ class LiveSensorSynchronizer: newest = values[-1].context.captured_at_epoch_ns oldest_allowed = newest - self._retention_ns while values and ( - len(values) > self._capacity - or values[0].context.captured_at_epoch_ns < oldest_allowed + len(values) > self._capacity or values[0].context.captured_at_epoch_ns < oldest_allowed ): values.popleft() removed += 1 @@ -837,9 +839,7 @@ class WorldStateProjector: if velocity_history_limit_s <= 0: raise ValueError("velocity history limit must be positive") self._velocity_history_limit_s = velocity_history_limit_s - self._track_history: dict[ - int, deque[tuple[float, tuple[float, float, float]]] - ] = {} + self._track_history: dict[int, deque[tuple[float, tuple[float, float, float]]]] = {} def project( self, @@ -950,9 +950,7 @@ class WorldStateProjector: math.sqrt( sum( (observed - expected) ** 2 - for observed, expected in zip( - observed_center, predicted, strict=True - ) + for observed, expected in zip(observed_center, predicted, strict=True) ) ) ) diff --git a/src/k1link/device_plugins/xgrids_k1/ble/wifi_provisioning.py b/src/k1link/device_plugins/xgrids_k1/ble/wifi_provisioning.py index ae7f002..c6e7a6f 100644 --- a/src/k1link/device_plugins/xgrids_k1/ble/wifi_provisioning.py +++ b/src/k1link/device_plugins/xgrids_k1/ble/wifi_provisioning.py @@ -163,13 +163,18 @@ async def read_wifi_status_once( device_macos_uuid: str, *, timeout_seconds: float = 20.0, + rediscover: bool = False, ) -> WifiStatusReadResult: """Read the K1's current DHCP status over BLE without writing a characteristic.""" if timeout_seconds <= 0: raise ValueError("timeout_seconds must be positive") async with asyncio.timeout(timeout_seconds + 5.0): - device = discovered_device(device_macos_uuid) + # A CoreBluetooth handle retained by an earlier scan is an optimization, + # not durable connection state. Recovery after sleep, Wi-Fi transition, + # or a completed provisioning GATT session must rediscover the device + # instead of repeatedly opening a stale handle. + device = None if rediscover else discovered_device(device_macos_uuid) if device is None: device = await BleakScanner.find_device_by_address( device_macos_uuid, @@ -183,9 +188,7 @@ async def read_wifi_status_once( async with BleakClient(device, timeout=timeout_seconds, pair=False) as client: service = client.services.get_service(SERVICE_UUID) - status_characteristic = client.services.get_characteristic( - STATUS_CHARACTERISTIC_UUID - ) + status_characteristic = client.services.get_characteristic(STATUS_CHARACTERISTIC_UUID) if service is None: raise ValueError(f"Reviewed K1 service not found: {SERVICE_UUID}") if status_characteristic is None: @@ -193,9 +196,7 @@ async def read_wifi_status_once( f"Reviewed K1 status characteristic not found: {STATUS_CHARACTERISTIC_UUID}" ) if status_characteristic.service_uuid != service.uuid: - raise ValueError( - "K1 status characteristic is attached to an unexpected service" - ) + raise ValueError("K1 status characteristic is attached to an unexpected service") if "read" not in status_characteristic.properties: raise ValueError("Reviewed K1 status characteristic is not readable") value = bytes(await client.read_gatt_char(status_characteristic)) diff --git a/src/k1link/device_plugins/xgrids_k1/facade.py b/src/k1link/device_plugins/xgrids_k1/facade.py index 1d62df4..30f3b93 100644 --- a/src/k1link/device_plugins/xgrids_k1/facade.py +++ b/src/k1link/device_plugins/xgrids_k1/facade.py @@ -5,8 +5,11 @@ import hashlib import hmac import importlib.util import json +import logging import secrets import socket +import subprocess +import sys import threading import time import unicodedata @@ -121,9 +124,12 @@ from k1link.web.plugin_runtime import ( XGRIDS_K1_PLUGIN_ID = "nodedc.device.xgrids-lixelkity-k1" XGRIDS_K1_PLUGIN_VERSION = "0.6.0" XGRIDS_K1_MODEL_ID = "xgrids.lixelkity-k1" -XGRIDS_K1_COMPATIBILITY_PROFILE_ID = ( - "xgrids.lixelkity-k1.fw-3.0.2.local-network.v2" -) +XGRIDS_K1_COMPATIBILITY_PROFILE_ID = "xgrids.lixelkity-k1.fw-3.0.2.local-network.v2" +DEFAULT_ACQUISITION_CAMERA_SOURCE: CameraSourceId = "sensor.camera.right" +CONTROL_MQTT_PORT = 1883 +CONTROL_ENDPOINT_PROBE_TIMEOUT_SECONDS = 1.5 + +logger = logging.getLogger(__name__) ConnectionMode = Literal["bridge", "quick-connect", "direct-connect"] ConnectionTopology = Literal["direct-lan", "device-ap", "controller-hotspot"] @@ -133,6 +139,23 @@ CONNECTION_TOPOLOGY_BY_MODE: dict[ConnectionMode, ConnectionTopology] = { "direct-connect": "controller-hotspot", } + +class ConnectionLeaseUnavailable(RuntimeError): + """The process-owned K1 network route cannot admit a new control session.""" + + def __init__(self, message: str, *, reason_code: str) -> None: + super().__init__(message) + self.reason_code = reason_code + + +class LocalAcquisitionLifecycleError(RuntimeError): + """A local producer lifecycle invariant failed before scanner authority.""" + + def __init__(self, message: str, *, reason_code: str) -> None: + super().__init__(message) + self.reason_code = reason_code + + ACTION_STATE_READ = "state.read" ACTION_DISCOVERY_SCAN = "discovery.scan" ACTION_DEVICE_INSPECT = "device.inspect" @@ -238,9 +261,7 @@ class ConnectRequest(StrictRequest): def validate_connection_topology(self) -> Self: expected = CONNECTION_TOPOLOGY_BY_MODE[self.connection_mode] if self.compatibility_attestation.topology != expected: - raise ValueError( - f"connection_mode={self.connection_mode} requires topology={expected}" - ) + raise ValueError(f"connection_mode={self.connection_mode} requires topology={expected}") if self.connection_mode == "quick-connect": if self.ssid is not None or self.password is not None: raise ValueError( @@ -252,9 +273,7 @@ class ConnectRequest(StrictRequest): if not 1 <= len(self.ssid.encode("utf-8")) <= 32: raise ValueError("SSID must contain between 1 and 32 UTF-8 bytes") if not 1 <= len(self.password.get_secret_value().encode("utf-8")) <= 64: - raise ValueError( - "Wi-Fi password must contain between 1 and 64 UTF-8 bytes" - ) + raise ValueError("Wi-Fi password must contain between 1 and 64 UTF-8 bytes") return self @@ -410,8 +429,11 @@ class XgridsK1CompatibilityService: self._device_id: str | None = None self._device_session_id: str | None = None self._device_session_opened_at: str | None = None + self._connection_lease_generation = 0 self._connection_verification: dict[str, Any] = { "status": "not-probed", + "lease_state": "disconnected", + "lease_generation": 0, "endpoint_validation": "not-performed", "network_reachability": "unknown", "observed_at": None, @@ -436,9 +458,7 @@ class XgridsK1CompatibilityService: self.live_perception_token_path, self._live_perception_token, ) = ensure_live_shadow_token(self.repository_root) - authority_loader = ( - application_authority_loader or MacOSKeychainApplicationAuthorityLoader() - ) + authority_loader = application_authority_loader or MacOSKeychainApplicationAuthorityLoader() self._calibration_snapshot_reader = ( calibration_snapshot_reader or DeviceCalibrationSnapshotReader( @@ -454,6 +474,7 @@ class XgridsK1CompatibilityService: self._application_control_session = InteractiveApplicationControlSession( authority_loader, transport_factory=self._application_control_transport, + scanning_observer=self._activate_default_acquisition_camera, ) self.runtime = VisualizationRuntime( normalizer=normalize_k1_message, @@ -493,6 +514,7 @@ class XgridsK1CompatibilityService: device_id = self._device_id device_session_id = self._device_session_id device_session_opened_at = self._device_session_opened_at + connection_lease_generation = self._connection_lease_generation connection_verification = dict(self._connection_verification) compatibility_attestation = ( dict(self._compatibility_attestation) @@ -515,15 +537,31 @@ class XgridsK1CompatibilityService: "stopping", "error", } + connection_reachability = connection_verification.get("network_reachability") + if connection_reachability == "unreachable": + device_connectivity = "offline" + elif connection_reachability == "degraded": + device_connectivity = "degraded" + elif k1_ip is not None: + device_connectivity = "connected" + else: + device_connectivity = "unknown" + if operation_phase is not None: phase = operation_phase message = operation_message elif runtime_active: phase = runtime["phase"] message = runtime["message"] - elif k1_ip is not None: + elif k1_ip is not None and device_connectivity != "offline": phase = "connected" message = runtime["message"] + elif connection_verification.get("reason_code") == ("connection_lease_host_route_mismatch"): + phase = "device_selected" + message = ( + "K1 получил адрес, но этот компьютер подключён к другой сети. " + "Подключите компьютер к той же локальной сети и повторите запуск." + ) elif selected_device_id is not None: phase = "device_selected" message = "Устройство выбрано. Теперь введите название и пароль Wi-Fi." @@ -605,12 +643,15 @@ class XgridsK1CompatibilityService: "device_id": device_id, "opened_at": device_session_opened_at, "compatibility_profile_id": active_profile_id, - "connectivity": "connected" if k1_ip is not None else "unknown", + "connectivity": device_connectivity, } if device_id is not None and device_session_id is not None else None ), - "connection_verification": connection_verification, + "connection_verification": { + **connection_verification, + "lease_generation": connection_lease_generation, + }, "sensor_catalog": _sensor_catalog( active_profile_id, device_session_id, @@ -694,11 +735,6 @@ class XgridsK1CompatibilityService: return self.state() async def connect(self, request: ConnectRequest) -> dict[str, Any]: - control_state = self._application_control_session.snapshot()["state"] - if control_state not in {"idle", "completed", "closed"}: - raise RuntimeError( - "сначала завершите или безопасно закройте текущую control-сессию K1" - ) known_ids = {str(item["device_id"]) for item in self.state()["devices"]} if request.device_id not in known_ids: raise ValueError("сначала найдите и выберите устройство через Bluetooth") @@ -707,6 +743,33 @@ class XgridsK1CompatibilityService: item for item in self.state()["devices"] if item["device_id"] == request.device_id ) selected_device_name = str(selected_device.get("name") or "").strip() + control_snapshot = self._application_control_session.snapshot() + try: + self._application_control_session.retire_for_network_change() + except RuntimeError as exc: + failure = control_snapshot.get("failure") + logger.warning( + "K1 network change blocked by retained control ownership", + extra={ + "event_code": "k1_network_change_blocked_by_control_session", + "reason_code": ( + str(failure.get("reason_code")) + if isinstance(failure, dict) and failure.get("reason_code") + else None + ), + "failed_phase": ( + str(failure.get("failed_phase")) + if isinstance(failure, dict) and failure.get("failed_phase") + else str(control_snapshot.get("state") or "unknown") + ), + "automatic_retry": False, + }, + ) + raise RuntimeError( + "сначала завершите текущую control-сессию K1; " + "после START новый сетевой путь допустим только при " + "подтверждённом STOP" + ) from exc if quick_connect and not selected_device_name: raise ValueError( "выбранный BLE-кандидат не сообщил имя точки доступа; " @@ -716,31 +779,21 @@ class XgridsK1CompatibilityService: quick_connect_host_profile_id(selected_device_name) if quick_connect else None ) host_wifi_helper_path = ( - self.repository_root - / "plugins" - / "xgrids-k1" - / "macos" - / "associate_wifi.swift" + self.repository_root / "plugins" / "xgrids-k1" / "macos" / "associate_wifi.swift" ) # Bridge and Direct Connect unwrap once at the BLE service boundary. # Quick Connect carries no browser/API credential. It first sends the # reviewed fixed AP-enable command, then the host-network adapter uses # the selected device's advertised name as its exact SSID and resolves # a device-scoped profile inside the OS credential store. - password = ( - "" - if request.password is None - else request.password.get_secret_value() - ) + password = "" if request.password is None else request.password.get_secret_value() request_fingerprint = self._request_fingerprint( ACTION_NETWORK_PROVISION, { "device_id": request.device_id, "ssid": request.ssid, "password": password if not quick_connect else None, - "host_wifi_profile": ( - quick_connect_profile_id if quick_connect else None - ), + "host_wifi_profile": (quick_connect_profile_id if quick_connect else None), "connection_mode": request.connection_mode, "compatibility_attestation": request.compatibility_attestation.model_dump( mode="json" @@ -779,9 +832,8 @@ class XgridsK1CompatibilityService: session_dir: Path | None = None network_change_attempted = False - operation_stage = ( - "device-ap-activation" if quick_connect else "ble-provisioning-write" - ) + retired_ingress_session_id: str | None = None + operation_stage = "device-ap-activation" if quick_connect else "ble-provisioning-write" try: if quick_connect: assert quick_connect_profile_id is not None @@ -816,6 +868,10 @@ class XgridsK1CompatibilityService: raise RuntimeError( "нельзя менять устройство или его сеть во время активной acquisition-сессии" ) + if self._acquisition_session_lease is not None: + raise RuntimeError( + "предыдущая evidence-сессия ещё не завершила локальную очистку" + ) self._provisioning_active = True # Once a new network operation is admitted, the previous route # and device session can no longer be represented as current. @@ -830,14 +886,34 @@ class XgridsK1CompatibilityService: self._device_session_opened_at = None self._connection_verification = { "status": "not-probed", + "lease_state": "disconnected", + "lease_generation": self._connection_lease_generation, "endpoint_validation": "not-performed", "network_reachability": "unknown", "observed_at": None, } + # A new admitted network route starts a new device context. + # Preserve immutable evidence and operation history, but never + # project the previous terminal acquisition into this route. + if self._acquisition_out_dir is not None: + retired_ingress_session_id = self._acquisition_out_dir.name + self._acquisition = None + self._acquisition_project_name = None + self._acquisition_mount_type = None + self._acquisition_gnss_mode = None + self._acquisition_out_dir = None + self._acquisition_start_operation_id = None + self._acquisition_stop_operation_id = None # A connection-mode change can replace both the device session and # its address. self._application_control.disarm() + # A terminal receiver error is process-local state, not the state + # of the newly selected Bridge route. stop() is idempotent once + # the source thread has already exited and clears phase/source. + self.runtime.stop() + if retired_ingress_session_id is not None: + self.live_perception_ingress.end_session(retired_ingress_session_id) # Revoke preview/producer state only after the active-acquisition # guard; a rejected network write must never stop evidence capture. self.camera_preview.stop_current() @@ -850,9 +926,7 @@ class XgridsK1CompatibilityService: self._set_operation("provisioning", operation_message) session_dir = _new_operation_session_dir( self.evidence_root, - "viewer_k1_ap_association" - if quick_connect - else "viewer_wifi_provisioning", + "viewer_k1_ap_association" if quick_connect else "viewer_wifi_provisioning", ) session_dir.mkdir(parents=True, exist_ok=False) self._operations.transition( @@ -929,25 +1003,20 @@ class XgridsK1CompatibilityService: "device_ap_activation_write_mode": activation["write_mode"], "host_wifi_adapter": association["adapter"], "host_wifi_profile_id": quick_connect_profile_id, - "host_wifi_profile_ready_before_device_write": profile_preflight[ - "available" - ], + "host_wifi_profile_ready_before_device_write": profile_preflight["available"], "host_wifi_profile_preflight_adapter": profile_preflight["adapter"], "host_wifi_profile_materialized_before_device_write": profile_preflight[ "profile_enrolled" ], "credential_provider_id": K1_FW302_CREDENTIAL_PROVIDER_ID, - "credential_provider_source": profile_preflight[ - "credential_source" - ], + "credential_provider_source": profile_preflight["credential_source"], "host_wifi_profile_enrolled_now": association["profile_enrolled"], "host_wifi_scan_attempt_count": association["scan_attempt_count"], "host_wifi_scan_elapsed_ms": association["scan_elapsed_ms"], "host_wifi_credential_source": association["credential_source"], "device_ap_ssid_source": "selected-ble-advertised-name", "credentials_resolved_by_plugin": ( - profile_preflight["credential_source"] - == "exact-firmware-profile" + profile_preflight["credential_source"] == "exact-firmware-profile" ), "credentials_persisted_by_host_adapter": True, } @@ -978,9 +1047,7 @@ class XgridsK1CompatibilityService: connection_manifest.update( { "k1_lan_address_observed": ipv4 is not None, - "k1_lan_address_admitted": ( - ipv4 is not None and not local_address_conflict - ), + "k1_lan_address_admitted": (ipv4 is not None and not local_address_conflict), "local_address_conflict": local_address_conflict, } ) @@ -994,6 +1061,15 @@ class XgridsK1CompatibilityService: "Устройство сообщило IPv4-адрес, который уже принадлежит этому компьютеру; " "адрес K1 не принят и автоматического повтора не было" ) + host_route_class = "device-ap" if quick_connect else _host_route_class(ipv4) + host_route_ready = host_route_class not in {"tunnel", "default-route"} + connection_manifest.update( + { + "host_route_class": host_route_class, + "host_route_ready": host_route_ready, + } + ) + write_json_atomic(session_dir / "manifest.redacted.json", connection_manifest) with self._lock: self._selected_device_id = request.device_id self._k1_ip = ipv4 @@ -1004,6 +1080,7 @@ class XgridsK1CompatibilityService: ) self._device_session_id = new_device_session_id() self._device_session_opened_at = _utc_now_iso() + self._connection_lease_generation += 1 self._compatibility_attestation = _attestation_snapshot( request.compatibility_attestation ) @@ -1011,20 +1088,40 @@ class XgridsK1CompatibilityService: XGRIDS_K1_COMPATIBILITY_PROFILE_ID ) self._connection_verification = { - "status": "not-probed", - "endpoint_validation": "not-performed", - "network_reachability": "unknown", - "observed_at": None, + "status": ("configured" if host_route_ready else "host-route-mismatch"), + "lease_state": "configured" if host_route_ready else "disconnected", + "lease_generation": self._connection_lease_generation, + "endpoint_validation": ( + "provisioning-status" if host_route_ready else "host-route" + ), + "network_reachability": ("unknown" if host_route_ready else "unreachable"), + **( + {} + if host_route_ready + else { + "reason_code": "connection_lease_host_route_mismatch", + "host_route_class": host_route_class, + } + ), + "write_performed": True, + "observed_at": _utc_now_iso(), } - self._operation_message = { - "bridge": "K1 подключён к общей сети и сообщил локальный адрес.", - "direct-connect": ( - "K1 подключён к хотспоту контроллера и сообщил локальный адрес." - ), - "quick-connect": ( - "Mission Core подключён к точке доступа K1; адрес K1 подтверждён." - ), - }[request.connection_mode] + self._operation_message = ( + { + "bridge": "K1 подключён к общей сети и сообщил локальный адрес.", + "direct-connect": ( + "K1 подключён к хотспоту контроллера и сообщил локальный адрес." + ), + "quick-connect": ( + "Mission Core подключён к точке доступа K1; адрес K1 подтверждён." + ), + }[request.connection_mode] + if host_route_ready + else ( + "K1 получил адрес, но этот компьютер подключён к другой сети. " + "Подключите компьютер к той же локальной сети." + ) + ) self._operations.transition( operation.operation_id, "succeeded", @@ -1037,10 +1134,10 @@ class XgridsK1CompatibilityService: "connection_mode": request.connection_mode, "topology": request.compatibility_attestation.topology, "address_source": ( - "reviewed-k1-ap-baseline" - if quick_connect - else "ble-wifi-status" + "reviewed-k1-ap-baseline" if quick_connect else "ble-wifi-status" ), + "host_route_class": host_route_class, + "host_route_ready": host_route_ready, }, evidence_refs=(f"evidence-session-{session_dir.name}",), ) @@ -1053,9 +1150,7 @@ class XgridsK1CompatibilityService: error=_operation_error( exc, category="transport" if quick_connect else "device", - side_effect_status=( - "unknown" if network_change_attempted else "none" - ), + side_effect_status=("unknown" if network_change_attempted else "none"), safe_to_retry=not network_change_attempted, ), evidence_refs=( @@ -1124,10 +1219,10 @@ class XgridsK1CompatibilityService: def verify_connection(self) -> dict[str, Any]: """Refresh the session-scoped DHCP address from the read-only BLE status.""" - self._refresh_live_lan_address() + self._refresh_live_lan_address(rediscover=True) return self.state() - def _refresh_live_lan_address(self) -> str: + def _refresh_live_lan_address(self, *, rediscover: bool = False) -> str: with self._lock: selected_device_id = self._selected_device_id connection_mode = self._connection_mode @@ -1136,9 +1231,7 @@ class XgridsK1CompatibilityService: if selected_device_id is None or connection_mode is None: raise ValueError("сначала выберите и подключите K1 через BLE/Wi-Fi") if acquisition is not None and acquisition.state not in TERMINAL_ACQUISITION_STATES: - raise RuntimeError( - "нельзя менять DHCP-привязку во время активной acquisition-сессии" - ) + raise RuntimeError("нельзя менять DHCP-привязку во время активной acquisition-сессии") control_state = str(self._application_control_session.snapshot()["state"]) if control_state not in {"idle", "completed", "closed", "failed"}: raise RuntimeError("нельзя менять DHCP-привязку при открытой control-сессии") @@ -1148,7 +1241,11 @@ class XgridsK1CompatibilityService: return AP_FALLBACK_IPV4 status_read = asyncio.run( - read_wifi_status_once(selected_device_id, timeout_seconds=20.0) + read_wifi_status_once( + selected_device_id, + timeout_seconds=20.0, + rediscover=rediscover, + ) ) observed_target = status_read["status"]["ipv4"] if observed_target is None or observed_target == AP_FALLBACK_IPV4: @@ -1168,11 +1265,14 @@ class XgridsK1CompatibilityService: if address_changed: self._device_session_id = new_device_session_id() self._device_session_opened_at = _utc_now_iso() + self._connection_lease_generation += 1 self._device_calibration = unavailable_device_calibration_snapshot( XGRIDS_K1_COMPATIBILITY_PROFILE_ID ) self._connection_verification = { "status": "live-address-observed", + "lease_state": "configured", + "lease_generation": self._connection_lease_generation, "endpoint_validation": "ble-wifi-status-read", "network_reachability": "not-probed", "address_changed": address_changed, @@ -1182,6 +1282,254 @@ class XgridsK1CompatibilityService: } return target + def _reuse_or_recover_control_target(self) -> tuple[str, dict[str, Any]]: + """Resolve one live process-owned route without repeating Wi-Fi writes.""" + + with self._lock: + selected_device_id = self._selected_device_id + connection_mode = self._connection_mode + current_target = self._k1_ip + device_session_id = self._device_session_id + lease_generation = self._connection_lease_generation + if ( + selected_device_id is None + or connection_mode is None + or current_target is None + or device_session_id is None + ): + raise ConnectionLeaseUnavailable( + "сначала настройте сетевое подключение K1", + reason_code="connection_lease_missing", + ) + current_target = validate_private_ipv4(current_target) + + if _control_endpoint_reachable(current_target): + observed_at = _utc_now_iso() + with self._lock: + if ( + self._selected_device_id != selected_device_id + or self._connection_mode != connection_mode + or self._k1_ip != current_target + or self._device_session_id != device_session_id + ): + raise ConnectionLeaseUnavailable( + "подключение K1 изменилось во время проверки", + reason_code="connection_lease_changed_during_probe", + ) + self._connection_verification = { + "status": "reachable", + "lease_state": "reachable", + "lease_generation": lease_generation, + "endpoint_validation": "mqtt-tcp-connect", + "network_reachability": "reachable", + "address_changed": False, + "write_performed": False, + "observed_at": observed_at, + } + logger.info( + "K1 process-owned connection lease reused", + extra={ + "event_code": "k1_connection_lease_reused", + "lease_generation": lease_generation, + "lease_state": "reachable", + "recovery_strategy": "existing-mqtt-endpoint", + "endpoint_reachable": True, + "address_changed": False, + "device_write_performed": False, + "automatic_retry": False, + }, + ) + return current_target, { + "lease_generation": lease_generation, + "connection_lease_reused": True, + "recovery_performed": False, + "address_changed": False, + "device_write_performed": False, + } + + if connection_mode == "quick-connect": + self._mark_connection_lease_unreachable( + reason_code="quick_connect_endpoint_unreachable", + lease_generation=lease_generation, + ) + raise ConnectionLeaseUnavailable( + "точка доступа K1 больше не доступна с этого компьютера; " + "выполните Quick Connect заново", + reason_code="quick_connect_endpoint_unreachable", + ) + + host_route_class = _host_route_class(current_target) + if host_route_class in {"default-route", "tunnel"}: + self._mark_connection_lease_unreachable( + reason_code="connection_lease_host_route_mismatch", + lease_generation=lease_generation, + endpoint_validation="host-route", + host_route_class=host_route_class, + ) + raise ConnectionLeaseUnavailable( + "Сканер подключён к другой локальной сети: его адрес получен, " + "но этот компьютер не имеет прямого маршрута к нему. " + "Подключите компьютер к той же сети, что и сканер. " + "Команды запуска и остановки не отправлялись.", + reason_code="connection_lease_host_route_mismatch", + ) + + recovered_route_class = "unknown" + try: + status_read = asyncio.run( + read_wifi_status_once( + selected_device_id, + timeout_seconds=20.0, + rediscover=True, + ) + ) + observed_target = status_read["status"]["ipv4"] + if observed_target is None or observed_target == AP_FALLBACK_IPV4: + raise ConnectionLeaseUnavailable( + "K1 не сообщил актуальный адрес общей сети", + reason_code="connection_lease_address_unavailable", + ) + recovered_target = validate_private_ipv4(observed_target) + if _target_is_local_ipv4(recovered_target): + raise ConnectionLeaseUnavailable( + "K1 сообщил адрес этого компьютера вместо адреса устройства", + reason_code="connection_lease_local_address_conflict", + ) + recovered_route_class = _host_route_class(recovered_target) + if recovered_route_class in {"default-route", "tunnel"}: + raise ConnectionLeaseUnavailable( + "Сканер подключён к другой локальной сети: его адрес получен, " + "но этот компьютер не имеет прямого маршрута к нему. " + "Подключите компьютер к той же сети, что и сканер. " + "Команды запуска и остановки не отправлялись.", + reason_code="connection_lease_host_route_mismatch", + ) + if not _control_endpoint_reachable(recovered_target): + raise ConnectionLeaseUnavailable( + "Сканер сообщил адрес общей сети, но управляющее соединение " + "с этого компьютера не открывается. Убедитесь, что оба " + "устройства подключены к одной сети и роутер не изолирует клиентов.", + reason_code="connection_lease_recovered_endpoint_unreachable", + ) + except Exception as exc: + reason_code = getattr(exc, "reason_code", None) + if not isinstance(reason_code, str) or not reason_code: + reason_code = "connection_lease_ble_recovery_failed" + self._mark_connection_lease_unreachable( + reason_code=reason_code, + lease_generation=lease_generation, + endpoint_validation=( + "host-route" + if reason_code == "connection_lease_host_route_mismatch" + else "mqtt-tcp-connect" + ), + host_route_class=( + recovered_route_class + if reason_code == "connection_lease_host_route_mismatch" + else None + ), + ) + logger.error( + "K1 process-owned connection lease recovery failed", + extra={ + "event_code": "k1_connection_lease_recovery_failed", + "reason_code": reason_code, + "lease_generation": lease_generation, + "lease_state": "disconnected", + "recovery_strategy": "read-only-ble-rediscovery", + "endpoint_reachable": False, + "host_route_class": ( + recovered_route_class + if reason_code == "connection_lease_host_route_mismatch" + else None + ), + "device_write_performed": False, + "automatic_retry": False, + }, + ) + if isinstance(exc, ConnectionLeaseUnavailable): + raise + raise ConnectionLeaseUnavailable( + "живое подключение K1 потеряно; повторное чтение состояния по Bluetooth не удалось", + reason_code=reason_code, + ) from exc + + address_changed = recovered_target != current_target + observed_at = str(status_read["observed_at_utc"]) + with self._lock: + if ( + self._selected_device_id != selected_device_id + or self._connection_mode != connection_mode + or self._device_session_id != device_session_id + ): + raise ConnectionLeaseUnavailable( + "подключение K1 изменилось во время восстановления", + reason_code="connection_lease_changed_during_recovery", + ) + self._k1_ip = recovered_target + # Any recovery proves that continuity of the previous route was + # lost, even when DHCP returns the same address. Rotate the + # device-session generation so browser/viewer state from before + # the gap cannot be merged with the recovered connection. + self._device_session_id = new_device_session_id() + self._device_session_opened_at = observed_at + self._connection_lease_generation += 1 + self._device_calibration = unavailable_device_calibration_snapshot( + XGRIDS_K1_COMPATIBILITY_PROFILE_ID + ) + lease_generation = self._connection_lease_generation + self._connection_verification = { + "status": "recovered", + "lease_state": "reachable", + "lease_generation": lease_generation, + "endpoint_validation": "ble-rediscovery+mqtt-tcp-connect", + "network_reachability": "reachable", + "address_changed": address_changed, + "write_performed": False, + "observed_at": observed_at, + } + logger.info( + "K1 process-owned connection lease recovered", + extra={ + "event_code": "k1_connection_lease_recovered", + "lease_generation": lease_generation, + "lease_state": "reachable", + "recovery_strategy": "read-only-ble-rediscovery", + "endpoint_reachable": True, + "address_changed": address_changed, + "device_write_performed": False, + "automatic_retry": False, + }, + ) + return recovered_target, { + "lease_generation": lease_generation, + "connection_lease_reused": not address_changed, + "recovery_performed": True, + "address_changed": address_changed, + "device_write_performed": False, + } + + def _mark_connection_lease_unreachable( + self, + *, + reason_code: str, + lease_generation: int, + endpoint_validation: str = "mqtt-tcp-connect", + host_route_class: str | None = None, + ) -> None: + with self._lock: + self._connection_verification = { + "status": "unreachable", + "lease_state": "disconnected", + "lease_generation": lease_generation, + "endpoint_validation": endpoint_validation, + "network_reachability": "unreachable", + "reason_code": reason_code, + **({"host_route_class": host_route_class} if host_route_class is not None else {}), + "write_performed": False, + "observed_at": _utc_now_iso(), + } + def read_device_calibration_snapshot(self) -> dict[str, Any]: """Read and seal the two reviewed factory YAML files without device mutation.""" @@ -1229,21 +1577,73 @@ class XgridsK1CompatibilityService: with self._lock: if self._provisioning_active: raise RuntimeError("нельзя открывать control-сессию во время настройки Wi-Fi") - self._refresh_live_lan_address() - with self._lock: - target = self._k1_ip + device_id = self._device_id + device_session_id = self._device_session_id attestation = self._compatibility_attestation acquisition = self._acquisition - if target is None or attestation is None: - raise RuntimeError("сначала подключите K1 и выберите exact-profile") - if acquisition is not None and acquisition.state not in TERMINAL_ACQUISITION_STATES: - raise RuntimeError("control-сессия должна быть открыта до подготовки acquisition") - self._application_control.disarm() - self._application_control_session.open( - host=validate_private_ipv4(target), - timezone_name=request.timezone_name, - confirmation=request.confirmation(), + operation, _ = self._operations.begin( + ACTION_APPLICATION_CONTROL_SESSION_OPEN, + device_id=device_id, + device_session_id=device_session_id, + deadline_seconds=30.0, ) + self._operations.transition( + operation.operation_id, + "running", + stage_code="connection-lease-check", + message_code="application-control.session.open.checking_connection_lease", + ) + try: + if attestation is None: + raise RuntimeError("сначала подключите K1 и выберите exact-profile") + if acquisition is not None and acquisition.state not in TERMINAL_ACQUISITION_STATES: + raise RuntimeError("control-сессия должна быть открыта до подготовки acquisition") + target, lease_result = self._reuse_or_recover_control_target() + self._application_control.disarm() + self._application_control_session.open( + host=target, + timezone_name=request.timezone_name, + confirmation=request.confirmation(), + ) + self._operations.transition( + operation.operation_id, + "succeeded", + stage_code="control-worker-accepted", + message_code="application-control.session.open.accepted", + result=lease_result, + ) + except Exception as exc: + control_snapshot = self._application_control_session.snapshot() + safe_to_retry = control_snapshot.get("can_open") is True + self._operations.transition_if_pending( + operation.operation_id, + "failed", + stage_code="prestart-failed", + message_code="application-control.session.open.failed", + error=_operation_error( + exc, + category="connection", + side_effect_status="none", + safe_to_retry=safe_to_retry, + ), + ) + reason_code = getattr(exc, "reason_code", None) + logger.error( + "K1 control session open failed before START", + extra={ + "event_code": "k1_control_session_open_prestart_failed", + "reason_code": ( + reason_code + if isinstance(reason_code, str) and reason_code + else type(exc).__name__ + ), + "modeling_command_attempted": False, + "safe_to_retry": safe_to_retry, + "device_write_performed": False, + "automatic_retry": False, + }, + ) + raise return self.state() @_serialized_acquisition_access @@ -1299,25 +1699,25 @@ class XgridsK1CompatibilityService: self._application_control.disarm() control_state = str(self._application_control_session.snapshot()["state"]) if control_state == "failed": - raise RuntimeError( - "control-сессия K1 завершилась ошибкой и требует ручной проверки" - ) - if control_state in { - "connecting", - "connection-ready", - "workspace-requested", - "project-requested", - "project-ready", - "start-requested", - "initializing", - "scanning", - "stop-requested", - "stopping", - "awaiting-standby-confirmation", - } and control_state != "workspace-ready": - raise RuntimeError( - "каноническая control-сессия должна ожидать сохранения проекта" - ) + raise RuntimeError("control-сессия K1 завершилась ошибкой и требует ручной проверки") + if ( + control_state + in { + "connecting", + "connection-ready", + "workspace-requested", + "project-requested", + "project-ready", + "start-requested", + "initializing", + "scanning", + "stop-requested", + "stopping", + "awaiting-standby-confirmation", + } + and control_state != "workspace-ready" + ): + raise RuntimeError("каноническая control-сессия должна ожидать сохранения проекта") control_mode: Literal["operator-manual", "plugin-commanded"] = ( "plugin-commanded" if control_state == "workspace-ready" else "operator-manual" ) @@ -1343,9 +1743,7 @@ class XgridsK1CompatibilityService: "compatibility attestation не совпадает с активным способом подключения" ) if target == AP_FALLBACK_IPV4 and connection_mode != "quick-connect": - raise ValueError( - "адрес точки доступа K1 разрешён только после Quick Connect" - ) + raise ValueError("адрес точки доступа K1 разрешён только после Quick Connect") if connection_mode == "quick-connect" and target != AP_FALLBACK_IPV4: raise ValueError("Quick Connect должен использовать подтверждённый AP-адрес K1") if control_mode == "plugin-commanded": @@ -1469,9 +1867,7 @@ class XgridsK1CompatibilityService: plugin_commanded = acquisition.control_mode == "plugin-commanded" if plugin_commanded: if request.physical_acceptance is None: - raise ValueError( - "START K1 требует явного подтверждения присутствия оператора" - ) + raise ValueError("START K1 требует явного подтверждения присутствия оператора") if self._application_control_session.snapshot()["state"] != "project-ready": raise RuntimeError("канонический диалог K1 ещё не готов принять START") if ( @@ -1512,6 +1908,8 @@ class XgridsK1CompatibilityService: owns_start = False cleanup_failed = False + failure_stage = "local-start-preflight" + start_checkpoint_released = False try: runtime_state = self.runtime.snapshot() with self._lock: @@ -1535,27 +1933,39 @@ class XgridsK1CompatibilityService: ) self._acquisition_start_operation_id = operation.operation_id owns_start = True + failure_stage = "stale-ingress-retirement" + self._retire_stale_live_perception_ingress( + next_session_id=out_dir.name, + runtime_state=runtime_state, + ) + failure_stage = "evidence-lease-acquire" lease = ActiveSessionLease.acquire(out_dir.parent, out_dir) with self._lock: if self._acquisition_session_lease is not None: lease.release() raise RuntimeError("evidence-сессия уже удерживается активным acquisition") self._acquisition_session_lease = lease + failure_stage = "live-perception-ingress" self.live_perception_ingress.begin_session(out_dir.name) self._modeling_control_safety.reset() + failure_stage = "local-receiver-start" self.runtime.start_live( acquisition.target_host, out_dir, duration_seconds=acquisition.duration_seconds, project_name=project_name, ) - self._arm_camera_recording(out_dir) if plugin_commanded: assert request.physical_acceptance is not None + failure_stage = "canonical-start-checkpoint" self._application_control_session.request_start( project_name=project_name, confirmation=request.physical_acceptance.confirmation(), ) + start_checkpoint_released = True + else: + failure_stage = "camera-recording" + self._arm_camera_recording(out_dir) except Exception as exc: if owns_start: with self._lock: @@ -1580,14 +1990,33 @@ class XgridsK1CompatibilityService: self._operations.transition( operation.operation_id, "failed", - stage_code="failed", + stage_code=f"{failure_stage}-failed", message_code="acquisition.start.failed", error=_operation_error( exc, category="stream" if owns_start else "conflict", - side_effect_status="unknown" if cleanup_failed else "none", + side_effect_status=( + "unknown" if start_checkpoint_released or cleanup_failed else "none" + ), + safe_to_retry=not start_checkpoint_released and not cleanup_failed, ), ) + reason_code = getattr(exc, "reason_code", None) + logger.error( + "K1 local acquisition start failed", + extra={ + "event_code": "k1_local_acquisition_start_failed", + "reason_code": ( + reason_code + if isinstance(reason_code, str) and reason_code + else type(exc).__name__ + ), + "failure_stage": failure_stage, + "start_checkpoint_released": start_checkpoint_released, + "safe_to_retry": not start_checkpoint_released and not cleanup_failed, + "automatic_retry": False, + }, + ) raise return self.state() @@ -1609,19 +2038,14 @@ class XgridsK1CompatibilityService: if request.operator_confirmed: if request.mode != "graceful": raise ValueError("operator_confirmed допустим только для graceful stop") - if ( - acquisition_state != "awaiting_external_stop" - and not terminal_manual_stop_recovery - ): + if acquisition_state != "awaiting_external_stop" and not terminal_manual_stop_recovery: raise ValueError("acquisition не ожидает подтверждения физической остановки") if request.operation_id is None or request.operation_id != expected_stop_operation_id: raise ValueError( "подтверждение остановки должно ссылаться на исходную stop-operation" ) if plugin_commanded: - raise ValueError( - "канонический STOP завершается автоматически по READY от K1" - ) + raise ValueError("канонический STOP завершается автоматически по READY от K1") elif ( request.mode == "graceful" and acquisition_state @@ -1639,15 +2063,9 @@ class XgridsK1CompatibilityService: raise ValueError( "graceful stop допустим только после подтверждённого потока point cloud" ) - if ( - plugin_commanded - and request.mode == "graceful" - and not request.operator_confirmed - ): + if plugin_commanded and request.mode == "graceful" and not request.operator_confirmed: if request.physical_acceptance is None: - raise ValueError( - "STOP K1 требует явного подтверждения присутствия оператора" - ) + raise ValueError("STOP K1 требует явного подтверждения присутствия оператора") if self._application_control_session.snapshot()["state"] != "scanning": raise RuntimeError("канонический диалог K1 ещё не готов принять STOP") @@ -1768,10 +2186,7 @@ class XgridsK1CompatibilityService: ), operator_instructions=() if plugin_commanded - else ( - "Дважды нажмите физическую кнопку устройства и " - "подтвердите остановку.", - ), + else ("Дважды нажмите физическую кнопку устройства и подтвердите остановку.",), ) self._acquisition_stop_operation_id = operation.operation_id self._operations.transition( @@ -1868,10 +2283,7 @@ class XgridsK1CompatibilityService: if not created: return self.state() try: - if ( - acquisition.control_mode == "plugin-commanded" - and acquisition.state == "prepared" - ): + if acquisition.control_mode == "plugin-commanded" and acquisition.state == "prepared": self._application_control_session.close_prestart() with self._lock: should_abort = acquisition.state not in TERMINAL_ACQUISITION_STATES @@ -2005,7 +2417,14 @@ class XgridsK1CompatibilityService: acquisition_active = ( acquisition is not None and acquisition.state not in TERMINAL_ACQUISITION_STATES ) - if acquisition_active: + plugin_commanded = ( + acquisition is not None and acquisition.control_mode == "plugin-commanded" + ) + device_scanning = ( + not plugin_commanded + or self._application_control_session.snapshot().get("state") == "scanning" + ) + if acquisition_active and device_scanning: if out_dir is None: raise RuntimeError("для acquisition не выделена evidence-сессия") self._arm_camera_recording(out_dir, require_session=True) @@ -2135,6 +2554,50 @@ class XgridsK1CompatibilityService: time.sleep(0.01) self.camera_preview.start_recording(out_dir) + @_serialized_acquisition_access + def _activate_default_acquisition_camera(self) -> None: + """Start the canonical right-camera archive after confirmed K1 START.""" + + with self._lock: + acquisition = self._acquisition + out_dir = self._acquisition_out_dir + device_session_id = self._device_session_id + if ( + acquisition is None + or acquisition.state in TERMINAL_ACQUISITION_STATES + or acquisition.control_mode != "plugin-commanded" + ): + return + if out_dir is None: + raise RuntimeError("для acquisition не выделена evidence-сессия") + if device_session_id is None: + raise RuntimeError("для acquisition нет активной device-сессии") + + target = self._camera_target_for_session(device_session_id) + camera = self.camera_preview.snapshot() + if camera.get("active_source_id") != DEFAULT_ACQUISITION_CAMERA_SOURCE: + self.camera_preview.select(DEFAULT_ACQUISITION_CAMERA_SOURCE, target) + self._arm_camera_recording(out_dir, require_session=True) + + camera = self.camera_preview.snapshot() + recording = camera.get("recording") + if ( + camera.get("active_source_id") != DEFAULT_ACQUISITION_CAMERA_SOURCE + or not isinstance(recording, dict) + or recording.get("active") is not True + or camera.get("phase") == "error" + ): + raise RuntimeError("правая камера K1 не перешла в обязательную evidence-запись") + logger.info( + "K1 right camera activated after confirmed SCANNING", + extra={ + "event_code": "k1_default_acquisition_camera_activated", + "camera_source_id": DEFAULT_ACQUISITION_CAMERA_SOURCE, + "evidence_session_id": out_dir.name, + "activation_trigger": "application-control-scanning", + }, + ) + def _stop_acquisition_sources( self, *, @@ -2175,6 +2638,43 @@ class XgridsK1CompatibilityService: if cleanup_complete: self._release_acquisition_session_lease() + def _retire_stale_live_perception_ingress( + self, + *, + next_session_id: str, + runtime_state: Mapping[str, Any], + ) -> None: + ingress = self.live_perception_ingress.snapshot() + if ingress.get("active") is not True: + return + current_session_id = ingress.get("session_id") + if current_session_id == next_session_id: + return + with self._lock: + acquisition_lease_active = self._acquisition_session_lease is not None + if ( + runtime_state.get("source_mode") != "idle" + or acquisition_lease_active + or not isinstance(current_session_id, str) + or not current_session_id + ): + raise LocalAcquisitionLifecycleError( + "Предыдущий локальный приём ещё не завершил освобождение ресурсов. " + "Команда запуска сканеру не отправлялась.", + reason_code="live_perception_ingress_busy", + ) + self.live_perception_ingress.end_session(current_session_id) + logger.warning( + "K1 stale live perception session retired before a new acquisition", + extra={ + "event_code": "k1_stale_live_perception_session_retired", + "reason_code": "terminal_session_ingress_stale", + "failure_stage": "stale-ingress-retirement", + "stale_session_retired": True, + "automatic_retry": False, + }, + ) + def _seal_acquisition_capture_clock(self) -> None: with self._lock: out_dir = self._acquisition_out_dir @@ -2318,9 +2818,7 @@ class XgridsK1CompatibilityService: and self._acquisition_session_lease is not None ) terminal_camera_status: Literal["complete", "failed"] = ( - "complete" - if current is not None and current.state == "completed" - else "failed" + "complete" if current is not None and current.state == "completed" else "failed" ) if terminal_canonical_cleanup: self._stop_acquisition_sources( @@ -2452,6 +2950,10 @@ class XgridsK1CompatibilityService: acquisition.transition("failed", message_code="acquisition.runtime_failed") camera_terminal_status = "failed" camera_failure_code = "runtime-failed" + # The source thread has already failed. Normalize its retained + # phase/source fields to idle after sealing so connection UI + # cannot mistake this terminal tail for a live receiver. + stop_runtime_for_camera_failure = True if waiting_for_start: failed_operation_id = self._acquisition_start_operation_id if waiting_for_stop: @@ -2530,6 +3032,10 @@ class XgridsK1CompatibilityService: message_code="acquisition.capture_clock_failed", ) finally: + with self._lock: + finalized_out_dir = self._acquisition_out_dir + if finalized_out_dir is not None: + self.live_perception_ingress.end_session(finalized_out_dir.name) if reconciliation_error is None: self._release_acquisition_session_lease() @@ -2771,9 +3277,7 @@ class XgridsK1PluginFacade: return await asyncio.to_thread(self.service.state) if action_id == ACTION_DEVICE_CALIBRATION_SNAPSHOT_READ: EmptyRequest.model_validate(payload) - return await asyncio.to_thread( - self.service.read_device_calibration_snapshot - ) + return await asyncio.to_thread(self.service.read_device_calibration_snapshot) if action_id == ACTION_NETWORK_PROVISION: connect_request = ConnectRequest.model_validate(payload) return await self.service.connect(connect_request) @@ -2806,9 +3310,7 @@ class XgridsK1PluginFacade: ) if action_id == ACTION_APPLICATION_CONTROL_SESSION_CLOSE: EmptyRequest.model_validate(payload) - return await asyncio.to_thread( - self.service.close_application_control_session - ) + return await asyncio.to_thread(self.service.close_application_control_session) if action_id == ACTION_ACQUISITION_PREPARE: prepare_request = PrepareAcquisitionRequest.model_validate(payload) return await asyncio.to_thread(self.service.prepare_acquisition, prepare_request) @@ -3090,6 +3592,60 @@ def _target_is_local_ipv4(target: str) -> bool: return source_address == target +def _control_endpoint_reachable( + target: str, + *, + timeout_seconds: float = CONTROL_ENDPOINT_PROBE_TIMEOUT_SECONDS, +) -> bool: + """Check only the confirmed K1 MQTT endpoint; send no protocol bytes.""" + + if timeout_seconds <= 0: + raise ValueError("control endpoint probe timeout must be positive") + target = validate_private_ipv4(target) + try: + with socket.create_connection( + (target, CONTROL_MQTT_PORT), + timeout=timeout_seconds, + ): + return True + except OSError: + return False + + +def _host_route_class(target: str) -> str: + """Classify the host route without transmitting packets to the target.""" + + target = validate_private_ipv4(target) + if sys.platform != "darwin": + return "unknown" + try: + result = subprocess.run( + ["route", "-n", "get", target], + check=False, + capture_output=True, + text=True, + timeout=2.0, + ) + except (OSError, subprocess.SubprocessError): + return "unknown" + if result.returncode != 0: + return "unknown" + fields: dict[str, str] = {} + for line in result.stdout.splitlines(): + key, separator, value = line.strip().partition(":") + if separator: + fields[key] = value.strip() + interface = fields.get("interface", "") + destination = fields.get("destination", "") + if interface.startswith(("utun", "ipsec", "ppp")): + return "tunnel" + if destination == "default": + return "default-route" + if interface: + return "direct-or-routed" + return "unknown" + + def _validate_installed_compatibility_profile(repository_root: Path) -> None: """Run the plugin-owned, fail-closed profile validator before activation.""" diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/application_mqtt.py b/src/k1link/device_plugins/xgrids_k1/protocol/application_mqtt.py index f8f7b0a..56885d6 100644 --- a/src/k1link/device_plugins/xgrids_k1/protocol/application_mqtt.py +++ b/src/k1link/device_plugins/xgrids_k1/protocol/application_mqtt.py @@ -243,6 +243,9 @@ class ApplicationMqttTransportSnapshot: latest_device_init_ready: bool | None latest_system_error_code: int | None latest_system_error_state: str | None + last_loop_result_code: int | None + last_loop_result_name: str | None + last_loop_phase: str | None operation_keys_consumed: int clean_session: bool = False keepalive_seconds: int = CONTROL_KEEPALIVE_SECONDS @@ -268,6 +271,9 @@ class ApplicationMqttTransportSnapshot: "latest_device_init_ready": self.latest_device_init_ready, "latest_system_error_code": self.latest_system_error_code, "latest_system_error_state": self.latest_system_error_state, + "last_loop_result_code": self.last_loop_result_code, + "last_loop_result_name": self.last_loop_result_name, + "last_loop_phase": self.last_loop_phase, "operation_keys_consumed": self.operation_keys_consumed, "clean_session": self.clean_session, "keepalive_seconds": self.keepalive_seconds, @@ -348,6 +354,9 @@ class ReviewedApplicationMqttTransport: self._latest_device_fault = False self._latest_system_error_code: int | None = None self._latest_system_error_state: str | None = None + self._last_loop_result_code: int | None = None + self._last_loop_result_name: str | None = None + self._last_loop_phase: str | None = None def open(self) -> ApplicationMqttTransportSnapshot: with self._lock: @@ -562,7 +571,9 @@ class ReviewedApplicationMqttTransport: except (OSError, RuntimeError, ValueError) as exc: self._fail_after_publish("control MQTT network loop failed", exc) if result != mqtt.MQTT_ERR_SUCCESS: - self._fail_after_publish("control MQTT network loop returned an error") + self._fail_after_publish( + self._record_loop_failure(result, phase="maintain-open") + ) self._discard_allowed_responses(allowed) def close(self) -> None: @@ -649,6 +660,9 @@ class ReviewedApplicationMqttTransport: latest_device_init_ready=self._latest_device_init_ready, latest_system_error_code=self._latest_system_error_code, latest_system_error_state=self._latest_system_error_state, + last_loop_result_code=self._last_loop_result_code, + last_loop_result_name=self._last_loop_result_name, + last_loop_phase=self._last_loop_phase, operation_keys_consumed=len(self._consumed_operation_keys), ) @@ -879,9 +893,13 @@ class ReviewedApplicationMqttTransport: self._fail_after_publish("control MQTT network loop failed", exc) self._fail_before_publish("control MQTT network loop failed", exc) if result != mqtt.MQTT_ERR_SUCCESS: + failure = self._record_loop_failure( + result, + phase="post-publish" if post_publish else "connect-subscribe", + ) if post_publish: - self._fail_after_publish("control MQTT network loop returned an error") - self._fail_before_publish("control MQTT network loop returned an error") + self._fail_after_publish(failure) + self._fail_before_publish(failure) def _service_once(self, *, post_publish: bool) -> None: with self._lock: @@ -898,9 +916,33 @@ class ReviewedApplicationMqttTransport: self._fail_after_publish("control MQTT network loop failed", exc) self._fail_before_publish("control MQTT network loop failed", exc) if result != mqtt.MQTT_ERR_SUCCESS: + failure = self._record_loop_failure( + result, + phase="post-publish-drain" if post_publish else "pre-publish-drain", + ) if post_publish: - self._fail_after_publish("control MQTT network loop returned an error") - self._fail_before_publish("control MQTT network loop returned an error") + self._fail_after_publish(failure) + self._fail_before_publish(failure) + + def _record_loop_failure( + self, + result: int, + *, + phase: str, + ) -> str: + result_code = int(result) + try: + result_name = mqtt.error_string(result_code) + except (TypeError, ValueError): + result_name = "unknown MQTT error" + with self._lock: + self._last_loop_result_code = result_code + self._last_loop_result_name = result_name + self._last_loop_phase = phase + return ( + "control MQTT network loop returned an error " + f"(phase={phase}, result={result_code}: {result_name})" + ) def _drain_responses( self, diff --git a/src/k1link/device_plugins/xgrids_k1/protocol/application_session.py b/src/k1link/device_plugins/xgrids_k1/protocol/application_session.py index a1e57ec..a185230 100644 --- a/src/k1link/device_plugins/xgrids_k1/protocol/application_session.py +++ b/src/k1link/device_plugins/xgrids_k1/protocol/application_session.py @@ -62,6 +62,7 @@ logger = logging.getLogger(__name__) TransportFactory = Callable[[str], ReviewedApplicationMqttTransport] +ScanningObserver = Callable[[], None] @dataclass(frozen=True, slots=True) @@ -97,10 +98,13 @@ class InteractiveApplicationControlSession: *, transport_factory: TransportFactory = ReviewedApplicationMqttTransport, epoch_seconds: Callable[[], int] = lambda: int(time.time()), + scanning_observer: ScanningObserver | None = None, ) -> None: self._authority_loader = authority_loader self._transport_factory = transport_factory self._epoch_seconds = epoch_seconds + self._scanning_observer = scanning_observer + self._scanning_observer_errors = 0 self._lock = threading.RLock() self._phase: ApplicationControlPhase = "idle" self._host: str | None = None @@ -215,6 +219,55 @@ class InteractiveApplicationControlSession: transport.close() return self.snapshot() + def retire_for_network_change(self) -> dict[str, object]: + """Retire local control ownership for one new explicit network action. + + A correlated STOP followed by a network-loop loss in SCAN_STOPPING is + not enough evidence to authorize another START. It is enough to retire + the dead local socket when the operator explicitly chooses a new + network route. Pre-START failures already classified safe for a fresh + click are admissible here for the same reason: no modeling command was + attempted. + """ + + with self._lock: + failure = self._failure + failed_network_change_admissible = ( + self._phase == "failed" + and failure is not None + and ( + failure.get("safe_to_retry") is True + or failure.get("network_change_admissible") is True + ) + ) + if self._phase not in {"idle", "completed", "closed"} and not ( + failed_network_change_admissible + ): + raise ApplicationAcceptanceError( + "control session cannot be retired for a network change" + ) + if not self._worker_retired_locked(): + raise ApplicationAcceptanceError( + "control session worker is still retiring" + ) + previous_phase = self._phase + previous_reason = ( + self._json_string(failure.get("reason_code")) + if failure is not None + else None + ) + self._reset_locked() + if previous_phase != "idle": + logger.info( + "K1 local control session retired for an explicit network change", + extra={ + "event_code": "k1_control_session_retired_for_network_change", + "reason_code": previous_reason, + "automatic_retry": False, + }, + ) + return self.snapshot() + def close(self) -> None: """Stop the process-owned socket without ever inventing a device STOP.""" @@ -247,6 +300,7 @@ class InteractiveApplicationControlSession: "scripted_transitions": False, "automatic_retry": False, "outcome_unknown": self._outcome_unknown, + "scanning_observer_errors": self._scanning_observer_errors, "failure": dict(self._failure) if self._failure is not None else None, "dialogue": ( dict(self._dialogue_snapshot) if self._dialogue_snapshot is not None else None @@ -310,6 +364,7 @@ class InteractiveApplicationControlSession: checkpoint=start_checkpoint, ) self._set_phase("scanning") + self._notify_scanning_observer() executor.maintain_active_until_stop_requested(self._stop_requested.is_set) stop_confirmation = self._stop_request() @@ -412,10 +467,52 @@ class InteractiveApplicationControlSession: ) == 1 ) - safe_to_retry = not outcome_unknown and ( - not transport_created - or (transport_snapshot_available and publish_attempts == 0) - or correlated_read_only_profile_mismatch + status_reconciled_prestart_failure = ( + outcome_unknown + and modeling_command_attempted is False + and transport_snapshot_available + and not diagnostic_evidence_unavailable + and self._json_int_or_none( + transport_snapshot.get("device_status_reports") + ) + not in {None, 0} + and self._json_string( + transport_snapshot.get("latest_device_session_state") + ) + == "ready" + and transport_snapshot.get("latest_device_project_bound") is False + and self._json_int_or_none( + transport_snapshot.get("latest_system_error_code") + ) + is None + ) + stop_acknowledged_network_change = ( + outcome_unknown + and failure_reason_code == "mqtt_network_loop_failed" + and failed_phase == "awaiting-standby-confirmation" + and dialogue_snapshot_available + and transport_snapshot_available + and dialogue_snapshot.get("stop_attempted") is True + and dialogue_snapshot.get("stop_complete") is True + and dialogue_stage == "stop-acknowledged" + and modeling_command_attempted is True + and self._json_string( + transport_snapshot.get("latest_device_session_state") + ) + == "scan_stopping" + and transport_snapshot.get("latest_device_project_bound") is True + and self._json_int_or_none( + transport_snapshot.get("latest_system_error_code") + ) + is None + ) + safe_to_retry = status_reconciled_prestart_failure or ( + not outcome_unknown + and ( + not transport_created + or (transport_snapshot_available and publish_attempts == 0) + or correlated_read_only_profile_mismatch + ) ) self._failure = { "code": type(exc).__name__, @@ -451,6 +548,37 @@ class InteractiveApplicationControlSession: else None ), "safe_to_retry": safe_to_retry, + **( + { + "status_reconciliation": { + "device_session_state": "ready", + "device_project_bound": False, + "system_error_code": None, + "decision": "safe-explicit-prestart-retry", + "automatic_retry": False, + } + } + if status_reconciled_prestart_failure + else {} + ), + **( + { + "network_change_admissible": True, + "network_change_reconciliation": { + "device_session_state": "scan_stopping", + "device_project_bound": True, + "system_error_code": None, + "stop_complete": True, + "standby_confirmed": False, + "decision": ( + "explicit-network-change-only-after-acknowledged-stop" + ), + "automatic_retry": False, + }, + } + if stop_acknowledged_network_change + else {} + ), } self._outcome_unknown = outcome_unknown self._dialogue_snapshot = dialogue_snapshot or None @@ -462,7 +590,8 @@ class InteractiveApplicationControlSession: "publish_attempts=%s modeling_command_attempted=%s " "diagnostic_snapshot_unavailable=%s " "diagnostic_evidence_unavailable=%s outcome_unknown=%s " - "safe_to_retry=%s", + "safe_to_retry=%s status_reconciliation=%s " + "network_change_reconciliation=%s", type(exc).__name__, failure_reason_code, failed_phase, @@ -474,6 +603,51 @@ class InteractiveApplicationControlSession: diagnostic_evidence_unavailable, outcome_unknown, safe_to_retry, + ( + "safe-explicit-prestart-retry" + if status_reconciled_prestart_failure + else None + ), + ( + "explicit-network-change-only-after-acknowledged-stop" + if stop_acknowledged_network_change + else None + ), + extra={ + "event_code": "k1_application_control_session_failed", + "reason_code": failure_reason_code, + "failed_phase": failed_phase, + "dialogue_stage": dialogue_stage, + "transport_state": self._json_string( + transport_snapshot.get("state") + ), + "publish_attempts": publish_attempts, + "modeling_command_attempted": modeling_command_attempted, + "outcome_unknown": outcome_unknown, + "safe_to_retry": safe_to_retry, + "status_reconciliation": ( + "safe-explicit-prestart-retry" + if status_reconciled_prestart_failure + else None + ), + "network_change_admissible": ( + stop_acknowledged_network_change + ), + "network_change_reconciliation": ( + "explicit-network-change-only-after-acknowledged-stop" + if stop_acknowledged_network_change + else None + ), + "mqtt_loop_result_code": self._json_int_or_none( + transport_snapshot.get("last_loop_result_code") + ), + "mqtt_loop_result_name": self._json_string( + transport_snapshot.get("last_loop_result_name") + ), + "mqtt_loop_phase": self._json_string( + transport_snapshot.get("last_loop_phase") + ), + }, ) finally: final_dialogue_snapshot: dict[str, object] | None = None @@ -552,6 +726,23 @@ class InteractiveApplicationControlSession: with self._lock: self._set_phase_locked(phase) + def _notify_scanning_observer(self) -> None: + observer = self._scanning_observer + if observer is None: + return + try: + observer() + except Exception: + with self._lock: + self._scanning_observer_errors += 1 + logger.exception( + "K1 confirmed SCANNING, but its acquisition observer failed", + extra={ + "event_code": "k1_application_scanning_observer_failed", + "automatic_retry": False, + }, + ) + @staticmethod def _executor_snapshot_safely( executor: PhysicalAcceptanceDialogueExecutor | None, diff --git a/src/k1link/device_plugins/xgrids_k1/viewer/runtime.py b/src/k1link/device_plugins/xgrids_k1/viewer/runtime.py index 63f76f0..88722a2 100644 --- a/src/k1link/device_plugins/xgrids_k1/viewer/runtime.py +++ b/src/k1link/device_plugins/xgrids_k1/viewer/runtime.py @@ -226,7 +226,7 @@ class VisualizationRuntime: raise RuntimeError("поток не завершился за отведённое время; повторите остановку") def close(self, *, wait_seconds: float = 5.0) -> None: - """Stop the active source and release the process-wide visual bridge.""" + """Stop the active source and release its acquisition-scoped bridge.""" with self._lock: self._closed = True thread = self._thread @@ -429,24 +429,25 @@ class VisualizationRuntime: bridge: RerunBridge | None = None try: with self._lock: - bridge = self._bridge if self._closed: publisher_aborted.set() if publisher_aborted.is_set(): publisher_ready.set() return - if bridge is None: - candidate = self._bridge_factory( - grpc_port=self._grpc_port, - metrics=self._metrics, - settings_provider=self._current_scene_settings, - ) - bridge = candidate - with self._lock: - if self._closed: - publisher_aborted.set() - else: - self._bridge = candidate + candidate = self._bridge_factory( + grpc_port=self._grpc_port, + metrics=self._metrics, + settings_provider=self._current_scene_settings, + ) + bridge = candidate + with self._lock: + if self._closed: + publisher_aborted.set() + else: + # Every acquisition owns a fresh Rerun recording/store. + # Reusing a RecordingStream across independent scans can + # leave a later WebViewer without a StoreInfo event. + self._bridge = candidate if publisher_aborted.is_set(): publisher_ready.set() return @@ -520,19 +521,17 @@ class VisualizationRuntime: finally: if bridge is not None: with self._lock: - close_bridge = self._closed and ( - self._bridge is bridge or publisher_aborted.is_set() - ) - if close_bridge and self._bridge is bridge: + if self._bridge is bridge: self._bridge = None self._rerun_grpc_url = None - if close_bridge: - try: - bridge.close() - except BaseException as exc: - # Process shutdown must not become a false successful - # idle state when the native bridge failed to close. - publisher_error.append(exc) + try: + bridge.close() + except BaseException as exc: + # A failed native disconnect must not become a false + # successful idle state or leak port 9876 into the next + # independent acquisition. + publisher_error.append(exc) + self._notify() def join_publisher() -> None: # Never orphan a publisher: the session thread remains its owner. @@ -596,6 +595,7 @@ class VisualizationRuntime: self._message = message self._foxglove_ws_url = None self._foxglove_viewer_url = None + self._rerun_grpc_url = None self._notify() def _finish_error(self, message: str) -> None: @@ -605,6 +605,7 @@ class VisualizationRuntime: self._message = message self._foxglove_ws_url = None self._foxglove_viewer_url = None + self._rerun_grpc_url = None self._notify() def _current_scene_settings(self) -> RerunSceneSettings: diff --git a/src/k1link/viewer/rerun_bridge.py b/src/k1link/viewer/rerun_bridge.py index a3c0bc8..7fcfb14 100644 --- a/src/k1link/viewer/rerun_bridge.py +++ b/src/k1link/viewer/rerun_bridge.py @@ -1,6 +1,8 @@ from __future__ import annotations +import logging import math +import socket import time from collections.abc import Callable, Mapping from contextlib import suppress @@ -31,6 +33,7 @@ TRAJECTORY_MIN_DISTANCE_METERS = 0.02 TRAJECTORY_PUBLISH_INTERVAL_NS = 500_000_000 LIVE_GRPC_BUFFER_LIMIT = "32MiB" DEFAULT_GRPC_PORT = 9876 +GRPC_PORT_SEARCH_SPAN = 128 DEFAULT_CORS_ORIGINS = ( "http://127.0.0.1:5173", "http://localhost:5173", @@ -40,6 +43,8 @@ DEFAULT_CORS_ORIGINS = ( "http://localhost:8000", ) +logger = logging.getLogger("k1link.device_plugins.xgrids_k1.viewer_receiver") + @dataclass(frozen=True, slots=True) class RerunSceneSettings: @@ -62,6 +67,34 @@ class RerunSceneSettings: SettingsProvider = Callable[[], RerunSceneSettings] +def _select_available_grpc_port( + preferred_port: int, + *, + search_span: int = GRPC_PORT_SEARCH_SPAN, +) -> int: + """Select a local Rerun port without reusing a still-served recording.""" + + if not 1 <= preferred_port <= 65_535: + raise ValueError("Rerun gRPC port must be between 1 and 65535") + if search_span < 1: + raise ValueError("Rerun gRPC port search span must be positive") + + last_port = min(preferred_port + search_span, 65_536) + for candidate in range(preferred_port, last_port): + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe: + try: + # Rerun binds all local interfaces. Probe the same address class + # so a previous recording retained by a late viewer is detected. + probe.bind(("0.0.0.0", candidate)) + except OSError: + continue + return candidate + raise RuntimeError( + "No local Rerun gRPC port is available in " + f"{preferred_port}..{last_port - 1}" + ) + + class RerunBridge: """Publish transport-neutral canonical envelopes to a Rerun recording.""" @@ -85,9 +118,20 @@ class RerunBridge: else: recording = recording_factory("nodedc_mission_core_spatial") try: + selected_grpc_port = _select_available_grpc_port(grpc_port) + if selected_grpc_port != grpc_port: + logger.info( + "Mission Core selected a new Rerun port because an earlier " + "viewer still owns the preferred listener", + extra={ + "event_code": "rerun_grpc_port_rotated", + "preferred_port": grpc_port, + "selected_port": selected_grpc_port, + }, + ) blueprint = _blueprint(self._settings) url = recording.serve_grpc( - grpc_port=grpc_port, + grpc_port=selected_grpc_port, default_blueprint=blueprint, # This is a reconnect cushion for the live preview, not the source # of record. Raw MQTT evidence is persisted independently. A large @@ -132,7 +176,7 @@ class RerunBridge: return self._url def begin_session(self, metrics: BridgeMetrics | None = None) -> None: - """Reset session state while keeping the process-lifetime server alive.""" + """Initialize the one acquisition owned by this recording server.""" if self._closed: raise RuntimeError("Rerun bridge is already closed") if metrics is not None: diff --git a/src/k1link/web/app.py b/src/k1link/web/app.py index cef5a91..98ef57d 100644 --- a/src/k1link/web/app.py +++ b/src/k1link/web/app.py @@ -53,8 +53,10 @@ from k1link.web.plugin_runtime import ( PluginRuntimeUnavailableError, ) from k1link.web.polygon_api import build_polygon_router, configured_polygon_runs_root +from k1link.web.runtime_diagnostics import configure_scanner_diagnostics from k1link.web.session_api import build_session_router from k1link.web.system_telemetry_api import build_system_telemetry_router +from k1link.web.viewer_diagnostics_api import build_viewer_diagnostics_router REPOSITORY_ROOT = Path(__file__).resolve().parents[3] INVALID_REQUEST_DETAIL = "Некорректные параметры запроса." @@ -274,6 +276,9 @@ async def _recording_preparation_reconciler() -> None: async def app_lifespan(_: FastAPI) -> AsyncIterator[None]: reconciler: asyncio.Task[None] | None = None try: + configure_scanner_diagnostics( + REPOSITORY_ROOT / ".runtime" / "mission-core" / "logs" + ) session_recording_preparation_manager.start() # Recovery is intentionally a one-shot startup phase. The archive # helper owns a cross-process lease, while ordinary catalog requests @@ -671,6 +676,7 @@ app.include_router( root_provider=lambda: REPOSITORY_ROOT / ".runtime" / "system", ) ) +app.include_router(build_viewer_diagnostics_router()) frontend_dist = REPOSITORY_ROOT / "apps" / "control-station" / "dist" diff --git a/src/k1link/web/runtime_diagnostics.py b/src/k1link/web/runtime_diagnostics.py new file mode 100644 index 0000000..1e61f4e --- /dev/null +++ b/src/k1link/web/runtime_diagnostics.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import json +import logging +import os +from datetime import UTC, datetime +from logging.handlers import RotatingFileHandler +from pathlib import Path +from typing import Final + +SCANNER_LOGGER_NAME: Final = "k1link.device_plugins.xgrids_k1" +SCANNER_DIAGNOSTIC_FILE: Final = "scanner-diagnostics.jsonl" +_HANDLER_MARKER: Final = "_mission_core_scanner_diagnostics_path" +_EXTRA_FIELDS: Final = ( + "event_code", + "reason_code", + "failed_phase", + "dialogue_stage", + "transport_state", + "publish_attempts", + "modeling_command_attempted", + "outcome_unknown", + "safe_to_retry", + "status_reconciliation", + "network_change_admissible", + "network_change_reconciliation", + "mqtt_loop_result_code", + "mqtt_loop_result_name", + "mqtt_loop_phase", + "automatic_retry", + "camera_source_id", + "evidence_session_id", + "activation_trigger", + "lease_generation", + "lease_state", + "recovery_strategy", + "endpoint_reachable", + "host_route_class", + "address_changed", + "device_write_performed", + "failure_stage", + "start_checkpoint_released", + "stale_session_retired", + "stream_id", + "backend_activity_sequence", + "viewer_range_max_ns", + "stalled_for_ms", + "recovery_attempt", + "preferred_port", + "selected_port", +) + + +class ScannerDiagnosticJsonFormatter(logging.Formatter): + """Serialize a bounded, secret-free scanner diagnostic record.""" + + def format(self, record: logging.LogRecord) -> str: + document: dict[str, object] = { + "timestamp_utc": datetime.fromtimestamp(record.created, UTC) + .isoformat() + .replace("+00:00", "Z"), + "level": record.levelname, + "logger": record.name, + "message": record.getMessage()[:1000], + } + for field in _EXTRA_FIELDS: + value = getattr(record, field, None) + if value is not None and isinstance(value, (str, int, bool)): + document[field] = value + return json.dumps(document, ensure_ascii=False, separators=(",", ":")) + + +def configure_scanner_diagnostics(logs_root: Path) -> Path: + """Install one private rotating JSONL handler for K1 field diagnostics.""" + + logs_root.mkdir(mode=0o700, parents=True, exist_ok=True) + os.chmod(logs_root, 0o700) + target = (logs_root / SCANNER_DIAGNOSTIC_FILE).resolve() + logger = logging.getLogger(SCANNER_LOGGER_NAME) + for handler in logger.handlers: + if getattr(handler, _HANDLER_MARKER, None) == str(target): + return target + + handler = RotatingFileHandler( + target, + maxBytes=5 * 1024 * 1024, + backupCount=5, + encoding="utf-8", + delay=False, + ) + os.chmod(target, 0o600) + handler.setLevel(logging.INFO) + handler.setFormatter(ScannerDiagnosticJsonFormatter()) + setattr(handler, _HANDLER_MARKER, str(target)) + logger.addHandler(handler) + logger.setLevel(logging.INFO) + return target diff --git a/src/k1link/web/viewer_diagnostics_api.py b/src/k1link/web/viewer_diagnostics_api.py new file mode 100644 index 0000000..dd259fc --- /dev/null +++ b/src/k1link/web/viewer_diagnostics_api.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import logging +from typing import Literal + +from fastapi import APIRouter, Response +from pydantic import BaseModel, ConfigDict, Field + +logger = logging.getLogger("k1link.device_plugins.xgrids_k1.viewer_receiver") + +LiveViewerEventCode = Literal[ + "live_receiver_stalled", + "live_receiver_restart_requested", + "live_receiver_recovered", + "live_receiver_recovery_exhausted", + "live_receiver_active_store_admitted", + "live_receiver_error", +] +LiveViewerFailureStage = Literal[ + "recording-open-timeout", + "viewer-start", + "module-load", + "receiver-stalled", +] + + +class LiveViewerDiagnosticEvent(BaseModel): + model_config = ConfigDict(extra="forbid", strict=True) + + schema_version: Literal["missioncore.live-viewer-diagnostic/v1"] + event_code: LiveViewerEventCode + failure_stage: LiveViewerFailureStage | None = None + stream_id: str | None = Field( + default=None, + min_length=1, + max_length=128, + pattern=r"^[A-Za-z0-9][A-Za-z0-9._:-]*$", + ) + backend_activity_sequence: int | None = Field(default=None, ge=0) + viewer_range_max_ns: int | None = Field(default=None, ge=0) + stalled_for_ms: int | None = Field(default=None, ge=0, le=600_000) + recovery_attempt: int | None = Field(default=None, ge=1, le=3) + + +def build_viewer_diagnostics_router() -> APIRouter: + router = APIRouter(prefix="/api/v1/viewer", tags=["viewer"]) + + @router.post("/live-diagnostics", status_code=204) + def record_live_diagnostic(event: LiveViewerDiagnosticEvent) -> Response: + logger.info( + "Mission Core live Rerun receiver diagnostic: event=%s stream=%s", + event.event_code, + event.stream_id, + extra={ + "event_code": event.event_code, + "failure_stage": event.failure_stage, + "stream_id": event.stream_id, + "backend_activity_sequence": event.backend_activity_sequence, + "viewer_range_max_ns": event.viewer_range_max_ns, + "stalled_for_ms": event.stalled_for_ms, + "recovery_attempt": event.recovery_attempt, + }, + ) + return Response(status_code=204) + + return router diff --git a/tests/test_live_perception.py b/tests/test_live_perception.py index 10c90e0..22b80ee 100644 --- a/tests/test_live_perception.py +++ b/tests/test_live_perception.py @@ -136,6 +136,31 @@ def test_live_ingress_allows_only_one_worker_consumer() -> None: ingress.open_consumer("worker-2") +def test_live_ingress_new_session_discards_queued_events_from_previous_session() -> None: + ingress = LivePerceptionIngress() + ingress.open_consumer("worker-1") + ingress.begin_session("session-1") + assert ingress.publish( + modality="lidar", + source_id="lixel/application/report/lio_pcl", + source_sequence=1, + captured_at_epoch_ns=1, + received_monotonic_ns=1, + payload=b"old-session", + ) + + ingress.end_session("session-1") + ingress.begin_session("session-2") + + events = [] + while (event := ingress.take_next("worker-1", timeout=0)) is not None: + events.append(event) + assert len(events) == 1 + assert events[0].session_id == "session-2" + assert events[0].modality == "control" + assert events[0].payload == b'{"event":"session-start"}' + + def test_live_result_round_trip_keeps_video_mask_boxes_and_shadow_authority() -> None: mask = np.zeros((600, 800), dtype=np.uint8) mask[100:120, 200:240] = 4 diff --git a/tests/test_rerun_bridge.py b/tests/test_rerun_bridge.py index 7627efa..9ca52d2 100644 --- a/tests/test_rerun_bridge.py +++ b/tests/test_rerun_bridge.py @@ -1,6 +1,8 @@ from __future__ import annotations import json +import logging +import socket import struct import threading import time @@ -20,6 +22,7 @@ from k1link.viewer.rerun_bridge import ( RerunSceneSettings, _live_time_panel, _point_colors, + _select_available_grpc_port, ) @@ -62,6 +65,33 @@ class DisconnectFailureRecording(FakeRecording): raise RuntimeError("synthetic disconnect failure") +def test_rerun_port_selection_skips_a_recording_still_held_by_a_viewer( + caplog: pytest.LogCaptureFixture, +) -> None: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as occupied: + occupied.bind(("0.0.0.0", 0)) + occupied.listen() + preferred_port = int(occupied.getsockname()[1]) + + selected_port = _select_available_grpc_port(preferred_port, search_span=8) + recording = FakeRecording() + with caplog.at_level( + logging.INFO, + logger="k1link.device_plugins.xgrids_k1.viewer_receiver", + ): + bridge = RerunBridge( + grpc_port=preferred_port, + recording_factory=lambda _: recording, # type: ignore[arg-type] + ) + + assert selected_port != preferred_port + assert preferred_port < selected_port < preferred_port + 8 + assert caplog.records[-1].event_code == "rerun_grpc_port_rotated" + assert caplog.records[-1].preferred_port == preferred_port + assert caplog.records[-1].selected_port == selected_port + bridge.close() + + def _message( topic: str, payload: bytes, @@ -266,7 +296,7 @@ def test_palettes_are_deterministic_and_custom_color_is_exact() -> None: assert custom_over_rgb.tolist() == [[16, 32, 48], [16, 32, 48]] -def test_runtime_reuses_one_bridge_across_sequential_sessions(tmp_path: Path) -> None: +def test_runtime_owns_fresh_bridge_for_each_sequential_session(tmp_path: Path) -> None: capture = tmp_path / "mqtt.raw.k1mqtt" point_topic = "RealtimePointcloud" pose_topic = "RealtimePath" @@ -300,10 +330,12 @@ def test_runtime_reuses_one_bridge_across_sequential_sessions(tmp_path: Path) -> encoding="utf-8", ) - recording = FakeRecording() + recordings: list[FakeRecording] = [] created: list[RerunBridge] = [] def bridge_factory(**kwargs: object) -> RerunBridge: + recording = FakeRecording() + recordings.append(recording) bridge = RerunBridge( recording_factory=lambda _: recording, # type: ignore[arg-type] **kwargs, # type: ignore[arg-type] @@ -329,8 +361,8 @@ def test_runtime_reuses_one_bridge_across_sequential_sessions(tmp_path: Path) -> assert snapshot["metrics"]["mqtt_to_publish_ms"] is None runtime.stop(wait_seconds=5.0) assert runtime.snapshot()["phase"] == "idle" - assert runtime.snapshot()["rerun_grpc_url"] == "rerun+http://127.0.0.1:9876/proxy" - assert recording.disconnected is False + assert runtime.snapshot()["rerun_grpc_url"] is None + assert recordings[0].disconnected is True runtime.start_replay(capture, speed=0.0) deadline = time.monotonic() + 5.0 @@ -340,13 +372,12 @@ def test_runtime_reuses_one_bridge_across_sequential_sessions(tmp_path: Path) -> time.sleep(0.05) runtime.stop(wait_seconds=5.0) - assert len(created) == 1 + assert len(created) == 2 assert runtime.snapshot()["metrics"]["pcl_frames"] == 1 - assert runtime.snapshot()["rerun_grpc_url"] == "rerun+http://127.0.0.1:9876/proxy" - assert recording.disconnected is False + assert runtime.snapshot()["rerun_grpc_url"] is None + assert all(recording.disconnected for recording in recordings) runtime.close() assert runtime.snapshot()["rerun_grpc_url"] is None - assert recording.disconnected is True def test_runtime_reports_bridge_close_failure_instead_of_false_idle(tmp_path: Path) -> None: @@ -385,18 +416,11 @@ def test_runtime_reports_bridge_close_failure_instead_of_false_idle(tmp_path: Pa time.sleep(0.01) snapshot = runtime.snapshot() - assert snapshot["phase"] == "idle" - assert snapshot["rerun_grpc_url"] == "rerun+http://127.0.0.1:9876/proxy" - assert recording.disconnected is False - - with pytest.raises(RuntimeError, match="synthetic disconnect failure"): - runtime.close() - - snapshot = runtime.snapshot() assert snapshot["phase"] == "error" assert "synthetic disconnect failure" in snapshot["message"] assert snapshot["rerun_grpc_url"] is None assert recording.disconnected is True + runtime.close() def test_close_during_blocked_factory_closes_the_late_bridge(tmp_path: Path) -> None: diff --git a/tests/test_viewer_diagnostics_api.py b/tests/test_viewer_diagnostics_api.py new file mode 100644 index 0000000..f52f556 --- /dev/null +++ b/tests/test_viewer_diagnostics_api.py @@ -0,0 +1,142 @@ +from __future__ import annotations + +import json +import logging +import stat +from collections.abc import Callable +from pathlib import Path +from typing import Any + +import pytest +from fastapi import APIRouter +from fastapi.routing import APIRoute +from pydantic import ValidationError + +from k1link.web.runtime_diagnostics import ( + SCANNER_LOGGER_NAME, + configure_scanner_diagnostics, +) +from k1link.web.viewer_diagnostics_api import ( + LiveViewerDiagnosticEvent, + build_viewer_diagnostics_router, +) + + +def _endpoint(router: APIRouter, path: str, method: str) -> Callable[..., Any]: + for route in router.routes: + if ( + isinstance(route, APIRoute) + and route.path == path + and route.methods is not None + and method in route.methods + ): + return route.endpoint + raise AssertionError(f"{method} {path} route is missing") + + +def test_private_scanner_diagnostics_are_durable_structured_and_bounded( + tmp_path: Path, +) -> None: + target = configure_scanner_diagnostics(tmp_path / "logs") + logger = logging.getLogger(f"{SCANNER_LOGGER_NAME}.test") + logger.error( + "field control failure", + extra={ + "event_code": "k1_application_control_session_failed", + "reason_code": "mqtt_network_loop_failed", + "mqtt_loop_result_code": 7, + "mqtt_loop_result_name": "The connection was lost.", + "mqtt_loop_phase": "post-publish-drain", + "automatic_retry": False, + "camera_source_id": "sensor.camera.right", + "evidence_session_id": "20260728T163450Z_viewer_live", + "activation_trigger": "application-control-scanning", + "network_change_admissible": True, + "network_change_reconciliation": ( + "explicit-network-change-only-after-acknowledged-stop" + ), + "lease_generation": 3, + "lease_state": "reachable", + "recovery_strategy": "existing-mqtt-endpoint", + "endpoint_reachable": True, + "address_changed": False, + "device_write_performed": False, + "preferred_port": 9876, + "selected_port": 9877, + "unapproved_secret_field": "must-not-be-written", + }, + ) + for handler in logging.getLogger(SCANNER_LOGGER_NAME).handlers: + handler.flush() + + document = json.loads(target.read_text(encoding="utf-8").splitlines()[-1]) + assert stat.S_IMODE(target.parent.stat().st_mode) == 0o700 + assert stat.S_IMODE(target.stat().st_mode) == 0o600 + assert document["event_code"] == "k1_application_control_session_failed" + assert document["reason_code"] == "mqtt_network_loop_failed" + assert document["mqtt_loop_result_code"] == 7 + assert document["mqtt_loop_phase"] == "post-publish-drain" + assert document["automatic_retry"] is False + assert document["camera_source_id"] == "sensor.camera.right" + assert document["evidence_session_id"] == "20260728T163450Z_viewer_live" + assert document["activation_trigger"] == "application-control-scanning" + assert document["network_change_admissible"] is True + assert document["network_change_reconciliation"] == ( + "explicit-network-change-only-after-acknowledged-stop" + ) + assert document["lease_generation"] == 3 + assert document["lease_state"] == "reachable" + assert document["recovery_strategy"] == "existing-mqtt-endpoint" + assert document["endpoint_reachable"] is True + assert document["address_changed"] is False + assert document["device_write_performed"] is False + assert document["preferred_port"] == 9876 + assert document["selected_port"] == 9877 + assert "unapproved_secret_field" not in document + + parent = logging.getLogger(SCANNER_LOGGER_NAME) + for handler in list(parent.handlers): + if getattr(handler, "baseFilename", None) == str(target): + parent.removeHandler(handler) + handler.close() + + +def test_live_viewer_diagnostic_endpoint_accepts_only_bounded_events( + caplog: pytest.LogCaptureFixture, +) -> None: + router = build_viewer_diagnostics_router() + endpoint = _endpoint(router, "/api/v1/viewer/live-diagnostics", "POST") + event = LiveViewerDiagnosticEvent( + schema_version="missioncore.live-viewer-diagnostic/v1", + event_code="live_receiver_stalled", + failure_stage="receiver-stalled", + stream_id="acquisition-123", + backend_activity_sequence=8_572, + viewer_range_max_ns=231_000_000_000, + stalled_for_ms=5_500, + recovery_attempt=1, + ) + + with caplog.at_level( + logging.INFO, + logger="k1link.device_plugins.xgrids_k1.viewer_receiver", + ): + response = endpoint(event) + + assert response.status_code == 204 + assert "event=live_receiver_stalled" in caplog.text + assert caplog.records[-1].failure_stage == "receiver-stalled" + with pytest.raises(ValidationError): + LiveViewerDiagnosticEvent.model_validate( + { + **event.model_dump(), + "source_url": "http://192.168.56.1/private", + } + ) + fallback = LiveViewerDiagnosticEvent( + schema_version="missioncore.live-viewer-diagnostic/v1", + event_code="live_receiver_active_store_admitted", + stream_id="acquisition-123", + backend_activity_sequence=8_573, + ) + assert fallback.failure_stage is None diff --git a/tests/test_wifi_provisioning.py b/tests/test_wifi_provisioning.py index 8ab6675..45bda9a 100644 --- a/tests/test_wifi_provisioning.py +++ b/tests/test_wifi_provisioning.py @@ -128,3 +128,60 @@ def test_read_wifi_status_once_reads_only_and_returns_current_dhcp_address( assert result["operation"] == "single_reviewed_wifi_status_read" assert result["write_performed"] is False assert result["status"]["ipv4"] == "10.255.254.77" + + +def test_read_wifi_status_recovery_rediscover_ignores_retained_handle( + monkeypatch: pytest.MonkeyPatch, +) -> None: + value = bytearray(54) + value[0] = 11 + value[1:12] = b"WIFI_CLIENT" + value[33] = 4 + value[34:38] = bytes((10, 255, 254, 77)) + value[50] = 1 + stale_handle = object() + recovered_handle = object() + characteristic = SimpleNamespace( + uuid=wifi_module.STATUS_CHARACTERISTIC_UUID, + service_uuid=wifi_module.SERVICE_UUID, + properties=["read"], + ) + service = SimpleNamespace(uuid=wifi_module.SERVICE_UUID) + + class FakeServices: + def get_service(self, uuid: str) -> object | None: + return service if uuid == wifi_module.SERVICE_UUID else None + + def get_characteristic(self, uuid: str) -> object | None: + return characteristic if uuid == wifi_module.STATUS_CHARACTERISTIC_UUID else None + + class FakeClient: + def __init__(self, device: object, **_kwargs: object) -> None: + assert device is recovered_handle + self.services = FakeServices() + self.name = "XGR-K1" + + async def __aenter__(self) -> Any: + return self + + async def __aexit__(self, *_args: object) -> None: + return None + + async def read_gatt_char(self, _characteristic: object) -> bytes: + return bytes(value) + + async def rediscover(*_args: object, **_kwargs: object) -> object: + return recovered_handle + + monkeypatch.setattr(wifi_module, "discovered_device", lambda _uuid: stale_handle) + monkeypatch.setattr(wifi_module.BleakScanner, "find_device_by_address", rediscover) + monkeypatch.setattr(wifi_module, "BleakClient", FakeClient) + + result = asyncio.run( + read_wifi_status_once( + "synthetic-corebluetooth-uuid", + rediscover=True, + ) + ) + + assert result["status"]["ipv4"] == "10.255.254.77" diff --git a/tests/test_xgrids_acquisition_lifecycle.py b/tests/test_xgrids_acquisition_lifecycle.py index 2021e78..f10d7bb 100644 --- a/tests/test_xgrids_acquisition_lifecycle.py +++ b/tests/test_xgrids_acquisition_lifecycle.py @@ -247,6 +247,8 @@ def test_verify_connection_refreshes_dynamic_dhcp_address_and_rotates_session( assert state["device_session"]["device_session_id"] != "old-device-session" assert state["connection_verification"] == { "status": "live-address-observed", + "lease_state": "configured", + "lease_generation": 1, "endpoint_validation": "ble-wifi-status-read", "network_reachability": "not-probed", "address_changed": True, @@ -283,7 +285,7 @@ def test_implicit_acquisition_target_uses_current_ble_dhcp_address( assert state["acquisition"]["target_host"] == "10.255.254.77" -def test_control_session_opens_against_current_ble_dhcp_address( +def test_control_session_reuses_reachable_process_owned_connection_lease( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: @@ -291,6 +293,8 @@ def test_control_session_opens_against_current_ble_dhcp_address( service._selected_device_id = "test-ble-transport" # noqa: SLF001 service._connection_mode = "bridge" # noqa: SLF001 service._k1_ip = "10.255.254.54" # noqa: SLF001 + service._device_id = "known-k1" # noqa: SLF001 + service._device_session_id = "known-session" # noqa: SLF001 service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001 opened_hosts: list[str] = [] @@ -302,12 +306,12 @@ def test_control_session_opens_against_current_ble_dhcp_address( opened_hosts.append(host) return self.snapshot() - async def fake_status_read(*_: object, **__: object) -> dict[str, Any]: - return _wifi_status_read("10.255.254.77") + async def forbidden_status_read(*_: object, **__: object) -> dict[str, Any]: + raise AssertionError("reachable connection lease must not reopen BLE") service._application_control_session = FakeOpenControlSession() # type: ignore[assignment] # noqa: SLF001 - monkeypatch.setattr(facade_module, "read_wifi_status_once", fake_status_read) - monkeypatch.setattr(facade_module, "_target_is_local_ipv4", lambda _target: False) + monkeypatch.setattr(facade_module, "read_wifi_status_once", forbidden_status_read) + monkeypatch.setattr(facade_module, "_control_endpoint_reachable", lambda _target: True) state = service.open_application_control_session( OpenApplicationControlSessionRequest( @@ -320,8 +324,241 @@ def test_control_session_opens_against_current_ble_dhcp_address( ) ) + assert opened_hosts == ["10.255.254.54"] + assert state["k1_ip"] == "10.255.254.54" + assert state["connection_verification"]["lease_state"] == "reachable" + assert state["connection_verification"]["network_reachability"] == "reachable" + open_operation = next( + operation + for operation in state["operations"] + if operation["action"] == "application-control.session.open" + ) + assert open_operation["status"] == "succeeded" + assert open_operation["result"] == { + "lease_generation": 0, + "connection_lease_reused": True, + "recovery_performed": False, + "address_changed": False, + "device_write_performed": False, + } + + +def test_reachable_connection_lease_supports_repeated_independent_control_sessions( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + service, _ = service_with_fake_runtime(tmp_path) + service._selected_device_id = "test-ble-transport" # noqa: SLF001 + service._connection_mode = "bridge" # noqa: SLF001 + service._k1_ip = "10.255.254.54" # noqa: SLF001 + service._device_id = "known-k1" # noqa: SLF001 + service._device_session_id = "known-session" # noqa: SLF001 + service._connection_lease_generation = 7 # noqa: SLF001 + service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001 + opened_hosts: list[str] = [] + + class FakeCompletedControlSession: + def snapshot(self) -> dict[str, object]: + return { + "state": "completed", + "can_open": True, + "can_confirm_standby": False, + } + + def open(self, *, host: str, **_: object) -> dict[str, object]: + opened_hosts.append(host) + return self.snapshot() + + async def forbidden_status_read(*_: object, **__: object) -> dict[str, Any]: + raise AssertionError("repeated scans must not repeat BLE or Wi-Fi setup") + + service._application_control_session = FakeCompletedControlSession() # type: ignore[assignment] # noqa: SLF001 + monkeypatch.setattr(facade_module, "read_wifi_status_once", forbidden_status_read) + monkeypatch.setattr(facade_module, "_control_endpoint_reachable", lambda _target: True) + request = OpenApplicationControlSessionRequest( + operator_present=True, + owner_controlled_device=True, + lixelgo_closed=True, + battery_storage_confirmed=True, + expected_physical_state_confirmed=True, + timezone_name="Europe/Moscow", + ) + + service.open_application_control_session(request) + state = service.open_application_control_session(request) + + assert opened_hosts == ["10.255.254.54", "10.255.254.54"] + assert state["device_session"]["device_session_id"] == "known-session" + assert state["connection_verification"]["lease_generation"] == 7 + open_operations = [ + operation + for operation in state["operations"] + if operation["action"] == "application-control.session.open" + ] + assert len(open_operations) == 2 + assert {operation["status"] for operation in open_operations} == {"succeeded"} + assert all( + operation["result"]["connection_lease_reused"] is True for operation in open_operations + ) + + +def test_control_session_recovers_changed_bridge_address_without_wifi_write( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + service, _ = service_with_fake_runtime(tmp_path) + service._selected_device_id = "test-ble-transport" # noqa: SLF001 + service._connection_mode = "bridge" # noqa: SLF001 + service._k1_ip = "10.255.254.54" # noqa: SLF001 + service._device_id = "known-k1" # noqa: SLF001 + service._device_session_id = "old-session" # noqa: SLF001 + service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001 + opened_hosts: list[str] = [] + status_calls: list[dict[str, object]] = [] + + class FakeOpenControlSession: + def snapshot(self) -> dict[str, object]: + return {"state": "idle", "can_confirm_standby": False} + + def open(self, *, host: str, **_: object) -> dict[str, object]: + opened_hosts.append(host) + return self.snapshot() + + async def fake_status_read(*_: object, **kwargs: object) -> dict[str, Any]: + status_calls.append(kwargs) + return _wifi_status_read("10.255.254.77") + + service._application_control_session = FakeOpenControlSession() # type: ignore[assignment] # noqa: SLF001 + monkeypatch.setattr(facade_module, "read_wifi_status_once", fake_status_read) + monkeypatch.setattr(facade_module, "_target_is_local_ipv4", lambda _target: False) + monkeypatch.setattr(facade_module, "_host_route_class", lambda _target: "direct-or-routed") + monkeypatch.setattr( + facade_module, + "_control_endpoint_reachable", + lambda target: target == "10.255.254.77", + ) + + state = service.open_application_control_session( + OpenApplicationControlSessionRequest( + operator_present=True, + owner_controlled_device=True, + lixelgo_closed=True, + battery_storage_confirmed=True, + expected_physical_state_confirmed=True, + timezone_name="Europe/Moscow", + ) + ) + + assert status_calls == [{"timeout_seconds": 20.0, "rediscover": True}] assert opened_hosts == ["10.255.254.77"] assert state["k1_ip"] == "10.255.254.77" + assert state["device_session"]["device_session_id"] != "old-session" + assert state["connection_verification"]["status"] == "recovered" + assert state["connection_verification"]["write_performed"] is False + open_operation = next( + operation + for operation in state["operations"] + if operation["action"] == "application-control.session.open" + ) + assert open_operation["result"]["recovery_performed"] is True + assert open_operation["result"]["address_changed"] is True + assert open_operation["result"]["device_write_performed"] is False + + +def test_control_session_prestart_failure_is_journaled_and_marks_lease_offline( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + service, _ = service_with_fake_runtime(tmp_path) + service._selected_device_id = "test-ble-transport" # noqa: SLF001 + service._connection_mode = "bridge" # noqa: SLF001 + service._k1_ip = "10.255.254.54" # noqa: SLF001 + service._device_id = "known-k1" # noqa: SLF001 + service._device_session_id = "known-session" # noqa: SLF001 + service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001 + + async def failed_status_read(*_: object, **__: object) -> dict[str, Any]: + raise TimeoutError("synthetic BLE recovery timeout") + + monkeypatch.setattr(facade_module, "read_wifi_status_once", failed_status_read) + monkeypatch.setattr(facade_module, "_control_endpoint_reachable", lambda _target: False) + monkeypatch.setattr(facade_module, "_host_route_class", lambda _target: "direct-or-routed") + + with pytest.raises( + facade_module.ConnectionLeaseUnavailable, + match="повторное чтение состояния", + ): + service.open_application_control_session( + OpenApplicationControlSessionRequest( + operator_present=True, + owner_controlled_device=True, + lixelgo_closed=True, + battery_storage_confirmed=True, + expected_physical_state_confirmed=True, + timezone_name="Europe/Moscow", + ) + ) + + state = service.state() + assert state["application_control_session"]["state"] == "idle" + assert state["device_session"]["connectivity"] == "offline" + assert state["connection_verification"]["lease_state"] == "disconnected" + open_operation = next( + operation + for operation in state["operations"] + if operation["action"] == "application-control.session.open" + ) + assert open_operation["status"] == "failed" + assert open_operation["error"] == { + "category": "connection", + "code": "connection_lease_ble_recovery_failed", + "retryable": False, + "safe_to_retry": True, + "side_effect_status": "none", + } + + +def test_bridge_route_mismatch_stops_before_ble_recovery_and_vendor_commands( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + service, _ = service_with_fake_runtime(tmp_path) + service._selected_device_id = "test-ble-transport" # noqa: SLF001 + service._connection_mode = "bridge" # noqa: SLF001 + service._k1_ip = "192.168.68.50" # noqa: SLF001 + service._device_id = "known-k1" # noqa: SLF001 + service._device_session_id = "known-session" # noqa: SLF001 + service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001 + + async def forbidden_status_read(*_: object, **__: object) -> dict[str, Any]: + raise AssertionError("host route mismatch must stop before BLE recovery") + + monkeypatch.setattr(facade_module, "_control_endpoint_reachable", lambda _target: False) + monkeypatch.setattr(facade_module, "_host_route_class", lambda _target: "tunnel") + monkeypatch.setattr(facade_module, "read_wifi_status_once", forbidden_status_read) + + with pytest.raises( + facade_module.ConnectionLeaseUnavailable, + match="другой локальной сети", + ): + service.open_application_control_session( + OpenApplicationControlSessionRequest( + operator_present=True, + owner_controlled_device=True, + lixelgo_closed=True, + battery_storage_confirmed=True, + expected_physical_state_confirmed=True, + timezone_name="Europe/Moscow", + ) + ) + + state = service.state() + assert state["connection_verification"]["reason_code"] == ( + "connection_lease_host_route_mismatch" + ) + assert state["connection_verification"]["endpoint_validation"] == "host-route" + assert state["connection_verification"]["host_route_class"] == "tunnel" + assert state["application_control_session"]["state"] == "idle" def test_prepare_creates_provisional_device_session_and_profiled_acquisition( @@ -384,18 +621,24 @@ def test_acquisition_is_unbounded_by_default_and_accepts_ten_hour_hint() -> None def test_connection_modes_require_their_exact_topology_attestation() -> None: - assert ConnectRequest( - device_id="synthetic-device", - connection_mode="quick-connect", - compatibility_attestation=QUICK_CONNECT_ATTESTATION, - ).connection_mode == "quick-connect" - assert ConnectRequest( - device_id="synthetic-device", - ssid="synthetic-network", - password=SecretStr(PRIMARY_TEST_CREDENTIAL), - connection_mode="direct-connect", - compatibility_attestation=DIRECT_CONNECT_ATTESTATION, - ).connection_mode == "direct-connect" + assert ( + ConnectRequest( + device_id="synthetic-device", + connection_mode="quick-connect", + compatibility_attestation=QUICK_CONNECT_ATTESTATION, + ).connection_mode + == "quick-connect" + ) + assert ( + ConnectRequest( + device_id="synthetic-device", + ssid="synthetic-network", + password=SecretStr(PRIMARY_TEST_CREDENTIAL), + connection_mode="direct-connect", + compatibility_attestation=DIRECT_CONNECT_ATTESTATION, + ).connection_mode + == "direct-connect" + ) with pytest.raises(ValidationError): ConnectRequest( device_id="synthetic-device", @@ -628,6 +871,91 @@ def test_plugin_commanded_acquisition_keeps_start_and_stop_as_explicit_actions( assert completed["acquisition"]["result"]["device_state"] == "ready" assert completed["last_operation"]["status"] == "succeeded" assert runtime.stop_calls == 1 + assert completed["live_perception_shadow"]["active"] is False + + +def test_next_scan_retires_stale_terminal_live_perception_ingress_before_start( + tmp_path: Path, +) -> None: + service, runtime = service_with_fake_runtime(tmp_path) + service.live_perception_ingress.begin_session("stale-completed-session") + prepared = service.prepare_acquisition( + PrepareAcquisitionRequest( + project_name=PROJECT_NAME, + host="192.168.1.20", + compatibility_attestation=ATTESTATION, + ) + ) + + state = service.start_acquisition( + StartAcquisitionRequest(acquisition_id=prepared["acquisition"]["acquisition_id"]) + ) + + assert runtime.start_calls + assert state["acquisition"]["state"] == "starting" + assert state["live_perception_shadow"]["active"] is True + assert state["live_perception_shadow"]["session_id"] != "stale-completed-session" + + +def test_confirmed_scanning_activates_and_records_right_camera( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + service, runtime = service_with_fake_runtime(tmp_path) + control = FakeInteractiveControlSession() + service._application_control_session = control # type: ignore[assignment] # noqa: SLF001 + service._k1_ip = "192.168.1.20" # noqa: SLF001 + service._compatibility_attestation = {"profile": "exact"} # noqa: SLF001 + + prepared = service.prepare_acquisition( + PrepareAcquisitionRequest( + project_name="TEST001", + host="192.168.1.20", + compatibility_attestation=ATTESTATION, + ) + ) + acquisition_id = prepared["acquisition"]["acquisition_id"] + service.start_acquisition( + StartAcquisitionRequest( + acquisition_id=acquisition_id, + physical_acceptance=PHYSICAL_ACCEPTANCE, + ) + ) + out_dir = service._acquisition_out_dir # noqa: SLF001 + assert out_dir is not None + out_dir.mkdir(parents=True) + events: list[tuple[str, object]] = [] + camera_state: dict[str, object] = { + "phase": "idle", + "active_source_id": None, + "recording": {"active": False}, + } + + def select_camera(source_id: str, target: str) -> dict[str, object]: + events.append(("select", (source_id, target))) + camera_state["phase"] = "selected" + camera_state["active_source_id"] = source_id + return dict(camera_state) + + def start_recording(session_dir: Path) -> dict[str, object]: + events.append(("record", session_dir)) + camera_state["recording"] = {"active": True} + camera_state["phase"] = "connecting" + return dict(camera_state) + + monkeypatch.setattr(service.camera_preview, "snapshot", lambda: dict(camera_state)) + monkeypatch.setattr(service.camera_preview, "select", select_camera) + monkeypatch.setattr(service.camera_preview, "start_recording", start_recording) + + service._activate_default_acquisition_camera() # noqa: SLF001 + + assert events == [ + ("select", ("sensor.camera.right", "192.168.1.20")), + ("record", out_dir), + ] + assert camera_state["active_source_id"] == "sensor.camera.right" + assert camera_state["recording"] == {"active": True} + assert runtime.source_mode == "live" def test_device_standby_retires_sources_after_terminal_local_stop_failure( @@ -1611,6 +1939,9 @@ def test_runtime_error_terminalizes_pending_graceful_stop(tmp_path: Path) -> Non assert failed["acquisition"]["state"] == "failed" assert stop_operation["status"] == "failed" assert stop_operation["error"]["side_effect_status"] == "unknown" + assert failed["source_mode"] == "idle" + assert failed["phase"] != "error" + assert runtime.stop_calls == 1 def test_receiver_completion_terminalizes_unconfirmed_graceful_stop( @@ -1970,6 +2301,11 @@ def test_network_provisioning_is_single_flight_and_secret_is_unwrapped_only_at_b } monkeypatch.setattr(facade_module, "provision_wifi_once", fake_provision) + monkeypatch.setattr( + facade_module, + "_host_route_class", + lambda _target: "direct-or-routed", + ) first = asyncio.create_task( service.connect( ConnectRequest( @@ -2005,6 +2341,117 @@ def test_network_provisioning_is_single_flight_and_secret_is_unwrapped_only_at_b assert {item["status"] for item in provision_operations} == {"succeeded", "failed"} +def test_bridge_network_change_retires_terminal_acquisition_and_receiver_error( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + service, runtime = service_with_fake_runtime(tmp_path) + service._devices = [{"device_id": "k1-a", "name": "XGR-TEST-A"}] # noqa: SLF001 + prepared = service.prepare_acquisition( + PrepareAcquisitionRequest( + project_name=PROJECT_NAME, + host="192.168.1.20", + compatibility_attestation=ATTESTATION, + ) + ) + service.abort_acquisition( + AbortAcquisitionRequest( + acquisition_id=prepared["acquisition"]["acquisition_id"], + ) + ) + runtime.phase = "error" + runtime.source_mode = "live" + stop_calls_before_connect = runtime.stop_calls + + async def fake_provision(*_: object, **__: object) -> dict[str, Any]: + return { + "started_at_utc": "2026-07-28T18:11:37Z", + "completed_at_utc": "2026-07-28T18:11:44Z", + "profile_id": "xgrids-k1-fw3-wifi-v1", + "outcome": "lan_address_observed", + "observations": [{"status": {"ipv4": "192.168.1.20"}}], + } + + monkeypatch.setattr(facade_module, "provision_wifi_once", fake_provision) + monkeypatch.setattr( + facade_module, + "_host_route_class", + lambda _target: "direct-or-routed", + ) + connected = asyncio.run( + service.connect( + ConnectRequest( + device_id="k1-a", + ssid="lab-network", + password=SecretStr(PRIMARY_TEST_CREDENTIAL), + connection_mode="bridge", + compatibility_attestation=ATTESTATION, + ) + ) + ) + + assert connected["connection_mode"] == "bridge" + assert connected["k1_ip"] == "192.168.1.20" + assert connected["acquisition"] is None + assert connected["source_mode"] == "idle" + assert connected["phase"] == "connected" + assert runtime.stop_calls == stop_calls_before_connect + 1 + + +def test_bridge_provisioning_reports_host_route_mismatch_without_hiding_device_success( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + service, _ = service_with_fake_runtime(tmp_path) + service._devices = [{"device_id": "k1-a", "name": "XGR-TEST-A"}] # noqa: SLF001 + + async def fake_provision(*_: object, **__: object) -> dict[str, Any]: + return { + "started_at_utc": "2026-07-28T19:43:03Z", + "completed_at_utc": "2026-07-28T19:43:16Z", + "profile_id": "xgrids-k1-fw3-wifi-v1", + "outcome": "lan_address_observed", + "observations": [{"status": {"ipv4": "192.168.68.50"}}], + } + + monkeypatch.setattr(facade_module, "provision_wifi_once", fake_provision) + monkeypatch.setattr(facade_module, "_host_route_class", lambda _target: "tunnel") + + connected = asyncio.run( + service.connect( + ConnectRequest( + device_id="k1-a", + ssid="lab-router", + password=SecretStr(PRIMARY_TEST_CREDENTIAL), + connection_mode="bridge", + compatibility_attestation=ATTESTATION, + ) + ) + ) + + assert connected["k1_ip"] == "192.168.68.50" + assert connected["phase"] == "device_selected" + assert "компьютер подключён к другой сети" in connected["message"] + assert connected["device_session"]["connectivity"] == "offline" + assert connected["connection_verification"] == { + "status": "host-route-mismatch", + "lease_state": "disconnected", + "lease_generation": 1, + "endpoint_validation": "host-route", + "network_reachability": "unreachable", + "reason_code": "connection_lease_host_route_mismatch", + "host_route_class": "tunnel", + "write_performed": True, + "observed_at": connected["connection_verification"]["observed_at"], + } + operation = next( + item for item in connected["operations"] if item["action"] == "network.provision" + ) + assert operation["status"] == "succeeded" + assert operation["result"]["host_route_ready"] is False + assert operation["result"]["host_route_class"] == "tunnel" + + def test_quick_connect_activates_the_device_ap_then_associates_the_host( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, @@ -2028,9 +2475,7 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host( ) @asynccontextmanager - async def fake_activation_session( - device_id: str, **_: object - ) -> AsyncIterator[dict[str, Any]]: + async def fake_activation_session(device_id: str, **_: object) -> AsyncIterator[dict[str, Any]]: nonlocal ble_session_open activation_calls.append(device_id) ble_session_open = True @@ -2092,9 +2537,7 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host( assert not ble_session_open assert len(association_calls) == 1 assert association_calls[0][0].name == "associate_wifi.swift" - assert association_calls[0][1] == facade_module.quick_connect_host_profile_id( - "XGR-TEST-A" - ) + assert association_calls[0][1] == facade_module.quick_connect_host_profile_id("XGR-TEST-A") assert association_calls[0][2] == "XGR-TEST-A" assert state["connection_mode"] == "quick-connect" assert state["k1_ip"] == "192.168.56.1" @@ -2102,9 +2545,7 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host( quick_sessions = sorted(service.evidence_root.glob("*viewer_k1_ap_association*")) assert len(quick_sessions) == 1 assert not (quick_sessions[0] / "provisioning.sensitive.json").exists() - redacted_manifest = (quick_sessions[0] / "manifest.redacted.json").read_text( - encoding="utf-8" - ) + redacted_manifest = (quick_sessions[0] / "manifest.redacted.json").read_text(encoding="utf-8") assert PRIMARY_TEST_CREDENTIAL not in redacted_manifest assert "host_wifi_profile_id" in redacted_manifest assert '"host_wifi_profile_ready_before_device_write": true' in redacted_manifest @@ -2112,9 +2553,12 @@ def test_quick_connect_activates_the_device_ap_then_associates_the_host( assert "credential_provider_id" in redacted_manifest assert "device_ap_activation_profile_id" in redacted_manifest assert (quick_sessions[0] / "ap-activation.redacted.json").exists() - assert service._camera_target_for_session( # noqa: SLF001 - state["device_session"]["device_session_id"] - ) == "192.168.56.1" + assert ( + service._camera_target_for_session( # noqa: SLF001 + state["device_session"]["device_session_id"] + ) + == "192.168.56.1" + ) prepared = service.prepare_acquisition( PrepareAcquisitionRequest( @@ -2158,6 +2602,11 @@ def test_direct_connect_reuses_the_reviewed_ble_provisioning_frame( forbidden_host_association, ) monkeypatch.setattr(facade_module, "_target_is_local_ipv4", lambda _target: False) + monkeypatch.setattr( + facade_module, + "_host_route_class", + lambda _target: "direct-or-routed", + ) state = asyncio.run( service.connect( @@ -2171,14 +2620,10 @@ def test_direct_connect_reuses_the_reviewed_ble_provisioning_frame( ) ) - assert provisioning_calls == [ - ("k1-a", "controller-hotspot", PRIMARY_TEST_CREDENTIAL) - ] + assert provisioning_calls == [("k1-a", "controller-hotspot", PRIMARY_TEST_CREDENTIAL)] assert state["connection_mode"] == "direct-connect" assert state["k1_ip"] == "172.20.10.2" - assert state["compatibility"]["attestation"]["topology"] == ( - "controller-hotspot" - ) + assert state["compatibility"]["attestation"]["topology"] == ("controller-hotspot") def test_quick_connect_missing_credential_provider_stops_before_ap_write( @@ -2232,9 +2677,7 @@ def test_quick_connect_missing_credential_provider_stops_before_ap_write( assert state["k1_ip"] == "192.168.1.20" assert state["connection_mode"] == "bridge" assert not list(service.evidence_root.glob("*viewer_k1_ap_association*")) - operation = next( - item for item in state["operations"] if item["action"] == "network.provision" - ) + operation = next(item for item in state["operations"] if item["action"] == "network.provision") assert operation["status"] == "failed" assert operation["error"]["side_effect_status"] == "none" assert operation["error"]["safe_to_retry"] is True @@ -2259,9 +2702,7 @@ def test_quick_connect_does_not_start_host_wifi_without_ap_ready_flag( ) @asynccontextmanager - async def not_ready_session( - *_: object, **__: object - ) -> AsyncIterator[dict[str, Any]]: + async def not_ready_session(*_: object, **__: object) -> AsyncIterator[dict[str, Any]]: yield { "profile_id": "xgrids-k1-fw3-quick-connect-ap-v1", "started_at_utc": "2026-07-19T15:00:00Z", @@ -2325,9 +2766,7 @@ def test_failed_connection_change_revokes_the_previous_route( ) @asynccontextmanager - async def fake_activation_session( - *_: object, **__: object - ) -> AsyncIterator[dict[str, Any]]: + async def fake_activation_session(*_: object, **__: object) -> AsyncIterator[dict[str, Any]]: yield { "profile_id": "xgrids-k1-fw3-quick-connect-ap-v1", "started_at_utc": "2026-07-19T15:00:00Z", @@ -2370,9 +2809,7 @@ def test_failed_connection_change_revokes_the_previous_route( assert state["compatibility"]["attestation"] is None quick_sessions = sorted(service.evidence_root.glob("*viewer_k1_ap_association*")) failure_evidence = json.loads( - (quick_sessions[0] / "host-wifi-association.redacted.json").read_text( - encoding="utf-8" - ) + (quick_sessions[0] / "host-wifi-association.redacted.json").read_text(encoding="utf-8") ) assert failure_evidence["reason_code"] == "network-not-found" assert failure_evidence["scan_attempt_count"] == 4 @@ -2413,9 +2850,7 @@ def test_network_provisioning_rejects_an_ipv4_owned_by_the_local_host( state = service.state() assert state["selected_device_id"] is None assert state["k1_ip"] is None - operation = next( - item for item in state["operations"] if item["action"] == "network.provision" - ) + operation = next(item for item in state["operations"] if item["action"] == "network.provision") assert operation["status"] == "failed" assert operation["error"]["safe_to_retry"] is False assert operation["error"]["side_effect_status"] == "unknown" diff --git a/tests/test_xgrids_application_mqtt.py b/tests/test_xgrids_application_mqtt.py index 7128633..71385cd 100644 --- a/tests/test_xgrids_application_mqtt.py +++ b/tests/test_xgrids_application_mqtt.py @@ -778,3 +778,36 @@ def test_post_publish_timeout_poisoned_transport_never_retries() -> None: required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"}, ) assert len(fake.publish_calls) == 1 + + +def test_network_loop_failure_keeps_exact_paho_result_and_phase() -> None: + class LoopFailureClient(FakeControlClient): + def loop(self, timeout: float) -> mqtt.MQTTErrorCode: + if self.publish_calls and not self.events: + return mqtt.MQTT_ERR_CONN_LOST + return super().loop(timeout) + + fake = LoopFailureClient() + transport = ReviewedApplicationMqttTransport( + "192.168.1.20", + client_factory=lambda: cast(mqtt.Client, fake), + ) + transport.open() + + with pytest.raises( + ApplicationCommandOutcomeUnknown, + match=r"phase=post-publish-drain, result=7: The connection was lost", + ): + transport.exchange_batch_once( + [_envelope()], + required_response_operation_keys={"bootstrap:1:DeviceInfoRequest"}, + ) + + snapshot = transport.snapshot().as_dict() + assert snapshot["state"] == "poisoned" + assert snapshot["last_loop_result_code"] == int(mqtt.MQTT_ERR_CONN_LOST) + assert snapshot["last_loop_result_name"] == mqtt.error_string( + int(mqtt.MQTT_ERR_CONN_LOST) + ) + assert snapshot["last_loop_phase"] == "post-publish-drain" + assert len(fake.publish_calls) == 1 diff --git a/tests/test_xgrids_application_session.py b/tests/test_xgrids_application_session.py index 3c9250d..b5a167c 100644 --- a/tests/test_xgrids_application_session.py +++ b/tests/test_xgrids_application_session.py @@ -170,6 +170,7 @@ def test_canonical_stages_require_operator_events_but_device_standby_does_not( monkeypatch: pytest.MonkeyPatch, ) -> None: FakeExecutor.records = [] + scanning_observed = threading.Event() monkeypatch.setattr( session_module, "PhysicalAcceptanceDialogueExecutor", @@ -181,6 +182,7 @@ def test_canonical_stages_require_operator_events_but_device_standby_does_not( loader, transport_factory=lambda _host: transport, # type: ignore[arg-type] epoch_seconds=lambda: 1_752_680_000, + scanning_observer=scanning_observed.set, ) session.open( @@ -202,6 +204,7 @@ def test_canonical_stages_require_operator_events_but_device_standby_does_not( session.request_start(project_name="TEST001", confirmation=_confirmation()) _wait_phase(session, "scanning") + assert scanning_observed.wait(timeout=1.0) assert FakeExecutor.records[-1] == "wait:stop" session.request_stop(confirmation=_confirmation()) @@ -210,6 +213,7 @@ def test_canonical_stages_require_operator_events_but_device_standby_does_not( assert completed["pending_operator_action"] is None assert loader.calls == 1 assert completed["automatic_retry"] is False + assert completed["scanning_observer_errors"] == 0 assert completed["scripted_transitions"] is False assert FakeExecutor.records == [ "connection:1-6", @@ -562,6 +566,210 @@ def test_start_outcome_unknown_blocks_reopen_even_when_transport_is_closed( assert failed["can_open"] is False assert failed["failure"]["modeling_command_attempted"] is True # type: ignore[index] assert failed["failure"]["safe_to_retry"] is False # type: ignore[index] + with pytest.raises( + session_module.ApplicationAcceptanceError, + match="cannot be retired", + ): + session.retire_for_network_change() + + +def test_acknowledged_stop_disconnect_allows_only_explicit_network_change( + monkeypatch: pytest.MonkeyPatch, +) -> None: + @dataclass + class StopDisconnectTransportSnapshot: + state: str + + def as_dict(self) -> dict[str, object]: + return { + "state": self.state, + "publish_attempts": 15, + "qos2_completions": 15, + "correlated_responses": 13, + "ignored_known_responses": 304, + "late_known_responses": 0, + "device_status_reports": 52, + "latest_device_session_state": "scan_stopping", + "latest_device_project_bound": True, + "latest_system_error_code": None, + "last_loop_result_code": 7, + "last_loop_result_name": "The connection was lost.", + "last_loop_phase": "maintain-open", + "automatic_retry": False, + "automatic_reconnect": False, + } + + class StopDisconnectTransport(FakeTransport): + def snapshot(self) -> StopDisconnectTransportSnapshot: + return StopDisconnectTransportSnapshot(self.state) + + class StopDisconnectExecutor(FakeExecutor): + def __init__(self, transport: StopDisconnectTransport) -> None: + super().__init__(transport) + self.stop_attempted = False + self.stop_complete = False + + def execute_canonical_start(self, *_args: object, **_kwargs: object) -> object: + self.records.append("start:11-14") + return object() + + def execute_canonical_stop(self, *_args: object, **_kwargs: object) -> object: + self.records.append("stop") + self.stop_attempted = True + self.stop_complete = True + return object() + + def maintain_post_stop_until_standby(self) -> None: + self.records.append("wait:device-standby") + self.transport.state = "poisoned" + raise session_module.ApplicationCommandOutcomeUnknown( + "control MQTT network loop returned an error", + reason_code="mqtt_network_loop_failed", + ) + + def snapshot(self) -> dict[str, object]: + return { + "dialogue_stage": ( + "stop-acknowledged" if self.stop_complete else "test-stage" + ), + "start_attempted": True, + "start_complete": True, + "stop_attempted": self.stop_attempted, + "stop_complete": self.stop_complete, + "response_evidence": [], + "automatic_retry": False, + } + + FakeExecutor.records = [] + monkeypatch.setattr( + session_module, + "PhysicalAcceptanceDialogueExecutor", + StopDisconnectExecutor, + ) + session = InteractiveApplicationControlSession( + FakeAuthorityLoader(), + transport_factory=lambda host: StopDisconnectTransport(host), # type: ignore[arg-type] + ) + + session.open( + host="192.168.1.20", + timezone_name="Europe/Moscow", + confirmation=_confirmation(), + ) + _wait_phase(session, "connection-ready") + session.enter_workspace() + _wait_phase(session, "workspace-ready") + session.open_project_prompt() + _wait_phase(session, "project-ready") + session.request_start(project_name="TEST001", confirmation=_confirmation()) + _wait_phase(session, "scanning") + session.request_stop(confirmation=_confirmation()) + failed = _wait_phase(session, "failed") + failed_thread = session._thread # noqa: SLF001 + assert failed_thread is not None + failed_thread.join(timeout=2.0) + assert not failed_thread.is_alive() + failed = session.snapshot() + + assert failed["can_open"] is False + assert failed["outcome_unknown"] is True + failure = failed["failure"] + assert isinstance(failure, dict) + assert failure["safe_to_retry"] is False + assert failure["network_change_admissible"] is True + assert failure["network_change_reconciliation"] == { + "device_session_state": "scan_stopping", + "device_project_bound": True, + "system_error_code": None, + "stop_complete": True, + "standby_confirmed": False, + "decision": "explicit-network-change-only-after-acknowledged-stop", + "automatic_retry": False, + } + + retired = session.retire_for_network_change() + assert retired["state"] == "idle" + assert retired["failure"] is None + assert retired["can_open"] is True + + +def test_prestart_loop_failure_allows_only_fresh_explicit_retry_after_ready_status( + monkeypatch: pytest.MonkeyPatch, +) -> None: + @dataclass + class ReconciledTransportSnapshot: + state: str + + def as_dict(self) -> dict[str, object]: + return { + "state": self.state, + "publish_attempts": 4, + "qos2_completions": 4, + "correlated_responses": 3, + "ignored_known_responses": 0, + "late_known_responses": 0, + "device_status_reports": 1, + "latest_device_session_state": "ready", + "latest_device_project_bound": False, + "latest_system_error_code": None, + "last_loop_result_code": 7, + "last_loop_result_name": "The connection was lost.", + "last_loop_phase": "post-publish-drain", + "automatic_retry": False, + "automatic_reconnect": False, + } + + class ReconciledTransport(FakeTransport): + def snapshot(self) -> ReconciledTransportSnapshot: + return ReconciledTransportSnapshot(self.state) + + class PrestartLoopFailureExecutor(FakeExecutor): + def run_connection_stage( + self, + _orchestrator: object, + ) -> LiveDeviceControlBinding: + raise session_module.ApplicationCommandOutcomeUnknown( + "control MQTT network loop returned an error", + reason_code="mqtt_network_loop_failed", + ) + + monkeypatch.setattr( + session_module, + "PhysicalAcceptanceDialogueExecutor", + PrestartLoopFailureExecutor, + ) + session = InteractiveApplicationControlSession( + FakeAuthorityLoader(), + transport_factory=lambda host: ReconciledTransport(host), # type: ignore[arg-type] + ) + + session.open( + host="192.168.1.20", + timezone_name="Europe/Moscow", + confirmation=_confirmation(), + ) + failed_thread = session._thread # noqa: SLF001 + assert failed_thread is not None + failed_thread.join(timeout=2.0) + assert not failed_thread.is_alive() + failed = session.snapshot() + + assert failed["state"] == "failed" + assert failed["outcome_unknown"] is True + assert failed["automatic_retry"] is False + assert failed["can_open"] is True + failure = failed["failure"] + assert isinstance(failure, dict) + assert failure["reason_code"] == "mqtt_network_loop_failed" + assert failure["modeling_command_attempted"] is False + assert failure["safe_to_retry"] is True + assert failure["status_reconciliation"] == { + "device_session_state": "ready", + "device_project_bound": False, + "system_error_code": None, + "decision": "safe-explicit-prestart-retry", + "automatic_retry": False, + } def test_unavailable_transport_snapshot_after_publish_blocks_reopen(