From 7d196db02bf42606fa9a26855a537f5ff4f7516c Mon Sep 17 00:00:00 2001 From: Miayui <774713819@qq.com> Date: Thu, 17 Sep 2026 11:04:05 +0800 Subject: [PATCH] feat(n7web): sync N7 scene cards and optimize runtime --- apps/web-console/app.js | 352 ++++++++++- apps/web-console/codepen-glass-theme.css | 5 +- apps/web-console/console-v2.js | 44 +- apps/web-console/energy-monitor.js | 5 +- apps/web-console/index.html | 8 +- apps/web-console/self-extension.js | 137 ++++- apps/web-console/service-worker.js | 12 +- apps/web-console/styles.css | 16 + domains/seat/src/seat_domain/validator.py | 15 +- .../application/assistant_capabilities.py | 23 +- .../application/mode_profile.py | 117 +++- .../application/mode_scene_catalog.py | 44 +- .../application/mode_scene_ingress.py | 449 ++++++++++++++ .../application/seat_service.py | 463 ++++++++++++++- .../infrastructure/can/mqtt_transport.py | 188 +++++- src/cognitive_agent/lan_scene_server.py | 23 +- src/cognitive_agent/server.py | 128 +++- tests/e2e/test_seat_web_api.py | 50 +- .../unit/test_assistant_capability_catalog.py | 20 +- .../unit/test_assistant_frontend_contract.py | 2 +- .../test_energy_monitor_frontend_contract.py | 10 +- .../test_frontend_performance_contract.py | 103 ++++ tests/unit/test_frontend_theme_contract.py | 4 +- .../unit/test_mode_card_dispatch_integrity.py | 61 ++ tests/unit/test_mode_gallery_grouping.py | 46 ++ tests/unit/test_mode_scene_ingress.py | 555 ++++++++++++++++++ tests/unit/test_mqtt_can_transport.py | 80 +++ tests/unit/test_nfc_frontend_contract.py | 4 +- tests/unit/test_start_n7web.py | 36 ++ tests/unit/test_vue_frontend_contract.py | 2 +- tools/start_n7web.py | 48 +- 31 files changed, 2924 insertions(+), 126 deletions(-) create mode 100644 src/cognitive_agent/application/mode_scene_ingress.py create mode 100644 tests/unit/test_frontend_performance_contract.py create mode 100644 tests/unit/test_mode_card_dispatch_integrity.py create mode 100644 tests/unit/test_mode_scene_ingress.py create mode 100644 tests/unit/test_start_n7web.py diff --git a/apps/web-console/app.js b/apps/web-console/app.js index a8827bf..5c67c1a 100755 --- a/apps/web-console/app.js +++ b/apps/web-console/app.js @@ -60,6 +60,8 @@ const state = { vehicleStateTurn: -1, seatTwin: null, seatActuatorState: {}, + seatRenderFingerprint: "", + domainRenderFingerprint: "", seatFeedbackSource: "simulated", seatFeedbackSources: { passenger: "simulated", driver: "unavailable" }, availableSeatAxes: null, @@ -76,6 +78,14 @@ const state = { modeRunTimer: null, modeHighlightTimer: null, latestModeRefreshKey: "", + modeProfileSync: { + initialized: false, + inFlight: false, + poller: null, + catalogRevision: "", + importedCount: 0, + rejectedCount: 0, + }, modeEditor: null, directCanCommandPending: false, }; @@ -90,6 +100,7 @@ const apiRequest = (input, options) => window.CognitiveAgentNetwork.request(inpu const HARDWARE_REQUEST_TIMEOUT_MS = 15 * 60 * 1000; const REQUEST_TIMEOUT_MS = 45 * 1000; const DEFAULT_PHYSICAL_STATE_POLL_INTERVAL_MS = 5000; +const MODE_PROFILE_SYNC_POLL_INTERVAL_MS = 1000; const AUTHORITATIVE_CH1_SOURCES = new Set(["physical_can_ch1", "wireless_mqtt_ch1"]); async function apiResponseError(response) { @@ -2025,6 +2036,7 @@ function applyTwinSceneTheme(twin) { canvas.dataset.themeRole = role; canvas.dataset.themeAccent = `#${palette.accent.toString(16).padStart(6, "0")}`; } + twin.requestRender?.({ hotspots: false }); } function updateTwinVehicleVisual(systems, timestamp) { @@ -2574,6 +2586,10 @@ function configureFixedTwinSeat(THREE, model, position, role) { return model; } +function requestSeatTwinRender(options = {}) { + state.seatTwin?.requestRender?.(options); +} + function setTwinSelection(selected) { const twin = state.seatTwin; if (!twin) return; @@ -2609,6 +2625,7 @@ function setTwinSelection(selected) { twin.view.targetX = 0; twin.view.yaw = -0.62; renderTwinReadout(); + requestSeatTwinRender(); } function twinDeviceForSeat(seat, actuatorState = state.seatTwin?.actuatorState || {}) { @@ -2734,6 +2751,7 @@ function updateSeatTwin(actuatorState) { }); }); renderTwinReadout(); + requestSeatTwinRender(); } const REFERENCE_VEHICLE_ROOT = "/assets/models/byd-tang/"; @@ -4080,7 +4098,7 @@ function initSeatTwin() { return; } renderer.autoClear = false; - renderer.setPixelRatio(Math.min(window.devicePixelRatio || 1, 2)); + renderer.setPixelRatio(Math.min(window.devicePixelRatio || 1, 1.5)); if ("outputColorSpace" in renderer) renderer.outputColorSpace = THREE.SRGBColorSpace; else renderer.outputEncoding = THREE.sRGBEncoding; renderer.shadowMap.enabled = true; @@ -4218,6 +4236,15 @@ function initSeatTwin() { view: { yaw: -0.62, pitch: 0.31, distance: 10.2, targetX: 0 }, dragging: false, pointer: { x: 0, y: 0 }, + animationFrame: null, + renderRequested: false, + settleFrames: 0, + inViewport: true, + hotspotsDirty: true, + hotspotLastLayoutAt: Number.NEGATIVE_INFINITY, + hotspotMetrics: new Map(), + hotspotElements: new Map(), + requestRender, }; applyTwinSceneTheme(state.seatTwin); window.addEventListener("cognitive-agent:module-change", () => { @@ -4284,6 +4311,7 @@ function initSeatTwin() { distance: 10.2, targetX: 0, }); + requestRender(); } let stageWasVisible = false; @@ -4294,6 +4322,8 @@ function initSeatTwin() { camera.aspect = width / height; camera.updateProjectionMatrix(); state.seatTwin.narrowViewport = width < 600; + state.seatTwin.hotspotMetrics.clear(); + requestRender({ settleFrames: 2 }); } function synchronizeTwinViewport({ reset = false } = {}) { @@ -4317,6 +4347,31 @@ function initSeatTwin() { stageObserver.observe(stage); state.seatTwin.resizeObserver = stageObserver; } + if (window.IntersectionObserver) { + const visibilityObserver = new IntersectionObserver(([entry]) => { + const twin = state.seatTwin; + if (!twin) return; + twin.inViewport = Boolean(entry?.isIntersecting); + if (!twin.inViewport && twin.animationFrame !== null) { + window.cancelAnimationFrame(twin.animationFrame); + twin.animationFrame = null; + } else if (twin.inViewport) { + requestRender({ settleFrames: 2 }); + } + }, { root: null, rootMargin: "100px" }); + visibilityObserver.observe(stage); + state.seatTwin.visibilityObserver = visibilityObserver; + } + document.addEventListener("visibilitychange", () => { + const twin = state.seatTwin; + if (!twin) return; + if (document.visibilityState !== "visible" && twin.animationFrame !== null) { + window.cancelAnimationFrame(twin.animationFrame); + twin.animationFrame = null; + return; + } + requestRender({ settleFrames: 2 }); + }); window.addEventListener("resize", scheduleViewportSync); window.addEventListener("cognitive-agent:module-change", scheduleViewportSync); canvas.addEventListener("webglcontextlost", (event) => { @@ -4342,6 +4397,7 @@ function initSeatTwin() { refreshIcons(); window.setTimeout(() => { fallback.hidden = true; }, 3200); } + requestRender({ settleFrames: 8 }); }); synchronizeTwinViewport({ reset: true }); stage.addEventListener("pointerdown", (event) => { @@ -4349,6 +4405,7 @@ function initSeatTwin() { state.seatTwin.pointer = { x: event.clientX, y: event.clientY }; stage.setPointerCapture(event.pointerId); stage.classList.add("dragging"); + requestRender({ settleFrames: 2 }); }); stage.addEventListener("pointermove", (event) => { const twin = state.seatTwin; @@ -4358,11 +4415,13 @@ function initSeatTwin() { twin.pointer = { x: event.clientX, y: event.clientY }; twin.view.yaw -= dx * 0.006; twin.view.pitch = Math.max(0.12, Math.min(0.68, twin.view.pitch + dy * 0.004)); + requestRender({ settleFrames: 2 }); }); const endDrag = () => { if (!state.seatTwin) return; state.seatTwin.dragging = false; stage.classList.remove("dragging"); + requestRender({ settleFrames: 2 }); }; stage.addEventListener("pointerup", endDrag); stage.addEventListener("pointercancel", endDrag); @@ -4370,6 +4429,7 @@ function initSeatTwin() { event.preventDefault(); const twin = state.seatTwin; twin.view.distance = Math.max(9.4, Math.min(15.2, twin.view.distance + event.deltaY * 0.006)); + requestRender({ settleFrames: 2 }); }, { passive: false }); $("#twinReset").addEventListener("click", resetView); $$(".twin-seat-button").forEach((button) => button.addEventListener("click", () => setTwinSelection(button.dataset.twinSeat))); @@ -4392,7 +4452,11 @@ function initSeatTwin() { scene.updateMatrixWorld(true); camera.updateMatrixWorld(); Object.entries(TWIN_DEVICE_HOTSPOT_LAYOUT).forEach(([deviceId, layout]) => { - const button = $(`[data-twin-device="${deviceId}"]`, hotspotLayer); + let button = twin.hotspotElements.get(deviceId); + if (!button?.isConnected) { + button = $(`[data-twin-device="${deviceId}"]`, hotspotLayer); + if (button) twin.hotspotElements.set(deviceId, button); + } if (!button || !twinDeviceWorldAnchor(THREE, twin, deviceId, hotspotWorldAnchor)) return; projectedHotspot.copy(hotspotWorldAnchor).project(camera); const anchorX = (projectedHotspot.x * 0.5 + 0.5) * width; @@ -4412,8 +4476,18 @@ function initSeatTwin() { button.dataset.anchorSource = "projected-3d"; const label = $(".v2-twin-hotspot-label", button); - const halfWidth = Math.max(44, (label?.offsetWidth || 94) / 2); - const halfHeight = Math.max(20, (label?.offsetHeight || 42) / 2); + let metrics = twin.hotspotMetrics.get(deviceId); + if (!metrics || metrics.label !== label) { + const measuredWidth = label?.offsetWidth || 94; + const measuredHeight = label?.offsetHeight || 42; + metrics = { + label, + halfWidth: Math.max(44, measuredWidth / 2), + halfHeight: Math.max(20, measuredHeight / 2), + }; + twin.hotspotMetrics.set(deviceId, metrics); + } + const { halfWidth, halfHeight } = metrics; const preferredX = anchorX + layout.labelOffset[0] * offsetScale; const preferredY = anchorY + layout.labelOffset[1] * offsetScale; const labelX = clampLabelCoordinate( @@ -4448,22 +4522,84 @@ function initSeatTwin() { window.CognitiveAgentConsoleV2?.positionTwinDevicePopover?.(); } + function queueTwinFrame() { + const twin = state.seatTwin; + if (!twin || twin.animationFrame !== null) return; + if (document.visibilityState !== "visible" || twin.inViewport === false) return; + twin.animationFrame = window.requestAnimationFrame(animate); + } + + function requestRender({ hotspots = true, settleFrames = 36 } = {}) { + const twin = state.seatTwin; + if (!twin) return; + twin.renderRequested = true; + twin.settleFrames = Math.max(twin.settleFrames || 0, Math.max(0, Number(settleFrames) || 0)); + if (hotspots) twin.hotspotsDirty = true; + queueTwinFrame(); + } + + function twinNeedsAnimation(twin) { + if (twin.dragging || twin.renderRequested || twin.hotspotsDirty || twin.settleFrames > 0) { + return true; + } + const movingSeat = Object.values(twin.models).some((model) => ( + Object.keys(model.positions).some( + (axis) => Math.abs(model.targets[axis] - model.positions[axis]) >= 0.002, + ) + || Object.keys(model.angles).some( + (axis) => Math.abs(model.angleTargets[axis] - model.angles[axis]) >= 0.002, + ) + || ["ventilation", "heating"].some((feature) => ( + Math.abs(model.thermalTargets[feature] - model.thermalLevels[feature]) >= 0.002 + || model.thermalTargets[feature] > 0.015 + || model.thermalLevels[feature] > 0.015 + )) + )); + if (movingSeat) return true; + const systems = twin.vehicleSystems; + if (!systems) return false; + const desiredAmbient = systems.targets.ambientPower ? systems.targets.ambientBrightness : 0; + const movingSystem = [ + [systems.current.ambientBrightness, desiredAmbient], + [systems.current.climateLevel, systems.targets.climateLevel], + [systems.current.roofTransmittance, systems.targets.roofTransmittance], + [systems.current.roofShadePosition, systems.targets.roofShadePosition], + [systems.current.lowBeam, systems.targets.lowBeam], + [systems.current.highBeam, systems.targets.highBeam], + [systems.current.fogLight, systems.targets.fogLight], + [systems.current.turnSignal, systems.targets.turnSignal], + ...Object.keys(systems.current.windowPositions).map((key) => [ + systems.current.windowPositions[key], + systems.targets.windowPositions[key], + ]), + ].some(([current, target]) => Math.abs(Number(target) - Number(current)) >= 0.002); + return movingSystem + || systems.targets.turnSignal > 0.015 + || systems.targets.climateLevel > 0.015 + || systems.airflow.level > 0.015; + } + function animate(timestamp = 0) { const twin = state.seatTwin; if (!twin) return; + twin.animationFrame = null; + if (document.visibilityState !== "visible" || twin.inViewport === false) return; if (!stageHasLayout() || canvas.dataset.contextLost === "true") { stageWasVisible = false; - window.requestAnimationFrame(animate); + twin.renderRequested = true; return; } if (!stageWasVisible) synchronizeTwinViewport({ reset: true }); + let geometryChanged = false; Object.values(twin.models).forEach((model) => { Object.keys(model.positions).forEach((axis) => { const delta = model.targets[axis] - model.positions[axis]; + if (Math.abs(delta) >= 0.002) geometryChanged = true; model.positions[axis] += Math.abs(delta) < 0.03 ? delta : delta * 0.085; }); Object.keys(model.angles).forEach((axis) => { const delta = model.angleTargets[axis] - model.angles[axis]; + if (Math.abs(delta) >= 0.002) geometryChanged = true; model.angles[axis] += Math.abs(delta) < 0.03 ? delta : delta * 0.085; }); model.slide.position.z = (model.positions.longitudinal - 50) * 0.012; @@ -4499,6 +4635,7 @@ function initSeatTwin() { ); updateTwinThermalVisual(model, timestamp); }); + if (geometryChanged) twin.hotspotsDirty = true; updateTwinVehicleVisual(twin.vehicleSystems, timestamp); const { view } = twin; // Fit the whole reference vehicle against the limiting horizontal field @@ -4519,7 +4656,11 @@ function initSeatTwin() { ); camera.lookAt(cameraTargetX, 1.14, -0.08); updateReferenceWindowOutlineLayers(twin.vehicleSystems, camera); - positionTwinDeviceHotspots(); + if (twin.hotspotsDirty && timestamp - twin.hotspotLastLayoutAt >= 100) { + positionTwinDeviceHotspots(); + twin.hotspotsDirty = false; + twin.hotspotLastLayoutAt = timestamp; + } canvas.dataset.framing = "whole-vehicle-centered"; renderer.clear(); camera.layers.set(0); @@ -4532,10 +4673,12 @@ function initSeatTwin() { scene.background = sceneBackground; camera.layers.set(0); if (!canvas.dataset.rendered) canvas.dataset.rendered = "true"; - window.requestAnimationFrame(animate); + twin.renderRequested = false; + twin.settleFrames = Math.max(0, twin.settleFrames - 1); + if (twinNeedsAnimation(twin)) queueTwinFrame(); } setTwinSelection("passenger"); - animate(); + requestRender({ settleFrames: 48 }); } function showToast(message) { @@ -5114,6 +5257,10 @@ function isUserModeProfile(mode) { return mode?.origin === "user" || mode?.deletable === true; } +function isN7SyncedModeProfile(mode) { + return mode?.origin === "n7_mqtt" || mode?.source_topic === "from_n7/scenes"; +} + function isDynamicModeProfile(mode) { const behavior = String(mode?.behavior || "").trim().toLowerCase(); if (["cyclic", "dynamic"].includes(behavior)) return true; @@ -5179,10 +5326,14 @@ function createModeGalleryCard(mode) { heading.className = "mode-gallery-card-heading"; const name = document.createElement("strong"); name.textContent = mode.name || mode.id; + const sourceBadge = document.createElement("span"); + sourceBadge.className = "mode-gallery-source-badge"; + sourceBadge.textContent = "N7 同步"; + sourceBadge.hidden = !isN7SyncedModeProfile(mode); const stateLabel = document.createElement("small"); stateLabel.className = "mode-gallery-state"; stateLabel.textContent = "点击执行"; - heading.append(name, stateLabel); + heading.append(name, sourceBadge, stateLabel); const description = document.createElement("span"); description.className = "mode-gallery-card-description"; description.textContent = mode.description || "整车场景联动配置"; @@ -5382,6 +5533,93 @@ function refreshModeProfileFromResult(result) { .catch(() => {}); } +function modeProfileSyncCount(value) { + const count = Number(value); + return Number.isFinite(count) && count >= 0 ? Math.trunc(count) : 0; +} + +function primeModeProfileSyncFromRuntime(runtime) { + const sync = state.modeProfileSync; + const catalogRevision = String( + runtime?.mode_scene_catalog?.catalog_revision || "", + ); + if (catalogRevision) sync.catalogRevision = catalogRevision; + if (sync.initialized) return; + const ingress = runtime?.vehicle_mqtt?.mode_scene_import; + if (!ingress || typeof ingress !== "object") return; + sync.importedCount = modeProfileSyncCount(ingress.imported_count); + sync.rejectedCount = modeProfileSyncCount(ingress.rejected_count); + sync.initialized = true; +} + +async function checkModeProfileSync() { + const sync = state.modeProfileSync; + if (document.visibilityState === "hidden" || sync.inFlight) return; + sync.inFlight = true; + try { + const response = await apiRequest("/api/mode-profiles/status", { + cache: "no-store", + }); + if (response.status === 401) { + window.location.replace("/login"); + return; + } + if (!response.ok) throw await apiResponseError(response); + const payload = await response.json(); + const ingress = payload?.mode_scene_ingress || {}; + const catalogRevision = String(payload?.catalog_revision || ""); + const importedCount = modeProfileSyncCount(ingress.imported_count); + const rejectedCount = modeProfileSyncCount(ingress.rejected_count); + if (!sync.initialized) { + sync.catalogRevision = catalogRevision; + sync.importedCount = importedCount; + sync.rejectedCount = rejectedCount; + sync.initialized = true; + return; + } + + const catalogChanged = Boolean( + catalogRevision && catalogRevision !== sync.catalogRevision, + ); + const imported = importedCount > sync.importedCount; + const rejected = rejectedCount > sync.rejectedCount; + sync.catalogRevision = catalogRevision || sync.catalogRevision; + sync.importedCount = importedCount; + sync.rejectedCount = rejectedCount; + + const sceneId = String(ingress.last_scene_id || "").trim(); + const sceneAlreadyVisible = sceneId + ? modeProfiles().some((mode) => String(mode?.id || "") === sceneId) + : false; + if (catalogChanged || (imported && !sceneAlreadyVisible)) { + await loadRuntime(); + } + if (imported) { + if (sceneId) revealModeProfile(sceneId); + const sceneName = String( + ingress.last_result?.scene_name || sceneId || "新模式", + ); + showToast(`已从 N7 同步模式卡片:${sceneName}`); + } else if (rejected) { + const reason = String(ingress.last_error || "消息格式不符合要求"); + showToast(`N7 模式卡片导入失败:${reason}`); + } + } catch (error) { + console.warn("N7 mode profile synchronization check failed", error); + } finally { + sync.inFlight = false; + } +} + +function startModeProfileSyncPolling() { + const sync = state.modeProfileSync; + window.clearInterval(sync.poller); + sync.poller = window.setInterval( + () => void checkModeProfileSync(), + MODE_PROFILE_SYNC_POLL_INTERVAL_MS, + ); +} + function openModeGallery() { const backdrop = $("#modeGalleryBackdrop"); if (!backdrop) return; @@ -6639,8 +6877,16 @@ function setModeChipsDisabled(disabled) { }); } +function hasExecutableModeCommandFlow(mode) { + const flow = mode?.command_flow; + if (!mode?.id || !flow || typeof flow !== "object") return false; + const plans = flow.plans; + return Boolean(flow.source_request_id) + || (plans && typeof plans === "object" && Object.keys(plans).length > 0); +} + async function executeSavedModeProfile(mode, { command = "" } = {}) { - if (!mode?.id || !mode?.command_flow?.source_request_id) return false; + if (!hasExecutableModeCommandFlow(mode)) return false; const modeCommand = String(command || `进入${mode.name || mode.id}`).trim(); const assistant = window.CognitiveAgentAssistant; const driverOnlyCh1 = isDriverOnlyCh1Runtime(); @@ -6747,7 +6993,7 @@ function applyModeCommand(command, modeProfileId = null) { setModeRailSelection(mode); closeModeGallery({ restoreFocus: false }); assistant?.open?.(); - if (mode?.command_flow?.source_request_id) { + if (hasExecutableModeCommandFlow(mode)) { void executeSavedModeProfile(mode, { command: modeCommand }); return true; } @@ -6809,6 +7055,7 @@ async function loadRuntime() { const response = await apiRequest("/api/runtime", { cache: "no-store" }); if (!response.ok) throw await apiResponseError(response); state.runtime = await response.json(); + primeModeProfileSyncFromRuntime(state.runtime); renderControlTransport(); applyRuntimeTargetPolicy(state.runtime); populateVehicleSelector(); @@ -7157,6 +7404,8 @@ function resetSeatState() { function resetVehicleStatusDisplay() { state.seatActuatorState = {}; + state.seatRenderFingerprint = ""; + state.domainRenderFingerprint = ""; resetSeatState(); const setText = (selector, value) => { const element = $(selector); @@ -7227,6 +7476,28 @@ function resetVehicleStatusDisplay() { summarySystem.textContent = "等待状态"; summarySystem.dataset.state = "waiting"; } + window.dispatchEvent(new CustomEvent("cognitive-agent:vehicle-state-rendered", { + detail: { scope: "reset", fingerprint: "" }, + })); +} + +const VEHICLE_RENDER_VOLATILE_KEYS = new Set([ + "age_ms", + "capture_sequence", + "captured_at", + "frame_timestamp", + "last_seen_at", + "observed_at", + "received_at", + "request_id", + "timestamp", + "updated_at", +]); + +function vehicleRenderFingerprint(value) { + return JSON.stringify(value, (key, item) => ( + VEHICLE_RENDER_VOLATILE_KEYS.has(key) ? undefined : item + )); } function mergeSeatActuatorState(previousState, patchState) { @@ -7246,20 +7517,6 @@ function mergeSeatActuatorState(previousState, patchState) { mergedSeat.position = mergedPosition; return; } - if (feature === "dynamic_mode") { - const item = $(`.seat-card[data-seat="${seat}"] [data-dynamic-mode]`); - if (!item) return; - const running = value.state === "running"; - item.classList.toggle("running", running); - const runtimeMinutes = Number(value.max_runtime_ms || 600000) / 60000; - const runtimeLabel = Number.isInteger(runtimeMinutes) - ? String(runtimeMinutes) - : runtimeMinutes.toFixed(1); - $("b", item).textContent = running - ? `${value.name} · 循环中 · 最长${runtimeLabel}分钟` - : "未运行"; - return; - } mergedSeat[feature] = value && typeof value === "object" && !Array.isArray(value) ? { ...(previousSeat[feature] || {}), ...value } : value; @@ -7270,13 +7527,15 @@ function mergeSeatActuatorState(previousState, patchState) { } function renderSeatState(actuatorState) { - if (!actuatorState) return; + if (!actuatorState) return false; const mergedActuatorState = mergeSeatActuatorState( state.seatActuatorState, actuatorState, ); + const renderFingerprint = vehicleRenderFingerprint(mergedActuatorState); state.seatActuatorState = mergedActuatorState; - resetSeatState(); + if (renderFingerprint === state.seatRenderFingerprint) return false; + state.seatRenderFingerprint = renderFingerprint; Object.entries(mergedActuatorState).forEach(([seat, functions]) => { const displayFunctions = seat === "driver" ? twinDeviceForSeat("driver", { driver: functions }) @@ -7302,6 +7561,20 @@ function renderSeatState(actuatorState) { : "CH1 状态已同步"; return; } + if (feature === "dynamic_mode") { + const item = $(`.seat-card[data-seat="${seat}"] [data-dynamic-mode]`); + if (!item) return; + const running = value.state === "running"; + item.classList.toggle("running", running); + const runtimeMinutes = Number(value.max_runtime_ms || 600000) / 60000; + const runtimeLabel = Number.isInteger(runtimeMinutes) + ? String(runtimeMinutes) + : runtimeMinutes.toFixed(1); + $("b", item).textContent = running + ? `${value.name} · 循环中 · 最长${runtimeLabel}分钟` + : "未运行"; + return; + } if (feature === "position") { Object.entries(value).forEach(([axis, motion]) => { const item = $(`.seat-card[data-seat="${seat}"] [data-axis="${axis}"]`); @@ -7335,10 +7608,17 @@ function renderSeatState(actuatorState) { }); }); updateSeatTwin(mergedActuatorState); + window.dispatchEvent(new CustomEvent("cognitive-agent:vehicle-state-rendered", { + detail: { scope: "seat", fingerprint: renderFingerprint }, + })); + return true; } function renderDomainState(domainState) { - if (!domainState) return; + if (!domainState) return false; + const renderFingerprint = vehicleRenderFingerprint(domainState); + if (renderFingerprint === state.domainRenderFingerprint) return false; + state.domainRenderFingerprint = renderFingerprint; const outletModeLabel = (value) => { const normalized = String(value || "").toLowerCase().replaceAll("-", "_"); const matrixLabels = { @@ -7723,6 +8003,11 @@ function renderDomainState(domainState) { }); } } + requestSeatTwinRender(); + window.dispatchEvent(new CustomEvent("cognitive-agent:vehicle-state-rendered", { + detail: { scope: "domain", fingerprint: renderFingerprint }, + })); + return true; } function renderedConfig(result) { @@ -9041,6 +9326,15 @@ window.addEventListener("cognitive-agent:request-complete", (event) => { setModeExecutionState(modeProfileId, succeeded ? "succeeded" : "failed"); }); +document.addEventListener("visibilitychange", () => { + if (document.visibilityState === "visible") void checkModeProfileSync(); +}); + +window.addEventListener("pagehide", () => { + window.clearInterval(state.modeProfileSync.poller); + state.modeProfileSync.poller = null; +}); + document.addEventListener("DOMContentLoaded", async () => { initGlassThemeToggle(); initJointTwinLayout(); @@ -9147,6 +9441,8 @@ document.addEventListener("DOMContentLoaded", async () => { $$('[data-download]').forEach((button) => button.addEventListener("click", () => downloadArtifact(button.dataset.download))); await loadAuthStatus(); await loadRuntime(); + startModeProfileSyncPolling(); + void checkModeProfileSync(); if (!isDriverOnlyCh1Runtime()) { await loadHardwareAccessStatus({ silent: true }); state.hardwareAccessPoller = window.setInterval( diff --git a/apps/web-console/codepen-glass-theme.css b/apps/web-console/codepen-glass-theme.css index e50279d..8e20291 100644 --- a/apps/web-console/codepen-glass-theme.css +++ b/apps/web-console/codepen-glass-theme.css @@ -466,6 +466,9 @@ body.console-v2 .v2-twin-device-popover { background: oklch(0.16 0.045 285 / 0.74) !important; border-color: var(--glass-line-strong) !important; box-shadow: 0 10px 28px oklch(0.04 0.035 286 / 0.32), inset 0 1px 0 var(--glass-highlight) !important; +} + +body.console-v2 .v2-twin-device-popover { -webkit-backdrop-filter: blur(14px) saturate(1.18); backdrop-filter: blur(14px) saturate(1.18); } @@ -526,8 +529,6 @@ body.console-v2 .v2-pet-nfc-field output { background: oklch(0.11 0.035 286 / 0.42) !important; border-color: var(--glass-line-strong) !important; box-shadow: inset 0 1px 0 oklch(1 0 0 / 0.08) !important; - -webkit-backdrop-filter: blur(10px) saturate(1.14); - backdrop-filter: blur(10px) saturate(1.14); } body.console-v2 :is(input, textarea, select):focus { diff --git a/apps/web-console/console-v2.js b/apps/web-console/console-v2.js index 38a5838..73be120 100644 --- a/apps/web-console/console-v2.js +++ b/apps/web-console/console-v2.js @@ -1265,6 +1265,7 @@ pollTimer: null, deviceFrame: null, deviceFingerprint: "", + deviceRenderDirty: true, activeDevice: null, pinnedDevice: null, previewClearTimer: null, @@ -1523,9 +1524,13 @@ `; } - function renderSyncedDevices() { + function renderSyncedDevices({ force = false } = {}) { const grid = $("#v2DeviceSyncGrid"); - if (!grid) return; + if (!grid) return false; + if (!force && !panelVisible("devices")) { + displaySync.deviceRenderDirty = true; + return false; + } grid.innerHTML = syncedDevices.map((device) => { const rows = deviceReadoutRows(device); const waiting = rows.some(([, value]) => value === "-"); @@ -1553,7 +1558,9 @@ $(`.v2-synced-device[data-device="${displaySync.activeDevice}"]`)?.classList.add("active"); } displaySync.deviceFingerprint = JSON.stringify(syncedDevices.map((device) => [device.id, deviceReadoutRows(device)])); + displaySync.deviceRenderDirty = false; window.lucide?.createIcons({ attrs: { "aria-hidden": "true" } }); + return true; } function storeDeviceControlDraft(input) { @@ -1641,11 +1648,17 @@ } function scheduleSyncedDeviceRender() { + if (!panelVisible("devices")) { + displaySync.deviceRenderDirty = true; + return; + } if (displaySync.deviceFrame) return; displaySync.deviceFrame = requestAnimationFrame(() => { displaySync.deviceFrame = null; const fingerprint = JSON.stringify(syncedDevices.map((device) => [device.id, deviceReadoutRows(device)])); - if (fingerprint === displaySync.deviceFingerprint && $("#v2DeviceSyncGrid")?.childElementCount) return; + if (!displaySync.deviceRenderDirty + && fingerprint === displaySync.deviceFingerprint + && $("#v2DeviceSyncGrid")?.childElementCount) return; renderSyncedDevices(); if (displaySync.activeDevice) presentTwinDevice(displaySync.activeDevice, { pinned: Boolean(displaySync.pinnedDevice), @@ -1748,7 +1761,11 @@ rightRail.toggleAttribute("inert", collapsed); rightRail.dataset.openPanels = openPanels.join(" ") || "none"; } + if (key === "devices" && visible) renderSyncedDevices({ force: true }); window.lucide?.createIcons({ attrs: { "aria-hidden": "true" } }); + window.dispatchEvent(new CustomEvent("cognitive-agent:side-panel-visibility", { + detail: { panel: key, visible: Boolean(visible) }, + })); window.dispatchEvent(new Event("resize")); } @@ -2550,19 +2567,10 @@ } buildTwinDeviceInteractions(); cabinBand.classList.add("v2-source-state-only"); - new MutationObserver(scheduleSyncedDeviceRender).observe(cabinBand, { - attributes: true, - childList: true, - characterData: true, - subtree: true, - }); - const stateBand = $(".state-band"); - if (stateBand) new MutationObserver(scheduleSyncedDeviceRender).observe(stateBand, { - attributes: true, - childList: true, - characterData: true, - subtree: true, - }); + window.addEventListener( + "cognitive-agent:vehicle-state-rendered", + scheduleSyncedDeviceRender, + ); } setSidePanelVisibility("devices", false); setSidePanelVisibility("extensions", false); @@ -2983,8 +2991,8 @@ }); window.addEventListener("cognitive-agent:vehicle-feedback-state", () => { syncRuntimePresentation(); - renderSyncedDevices(); - if (displaySync.activeDevice) { + scheduleSyncedDeviceRender(); + if (panelVisible("devices") && displaySync.activeDevice) { presentTwinDevice(displaySync.activeDevice, { pinned: Boolean(displaySync.pinnedDevice), selectSeat: false, diff --git a/apps/web-console/energy-monitor.js b/apps/web-console/energy-monitor.js index b90ec29..17f5450 100644 --- a/apps/web-console/energy-monitor.js +++ b/apps/web-console/energy-monitor.js @@ -869,7 +869,10 @@ createLayer(); seedHistory(); renderMonitor(); - monitor.timer = window.setInterval(() => sampleNow(), SAMPLE_INTERVAL_MS); + monitor.timer = window.setInterval( + () => sampleNow({ render: monitor.open }), + SAMPLE_INTERVAL_MS, + ); document.addEventListener("keydown", captureModalKeys, true); window.addEventListener("cognitive-agent:power-telemetry", (event) => ingestPowerTelemetry(event.detail || {}, { sample: false })); const existingUpdater = window.updateDevicePowerTelemetry; diff --git a/apps/web-console/index.html b/apps/web-console/index.html index e761b18..1d9d711 100755 --- a/apps/web-console/index.html +++ b/apps/web-console/index.html @@ -38,14 +38,14 @@ - + - - + + - +
diff --git a/apps/web-console/self-extension.js b/apps/web-console/self-extension.js index 18c28d1..658c7f5 100644 --- a/apps/web-console/self-extension.js +++ b/apps/web-console/self-extension.js @@ -389,6 +389,11 @@ let selectedPetCardId = null; const petCardDrafts = new Map(); let nfcPollTimer = null; + let nfcDialogFingerprint = ""; + let nfcDialogDirty = true; + let unifiedExtensionFingerprint = ""; + let unifiedExtensionDirty = true; + let gscuHealthFingerprint = ""; let legacySameOriginChecked = false; let legacyMigrationCompleted = false; const legacyBridgeAttempts = new Set(); @@ -3673,12 +3678,50 @@ return reader?.error || "请连接 NFC 读卡器并确认 CH340 / CH341 驱动可用"; } - function renderNfcDialog() { + function currentNfcDialogFingerprint() { + return JSON.stringify({ + revision: nfcSnapshot.revision, + reader: nfcSnapshot.reader, + card: nfcSnapshot.card, + petRevision: petCardSnapshot.revision, + pets: petCardSnapshot.cards, + publisher: petCardSnapshot.publisher, + devices: state.devices.map((device) => ({ + id: device.id, + name: device.name, + type: device.type, + cardUid: device.cardUid || null, + })), + capabilities: state.capabilities.map((capability) => ({ + id: capability.id, + deviceId: capability.deviceId, + name: capability.name, + aliases: capability.aliases, + slot: capability.slot, + status: capability.status, + verified: capability.verified, + enabled: capability.enabled !== false, + })), + infraredCommands: state.infraredCommands, + }); + } + + function renderNfcDialog({ force = false } = {}) { + const dialog = $("#extensionNfcDialog"); + if (!dialog) return false; + if (!force && !dialog.open) { + nfcDialogDirty = true; + return false; + } + const renderFingerprint = currentNfcDialogFingerprint(); + if (!force && !nfcDialogDirty && renderFingerprint === nfcDialogFingerprint) { + return false; + } const stage = $("#extensionNfcStage"); const body = $("#extensionNfcBody"); const status = $("#extensionNfcReaderStatus"); const badge = $("#extensionNfcReaderBadge"); - if (!stage || !body || !status || !badge) return; + if (!stage || !body || !status || !badge) return false; const reader = nfcSnapshot.reader || {}; const card = nfcSnapshot.card; @@ -3725,7 +3768,10 @@ ${capabilities.length ? `
${capabilities.map((capability) => ``).join("")}
` : `
该设备 ID 暂无有效命令槽位可能已删除或改绑到其他设备 ID,请重新检查红外设置。
`}
复选框状态会持久化为自然语言执行白名单。
`; } + nfcDialogFingerprint = renderFingerprint; + nfcDialogDirty = false; window.lucide?.createIcons({ attrs: { "aria-hidden": "true" } }); + return true; } function buildUnifiedNfcDialog() { @@ -3891,8 +3937,8 @@ } function openNfcDialog() { - renderNfcDialog(); openDialog("extensionNfcDialog"); + renderNfcDialog({ force: true }); void refreshNfcStatus(); } @@ -3969,10 +4015,73 @@ if (!mount) return false; let snapshot = null; - const render = () => { + const extensionViewFingerprint = (health) => JSON.stringify({ + transport: health.transport, + connected: health.connected, + peerConnected: health.peer_connected, + peer: health.peer, + listening: health.listening, + message: health.message, + extensionCatalog: (snapshot?.extension?.catalog || []).map((item) => ({ + id: item.id, + deviceId: item.device_id, + deviceName: item.device_name, + deviceType: item.device_type, + typeLabel: item.type_label, + action: item.action, + name: item.name, + cardUid: item.card_uid, + status: item.status, + enabled: item.enabled, + })), + lastAction: snapshot?.extension?.last_action ? { + command: snapshot.extension.last_action.command, + type: snapshot.extension.last_action.type, + } : null, + lastResult: snapshot?.extension?.result ? { + status: snapshot.extension.result.status, + message: snapshot.extension.result.message, + } : null, + nfc: { + revision: nfcSnapshot.revision, + reader: nfcSnapshot.reader, + card: nfcSnapshot.card, + }, + pets: petCardSnapshot, + localDevices: state.devices.map((device) => ({ + id: device.id, + name: device.name, + type: device.type, + cardUid: device.cardUid || null, + })), + localCapabilities: state.capabilities.map((capability) => ({ + id: capability.id, + deviceId: capability.deviceId, + name: capability.name, + slot: capability.slot, + status: capability.status, + verified: capability.verified, + enabled: capability.enabled !== false, + })), + infraredCommands: state.infraredCommands, + }); + + const render = ({ force = false } = {}) => { const health = snapshot?.extension_health || {}; healthState = health; - if ($("#extensionGscuTcpDialog")?.open) renderGscuTcpDialog(health, false); + const nextGscuHealthFingerprint = JSON.stringify({ + transport: health.transport, + connected: health.connected, + peerConnected: health.peer_connected, + peer: health.peer, + listening: health.listening, + message: health.message, + }); + if ($("#extensionGscuTcpDialog")?.open + && (force || nextGscuHealthFingerprint !== gscuHealthFingerprint)) { + renderGscuTcpDialog(health, false); + } + gscuHealthFingerprint = nextGscuHealthFingerprint; const embeddedNfc = health.nfc; if (nfcSnapshot.revision < 0 && embeddedNfc) { nfcSnapshot = { @@ -3989,6 +4098,16 @@ } } if (health.pet_cards) applyPetCardSnapshot(health.pet_cards, { render: false }); + const panel = mount.closest("#v2ExtensionPanel"); + if (!force && (panel?.hidden || panel?.classList.contains("is-collapsed"))) { + unifiedExtensionDirty = true; + return false; + } + const renderFingerprint = extensionViewFingerprint(health); + if (!force + && !unifiedExtensionDirty + && renderFingerprint === unifiedExtensionFingerprint + && mount.childElementCount) return false; const physicalConnected = health.transport !== "simulation" && Boolean(health.connected || health.peer_connected); const listening = Boolean(health.listening); @@ -4114,7 +4233,10 @@
最近联动${escapeHtml(visibleLastAction?.command || visibleLastAction?.type || "等待 NFC 外设感应")}
${escapeHtml(visibleLastAction ? (lastResult?.status || lastResult?.message || "STANDBY") : "STANDBY")} `; + unifiedExtensionFingerprint = renderFingerprint; + unifiedExtensionDirty = false; window.lucide?.createIcons({ attrs: { "aria-hidden": "true" } }); + return true; }; renderUnifiedExtensionView = render; @@ -4125,6 +4247,11 @@ } render(); }); + window.addEventListener("cognitive-agent:side-panel-visibility", (event) => { + if (event.detail?.panel === "extensions" && event.detail?.visible) { + render({ force: true }); + } + }); buildDialogs(); buildUnifiedNfcDialog(); bindUnifiedNfcEvents(mount); diff --git a/apps/web-console/service-worker.js b/apps/web-console/service-worker.js index ceb00e9..e864fb1 100644 --- a/apps/web-console/service-worker.js +++ b/apps/web-console/service-worker.js @@ -1,4 +1,4 @@ -const CACHE_VERSION = "cognitive-agent-pad-v317-nfc-auth-routing"; +const CACHE_VERSION = "cognitive-agent-pad-v321-n7-dispatch"; const SHELL_CACHE = `${CACHE_VERSION}-shell`; const RUNTIME_CACHE = `${CACHE_VERSION}-runtime`; const APP_SHELL = [ @@ -6,22 +6,22 @@ const APP_SHELL = [ "/index.html", "/styles.css?v=dual-seat-23", "/network.js?v=android-pad-2", - "/app.js?v=dual-seat-141", + "/app.js?v=dual-seat-145-n7-dispatch", "/gesture-control.js?v=android-pad-9", "/pwa.js?v=android-pad-3", "/console-v2.css?v=delivery-state-69", - "/console-v2.js?v=local-first-76", + "/console-v2.js?v=local-first-77-performance", "/self-extension.css?v=pet-settings-52", - "/self-extension.js?v=pet-settings-52", + "/self-extension.js?v=pet-settings-53-performance", "/speech-input.css?v=sensevoice-7", "/speech-input.js?v=sensevoice-13", "/speech-audio-worklet.js?v=sensevoice-5", "/assistant-chat.css?v=codex-assistant-29", "/assistant-chat.js?v=codex-assistant-34", "/energy-monitor.css?v=energy-monitor-4", - "/energy-monitor.js?v=energy-monitor-4", + "/energy-monitor.js?v=energy-monitor-5-performance", "/ui-consistency.css?v=spatial-ui-8", - "/codepen-glass-theme.css?v=creative-cloud-5", + "/codepen-glass-theme.css?v=creative-cloud-6-performance", "/vue-ui.js?v=vue-presentation-3", "/assets/pets/self-extension-cat.png?v=3", "/assets/pets/self-extension-cat-entrance.png?v=3", diff --git a/apps/web-console/styles.css b/apps/web-console/styles.css index ddfd2da..e46fe65 100755 --- a/apps/web-console/styles.css +++ b/apps/web-console/styles.css @@ -724,6 +724,22 @@ body.mode-editor-open { font-weight: 700; } +.mode-gallery-source-badge { + flex: 0 0 auto; + padding: 2px 5px; + color: #315eb8; + background: rgba(91, 132, 222, 0.14); + border: 1px solid rgba(91, 132, 222, 0.24); + border-radius: 999px; + font-size: 7px; + font-weight: 800; + line-height: 1; +} + +.mode-gallery-source-badge[hidden] { + display: none; +} + .mode-gallery-card-description, .mode-gallery-card-meta { overflow: hidden; diff --git a/domains/seat/src/seat_domain/validator.py b/domains/seat/src/seat_domain/validator.py index 65cdd55..5276dfc 100755 --- a/domains/seat/src/seat_domain/validator.py +++ b/domains/seat/src/seat_domain/validator.py @@ -40,7 +40,20 @@ def validate_semantics( for index, deferred in enumerate(parsed.get("deferred_requests", [])): domain = str(deferred.get("domain")) capability = str(deferred.get("capability")) - if domain not in detected_domains or not capability.startswith(f"{domain}."): + # An N7-synchronised scene may use a vehicle-owned capability namespace + # that intentionally differs from its routing domain (for example + # ``domain=local_vehicle`` with ``capability=local_scene.seat_preset``). + # It is not executed by this validator's local backend: the original + # action is forwarded unchanged to ``transto_n7/ctl``. Keep the strict + # namespace check for every ordinary deferred request, while allowing + # this explicitly marked transfer backend to retain the source contract. + n7_transfer = ( + deferred.get("execution_backend") == "n7_vehicle_transfer" + and deferred.get("rejection_code") == "PC_CAPABILITY_UNAVAILABLE" + ) + if domain not in detected_domains or ( + not n7_transfer and not capability.startswith(f"{domain}.") + ): issues.append( { "severity": "error", diff --git a/src/cognitive_agent/application/assistant_capabilities.py b/src/cognitive_agent/application/assistant_capabilities.py index 5b48cf4..f084f92 100644 --- a/src/cognitive_agent/application/assistant_capabilities.py +++ b/src/cognitive_agent/application/assistant_capabilities.py @@ -2225,8 +2225,24 @@ class AssistantCapabilityCatalog: operations = _safe_mode_operations( [item for item in requests if isinstance(item, Mapping)] ) - callable_mode = bool(operation_ids) and all( - capability_id in callable_capabilities for capability_id in operation_ids + n7_transfer_operation_ids = [ + capability_id + for capability_id in operation_ids + if capability_id not in callable_capabilities + ] + command_flow = raw.get("command_flow") + command_plans = ( + command_flow.get("plans") + if isinstance(command_flow, Mapping) + else None + ) + n7_transfer_ready = ( + raw.get("source_topic") == "from_n7/scenes" + and isinstance(command_plans, Mapping) + and bool(command_plans) + ) + callable_mode = bool(operation_ids) and ( + not n7_transfer_operation_ids or n7_transfer_ready ) modes.append( { @@ -2240,6 +2256,9 @@ class AssistantCapabilityCatalog: "operation_ids": operation_ids, "operations": operations, "assistant_callable": callable_mode, + "n7_transfer_operation_ids": ( + n7_transfer_operation_ids if n7_transfer_ready else [] + ), "all_operations_have_native_binding": bool(operation_ids) and all( capability_id in native_capabilities for capability_id in operation_ids diff --git a/src/cognitive_agent/application/mode_profile.py b/src/cognitive_agent/application/mode_profile.py index 2f0648b..4dcc597 100644 --- a/src/cognitive_agent/application/mode_profile.py +++ b/src/cognitive_agent/application/mode_profile.py @@ -14,8 +14,31 @@ from typing import Any import yaml _MATCH_FORMATTING = re.compile(r"[\s()()]+") +_MODE_INVOCATION_FORMATTING = re.compile(r"[\s,。!?,.!?::;;()()]+") _NON_ASCII_SLUG = re.compile(r"[^a-z0-9]+") _MODE_ID = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") +_EXACT_MODE_INVOCATION_PREFIXES = ( + "", + "进入", + "执行", + "执行一下", + "开始", + "开始执行", + "启动", + "打开", + "切换到", + "切换进入", + "请进入", + "请执行", + "请执行一下", + "请启动", + "请打开", + "帮我进入", + "帮我执行", + "请帮我进入", + "请帮我执行", +) +_EXACT_MODE_INVOCATION_SUFFIXES = ("", "吧", "一下", "试试") _MODE_REFERENCE_VERBS = ( "切换进入", "切换到", @@ -79,6 +102,34 @@ def _normalize_match_text(value: str) -> str: return _MATCH_FORMATTING.sub("", value.casefold()) +def _exact_mode_invocation_matches( + utterance: str, + candidates: list[object], +) -> bool: + """Accept a synchronized card only as the complete control intent. + + N7 can legitimately name a card ``影院模式`` while the same words are a + parameter of another capability (for example an ambient-light mode). A + synchronized card therefore uses an exact, bounded invocation grammar + rather than ordinary substring matching. + """ + + compact = _MODE_INVOCATION_FORMATTING.sub("", str(utterance)).casefold() + if not compact: + return False + invocations = { + _MODE_INVOCATION_FORMATTING.sub( + "", + f"{prefix}{candidate}{suffix}", + ).casefold() + for candidate in candidates + if str(candidate).strip() + for prefix in _EXACT_MODE_INVOCATION_PREFIXES + for suffix in _EXACT_MODE_INVOCATION_SUFFIXES + } + return compact in invocations + + def _mode_candidate_matches(utterance: str, candidate: object) -> bool: """Match a mode reference without treating device nouns as mode aliases. @@ -156,6 +207,20 @@ def _clean_command_flow(value: Any) -> dict[str, Any] | None: return json.loads(encoded) +def _clean_sync_origin(value: Any) -> str: + cleaned = str(value or "").strip() + if cleaned not in {"", "from_n7/scenes"}: + raise ValueError("mode sync_origin is not supported") + return cleaned + + +def _clean_upstream_origin(value: Any) -> str: + cleaned = str(value or "").strip() + if len(cleaned) > 64: + raise ValueError("mode upstream_origin must be at most 64 characters") + return cleaned + + def _mode_aliases(name: str, aliases: list[str] | None = None) -> list[str]: values = [name, *(aliases or [])] if name.endswith("模式") and len(name) > 2: @@ -451,6 +516,8 @@ class ModeProfile: user_defined: bool = False instruction: str = "" command_flow: dict[str, Any] | None = None + sync_origin: str = "" + upstream_origin: str = "" @classmethod def from_yaml(cls, path: Path, profile_root: Path | None = None) -> "ModeProfile": @@ -487,6 +554,10 @@ class ModeProfile: user_defined=_is_user_profile_path(path, profile_root), instruction=_clean_instruction(document.get("instruction", "")), command_flow=_clean_command_flow(document.get("command_flow")), + sync_origin=_clean_sync_origin(document.get("sync_origin")), + upstream_origin=_clean_upstream_origin( + document.get("upstream_origin") + ), ) def public_dict(self) -> dict[str, Any]: @@ -508,11 +579,20 @@ class ModeProfile: "vehicle_ids": list(self.vehicle_ids), }, "behavior": "cyclic" if is_cyclic else "static", - "origin": "user" if self.user_defined else "preset", + "origin": ( + "n7_mqtt" + if self.sync_origin == "from_n7/scenes" + else "user" + if self.user_defined + else "preset" + ), "deletable": self.user_defined, "instruction": self.instruction, "command_flow": self.command_flow, } + if self.sync_origin: + result["source_topic"] = self.sync_origin + result["upstream_origin"] = self.upstream_origin or None if self.timeline is not None: result["timeline"] = self.timeline.public_dict() return result @@ -588,7 +668,11 @@ class ModeProfileRegistry: public = [] for profile in profiles: item = profile.public_dict() - if profile.user_defined and profile.id in preset_ids: + if ( + profile.user_defined + and profile.id in preset_ids + and not profile.sync_origin + ): item["origin"] = "customized" item["preset_override"] = True public.append(item) @@ -628,6 +712,19 @@ class ModeProfileRegistry: matched_ids: set[str] = set() for profile in self._visible_candidates_for_vehicle(vehicle_id): candidates = [profile.id, profile.name, *profile.aliases] + effective_profile = selected[profile.id] + if ( + effective_profile.sync_origin == "from_n7/scenes" + and not _exact_mode_invocation_matches( + utterance, + [ + effective_profile.id, + effective_profile.name, + *effective_profile.aliases, + ], + ) + ): + continue if any( candidate and _mode_candidate_matches(haystack, candidate) for candidate in candidates @@ -635,7 +732,7 @@ class ModeProfileRegistry: if profile.id in matched_ids: continue matched_ids.add(profile.id) - matches.append(selected[profile.id]) + matches.append(effective_profile) return matches def resolve_exact_name( @@ -684,6 +781,8 @@ class ModeProfileRegistry: "timeline": profile.timeline.to_yaml() if profile.timeline is not None else None, "instruction": profile.instruction, "command_flow": profile.command_flow, + "sync_origin": profile.sync_origin, + "upstream_origin": profile.upstream_origin, } canonical = json.dumps( document, @@ -714,6 +813,8 @@ class ModeProfileRegistry: instruction: str = "", command_flow: dict[str, Any] | None = None, timeline: ModeTimeline | None = None, + sync_origin: str = "", + upstream_origin: str = "", expected_revision: str | None = None, create_only: bool = False, ) -> ModeProfile: @@ -729,6 +830,8 @@ class ModeProfileRegistry: profile_id = _clean_mode_id(mode_id or _slugify_mode_id(clean_name)) clean_instruction = _clean_instruction(instruction) clean_command_flow = _clean_command_flow(command_flow) + clean_sync_origin = _clean_sync_origin(sync_origin) + clean_upstream_origin = _clean_upstream_origin(upstream_origin) scope = ( {"vehicle_ids": [vehicle_id]} if vehicle_id @@ -751,6 +854,9 @@ class ModeProfileRegistry: document["command_flow"] = clean_command_flow if timeline is not None: document["timeline"] = timeline.to_yaml() + if clean_sync_origin: + document["sync_origin"] = clean_sync_origin + document["upstream_origin"] = clean_upstream_origin path = ( self.user_root / "vehicles" / vehicle_id / "user" / f"{profile_id}.yaml" if vehicle_id @@ -891,6 +997,8 @@ class ModeProfileRegistry: instruction=profile.instruction, command_flow=profile.command_flow, timeline=profile.timeline, + sync_origin=profile.sync_origin, + upstream_origin=profile.upstream_origin, ) backup_path = _initial_backup_path(Path(updated.source)) if not backup_path.exists(): @@ -1024,6 +1132,9 @@ def _mode_profile_document( document["instruction"] = profile.instruction if profile.command_flow is not None: document["command_flow"] = profile.command_flow + if profile.sync_origin: + document["sync_origin"] = profile.sync_origin + document["upstream_origin"] = profile.upstream_origin return document diff --git a/src/cognitive_agent/application/mode_scene_catalog.py b/src/cognitive_agent/application/mode_scene_catalog.py index a4d72ce..ff0f626 100644 --- a/src/cognitive_agent/application/mode_scene_catalog.py +++ b/src/cognitive_agent/application/mode_scene_catalog.py @@ -242,11 +242,41 @@ def _scene_from_profile( public_profile: dict[str, Any], ) -> dict[str, Any]: effective_operations = _effective_scene_operations(profile) + source_scene = ( + profile.command_flow.get("source_scene") + if profile.sync_origin == "from_n7/scenes" + and isinstance(profile.command_flow, dict) + else None + ) + source_actions = ( + source_scene.get("actions") + if isinstance(source_scene, dict) + and isinstance(source_scene.get("actions"), list) + else None + ) + actions = ( + deepcopy(source_actions) + if source_actions is not None + and len(source_actions) == len(effective_operations) + else [ + _scene_action( + operation, + target=target, + operation_ref=operation_ref, + ) + for operation, target, operation_ref in effective_operations + ] + ) scene = { "id": profile.id, "label": profile.name, "name": profile.name, - "background_time": _background_time(profile), + "background_time": ( + str(source_scene.get("background_time") or "").strip() + if isinstance(source_scene, dict) + else "" + ) + or _background_time(profile), "description": profile.description, "icon": profile.icon, "aliases": list(profile.aliases), @@ -257,19 +287,15 @@ def _scene_from_profile( "operation_count": len(effective_operations), "scope": deepcopy(public_profile.get("scope") or {}), "profile_revision": registry.profile_revision(profile), - "actions": [ - _scene_action( - operation, - target=target, - operation_ref=operation_ref, - ) - for operation, target, operation_ref in effective_operations - ], + "actions": actions, } if public_profile.get("preset_override") is True: scene["preset_override"] = True if profile.timeline is not None: scene["timeline"] = profile.timeline.public_dict() + if profile.sync_origin: + scene["source_topic"] = profile.sync_origin + scene["upstream_origin"] = profile.upstream_origin or None return scene diff --git a/src/cognitive_agent/application/mode_scene_ingress.py b/src/cognitive_agent/application/mode_scene_ingress.py new file mode 100644 index 0000000..b1bde78 --- /dev/null +++ b/src/cognitive_agent/application/mode_scene_ingress.py @@ -0,0 +1,449 @@ +"""Validate and persist N7-originated mode cards outside the MQTT callback.""" + +from __future__ import annotations + +import hashlib +import json +import queue +import re +import threading +import time +from collections.abc import Callable, Mapping +from copy import deepcopy +from datetime import UTC, datetime +from typing import Any + +from cognitive_agent.infrastructure.can.gateway import CanGatewayError +from cognitive_agent.infrastructure.can.mqtt_transport import ( + MQTT_MODE_SCENE_INGRESS_TOPIC, + MQTT_MODE_SCENE_MAX_PAYLOAD_BYTES, + mqtt_mode_scene_payload, +) + + +DEFAULT_MODE_SCENE_INGRESS_QUEUE_SIZE = 32 +DEFAULT_MODE_SCENE_DEDUPLICATION_WINDOW_S = 5.0 +MAX_MODE_SCENE_ACTIONS = 256 +MAX_MODE_SCENE_AT_MS = 24 * 60 * 60 * 1000 +_MODE_ID_PATTERN = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") + + +class ModeSceneIngressError(ValueError): + """Raised when an inbound ``from_n7/scenes`` card is not safe to import.""" + + +def _timestamp() -> str: + return datetime.now(UTC).isoformat(timespec="milliseconds").replace("+00:00", "Z") + + +def _json_object_without_duplicate_keys( + pairs: list[tuple[str, Any]], +) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise ModeSceneIngressError( + f"from_n7/scenes contains duplicate key: {key}" + ) + result[key] = value + return result + + +def _required_text( + value: Any, + *, + field: str, + maximum: int, +) -> str: + if not isinstance(value, str): + raise ModeSceneIngressError(f"from_n7/scenes {field} must be text") + cleaned = value.strip() + if not cleaned or len(cleaned) > maximum: + raise ModeSceneIngressError( + f"from_n7/scenes {field} must contain between 1 and {maximum} characters" + ) + return cleaned + + +def parse_mode_scene_ingress_payload(raw_payload: bytes | str) -> dict[str, Any]: + """Parse one card using the exact outbound scene contract plus hard limits.""" + + if isinstance(raw_payload, bytes): + encoded = raw_payload + try: + text = raw_payload.decode("utf-8") + except UnicodeDecodeError as exc: + raise ModeSceneIngressError( + "from_n7/scenes payload must be UTF-8 JSON" + ) from exc + elif isinstance(raw_payload, str): + text = raw_payload + encoded = raw_payload.encode("utf-8") + else: + raise ModeSceneIngressError( + "from_n7/scenes payload must be UTF-8 JSON" + ) + if not encoded or len(encoded) > MQTT_MODE_SCENE_MAX_PAYLOAD_BYTES: + raise ModeSceneIngressError( + "from_n7/scenes payload must contain 1-512 KiB" + ) + + try: + scene = json.loads( + text, + object_pairs_hook=_json_object_without_duplicate_keys, + ) + except ModeSceneIngressError: + raise + except (json.JSONDecodeError, TypeError, ValueError) as exc: + raise ModeSceneIngressError( + "from_n7/scenes payload must be valid JSON" + ) from exc + if not isinstance(scene, Mapping): + raise ModeSceneIngressError( + "from_n7/scenes payload must contain one scene object" + ) + try: + mqtt_mode_scene_payload(scene) + except CanGatewayError as exc: + raise ModeSceneIngressError(str(exc)) from exc + + normalized = deepcopy(dict(scene)) + scene_id = _required_text( + normalized.get("id"), field="id", maximum=64 + ) + if not _MODE_ID_PATTERN.fullmatch(scene_id): + raise ModeSceneIngressError( + "from_n7/scenes id must use lowercase letters, numbers, underscores, or hyphens" + ) + normalized["id"] = scene_id + normalized["name"] = _required_text( + normalized.get("name"), field="name", maximum=64 + ) + normalized["label"] = _required_text( + normalized.get("label"), field="label", maximum=64 + ) + normalized["icon"] = _required_text( + normalized.get("icon"), field="icon", maximum=64 + ) + normalized["background_time"] = _required_text( + normalized.get("background_time"), + field="background_time", + maximum=32, + ) + normalized["origin"] = _required_text( + normalized.get("origin"), field="origin", maximum=64 + ) + normalized["behavior"] = _required_text( + normalized.get("behavior"), field="behavior", maximum=32 + ) + normalized["profile_revision"] = _required_text( + normalized.get("profile_revision"), + field="profile_revision", + maximum=128, + ) + if not isinstance(normalized.get("deletable"), bool): + raise ModeSceneIngressError( + "from_n7/scenes deletable must be boolean" + ) + description = normalized.get("description") + if not isinstance(description, str) or len(description) > 500: + raise ModeSceneIngressError( + "from_n7/scenes description must be text with at most 500 characters" + ) + aliases = normalized.get("aliases") + if ( + not isinstance(aliases, list) + or len(aliases) > 64 + or any( + not isinstance(alias, str) + or not alias.strip() + or len(alias.strip()) > 64 + for alias in aliases + ) + ): + raise ModeSceneIngressError( + "from_n7/scenes aliases must contain at most 64 non-empty text values" + ) + normalized["aliases"] = [alias.strip() for alias in aliases] + domains = normalized.get("domains") + if ( + not isinstance(domains, list) + or not 1 <= len(domains) <= 64 + or any( + not isinstance(domain, str) + or not domain.strip() + or len(domain.strip()) > 64 + for domain in domains + ) + ): + raise ModeSceneIngressError( + "from_n7/scenes domains must contain 1-64 non-empty text values" + ) + normalized["domains"] = [domain.strip() for domain in domains] + + scope = normalized.get("scope") + if not isinstance(scope, Mapping): + raise ModeSceneIngressError("from_n7/scenes scope must be an object") + vehicle_ids = scope.get("vehicle_ids", []) + if "common" in scope and not isinstance(scope.get("common"), bool): + raise ModeSceneIngressError( + "from_n7/scenes scope.common must be boolean" + ) + if not isinstance(vehicle_ids, list) or len(vehicle_ids) > 64 or any( + not isinstance(vehicle_id, str) + or not vehicle_id.strip() + or len(vehicle_id.strip()) > 64 + for vehicle_id in vehicle_ids + ): + raise ModeSceneIngressError( + "from_n7/scenes scope.vehicle_ids must be an array of vehicle ids" + ) + + actions = normalized.get("actions") + if not isinstance(actions, list) or not 1 <= len(actions) <= MAX_MODE_SCENE_ACTIONS: + raise ModeSceneIngressError( + f"from_n7/scenes actions must contain 1-{MAX_MODE_SCENE_ACTIONS} items" + ) + for index, action in enumerate(actions): + if not isinstance(action, Mapping): + raise ModeSceneIngressError( + f"from_n7/scenes action {index} must be an object" + ) + at_ms = action.get("at_ms") + if ( + isinstance(at_ms, bool) + or not isinstance(at_ms, int) + or not 0 <= at_ms <= MAX_MODE_SCENE_AT_MS + ): + raise ModeSceneIngressError( + f"from_n7/scenes action {index} at_ms must be in range 0-{MAX_MODE_SCENE_AT_MS}" + ) + for field, maximum in ( + ("type", 64), + ("domain", 64), + ("capability", 128), + ("target_scope", 128), + ("operation", 64), + ): + action[field] = _required_text( + action.get(field), + field=f"actions[{index}].{field}", + maximum=maximum, + ) + if not isinstance(action.get("parameters"), Mapping): + raise ModeSceneIngressError( + f"from_n7/scenes action {index} parameters must be an object" + ) + target = action.get("target") + if target is not None and not isinstance(target, Mapping): + raise ModeSceneIngressError( + f"from_n7/scenes action {index} target must be an object" + ) + operation_ref = action.get("operation_ref") + if operation_ref is not None and ( + not isinstance(operation_ref, str) + or not operation_ref.strip() + or len(operation_ref.strip()) > 128 + ): + raise ModeSceneIngressError( + f"from_n7/scenes action {index} operation_ref must be non-empty text" + ) + return normalized + + +class ModeSceneIngressConsumer: + """Deduplicate and persist inbound cards on one bounded worker thread.""" + + def __init__( + self, + importer: Callable[[Mapping[str, Any]], Mapping[str, Any]], + *, + queue_size: int = DEFAULT_MODE_SCENE_INGRESS_QUEUE_SIZE, + deduplication_window_s: float = DEFAULT_MODE_SCENE_DEDUPLICATION_WINDOW_S, + ) -> None: + if queue_size <= 0: + raise ValueError("queue_size must be greater than zero") + if deduplication_window_s <= 0: + raise ValueError("deduplication_window_s must be greater than zero") + self._importer = importer + self._queue: queue.Queue[tuple[bytes, float, str]] = queue.Queue(queue_size) + self._deduplication_window_s = float(deduplication_window_s) + self._recent_digests: dict[str, float] = {} + self._lock = threading.RLock() + self._stop = threading.Event() + self._thread: threading.Thread | None = None + self._status: dict[str, Any] = { + "state": "stopped", + "topic": MQTT_MODE_SCENE_INGRESS_TOPIC, + "received_count": 0, + "processed_count": 0, + "imported_count": 0, + "duplicate_count": 0, + "rejected_count": 0, + "last_scene_id": None, + "last_received_at": None, + "last_completed_at": None, + "last_latency_ms": None, + "last_result": None, + "last_error": None, + } + + def start(self) -> None: + if self._thread and self._thread.is_alive(): + return + self._stop.clear() + with self._lock: + self._status.update(state="waiting", last_error=None) + self._thread = threading.Thread( + target=self._run, + name="n7-mode-scene-ingress", + daemon=True, + ) + self._thread.start() + + def close(self) -> None: + self._stop.set() + thread = self._thread + if thread and thread.is_alive(): + thread.join(timeout=2.0) + self._thread = None + with self._lock: + self._status["state"] = "stopped" + + def submit(self, raw_payload: bytes) -> None: + """Queue broker bytes immediately; parsing and disk I/O happen later.""" + + if self._stop.is_set(): + raise RuntimeError("mode scene ingress consumer is stopped") + payload = bytes(raw_payload) + if not payload or len(payload) > MQTT_MODE_SCENE_MAX_PAYLOAD_BYTES: + raise ModeSceneIngressError( + "from_n7/scenes payload must contain 1-512 KiB" + ) + received_at = _timestamp() + try: + self._queue.put_nowait((payload, time.monotonic(), received_at)) + except queue.Full as exc: + with self._lock: + self._status.update( + state="queue_full", + last_error="from_n7/scenes queue is full", + ) + raise RuntimeError("from_n7/scenes queue is full") from exc + with self._lock: + self._status.update( + state="queued", + received_count=int(self._status["received_count"]) + 1, + last_received_at=received_at, + last_error=None, + ) + + def snapshot(self) -> dict[str, Any]: + with self._lock: + return { + **deepcopy(self._status), + "queue_depth": self._queue.qsize(), + "queue_capacity": self._queue.maxsize, + "deduplication_window_ms": round( + self._deduplication_window_s * 1000 + ), + } + + def _is_duplicate(self, digest: str, now: float) -> bool: + cutoff = now - self._deduplication_window_s + self._recent_digests = { + key: observed_at + for key, observed_at in self._recent_digests.items() + if observed_at >= cutoff + } + previous = self._recent_digests.get(digest) + return previous is not None and previous >= cutoff + + def _process(self, item: tuple[bytes, float, str]) -> None: + raw_payload, received_monotonic, received_at = item + with self._lock: + self._status.update( + state="validating", + last_received_at=received_at, + last_error=None, + ) + try: + scene = parse_mode_scene_ingress_payload(raw_payload) + canonical = json.dumps( + scene, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ) + digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest() + now = time.monotonic() + with self._lock: + duplicate = self._is_duplicate(digest, now) + if duplicate: + with self._lock: + self._status.update( + state="duplicate_ignored", + processed_count=int(self._status["processed_count"]) + 1, + duplicate_count=int(self._status["duplicate_count"]) + 1, + last_scene_id=scene["id"], + last_completed_at=_timestamp(), + last_latency_ms=round( + max(0.0, now - received_monotonic) * 1000 + ), + last_error=None, + ) + return + result = self._importer(deepcopy(scene)) + if not isinstance(result, Mapping): + raise TypeError("mode scene importer returned an invalid result") + with self._lock: + self._recent_digests[digest] = time.monotonic() + except Exception as exc: # noqa: BLE001 - reject one card, keep worker alive + with self._lock: + self._status.update( + state="rejected", + processed_count=int(self._status["processed_count"]) + 1, + rejected_count=int(self._status["rejected_count"]) + 1, + last_completed_at=_timestamp(), + last_latency_ms=round( + max(0.0, time.monotonic() - received_monotonic) * 1000 + ), + last_result=None, + last_error=str(exc)[:240], + ) + return + + with self._lock: + self._status.update( + state="imported", + processed_count=int(self._status["processed_count"]) + 1, + imported_count=int(self._status["imported_count"]) + 1, + last_scene_id=scene["id"], + last_completed_at=_timestamp(), + last_latency_ms=round( + max(0.0, time.monotonic() - received_monotonic) * 1000 + ), + last_result=deepcopy(dict(result)), + last_error=None, + ) + + def _run(self) -> None: + while not self._stop.is_set(): + try: + item = self._queue.get(timeout=0.1) + except queue.Empty: + continue + try: + self._process(item) + finally: + self._queue.task_done() + with self._lock: + if self._queue.empty() and self._status["state"] in { + "imported", + "duplicate_ignored", + }: + # Keep the outcome visible while accurately reporting that + # the worker is ready for the next broker message. + self._status["worker_state"] = "waiting" diff --git a/src/cognitive_agent/application/seat_service.py b/src/cognitive_agent/application/seat_service.py index 17ce329..e632c27 100755 --- a/src/cognitive_agent/application/seat_service.py +++ b/src/cognitive_agent/application/seat_service.py @@ -2,14 +2,16 @@ from __future__ import annotations +import hashlib import json import os import re import sys +import threading import time import uuid from collections import Counter -from collections.abc import Callable +from collections.abc import Callable, Mapping from concurrent.futures import ThreadPoolExecutor from copy import deepcopy from datetime import UTC, datetime @@ -50,6 +52,10 @@ from cognitive_agent.application.domain_runtime import ConfiguredDomainRuntime from cognitive_agent.application.mode_profile import ( # noqa: E402 ModeProfile, ModeProfileRegistry, + ModeTimeline, +) +from cognitive_agent.application.mode_scene_ingress import ( # noqa: E402 + ModeSceneIngressConsumer, ) from cognitive_agent.application.mode_scene_catalog import ( # noqa: E402 DEFAULT_N7_SCENE_SERVER_ID, @@ -130,7 +136,9 @@ from cognitive_agent.infrastructure.can.hardware import ( # noqa: E402 VehicleMqttSettings, ) from cognitive_agent.infrastructure.can.mqtt_transport import ( # noqa: E402 + MQTT_MODE_SCENE_INGRESS_TOPIC, MQTT_MODE_SCENE_TOPIC, + MQTT_N7_CONTROL_TRANSFER_TOPIC, MqttCanTransport, ) from cognitive_agent.infrastructure.llm.openai_responses import ( # noqa: E402 @@ -446,6 +454,14 @@ class SeatDemoService: lambda scene: self._vehicle_mqtt_transport.publish_mode_scene(scene), topic=MQTT_MODE_SCENE_TOPIC, ) + self._mode_scene_ingress = ModeSceneIngressConsumer( + self.import_n7_mode_scene, + ) + self._vehicle_mqtt_transport.set_mode_scene_ingress_handler( + self._mode_scene_ingress.submit + ) + self._n7_transfer_lock = threading.RLock() + self._n7_transfer_publications: dict[str, float] = {} self.scene_templates = SceneTemplateRegistry( repository_root / "config" / "scene_templates" ) @@ -546,6 +562,18 @@ class SeatDemoService: def mode_scene_catalog_status(self) -> dict[str, Any]: return deepcopy(self._mode_scene_catalog_status) + def mode_profile_sync_status(self) -> dict[str, Any]: + """Return the lightweight state used by the browser's scene watcher.""" + + catalog = self.mode_scene_catalog_status() + return { + "catalog_revision": catalog.get("catalog_revision"), + "scene_count": int(catalog.get("scene_count") or 0), + "catalog_state": str(catalog.get("state") or "unknown"), + "catalog_error": catalog.get("last_error"), + "mode_scene_ingress": self._mode_scene_ingress.snapshot(), + } + def publish_saved_mode_scene( self, mode_id: str, @@ -591,6 +619,142 @@ class SeatDemoService: return self.publish_saved_mode_scene(mode_id, vehicle_id) + @staticmethod + def _imported_scene_function_configs( + scene: dict[str, Any], + ) -> dict[str, dict[str, Any]]: + """Retain only local peripheral metadata needed for later authorization.""" + + extension_operations: list[dict[str, Any]] = [] + for index, action in enumerate(scene.get("actions", [])): + if ( + not isinstance(action, dict) + or action.get("domain") != "self_extension" + or action.get("capability") != "infrared.emit" + ): + continue + target = action.get("target") + target_document = ( + deepcopy(target) + if isinstance(target, dict) + else {"device": str(action.get("target_scope") or "")} + ) + extension_operations.append( + { + "id": str(action.get("operation_ref") or f"n7-ir-{index + 1}"), + "domain": "self_extension", + "capability": "infrared.emit", + "target_scope": str(action.get("target_scope") or ""), + "target": target_document, + "operation": str(action.get("operation") or "emit"), + "parameters": deepcopy(dict(action.get("parameters") or {})), + "parse_source": "from_n7_scene", + } + ) + if not extension_operations: + return {} + return { + "self_extension": { + "apiVersion": "cognitive-agent.io/v1alpha1", + "kind": "FunctionConfig", + "metadata": { + "id": f"from-n7-{scene['id']}-self-extension", + "name": f"{scene['name']} imported infrared actions", + "labels": {"source_topic": MQTT_MODE_SCENE_INGRESS_TOPIC}, + }, + "spec": { + "domain": "self_extension", + "operations": extension_operations, + }, + } + } + + def import_n7_mode_scene( + self, + scene: Mapping[str, Any], + ) -> dict[str, Any]: + """Upsert one validated N7 card without publishing it back to the broker.""" + + document = deepcopy(dict(scene)) + actions = document.get("actions") + if not isinstance(actions, list) or not actions: + raise ValueError("from_n7/scenes card must contain executable actions") + operations = [ + { + "domain": str(action["domain"]), + "capability": str(action["capability"]), + "target_scope": str(action["target_scope"]), + "operation": str(action["operation"]), + "parameters": deepcopy(dict(action["parameters"])), + "reason": ( + f"由 {MQTT_MODE_SCENE_INGRESS_TOPIC} 导入的模式动作。" + ), + } + for action in actions + ] + timeline = None + raw_timeline = document.get("timeline") + if raw_timeline is not None: + if not isinstance(raw_timeline, dict): + raise ValueError("from_n7/scenes timeline must be an object") + timeline = ModeTimeline.from_dict(raw_timeline) + command_flow = { + "schema": "cognitive-agent-mode-command-flow/v1", + "source_topic": MQTT_MODE_SCENE_INGRESS_TOPIC, + "imported_at": _now(), + "source_scene": document, + "function_configs": self._imported_scene_function_configs(document), + # The execution path recompiles the canonical ModeOperation list + # against this PC's capability registry. This marker makes the + # card directly executable while avoiding trust in foreign CAN bytes. + "plans": { + "from_n7_scene": { + "spec": { + "commands": [ + { + "operation_ref": str( + action.get("operation_ref") or f"action-{index + 1}" + ), + "at_ms": int(action.get("at_ms", 0)), + } + for index, action in enumerate(actions) + ] + } + } + }, + "execution_steps": [], + } + scope = document.get("scope") or {} + vehicle_ids = scope.get("vehicle_ids", []) if isinstance(scope, dict) else [] + common = bool(scope.get("common", not vehicle_ids)) if isinstance(scope, dict) else False + resolved_vehicle_id = ( + None + if common and not vehicle_ids + else self.vehicle_profiles.default().id + ) + saved = self.mode_profiles.save_mode( + str(document["name"]), + operations, + mode_id=str(document["id"]), + aliases=[str(alias) for alias in document.get("aliases", [])], + vehicle_id=resolved_vehicle_id, + description=str(document.get("description") or "由 N7 同步的模式卡片。"), + icon=str(document.get("icon") or "radio-tower"), + instruction=f"进入{document['name']}", + command_flow=command_flow, + timeline=timeline, + sync_origin=MQTT_MODE_SCENE_INGRESS_TOPIC, + upstream_origin=str(document.get("origin") or ""), + ) + return { + "status": "imported", + "scene_id": saved.id, + "scene_name": saved.name, + "source_topic": MQTT_MODE_SCENE_INGRESS_TOPIC, + "profile_revision": self.mode_profiles.profile_revision(saved), + "storage": "persistent_yaml", + } + @property def driver_only_ch1_execution_active(self) -> bool: """Return whether deployment policy authorizes only the CH1 vehicle path.""" @@ -646,6 +810,7 @@ class SeatDemoService: "last_error": None, } vehicle_mqtt_info = self._vehicle_mqtt_transport.public_info() + vehicle_mqtt_info["mode_scene_import"] = self._mode_scene_ingress.snapshot() return { "service": "cabin-cognitive-orchestrator", "domain": "cabin", @@ -807,6 +972,207 @@ class SeatDemoService: resolved_vehicle_id = vehicle_id or self.vehicle_profiles.default().id return self.mode_profiles.get_profile(mode_id, resolved_vehicle_id) + @staticmethod + def _fallback_scene_action( + operation: Any, + *, + index: int, + ) -> dict[str, Any]: + return { + "type": "control", + "at_ms": 0, + "domain": operation.domain, + "capability": operation.capability, + "target_scope": operation.target_scope, + "operation": operation.operation, + "parameters": deepcopy(operation.parameters), + "operation_ref": f"action-{index + 1}", + } + + def _mode_requests_and_n7_transfers( + self, + profile: ModeProfile, + ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: + """Separate locally understood operations from N7-only capabilities.""" + + requests = self.mode_profiles.requests_for_profiles([profile]) + if profile.sync_origin != MQTT_MODE_SCENE_INGRESS_TOPIC: + return requests, [] + source_scene = ( + profile.command_flow.get("source_scene") + if isinstance(profile.command_flow, dict) + else None + ) + source_actions = ( + source_scene.get("actions") + if isinstance(source_scene, dict) + and isinstance(source_scene.get("actions"), list) + else [] + ) + local_requests: list[dict[str, Any]] = [] + transfer_actions: list[dict[str, Any]] = [] + for index, (operation, request) in enumerate( + zip(profile.operations, requests, strict=True) + ): + if ( + operation.domain == "self_extension" + and operation.capability == "infrared.emit" + ): + # The server's durable NFC allowlist owns this local TCP path. + # Authorization failures must not be converted into N7 retries. + continue + registered = self.model_plan_adapter.capabilities.get( + operation.capability + ) + native_candidate = { + "domain": operation.domain, + "capability": operation.capability, + "target": operation.target_scope, + "parameters": deepcopy(operation.parameters), + "operation": operation.operation, + } + if registered is not None or self.native_can_runtime.supports( + native_candidate + ): + local_requests.append(request) + continue + raw_action = ( + source_actions[index] + if index < len(source_actions) + and isinstance(source_actions[index], dict) + else None + ) + action = ( + deepcopy(raw_action) + if raw_action is not None + else self._fallback_scene_action(operation, index=index) + ) + transfer_actions.append(action) + return local_requests, transfer_actions + + @staticmethod + def _has_local_mode_peripheral(profile: ModeProfile) -> bool: + return any( + operation.domain == "self_extension" + and operation.capability == "infrared.emit" + for operation in profile.operations + ) + + @staticmethod + def _n7_transfer_deferred_requests( + actions: list[dict[str, Any]], + ) -> list[dict[str, Any]]: + return [ + { + "domain": str(action.get("domain") or ""), + "capability": str(action.get("capability") or ""), + "target_scope": str(action.get("target_scope") or ""), + "desired_change": str(action.get("operation") or "set"), + "parameters": deepcopy(dict(action.get("parameters") or {})), + "reason": "当前 PC 未注册该能力,将由 N7 端继续处理。", + "execution_backend": "n7_vehicle_transfer", + "rejection_code": "PC_CAPABILITY_UNAVAILABLE", + "transport_submitted": False, + } + for action in actions + ] + + def _finalize_n7_control_transfer( + self, + result: dict[str, Any], + *, + execution_mode: str, + local_preview: bool, + ) -> dict[str, Any]: + """Forward only N7-originated actions never submitted by this PC.""" + + actions = result.get("n7_transfer_candidates") + scene = result.get("n7_source_scene") + if ( + execution_mode != "hardware" + or local_preview + or not isinstance(actions, list) + or not actions + or not isinstance(scene, dict) + ): + return result + if str(result.get("status") or "") not in { + "succeeded", + "partially_succeeded", + "capability_unavailable", + }: + return result + if result.get("execution_outcome") == "failed_or_blocked": + return result + request_id = str(result.get("request_id") or "").strip() + scene_id = str(scene.get("id") or "").strip() + scene_name = str(scene.get("name") or "").strip() + if not request_id or not scene_id or not scene_name: + return result + transfer_actions = [deepcopy(action) for action in actions if isinstance(action, dict)] + if not transfer_actions: + return result + payload = { + "scene_id": scene_id, + "scene_name": scene_name, + "request_id": request_id, + "operation_count": len(transfer_actions), + "actions": transfer_actions, + } + digest = json.dumps( + payload, + ensure_ascii=False, + sort_keys=True, + separators=(",", ":"), + ) + publication_key = f"{request_id}:{hashlib.sha256(digest.encode('utf-8')).hexdigest()}" + now = time.monotonic() + with self._n7_transfer_lock: + cutoff = now - 300.0 + self._n7_transfer_publications = { + key: published_at + for key, published_at in self._n7_transfer_publications.items() + if published_at >= cutoff + } + if publication_key in self._n7_transfer_publications: + result["n7_control_transfer"] = { + "state": "duplicate_ignored", + "topic": MQTT_N7_CONTROL_TRANSFER_TOPIC, + "scene_id": scene_id, + "operation_count": len(transfer_actions), + } + return result + try: + publication = self._vehicle_mqtt_transport.publish_n7_control_transfer( + payload + ) + except Exception as exc: # noqa: BLE001 - local execution result remains valid + result["n7_control_transfer"] = { + "state": "publish_failed", + "topic": MQTT_N7_CONTROL_TRANSFER_TOPIC, + "scene_id": scene_id, + "operation_count": len(transfer_actions), + "last_error": str(exc)[:240], + } + return result + with self._n7_transfer_lock: + self._n7_transfer_publications[publication_key] = now + result["n7_control_transfer"] = { + "state": "published", + "scene_id": scene_id, + "operation_count": len(transfer_actions), + "payload": payload, + **dict(publication), + } + result["decision_summary"] = [ + *list(result.get("decision_summary") or []), + ( + f"当前 PC 未注册的 {len(transfer_actions)} 条 N7 来源动作已发送至 " + f"{MQTT_N7_CONTROL_TRANSFER_TOPIC};PC 已提交或被安全门禁拦截的动作未转发。" + ), + ] + return result + def execute_mode_profile_card( self, mode_id: str, @@ -1183,6 +1549,10 @@ class SeatDemoService: instruction=clean_instruction, command_flow=command_flow, timeline=profile.timeline if profile is not None else None, + sync_origin=profile.sync_origin if profile is not None else "", + upstream_origin=( + profile.upstream_origin if profile is not None else "" + ), expected_revision=draft_existing_mode_revision or None, create_only=not bool(clean_mode_id), ) @@ -1379,6 +1749,7 @@ class SeatDemoService: if self._vehicle_state_monitor is not None: self._vehicle_state_monitor.start() self._mode_scene_mqtt_publisher.start() + self._mode_scene_ingress.start() self._vehicle_mqtt_transport.start() def close(self) -> None: @@ -1386,6 +1757,7 @@ class SeatDemoService: self._vehicle_state_monitor.close() self._mode_scene_mqtt_publisher.close() self._vehicle_mqtt_transport.close() + self._mode_scene_ingress.close() def publish_pet_message(self, payload: dict[str, str]) -> dict[str, Any]: """Publish one NFC-authorized pet card through the shared vehicle broker.""" @@ -2416,7 +2788,7 @@ class SeatDemoService: # The head unit owns both physical CAN channels in wireless mode. # Local T2G/ZLG hardware access, firmware profiles and the local # physical-bench lease are unrelated to MQTT relay execution. - return self._process_once( + result = self._process_once( utterance, provider, conversation_id, @@ -2438,11 +2810,16 @@ class SeatDemoService: mode_profile_draft_context=_mode_profile_draft_context, cloud_instruction_context=_cloud_instruction_context, ) + return self._finalize_n7_control_transfer( + result, + execution_mode=execution_mode, + local_preview=_local_preview, + ) if hardware_access_bypass or self._driver_only_vehicle_execution_active( control_transport ): with self.hardware_execution.reserve(): - return self._process_once( + result = self._process_once( utterance, provider, conversation_id, @@ -2464,10 +2841,15 @@ class SeatDemoService: mode_profile_draft_context=_mode_profile_draft_context, cloud_instruction_context=_cloud_instruction_context, ) + return self._finalize_n7_control_transfer( + result, + execution_mode=execution_mode, + local_preview=_local_preview, + ) authorized_client_id = self.hardware_access.ensure_authorized(hardware_client_id) with self.hardware_execution.reserve(): with self.hardware_access.use(authorized_client_id): - return self._process_once( + result = self._process_once( utterance, provider, conversation_id, @@ -2489,6 +2871,11 @@ class SeatDemoService: mode_profile_draft_context=_mode_profile_draft_context, cloud_instruction_context=_cloud_instruction_context, ) + return self._finalize_n7_control_transfer( + result, + execution_mode=execution_mode, + local_preview=_local_preview, + ) def _process_once( self, @@ -2768,8 +3155,8 @@ class SeatDemoService: traces, ) if direct_mode_profile is not None: - mode_requests = self.mode_profiles.requests_for_profiles( - [direct_mode_profile] + mode_requests, n7_transfer_actions = ( + self._mode_requests_and_n7_transfers(direct_mode_profile) ) detected_domains = list( dict.fromkeys( @@ -2814,6 +3201,28 @@ class SeatDemoService: *list(parsed.get("native_can_actions", [])), *saved_native_actions, ] + if n7_transfer_actions: + parsed["deferred_requests"] = [ + *list(parsed.get("deferred_requests", [])), + *self._n7_transfer_deferred_requests( + n7_transfer_actions + ), + ] + parsed["_n7_transfer_actions"] = n7_transfer_actions + parsed["_n7_scene_profile"] = { + "id": direct_mode_profile.id, + "name": direct_mode_profile.name, + } + parsed["detected_domains"] = self._merge_domains( + list(parsed.get("detected_domains", [])), + [ + str(action.get("domain") or "") + for action in n7_transfer_actions + if str(action.get("domain") or "") + ], + ) + if self._has_local_mode_peripheral(direct_mode_profile): + parsed["_saved_mode_local_peripheral"] = True provider_used = "saved_mode_command_flow" provider_notice = None elif direct_can_batch is not None: @@ -2951,6 +3360,12 @@ class SeatDemoService: ) ), ) + n7_transfer_actions = parsed.get("_n7_transfer_actions", []) + n7_scene_profile = parsed.get("_n7_scene_profile") + if isinstance(n7_transfer_actions, list) and n7_transfer_actions: + result["n7_transfer_candidates"] = deepcopy(n7_transfer_actions) + if isinstance(n7_scene_profile, dict): + result["n7_source_scene"] = deepcopy(n7_scene_profile) result["parsed_control_signatures"] = [ list(self._control_signature(action)) for action in [ @@ -2969,7 +3384,10 @@ class SeatDemoService: ) ) if ( - allow_empty_control_result + ( + allow_empty_control_result + or parsed.get("_saved_mode_local_peripheral") is True + ) and not has_control and not parsed.get("ambiguities") ): @@ -3019,7 +3437,11 @@ class SeatDemoService: "hardware mode refuses virtual or unverified vehicle actions: " + ", ".join(unsupported) ) - if not parsed.get("actions") and not parsed.get("native_can_actions"): + if ( + not parsed.get("actions") + and not parsed.get("native_can_actions") + and not parsed.get("_n7_transfer_actions") + ): raise HardwareSafetyError( "hardware mode requires at least one verified seat or vehicle CAN operation" ) @@ -4070,7 +4492,10 @@ class SeatDemoService: if safety_ambiguities: return parsed - mode_requests = self.mode_profiles.requests_for_profiles(matched_modes) + matched_profile = matched_modes[0] + mode_requests, n7_transfer_actions = ( + self._mode_requests_and_n7_transfers(matched_profile) + ) detected_domains = [] for request in mode_requests: if request["domain"] not in detected_domains: @@ -4136,6 +4561,26 @@ class SeatDemoService: float(mode_parsed.get("confidence", 0)), float(parsed.get("confidence", 0)), ) + if n7_transfer_actions: + merged["deferred_requests"] = self._merge_unique( + list(merged.get("deferred_requests", [])), + self._n7_transfer_deferred_requests(n7_transfer_actions), + ) + merged["_n7_transfer_actions"] = deepcopy(n7_transfer_actions) + merged["_n7_scene_profile"] = { + "id": matched_profile.id, + "name": matched_profile.name, + } + merged["detected_domains"] = self._merge_domains( + list(merged.get("detected_domains", [])), + [ + str(action.get("domain") or "") + for action in n7_transfer_actions + if str(action.get("domain") or "") + ], + ) + if self._has_local_mode_peripheral(matched_profile): + merged["_saved_mode_local_peripheral"] = True return merged @staticmethod diff --git a/src/cognitive_agent/infrastructure/can/mqtt_transport.py b/src/cognitive_agent/infrastructure/can/mqtt_transport.py index 2ba16e4..060652c 100644 --- a/src/cognitive_agent/infrastructure/can/mqtt_transport.py +++ b/src/cognitive_agent/infrastructure/can/mqtt_transport.py @@ -43,6 +43,11 @@ MQTT_EXTENSION_CONTROL_TOPIC = "extension/ctl" MQTT_EXTENSION_CONTROL_QOS = 1 MQTT_EXTENSION_CONTROL_MAX_PAYLOAD_BYTES = 256 * 1024 MQTT_MODE_SCENE_TOPIC = "from_pc/scenes" +MQTT_MODE_SCENE_INGRESS_TOPIC = "from_n7/scenes" +MQTT_MODE_SCENE_INGRESS_QOS = 1 +MQTT_MODE_SCENE_MAX_PAYLOAD_BYTES = 512 * 1024 +MQTT_N7_CONTROL_TRANSFER_TOPIC = "transto_n7/ctl" +MQTT_N7_CONTROL_TRANSFER_QOS = 1 MQTT_POLL_INTERVAL_MS = 500 WIRELESS_MQTT_CH1_SOURCE = "wireless_mqtt_ch1" MQTT_VALID_FD_PAYLOAD_LENGTHS = frozenset( @@ -727,11 +732,86 @@ def mqtt_mode_scene_payload(payload: Mapping[str, Any]) -> str: ) except (TypeError, ValueError) as exc: raise CanGatewayError("mode scene MQTT payload is not valid JSON") from exc - if len(raw_payload.encode("utf-8")) > 512 * 1024: + if len(raw_payload.encode("utf-8")) > MQTT_MODE_SCENE_MAX_PAYLOAD_BYTES: raise CanGatewayError("mode scene MQTT payload exceeds 512 KiB") return raw_payload +_N7_CONTROL_TRANSFER_REQUIRED_KEYS = frozenset( + { + "scene_id", + "scene_name", + "request_id", + "operation_count", + "actions", + } +) + + +def mqtt_n7_control_transfer_payload(payload: Mapping[str, Any]) -> str: + """Serialize PC-unavailable actions for the originating N7 scene.""" + + if not isinstance(payload, Mapping): + raise CanGatewayError("N7 control transfer payload must be a JSON object") + missing = sorted(_N7_CONTROL_TRANSFER_REQUIRED_KEYS - set(payload)) + if missing: + raise CanGatewayError( + "N7 control transfer payload is missing required fields: " + + ", ".join(missing) + ) + for key, maximum in (("scene_id", 64), ("scene_name", 64), ("request_id", 128)): + value = payload.get(key) + if not isinstance(value, str) or not value.strip() or len(value) > maximum: + raise CanGatewayError( + f"N7 control transfer {key} must contain between 1 and {maximum} characters" + ) + operation_count = payload.get("operation_count") + actions = payload.get("actions") + if ( + isinstance(operation_count, bool) + or not isinstance(operation_count, int) + or operation_count < 1 + ): + raise CanGatewayError( + "N7 control transfer operation_count must be a positive integer" + ) + if not isinstance(actions, list) or len(actions) != operation_count: + raise CanGatewayError( + "N7 control transfer actions must match operation_count" + ) + for action in actions: + if not isinstance(action, Mapping): + raise CanGatewayError( + "N7 control transfer actions must contain JSON objects" + ) + missing_action_keys = sorted( + _MODE_SCENE_ACTION_REQUIRED_KEYS - set(action) + ) + if missing_action_keys: + raise CanGatewayError( + "N7 control transfer action is missing required fields: " + + ", ".join(missing_action_keys) + ) + if not isinstance(action.get("parameters"), Mapping): + raise CanGatewayError( + "N7 control transfer action parameters must be an object" + ) + try: + raw_payload = json.dumps( + deepcopy(dict(payload)), + ensure_ascii=False, + separators=(",", ":"), + allow_nan=False, + ) + except (TypeError, ValueError) as exc: + raise CanGatewayError( + "N7 control transfer payload is not valid JSON" + ) from exc + if len(raw_payload.encode("utf-8")) > MQTT_MODE_SCENE_MAX_PAYLOAD_BYTES: + raise CanGatewayError("N7 control transfer payload exceeds 512 KiB") + return raw_payload + + def _validate_native_seat_motion_control(data: bytes) -> None: """Validate the firmware-native 0x300 semantic runtime subset. @@ -863,6 +943,13 @@ class MqttCanTransport: self._extension_control_retained_count = 0 self._extension_control_last_received_at: str | None = None self._extension_control_last_error: str | None = None + self._mode_scene_ingress_handler: Callable[[bytes], None] | None = None + self._mode_scene_ingress_received_count = 0 + self._mode_scene_ingress_dispatched_count = 0 + self._mode_scene_ingress_rejected_count = 0 + self._mode_scene_ingress_retained_count = 0 + self._mode_scene_ingress_last_received_at: str | None = None + self._mode_scene_ingress_last_error: str | None = None def _build_client(self) -> Any: if self.client_factory is not None: @@ -968,6 +1055,18 @@ class MqttCanTransport: self._extension_control_rejection_handler = handler self._condition.notify_all() + def set_mode_scene_ingress_handler( + self, + handler: Callable[[bytes], None] | None, + ) -> None: + """Register the non-blocking application queue for ``from_n7/scenes``.""" + + if handler is not None and not callable(handler): + raise TypeError("mode scene ingress handler must be callable") + with self._condition: + self._mode_scene_ingress_handler = handler + self._condition.notify_all() + def _notify_extension_control_rejection(self, error: str, kind: str) -> None: with self._condition: handler = self._extension_control_rejection_handler @@ -1003,6 +1102,7 @@ class MqttCanTransport: subscriptions = ( (self.settings.condition_topic, self.settings.qos), (MQTT_EXTENSION_CONTROL_TOPIC, MQTT_EXTENSION_CONTROL_QOS), + (MQTT_MODE_SCENE_INGRESS_TOPIC, MQTT_MODE_SCENE_INGRESS_QOS), ) failed_subscription: tuple[str, int] | None = None try: @@ -1054,6 +1154,9 @@ class MqttCanTransport: if topic == MQTT_EXTENSION_CONTROL_TOPIC: self._dispatch_extension_control_message(message) return + if topic == MQTT_MODE_SCENE_INGRESS_TOPIC: + self._dispatch_mode_scene_ingress_message(message) + return if topic != self.settings.condition_topic: return try: @@ -1134,6 +1237,45 @@ class MqttCanTransport: self._condition.notify_all() self._notify_extension_control_rejection(error, rejection_kind) + def _dispatch_mode_scene_ingress_message(self, message: Any) -> None: + """Forward a fresh scene payload without blocking the MQTT network loop.""" + + received_at = _now() + with self._condition: + self._mode_scene_ingress_received_count += 1 + self._mode_scene_ingress_last_received_at = received_at + handler = self._mode_scene_ingress_handler + retained = bool(getattr(message, "retain", False)) + if retained: + self._mode_scene_ingress_retained_count += 1 + + try: + if retained: + raise CanGatewayError( + "retained from_n7/scenes messages are not imported" + ) + payload = bytes(message.payload) + if not payload or len(payload) > MQTT_MODE_SCENE_MAX_PAYLOAD_BYTES: + raise CanGatewayError( + "from_n7/scenes payload must contain 1-512 KiB" + ) + if handler is None: + raise CanGatewayError( + "from_n7/scenes handler is not configured" + ) + handler(payload) + except Exception as exc: # noqa: BLE001 - isolate MQTT network loop + with self._condition: + self._mode_scene_ingress_rejected_count += 1 + self._mode_scene_ingress_last_error = str(exc)[:240] + self._condition.notify_all() + return + + with self._condition: + self._mode_scene_ingress_dispatched_count += 1 + self._mode_scene_ingress_last_error = None + self._condition.notify_all() + def _record_frame(self, frame: CanFrame) -> None: if ( frame.channel not in MQTT_ACCEPTED_FEEDBACK_IDS @@ -1431,6 +1573,38 @@ class MqttCanTransport: "message_id": getattr(result, "mid", None), } + def publish_n7_control_transfer( + self, + payload: Mapping[str, Any], + ) -> dict[str, Any]: + """Queue PC-unavailable scene actions for execution by the N7 side.""" + + raw_payload = mqtt_n7_control_transfer_payload(payload) + with self._condition: + client = self._client + state = self._state + if client is None or state != "online": + raise CanGatewayError("N7 control transfer MQTT publisher is not connected") + result = client.publish( + MQTT_N7_CONTROL_TRANSFER_TOPIC, + raw_payload, + qos=MQTT_N7_CONTROL_TRANSFER_QOS, + retain=False, + ) + if int(getattr(result, "rc", 0)) != 0: + raise CanGatewayError( + f"N7 control transfer MQTT publish failed with code {result.rc}" + ) + is_published = getattr(result, "is_published", None) + broker_confirmed = bool(is_published()) if callable(is_published) else False + return { + "topic": MQTT_N7_CONTROL_TRANSFER_TOPIC, + "qos": MQTT_N7_CONTROL_TRANSFER_QOS, + "retain": False, + "delivery_status": "published" if broker_confirmed else "queued", + "message_id": getattr(result, "mid", None), + } + def _receive_after( self, cursor: int, @@ -1500,6 +1674,18 @@ class MqttCanTransport: "last_received_at": self._extension_control_last_received_at, "last_error": self._extension_control_last_error, } + info["mode_scene_ingress"] = { + "mode": "persistent_mqtt_subscription", + "topic": MQTT_MODE_SCENE_INGRESS_TOPIC, + "qos": MQTT_MODE_SCENE_INGRESS_QOS, + "handler_configured": self._mode_scene_ingress_handler is not None, + "received_count": self._mode_scene_ingress_received_count, + "dispatched_count": self._mode_scene_ingress_dispatched_count, + "rejected_count": self._mode_scene_ingress_rejected_count, + "retained_count": self._mode_scene_ingress_retained_count, + "last_received_at": self._mode_scene_ingress_last_received_at, + "last_error": self._mode_scene_ingress_last_error, + } return info diff --git a/src/cognitive_agent/lan_scene_server.py b/src/cognitive_agent/lan_scene_server.py index bb2415c..e844b33 100644 --- a/src/cognitive_agent/lan_scene_server.py +++ b/src/cognitive_agent/lan_scene_server.py @@ -5,6 +5,7 @@ from __future__ import annotations import argparse import json import os +import socket import threading from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -64,7 +65,16 @@ class SceneCatalogSource: class N7SceneHttpServer(ThreadingHTTPServer): daemon_threads = True - allow_reuse_address = True + allow_reuse_address = os.name != "nt" + + def server_bind(self) -> None: + if os.name == "nt" and hasattr(socket, "SO_EXCLUSIVEADDRUSE"): + self.socket.setsockopt( + socket.SOL_SOCKET, + socket.SO_EXCLUSIVEADDRUSE, + 1, + ) + super().server_bind() def __init__(self, address: tuple[str, int], source: SceneCatalogSource) -> None: self.catalog_source = source @@ -145,7 +155,16 @@ class N7SceneRequestHandler(BaseHTTPRequestHandler): class N7SceneUdpServer(ThreadingUDPServer): daemon_threads = True - allow_reuse_address = True + allow_reuse_address = os.name != "nt" + + def server_bind(self) -> None: + if os.name == "nt" and hasattr(socket, "SO_EXCLUSIVEADDRUSE"): + self.socket.setsockopt( + socket.SOL_SOCKET, + socket.SO_EXCLUSIVEADDRUSE, + 1, + ) + super().server_bind() class N7SceneDiscoveryHandler(BaseRequestHandler): diff --git a/src/cognitive_agent/server.py b/src/cognitive_agent/server.py index e5418f9..298e8c5 100755 --- a/src/cognitive_agent/server.py +++ b/src/cognitive_agent/server.py @@ -358,6 +358,9 @@ class SeatDemoHandler(BaseHTTPRequestHandler): if path == "/api/assistant/capabilities": self._json(self._assistant_capability_catalog().snapshot()) return + if path == "/api/mode-profiles/status": + self._json(self.service.mode_profile_sync_status()) + return if path == "/api/runtime": self._json(self.service.runtime_info()) return @@ -1592,6 +1595,107 @@ class SeatDemoHandler(BaseHTTPRequestHandler): spec["status"] = str(result.get("status") or "blocked") return result + def _exact_n7_mode_invocation( + self, + utterance: str, + vehicle_id: str | None, + ) -> object | None: + """Resolve a standalone spoken/text invocation of one N7-synced card.""" + + if not str(utterance).strip(): + return None + registry = getattr(self.service, "mode_profiles", None) + match_text = getattr(registry, "match_text", None) + if not callable(match_text): + # Lightweight service doubles used by peripheral-only routes do + # not own a mode registry. They must keep their existing routing + # behaviour instead of failing before the peripheral parser runs. + return None + matches = [ + profile + for profile in match_text( + utterance, + vehicle_id=vehicle_id, + ) + if getattr(profile, "sync_origin", "") == "from_n7/scenes" + ] + # ``ModeProfileRegistry.match_text`` applies the bounded, exact + # invocation grammar to N7-synchronised cards. Keeping that rule in + # one place prevents text, speech and card-click routes from drifting. + return matches[0] if len(matches) == 1 else None + + def _execute_n7_named_mode_request( + self, + path: str, + common_arguments: dict[str, object], + ) -> dict[str, object] | None: + utterance = str(common_arguments.get("utterance") or "").strip() + vehicle_id = ( + str(common_arguments["vehicle_id"]) + if common_arguments.get("vehicle_id") + else None + ) + profile = self._exact_n7_mode_invocation(utterance, vehicle_id) + if profile is None: + return None + execution_mode = ( + "dry_run" + if path == "/api/requests/preview" + else str(common_arguments.get("execution_mode") or "dry_run") + ) + conversation_id = ( + str(common_arguments["conversation_id"]) + if common_arguments.get("conversation_id") + else None + ) + entries = self._saved_mode_infrared_entries(profile) + can_result = self.service.execute_mode_profile_card( + str(getattr(profile, "id")), + conversation_id=conversation_id, + execution_mode=execution_mode, + approval_token=( + str(common_arguments["approval_token"]) + if common_arguments.get("approval_token") + else None + ), + vehicle_id=vehicle_id, + simulation_fault=str( + common_arguments.get("simulation_fault") or "normal" + ), + hardware_client_id=( + str(common_arguments["hardware_client_id"]) + if common_arguments.get("hardware_client_id") + else None + ), + hardware_access_bypass=bool( + common_arguments.get("hardware_access_bypass", True) + ), + control_transport=str( + common_arguments.get("control_transport") or "wired_can" + ), + ) + if not entries: + can_result["utterance"] = utterance + return can_result + infrared_result = self._execute_local_peripheral_actions( + utterance, + entries, + provider="saved_mode_command_flow", + conversation_id=conversation_id, + execution_mode=execution_mode, + ) + result = self._merge_can_and_peripheral_results( + utterance, + can_result, + infrared_result, + entries, + ) + result["provider_used"] = "saved_mode_command_flow" + result["orchestration_route"] = "saved_mode_command_flow" + result["mode_profile"] = profile.public_dict() + self.service.persist_composed_result(result) + return result + def _execute_request_with_peripherals( self, path: str, @@ -1626,6 +1730,12 @@ class SeatDemoHandler(BaseHTTPRequestHandler): **mode_arguments, _on_commands_validated=callback, ) + n7_mode_result = self._execute_n7_named_mode_request( + path, + common_arguments, + ) + if n7_mode_result is not None: + return n7_mode_result catalog = self._peripheral_routing_catalog() nfc_snapshot = self.nfc.snapshot() partition = partition_active_peripheral_intent( @@ -4198,6 +4308,22 @@ class SeatDemoHandler(BaseHTTPRequestHandler): ) +class N7WebHttpServer(ThreadingHTTPServer): + """Prevent multiple Windows processes from silently sharing one port.""" + + daemon_threads = True + allow_reuse_address = os.name != "nt" + + def server_bind(self) -> None: + if os.name == "nt" and hasattr(socket, "SO_EXCLUSIVEADDRUSE"): + self.socket.setsockopt( + socket.SOL_SOCKET, + socket.SO_EXCLUSIVEADDRUSE, + 1, + ) + super().server_bind() + + def main() -> None: parser = argparse.ArgumentParser(description="Run the cabin orchestration demo") parser.add_argument("--host", default=os.getenv("HOST", "127.0.0.1")) @@ -4221,7 +4347,7 @@ def main() -> None: args.scene_catalog, server_id=args.scene_server_id, ) - server = ThreadingHTTPServer((args.host, args.port), SeatDemoHandler) + server = N7WebHttpServer((args.host, args.port), SeatDemoHandler) SeatDemoHandler.speech.start() SeatDemoHandler.assistant.start( f"http://127.0.0.1:{server.server_port}/api/internal/codex/vehicle-control" diff --git a/tests/e2e/test_seat_web_api.py b/tests/e2e/test_seat_web_api.py index fe80c8e..cafa5a5 100755 --- a/tests/e2e/test_seat_web_api.py +++ b/tests/e2e/test_seat_web_api.py @@ -151,6 +151,12 @@ class SeatWebApiTests(unittest.TestCase): with self.assertRaises(urllib.error.HTTPError) as caught: unauthenticated.open(f"{self.base_url}/api/runtime", timeout=3) self.assertEqual(caught.exception.code, 401) + with self.assertRaises(urllib.error.HTTPError) as sync_status_denied: + unauthenticated.open( + f"{self.base_url}/api/mode-profiles/status", + timeout=3, + ) + self.assertEqual(sync_status_denied.exception.code, 401) with self.assertRaises(urllib.error.HTTPError) as state_denied: unauthenticated.open( @@ -292,7 +298,9 @@ class SeatWebApiTests(unittest.TestCase): self.assertGreaterEqual(catalog["scene_count"], 13) self.assertEqual(catalog["scene_count"], len(catalog["scenes"])) office = next(scene for scene in catalog["scenes"] if scene["id"] == "office") - self.assertIn(office["origin"], {"preset", "customized"}) + self.assertIn(office["origin"], {"preset", "customized", "n7_mqtt"}) + if office["origin"] == "n7_mqtt": + self.assertEqual(office["source_topic"], "from_n7/scenes") self.assertTrue(office["actions"]) self.assertIn("capability", office["actions"][0]) self.assertNotIn("command_flow", json.dumps(catalog, ensure_ascii=False)) @@ -306,6 +314,20 @@ class SeatWebApiTests(unittest.TestCase): self.assertEqual(response.headers["Access-Control-Allow-Origin"], "*") self.assertIn("GET", response.headers["Access-Control-Allow-Methods"]) + def test_mode_profile_sync_status_is_lightweight_and_authenticated(self) -> None: + with self.opener.open( + f"{self.base_url}/api/mode-profiles/status", + timeout=3, + ) as response: + payload = json.loads(response.read().decode("utf-8")) + + self.assertEqual(response.status, 200) + self.assertIn("catalog_revision", payload) + self.assertIn("scene_count", payload) + self.assertIn("mode_scene_ingress", payload) + self.assertIn("imported_count", payload["mode_scene_ingress"]) + self.assertNotIn("modes", payload) + def test_user_mode_delete_endpoint_removes_only_user_profiles(self) -> None: original_registry = self.service.mode_profiles with tempfile.TemporaryDirectory() as directory: @@ -932,14 +954,14 @@ operations: vite_entry = (web_console_root / "src" / "main.js").read_text( encoding="utf-8" ) - with self.opener.open(f"{self.base_url}/app.js?v=dual-seat-141", timeout=3) as response: + with self.opener.open(f"{self.base_url}/app.js?v=dual-seat-145-n7-dispatch", timeout=3) as response: app_script = response.read().decode("utf-8") with self.opener.open( f"{self.base_url}/gesture-control.js?v=android-pad-9", timeout=3 ) as response: gesture_script = response.read().decode("utf-8") with self.opener.open( - f"{self.base_url}/console-v2.js?v=local-first-76", timeout=3 + f"{self.base_url}/console-v2.js?v=local-first-77-performance", timeout=3 ) as response: console_script = response.read().decode("utf-8") with self.opener.open( @@ -959,7 +981,7 @@ operations: ) as response: speech_worklet = response.read().decode("utf-8") with self.opener.open( - f"{self.base_url}/self-extension.js?v=pet-settings-52", timeout=3 + f"{self.base_url}/self-extension.js?v=pet-settings-53-performance", timeout=3 ) as response: extension_script = response.read().decode("utf-8") with self.opener.open( @@ -971,7 +993,7 @@ operations: ) as response: assistant_script = response.read().decode("utf-8") with self.opener.open( - f"{self.base_url}/energy-monitor.js?v=energy-monitor-4", timeout=3 + f"{self.base_url}/energy-monitor.js?v=energy-monitor-5-performance", timeout=3 ) as response: energy_script = response.read().decode("utf-8") with self.opener.open( @@ -1018,10 +1040,10 @@ operations: self.assertIn("CMD 0x431 · FB 0x41A · CH1", template) self.assertIn("CMD 0x431 · FB 0x388 · CH1", template) self.assertNotIn("透明车体 · 实体设备", template) - self.assertIn('src="/app.js?v=dual-seat-141"', html) + self.assertIn('src="/app.js?v=dual-seat-145-n7-dispatch"', html) self.assertIn('import "../styles.css";', vite_entry) self.assertNotIn('class="vehicle-system-hud"', template) - self.assertIn('src="/console-v2.js?v=local-first-76"', html) + self.assertIn('src="/console-v2.js?v=local-first-77-performance"', html) self.assertIn('import "../console-v2.css";', vite_entry) self.assertIn('src="/speech-input.js?v=sensevoice-13"', html) self.assertIn('import "../speech-input.css";', vite_entry) @@ -1030,7 +1052,7 @@ operations: self.assertIn('', html) self.assertIn('id="v2BootScreen"', template) self.assertIn('正在加载车机控制与实时状态', template) - self.assertIn('src="/self-extension.js?v=pet-settings-52"', html) + self.assertIn('src="/self-extension.js?v=pet-settings-53-performance"', html) self.assertIn('import "../self-extension.css";', vite_entry) self.assertNotIn("self-calibration", template) self.assertNotIn("self-repair", template) @@ -1041,7 +1063,7 @@ operations: self.assertIn('import "../assistant-chat.css";', vite_entry) self.assertIn('src="/assistant-chat.js?v=codex-assistant-34"', html) self.assertIn('import "../energy-monitor.css";', vite_entry) - self.assertIn('src="/energy-monitor.js?v=energy-monitor-4"', html) + self.assertIn('src="/energy-monitor.js?v=energy-monitor-5-performance"', html) self.assertIn('import "../codepen-glass-theme.css";', vite_entry) self.assertEqual(extension_control_status["source_label"], "车端 MQTT") self.assertIn("action_statuses", extension_control_status) @@ -1331,16 +1353,16 @@ operations: self.assertIn('canvas.addEventListener("webglcontextlost"', app_script) self.assertIn('"cognitive-agent:module-change"', app_script) self.assertIn( - 'const CACHE_VERSION = "cognitive-agent-pad-v317-nfc-auth-routing"', + 'const CACHE_VERSION = "cognitive-agent-pad-v321-n7-dispatch"', service_worker, ) self.assertIn('/console-v2.css?v=delivery-state-69', service_worker) - self.assertIn('/console-v2.js?v=local-first-76', service_worker) + self.assertIn('/console-v2.js?v=local-first-77-performance', service_worker) self.assertIn('/assistant-chat.css?v=codex-assistant-29', service_worker) self.assertIn('/assistant-chat.js?v=codex-assistant-34', service_worker) self.assertIn('/energy-monitor.css?v=energy-monitor-4', service_worker) - self.assertIn('/energy-monitor.js?v=energy-monitor-4', service_worker) - self.assertIn('"/app.js?v=dual-seat-141"', service_worker) + self.assertIn('/energy-monitor.js?v=energy-monitor-5-performance', service_worker) + self.assertIn('"/app.js?v=dual-seat-145-n7-dispatch"', service_worker) self.assertIn('"/speech-input.js?v=sensevoice-13"', service_worker) self.assertIn('"/speech-input.css?v=sensevoice-7"', service_worker) self.assertIn('"/speech-audio-worklet.js?v=sensevoice-5"', service_worker) @@ -1800,7 +1822,7 @@ operations: ) as response: extension_script = response.read().decode("utf-8") with self.opener.open( - f"{self.base_url}/app.js?v=dual-seat-141", timeout=3 + f"{self.base_url}/app.js?v=dual-seat-145-n7-dispatch", timeout=3 ) as response: app_script = response.read().decode("utf-8") diff --git a/tests/unit/test_assistant_capability_catalog.py b/tests/unit/test_assistant_capability_catalog.py index bd211ec..1ed6108 100644 --- a/tests/unit/test_assistant_capability_catalog.py +++ b/tests/unit/test_assistant_capability_catalog.py @@ -16,8 +16,13 @@ from cognitive_agent.generated.can_semantic_protocol import ( @pytest.fixture(scope="module") -def vehicle_service() -> SeatDemoService: - return SeatDemoService() +def vehicle_service(tmp_path_factory: pytest.TempPathFactory) -> SeatDemoService: + service = SeatDemoService() + service.mode_profiles = type(service.mode_profiles)( + service.root / "config" / "modes", + user_root=tmp_path_factory.mktemp("assistant-capability-modes"), + ) + return service def _catalog( @@ -269,7 +274,16 @@ def test_all_modes_are_discoverable_and_reference_callable_capabilities( "zero_gravity", } assert all(item["assistant_callable"] is True for item in modes) - assert all(set(item["operation_ids"]) <= callable_ids for item in modes) + assert all( + set(item["operation_ids"]) + <= callable_ids | set(item.get("n7_transfer_operation_ids", [])) + for item in modes + ) + assert all( + not item.get("n7_transfer_operation_ids") + or item.get("origin") == "n7_mqtt" + for item in modes + ) assert ( next(item for item in modes if item["id"] == "baby_nap")[ "all_operations_have_native_binding" diff --git a/tests/unit/test_assistant_frontend_contract.py b/tests/unit/test_assistant_frontend_contract.py index a60303b..4877c58 100644 --- a/tests/unit/test_assistant_frontend_contract.py +++ b/tests/unit/test_assistant_frontend_contract.py @@ -27,7 +27,7 @@ def test_console_contains_ai_assistant_surface(): def test_codex_assistant_assets_are_versioned_together_in_the_service_worker(): service_worker = (ROOT / "apps/web-console/service-worker.js").read_text(encoding="utf-8") assert ( - 'const CACHE_VERSION = "cognitive-agent-pad-v317-nfc-auth-routing"' + 'const CACHE_VERSION = "cognitive-agent-pad-v321-n7-dispatch"' in service_worker ) assert '"/assistant-chat.css?v=codex-assistant-29"' in service_worker diff --git a/tests/unit/test_energy_monitor_frontend_contract.py b/tests/unit/test_energy_monitor_frontend_contract.py index e3673f2..959cc6a 100644 --- a/tests/unit/test_energy_monitor_frontend_contract.py +++ b/tests/unit/test_energy_monitor_frontend_contract.py @@ -32,16 +32,16 @@ def test_energy_monitor_assets_are_versioned_and_cached() -> None: for asset in ( "/energy-monitor.css?v=energy-monitor-4", - "/energy-monitor.js?v=energy-monitor-4", + "/energy-monitor.js?v=energy-monitor-5-performance", ): assert f'"{asset}"' in service_worker assert 'import "../energy-monitor.css";' in vite_entry - assert "/energy-monitor.js?v=energy-monitor-4" in html + assert "/energy-monitor.js?v=energy-monitor-5-performance" in html assert html.index("/assistant-chat.js?v=codex-assistant-34") < html.index( - "/energy-monitor.js?v=energy-monitor-4" + "/energy-monitor.js?v=energy-monitor-5-performance" ) assert ( - 'const CACHE_VERSION = "cognitive-agent-pad-v316-sidebar-accordion"' + 'const CACHE_VERSION = "cognitive-agent-pad-v321-n7-dispatch"' in service_worker ) @@ -140,7 +140,7 @@ def test_energy_monitor_samples_integrates_and_bounds_its_curves() -> None: assert 'id="energyMonitorChartLine"' in script assert 'id="energyMonitorChartArea"' in script assert "polylinePoints(values, maximum" in script - assert 'window.setInterval(() => sampleNow(), SAMPLE_INTERVAL_MS)' in script + assert '() => sampleNow({ render: monitor.open })' in script assert 'ingestPowerTelemetry(event.detail || {}, { sample: false })' in script assert 'ingestPowerTelemetry(payload, { sample: false })' in script assert 'new CustomEvent("cognitive-agent:energy-snapshot"' in script diff --git a/tests/unit/test_frontend_performance_contract.py b/tests/unit/test_frontend_performance_contract.py new file mode 100644 index 0000000..562a87c --- /dev/null +++ b/tests/unit/test_frontend_performance_contract.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[2] +WEB = ROOT / "apps" / "web-console" + + +def source(name: str) -> str: + return (WEB / name).read_text(encoding="utf-8") + + +def section(content: str, start: str, end: str) -> str: + start_index = content.index(start) + end_index = content.index(end, start_index) + return content[start_index:end_index] + + +def test_vehicle_twin_renders_on_demand_and_throttles_hotspot_layout() -> None: + app = source("app.js") + animation = section(app, " function animate(timestamp = 0) {", " setTwinSelection(") + + assert "function requestRender({ hotspots = true, settleFrames = 36 } = {})" in app + assert "if (twinNeedsAnimation(twin)) queueTwinFrame();" in animation + assert "window.requestAnimationFrame(animate)" not in animation + assert "document.visibilityState !== \"visible\"" in app + assert "new IntersectionObserver" in app + assert "timestamp - twin.hotspotLastLayoutAt >= 100" in animation + assert "hotspotMetrics: new Map()" in app + assert "label?.offsetWidth" in app + assert "renderer.setPixelRatio(Math.min(window.devicePixelRatio || 1, 1.5))" in app + + +def test_unchanged_vehicle_state_skips_dom_and_device_panel_rebuilds() -> None: + app = source("app.js") + console = source("console-v2.js") + seat_render = section(app, "function renderSeatState(actuatorState) {", "\nfunction renderDomainState") + + assert "vehicleRenderFingerprint(mergedActuatorState)" in seat_render + assert "renderFingerprint === state.seatRenderFingerprint" in seat_render + assert "resetSeatState();" not in seat_render + assert 'new CustomEvent("cognitive-agent:vehicle-state-rendered"' in app + assert '"cognitive-agent:vehicle-state-rendered"' in console + assert "observe(cabinBand" not in console + assert "observe(stateBand" not in console + assert 'if (!panelVisible("devices"))' in console + assert "renderSyncedDevices({ force: true })" in console + + +def test_nfc_polling_keeps_its_frequency_without_rendering_hidden_surfaces() -> None: + extension = source("self-extension.js") + + assert "const NFC_POLL_INTERVAL_MS = 700" in extension + assert "currentNfcDialogFingerprint()" in extension + assert "if (!force && !dialog.open)" in extension + assert "renderFingerprint === nfcDialogFingerprint" in extension + assert 'panel?.classList.contains("is-collapsed")' in extension + assert "renderFingerprint === unifiedExtensionFingerprint" in extension + assert '"cognitive-agent:side-panel-visibility"' in extension + + +def test_hidden_energy_monitor_samples_without_repainting() -> None: + monitor = source("energy-monitor.js") + + assert "SAMPLE_INTERVAL_MS = 1000" in monitor + assert "() => sampleNow({ render: monitor.open })" in monitor + assert "if (next)" in monitor + assert "renderMonitor();" in section(monitor, " function setOpen(open) {", "\n function selectSeries") + + +def test_high_frequency_fields_and_hotspots_do_not_add_compositor_blur() -> None: + theme = source("codepen-glass-theme.css") + field_rule = section( + theme, + "body.console-v2 :is(input, textarea, select),", + "body.console-v2 :is(input, textarea, select):focus", + ) + + assert "backdrop-filter" not in field_rule + assert "body.console-v2 .v2-twin-device-popover {" in theme + shared_hotspot_rule = section( + theme, + "body.console-v2 .v2-twin-hotspot-label,", + "body.console-v2 .v2-twin-device-popover {\n -webkit-backdrop-filter", + ) + assert "backdrop-filter" not in shared_hotspot_rule + + +def test_performance_assets_have_a_new_cache_generation() -> None: + html = source("index.html") + worker = source("service-worker.js") + + for asset in ( + "/app.js?v=dual-seat-145-n7-dispatch", + "/console-v2.js?v=local-first-77-performance", + "/self-extension.js?v=pet-settings-53-performance", + "/energy-monitor.js?v=energy-monitor-5-performance", + ): + assert asset in html + assert f'"{asset}"' in worker + assert 'const CACHE_VERSION = "cognitive-agent-pad-v321-n7-dispatch"' in worker + assert '"/codepen-glass-theme.css?v=creative-cloud-6-performance"' in worker diff --git a/tests/unit/test_frontend_theme_contract.py b/tests/unit/test_frontend_theme_contract.py index 6f6b2b6..239d0e6 100644 --- a/tests/unit/test_frontend_theme_contract.py +++ b/tests/unit/test_frontend_theme_contract.py @@ -47,12 +47,12 @@ def test_theme_asset_versions_are_consistent_with_the_pwa_shell() -> None: ('import "../console-v2.css"', "/console-v2.css?v=delivery-state-69"), ('import "../self-extension.css"', "/self-extension.css?v=pet-settings-52"), ('import "../speech-input.css"', "/speech-input.css?v=sensevoice-7"), - ('import "../codepen-glass-theme.css"', "/codepen-glass-theme.css?v=creative-cloud-5"), + ('import "../codepen-glass-theme.css"', "/codepen-glass-theme.css?v=creative-cloud-6-performance"), ): assert source_asset in main assert f'"{cached_asset}"' in service_worker assert 'import "./styles/design-system.css"' in main - assert 'const CACHE_VERSION = "cognitive-agent-pad-v316-sidebar-accordion"' in service_worker + assert 'const CACHE_VERSION = "cognitive-agent-pad-v321-n7-dispatch"' in service_worker def test_reference_glass_theme_covers_the_complete_workspace() -> None: diff --git a/tests/unit/test_mode_card_dispatch_integrity.py b/tests/unit/test_mode_card_dispatch_integrity.py new file mode 100644 index 0000000..f50b1ea --- /dev/null +++ b/tests/unit/test_mode_card_dispatch_integrity.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from cognitive_agent.application.mode_profile import ModeProfileRegistry +from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService + + +def test_all_repository_mode_cards_preserve_every_operation_in_dry_run( + tmp_path: Path, +) -> None: + """Every visible preset must reach a validated executable configuration.""" + + service = SeatDemoService(REPOSITORY_ROOT) + service.mode_profiles = ModeProfileRegistry( + REPOSITORY_ROOT / "config" / "modes", + user_root=tmp_path / "user-modes", + ) + service.artifact_root = tmp_path / "runs" + vehicle_id = "denza_n7" + failures: list[dict[str, object]] = [] + + for card in service.mode_profiles.public_modes(vehicle_id): + profile = service.mode_profiles.get_profile(str(card["id"]), vehicle_id) + result = service.process( + profile.instruction or f"进入{profile.name}", + provider="local", + conversation_id=f"conv-mode-integrity-{profile.id.replace('_', '-')}", + execution_mode="dry_run", + vehicle_id=vehicle_id, + control_transport="wireless_mqtt", + ) + compiled_operations = [ + operation + for config in result.get("function_configs", {}).values() + for operation in config.get("spec", {}).get("operations", []) + ] + validation_errors = [ + issue + for issue in result.get("validation", []) + if issue.get("severity") == "error" + ] + if ( + result.get("status") != "succeeded" + or validation_errors + or result.get("deferred_requests") + or len(compiled_operations) != len(profile.operations) + ): + failures.append( + { + "id": profile.id, + "expected_operations": len(profile.operations), + "compiled_operations": len(compiled_operations), + "status": result.get("status"), + "deferred_requests": result.get("deferred_requests"), + "validation_errors": validation_errors, + } + ) + + assert not failures, json.dumps(failures, ensure_ascii=False, indent=2) diff --git a/tests/unit/test_mode_gallery_grouping.py b/tests/unit/test_mode_gallery_grouping.py index cf36abd..9e4b8de 100644 --- a/tests/unit/test_mode_gallery_grouping.py +++ b/tests/unit/test_mode_gallery_grouping.py @@ -310,6 +310,11 @@ const context = { setModeRailSelection(mode) { calls.push(["select", mode?.id]); }, closeModeGallery(options) { calls.push(["close", options.restoreFocus]); }, showToast(message) { calls.push(["toast", message]); }, + hasExecutableModeCommandFlow(mode) { + const flow = mode?.command_flow; + return Boolean(flow?.source_request_id) + || Boolean(flow?.plans && Object.keys(flow.plans).length); + }, executeSavedModeProfile(mode, options) { calls.push(["execute", mode.id, options.command]); return Promise.resolve(true); @@ -346,6 +351,42 @@ console.log(JSON.stringify({ assert "executeSavedModeProfile" in snippet +def test_n7_mode_command_flow_is_directly_executable_without_source_request_id() -> None: + app = (WEB / "app.js").read_text(encoding="utf-8") + start = app.index("function hasExecutableModeCommandFlow") + end = app.index("\n\nasync function executeSavedModeProfile", start) + helper = app[start:end] + script = helper + r""" +console.log(JSON.stringify({ + imported: hasExecutableModeCommandFlow({ + id: "cinema", + command_flow: { + source_topic: "from_n7/scenes", + plans: { from_n7_scene: { spec: { commands: [] } } }, + }, + }), + generated: hasExecutableModeCommandFlow({ + id: "reading-user", + command_flow: { source_request_id: "req-reading" }, + }), + missing: hasExecutableModeCommandFlow({ id: "office" }), +})); +""" + completed = subprocess.run( + [node_runtime(), "-e", script], + check=True, + capture_output=True, + text=True, + encoding="utf-8", + ) + + assert json.loads(completed.stdout) == { + "imported": True, + "generated": True, + "missing": False, + } + + def test_mode_card_without_saved_flow_falls_back_to_assistant_submission() -> None: app = (WEB / "app.js").read_text(encoding="utf-8") start = app.index("function applyModeCommand") @@ -378,6 +419,11 @@ const context = { setModeExecutionState(modeId, status) { calls.push(["state", modeId, status]); }, closeModeGallery(options) { calls.push(["close", options.restoreFocus]); }, showToast(message) { calls.push(["toast", message]); }, + hasExecutableModeCommandFlow(mode) { + const flow = mode?.command_flow; + return Boolean(flow?.source_request_id) + || Boolean(flow?.plans && Object.keys(flow.plans).length); + }, $(selector) { return { "#utterance": input, "#requestForm": form, "#runButton": runButton }[selector]; }, diff --git a/tests/unit/test_mode_scene_ingress.py b/tests/unit/test_mode_scene_ingress.py new file mode 100644 index 0000000..3fbfd51 --- /dev/null +++ b/tests/unit/test_mode_scene_ingress.py @@ -0,0 +1,555 @@ +from __future__ import annotations + +import json +import time +from copy import deepcopy +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from cognitive_agent.application.assistant_capabilities import ( + AssistantCapabilityCatalog, +) +from cognitive_agent.application.mode_profile import ModeProfileRegistry +from cognitive_agent.application.mode_scene_ingress import ( + ModeSceneIngressConsumer, + ModeSceneIngressError, + parse_mode_scene_ingress_payload, +) +from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService + + +def scene_card(*, temperature: int = 24) -> dict[str, object]: + return { + "id": "n7-reading", + "label": "N7 阅读模式", + "name": "N7 阅读模式", + "background_time": "evening", + "description": "由车端同步的阅读场景。", + "icon": "book-open", + "aliases": ["N7 阅读模式", "车端阅读"], + "origin": "user", + "deletable": True, + "behavior": "static", + "domains": ["climate"], + "operation_count": 1, + "scope": {"common": False, "vehicle_ids": ["denza_n7"]}, + "profile_revision": f"remote-{temperature}", + "actions": [ + { + "type": "climate_temperature", + "at_ms": 250, + "domain": "climate", + "capability": "climate.set_temperature", + "target_scope": "passenger", + "operation": "set", + "parameters": {"temperature_c": temperature}, + "zone": "passenger", + "celsius": temperature, + } + ], + } + + +def cinema_scene_card() -> dict[str, object]: + """Representative N7 card containing local and head-unit-only actions.""" + + actions: list[dict[str, object]] = [ + *[ + { + "type": "window_position", + "at_ms": 0, + "domain": "window", + "capability": "window.set_position", + "target_scope": target, + "operation": "set", + "parameters": {"position_pct": 0}, + } + for target in ("front_left", "front_right", "rear_left", "rear_right") + ], + { + "type": "roof_shade_position", + "at_ms": 0, + "domain": "roof", + "capability": "roof.set_shade_position", + "target_scope": "vehicle", + "operation": "set", + "parameters": {"position_pct": 0}, + }, + { + "type": "seat_preset", + "at_ms": 0, + "domain": "local_vehicle", + "capability": "local_scene.seat_preset", + "target_scope": "vehicle", + "operation": "set", + "parameters": {"targets_pulse": [246, 112, 405, 342]}, + }, + { + "type": "seat_ventilation", + "at_ms": 0, + "domain": "seat", + "capability": "seat.set_ventilation", + "target_scope": "passenger", + "operation": "set", + "parameters": {"state": "off"}, + }, + { + "type": "climate_power", + "at_ms": 0, + "domain": "climate", + "capability": "climate.set_power", + "target_scope": "vehicle", + "operation": "set", + "parameters": {"state": "on"}, + }, + { + "type": "climate_fan", + "at_ms": 0, + "domain": "climate", + "capability": "climate.set_fan_level", + "target_scope": "vehicle", + "operation": "set", + "parameters": {"level": 2}, + }, + { + "type": "climate_outlet_mode", + "at_ms": 0, + "domain": "climate", + "capability": "climate.set_outlet_mode", + "target_scope": "vehicle", + "operation": "set", + "parameters": {"mode": "face"}, + }, + { + "type": "climate_cycle_mode", + "at_ms": 0, + "domain": "climate", + "capability": "climate.set_cycle_mode", + "target_scope": "vehicle", + "operation": "set", + "parameters": {"mode": "internal"}, + }, + { + "type": "climate_control_mode", + "at_ms": 0, + "domain": "climate", + "capability": "climate.set_control_mode", + "target_scope": "vehicle", + "operation": "set", + "parameters": {"mode": "manual"}, + }, + { + "type": "climate_temperature", + "at_ms": 0, + "domain": "climate", + "capability": "climate.set_temperature", + "target_scope": "vehicle", + "operation": "set", + "parameters": {"temperature_c": 25}, + }, + { + "type": "interior_light_reading_power", + "at_ms": 0, + "domain": "interior_light", + "capability": "interior_light.set_reading_power", + "target_scope": "all", + "operation": "set", + "parameters": {"state": "off"}, + }, + { + "type": "ambient_light_power", + "at_ms": 0, + "domain": "ambient_light", + "capability": "ambient_light.set_power", + "target_scope": "vehicle", + "operation": "set", + "parameters": {"state": "on"}, + }, + { + "type": "ambient_light_color", + "at_ms": 0, + "domain": "ambient_light", + "capability": "ambient_light.set_color_index", + "target_scope": "vehicle", + "operation": "set", + "parameters": {"color_index": 36}, + }, + { + "type": "ambient_light_brightness", + "at_ms": 0, + "domain": "ambient_light", + "capability": "ambient_light.set_brightness_level", + "target_scope": "vehicle", + "operation": "set", + "parameters": {"brightness_level": 4}, + }, + ] + return { + "id": "cinema", + "label": "Cinema mode", + "name": "Cinema mode", + "background_time": "night", + "description": "Representative N7 mixed-domain scene.", + "icon": "bookmark-plus", + "aliases": ["Cinema mode"], + "origin": "headunit", + "deletable": True, + "behavior": "static", + "domains": list(dict.fromkeys(str(item["domain"]) for item in actions)), + "operation_count": len(actions), + "scope": {"common": False, "vehicle_ids": ["denza_n7"]}, + "actions": actions, + "profile_revision": "cinema-test-revision", + } + + +def wait_until(predicate, timeout: float = 1.5) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(0.01) + return bool(predicate()) + + +def configured_service(tmp_path: Path) -> SeatDemoService: + service = SeatDemoService(REPOSITORY_ROOT) + service.mode_profiles = ModeProfileRegistry( + tmp_path / "presets", + user_root=tmp_path / "profiles", + ) + service.artifact_root = tmp_path / "runs" + return service + + +def test_parser_uses_exact_single_card_contract_and_rejects_duplicate_keys() -> None: + scene = scene_card() + parsed = parse_mode_scene_ingress_payload( + json.dumps(scene, ensure_ascii=False).encode("utf-8") + ) + assert parsed == scene + + duplicate = ( + '{"id":"one","id":"two","label":"x","name":"x",' + '"background_time":"noon","description":"",' + '"icon":"x","aliases":[],"origin":"user",' + '"deletable":true,"behavior":"static","domains":["x"],' + '"operation_count":1,"scope":{},"profile_revision":"x",' + '"actions":[]}' + ) + with pytest.raises(ModeSceneIngressError, match="duplicate key"): + parse_mode_scene_ingress_payload(duplicate) + + +def test_three_identical_frames_are_imported_once() -> None: + imports: list[dict[str, object]] = [] + consumer = ModeSceneIngressConsumer( + lambda scene: imports.append(dict(scene)) or {"status": "imported"}, + deduplication_window_s=1.0, + ) + payload = json.dumps(scene_card(), ensure_ascii=False).encode("utf-8") + consumer.start() + try: + for _ in range(3): + consumer.submit(payload) + assert wait_until(lambda: consumer.snapshot()["processed_count"] == 3) + assert imports == [scene_card()] + status = consumer.snapshot() + assert status["imported_count"] == 1 + assert status["duplicate_count"] == 2 + assert status["rejected_count"] == 0 + finally: + consumer.close() + + +def test_failed_import_does_not_suppress_identical_retry() -> None: + importer = MagicMock( + side_effect=[ + RuntimeError("temporary persistence failure"), + {"status": "imported"}, + ] + ) + consumer = ModeSceneIngressConsumer( + importer, + deduplication_window_s=1.0, + ) + payload = json.dumps(scene_card(), ensure_ascii=False).encode("utf-8") + consumer.start() + try: + consumer.submit(payload) + assert wait_until(lambda: consumer.snapshot()["processed_count"] == 1) + first_status = consumer.snapshot() + assert first_status["rejected_count"] == 1 + assert first_status["imported_count"] == 0 + + consumer.submit(payload) + assert wait_until(lambda: consumer.snapshot()["processed_count"] == 2) + retry_status = consumer.snapshot() + assert retry_status["rejected_count"] == 1 + assert retry_status["imported_count"] == 1 + assert retry_status["duplicate_count"] == 0 + assert importer.call_count == 2 + + consumer.submit(payload) + assert wait_until(lambda: consumer.snapshot()["processed_count"] == 3) + final_status = consumer.snapshot() + assert final_status["duplicate_count"] == 1 + assert importer.call_count == 2 + finally: + consumer.close() + + +def test_import_upserts_yaml_refreshes_catalog_and_does_not_echo( + tmp_path: Path, +) -> None: + service = configured_service(tmp_path) + service.configure_mode_scene_catalog(tmp_path / "scenes.json") + outbound = MagicMock() + service._vehicle_mqtt_transport.publish_mode_scene = outbound + + first = service.import_n7_mode_scene(scene_card(temperature=24)) + updated = service.import_n7_mode_scene(scene_card(temperature=26)) + + assert first["scene_id"] == "n7-reading" + assert updated["scene_id"] == "n7-reading" + profiles = service.mode_profiles.public_modes("denza_n7") + assert len(profiles) == 1 + profile = profiles[0] + assert profile["origin"] == "n7_mqtt" + assert profile["source_topic"] == "from_n7/scenes" + assert profile["upstream_origin"] == "user" + assert profile["operations"][0]["parameters"] == {"temperature_c": 26} + + catalog = service.mode_scene_catalog("denza_n7") + assert catalog["scene_count"] == 1 + card = catalog["scenes"][0] + assert card["origin"] == "n7_mqtt" + assert card["source_topic"] == "from_n7/scenes" + assert card["actions"] == scene_card(temperature=26)["actions"] + assert card["actions"][0]["at_ms"] == 250 + assert outbound.call_count == 0 + + yaml_path = Path(service.mode_profiles.get_profile("n7-reading", "denza_n7").source) + yaml_text = yaml_path.read_text(encoding="utf-8") + assert "sync_origin: from_n7/scenes" in yaml_text + assert "upstream_origin: user" in yaml_text + + +def test_unknown_n7_action_is_forwarded_once_only_during_hardware_execution( + tmp_path: Path, +) -> None: + service = configured_service(tmp_path) + unknown = scene_card() + unknown.update( + domains=["media"], + operation_count=1, + actions=[ + { + "type": "media_play_favorite", + "at_ms": 0, + "domain": "media", + "capability": "media.play_favorite", + "target_scope": "vehicle", + "operation": "play", + "parameters": {"playlist": "focus"}, + } + ], + ) + service.import_n7_mode_scene(unknown) + publisher = MagicMock( + return_value={ + "topic": "transto_n7/ctl", + "qos": 1, + "retain": False, + "delivery_status": "published", + "message_id": 7, + } + ) + service._vehicle_mqtt_transport.publish_n7_control_transfer = publisher + + dry_run = service.execute_mode_profile_card( + "n7-reading", + execution_mode="dry_run", + control_transport="wireless_mqtt", + ) + assert dry_run["status"] == "capability_unavailable" + publisher.assert_not_called() + + result = service.execute_mode_profile_card( + "n7-reading", + execution_mode="hardware", + control_transport="wireless_mqtt", + ) + + assert result["status"] == "capability_unavailable" + assert result["n7_control_transfer"]["state"] == "published" + publisher.assert_called_once() + payload = publisher.call_args.args[0] + assert payload == { + "scene_id": "n7-reading", + "scene_name": "N7 阅读模式", + "request_id": result["request_id"], + "operation_count": 1, + "actions": deepcopy(unknown["actions"]), + } + assert result["deferred_requests"][0]["transport_submitted"] is False + + +def test_mixed_n7_scene_compiles_every_local_action_and_preserves_transfer( + tmp_path: Path, +) -> None: + service = configured_service(tmp_path) + scene = cinema_scene_card() + service.import_n7_mode_scene(scene) + + result = service.execute_mode_profile_card( + "cinema", + execution_mode="dry_run", + control_transport="wireless_mqtt", + ) + + assert result["status"] == "partially_succeeded" + assert not [ + issue + for issue in result["validation"] + if issue.get("severity") == "error" + ] + compiled_operations = [ + operation + for config in result["function_configs"].values() + for operation in config["spec"]["operations"] + ] + assert len(compiled_operations) == 16 + assert all( + operation.get("parse_source") == "saved_mode_command_flow" + for operation in compiled_operations + ) + assert { + operation["target"]["zone"] + for operation in compiled_operations + if operation["capability"] == "window.set_position" + } == {"front_left", "front_right", "rear_left", "rear_right"} + transferred = scene["actions"][5] + assert result["n7_transfer_candidates"] == [transferred] + assert result["deferred_requests"] == [ + { + "domain": "local_vehicle", + "capability": "local_scene.seat_preset", + "target_scope": "vehicle", + "desired_change": "set", + "parameters": {"targets_pulse": [246, 112, 405, 342]}, + "reason": "当前 PC 未注册该能力,将由 N7 端继续处理。", + "execution_backend": "n7_vehicle_transfer", + "rejection_code": "PC_CAPABILITY_UNAVAILABLE", + "transport_submitted": False, + } + ] + assert len(compiled_operations) + len(result["n7_transfer_candidates"]) == len( + scene["actions"] + ) + + text_result = service.process( + "进入Cinema mode模式", + provider="local", + conversation_id="conv-n7-cinema-text", + execution_mode="dry_run", + vehicle_id="denza_n7", + control_transport="wireless_mqtt", + ) + assert text_result["status"] == "partially_succeeded" + assert text_result["mode_profile"] is None + assert [item["id"] for item in text_result["matched_modes"]] == ["cinema"] + assert len(text_result["n7_transfer_candidates"]) == 1 + assert service.mode_profiles.match_text( + "氛围灯设置为Cinema mode模式", + vehicle_id="denza_n7", + ) == [] + + catalog_mode = next( + mode + for mode in AssistantCapabilityCatalog(service).snapshot("denza_n7")["modes"] + if mode["id"] == "cinema" + ) + assert catalog_mode["assistant_callable"] is True + assert catalog_mode["n7_transfer_operation_ids"] == [ + "local_scene.seat_preset" + ] + + +@pytest.mark.parametrize( + ("domain", "target_scope", "parameters"), + [ + ("media", "passenger", {"temperature_c": 24}), + ("climate", "unsupported_zone", {"temperature_c": 24}), + ("climate", "passenger", {"unsupported_parameter": 24}), + ], +) +def test_registered_but_invalid_action_is_not_forwarded_to_n7( + tmp_path: Path, + domain: str, + target_scope: str, + parameters: dict[str, object], +) -> None: + service = configured_service(tmp_path) + invalid = scene_card() + invalid["actions"] = [ + { + "type": "climate_temperature", + "at_ms": 0, + "domain": domain, + "capability": "climate.set_temperature", + "target_scope": target_scope, + "operation": "set", + "parameters": parameters, + } + ] + service.import_n7_mode_scene(invalid) + profile = service.get_mode_profile_card("n7-reading", "denza_n7") + + local_requests, transfer_actions = service._mode_requests_and_n7_transfers(profile) + + assert len(local_requests) == 1 + assert transfer_actions == [] + + +def test_frontend_marks_n7_synchronized_cards() -> None: + script = (REPOSITORY_ROOT / "apps" / "web-console" / "app.js").read_text( + encoding="utf-8" + ) + styles = ( + REPOSITORY_ROOT / "apps" / "web-console" / "styles.css" + ).read_text(encoding="utf-8") + assert 'mode?.origin === "n7_mqtt"' in script + assert 'sourceBadge.textContent = "N7 同步"' in script + assert ".mode-gallery-source-badge" in styles + + +def test_mode_profile_sync_status_exposes_catalog_and_ingress_counters( + tmp_path: Path, +) -> None: + service = configured_service(tmp_path) + service.configure_mode_scene_catalog(tmp_path / "scenes.json") + + status = service.mode_profile_sync_status() + + assert len(status["catalog_revision"]) == 16 + assert status["scene_count"] == 0 + assert status["catalog_state"] == "ready" + assert status["mode_scene_ingress"]["imported_count"] == 0 + + +def test_frontend_polls_lightweight_status_and_refreshes_on_n7_import() -> None: + script = (REPOSITORY_ROOT / "apps" / "web-console" / "app.js").read_text( + encoding="utf-8" + ) + + assert 'apiRequest("/api/mode-profiles/status"' in script + assert "MODE_PROFILE_SYNC_POLL_INTERVAL_MS = 1000" in script + assert 'document.visibilityState === "hidden"' in script + assert "if (catalogChanged || (imported && !sceneAlreadyVisible))" in script + assert "await loadRuntime()" in script + assert "revealModeProfile(sceneId)" in script + assert "已从 N7 同步模式卡片" in script + assert "N7 模式卡片导入失败" in script diff --git a/tests/unit/test_mqtt_can_transport.py b/tests/unit/test_mqtt_can_transport.py index ce555d3..cf9f74e 100644 --- a/tests/unit/test_mqtt_can_transport.py +++ b/tests/unit/test_mqtt_can_transport.py @@ -19,11 +19,14 @@ from cognitive_agent.infrastructure.can.mqtt_transport import ( MQTT_CAN_SCHEMA, MQTT_EXTENSION_CONTROL_TOPIC, MQTT_EXTENSION_TOY_TOPIC, + MQTT_MODE_SCENE_INGRESS_TOPIC, MQTT_MODE_SCENE_TOPIC, + MQTT_N7_CONTROL_TRANSFER_TOPIC, MQTT_PET_TOPIC, MqttCanTransport, mqtt_extension_toy_payload, mqtt_mode_scene_payload, + mqtt_n7_control_transfer_payload, mqtt_pet_payload, parse_mqtt_can_payload, ) @@ -739,6 +742,82 @@ class MqttCanTransportTests(unittest.TestCase): } ) + def test_n7_mode_scene_ingress_dispatches_raw_bytes_and_rejects_retained( + self, + ) -> None: + received: list[bytes] = [] + self.transport.set_mode_scene_ingress_handler(received.append) + scene = { + "id": "n7-reading", + "label": "N7 阅读模式", + "name": "N7 阅读模式", + "background_time": "noon", + "description": "车端同步模式。", + "icon": "book-open", + "aliases": ["N7 阅读模式"], + "origin": "user", + "deletable": True, + "behavior": "static", + "domains": ["interior_light"], + "operation_count": 1, + "scope": {"common": False, "vehicle_ids": ["denza_n7"]}, + "profile_revision": "remote-revision", + "actions": [ + { + "type": "interior_light_reading_power", + "at_ms": 0, + "domain": "interior_light", + "capability": "interior_light.set_reading_power", + "target_scope": "all", + "operation": "set", + "parameters": {"state": "on"}, + } + ], + } + + self.client.emit(MQTT_MODE_SCENE_INGRESS_TOPIC, scene) + self.client.emit(MQTT_MODE_SCENE_INGRESS_TOPIC, scene, retain=True) + + self.assertEqual(len(received), 1) + self.assertEqual(json.loads(received[0]), scene) + status = self.transport.public_info()["mode_scene_ingress"] + self.assertTrue(status["handler_configured"]) + self.assertEqual(status["received_count"], 2) + self.assertEqual(status["dispatched_count"], 1) + self.assertEqual(status["rejected_count"], 1) + self.assertEqual(status["retained_count"], 1) + + def test_n7_control_transfer_uses_fixed_qos_one_topic_and_never_retains( + self, + ) -> None: + payload = { + "scene_id": "n7-reading", + "scene_name": "N7 阅读模式", + "request_id": "req-20260916000000-1234567890abcdef", + "operation_count": 1, + "actions": [ + { + "type": "n7_media", + "at_ms": 0, + "domain": "media", + "capability": "media.play_favorite", + "target_scope": "vehicle", + "operation": "play", + "parameters": {"playlist": "focus"}, + } + ], + } + + publication = self.transport.publish_n7_control_transfer(payload) + + topic, raw_payload, qos, retain = self.client.published[-1] + self.assertEqual(topic, MQTT_N7_CONTROL_TRANSFER_TOPIC) + self.assertEqual(raw_payload, mqtt_n7_control_transfer_payload(payload)) + self.assertEqual(json.loads(raw_payload), payload) + self.assertEqual(qos, 1) + self.assertFalse(retain) + self.assertEqual(publication["delivery_status"], "published") + def test_condition_frames_use_existing_decoder_and_snapshot_contract(self) -> None: self.client.emit( "n7/n7-headunit/can/state", @@ -775,6 +854,7 @@ class MqttCanTransportTests(unittest.TestCase): [ ("n7/n7-headunit/can/state", 1), (MQTT_EXTENSION_CONTROL_TOPIC, 1), + (MQTT_MODE_SCENE_INGRESS_TOPIC, 1), ], ) diff --git a/tests/unit/test_nfc_frontend_contract.py b/tests/unit/test_nfc_frontend_contract.py index 64834f1..bc0c483 100644 --- a/tests/unit/test_nfc_frontend_contract.py +++ b/tests/unit/test_nfc_frontend_contract.py @@ -61,7 +61,7 @@ def test_unified_console_exposes_device_id_based_nfc_authorization() -> None: assert '.v2-nfc-command-list' in styles assert '.v2-nfc-command-option' in styles assert '.v2-nfc-select-all' in styles - assert 'self-extension.js?v=pet-settings-52' in html + assert 'self-extension.js?v=pet-settings-53-performance' in html assert 'import "../self-extension.css";' in vite_entry @@ -209,5 +209,5 @@ def test_main_parsed_control_panel_renders_validated_infrared_operations() -> No assert "下发控制指令" in console assert "已解析控制指令" not in console assert 'li[data-can-channel="IR"]' in styles - assert "console-v2.js?v=local-first-76" in html + assert "console-v2.js?v=local-first-77-performance" in html assert 'speech_semantic: "语音语义"' in console diff --git a/tests/unit/test_start_n7web.py b/tests/unit/test_start_n7web.py new file mode 100644 index 0000000..7c0c34d --- /dev/null +++ b/tests/unit/test_start_n7web.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +import socket +from types import SimpleNamespace + +import pytest + +from tools import start_n7web + + +def test_tcp_port_probe_detects_an_existing_listener() -> None: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener: + listener.bind(("127.0.0.1", 0)) + listener.listen(1) + + assert start_n7web._tcp_port_is_listening( # noqa: SLF001 + "0.0.0.0", + listener.getsockname()[1], + ) + + +def test_startup_rejects_a_duplicate_service_port(monkeypatch: pytest.MonkeyPatch) -> None: + args = SimpleNamespace( + host="127.0.0.1", + port=8000, + scene_bind="0.0.0.0", + scene_port=8090, + ) + monkeypatch.setattr( + start_n7web, + "_tcp_port_is_listening", + lambda _host, port: port == 8000, + ) + + with pytest.raises(RuntimeError, match="stop the previous N7Web instance"): + start_n7web._ensure_service_ports_available(args) # noqa: SLF001 diff --git a/tests/unit/test_vue_frontend_contract.py b/tests/unit/test_vue_frontend_contract.py index 8932297..50b6efb 100644 --- a/tests/unit/test_vue_frontend_contract.py +++ b/tests/unit/test_vue_frontend_contract.py @@ -31,7 +31,7 @@ def test_vue_vite_application_is_the_primary_frontend_entry() -> None: assert package["scripts"]["build"] == "vite build" assert 'import vue from "@vitejs/plugin-vue"' in vite assert 'outDir: outputRoot' in vite - assert 'const CACHE_VERSION = "cognitive-agent-pad-v317-nfc-auth-routing"' in service_worker + assert 'const CACHE_VERSION = "cognitive-agent-pad-v321-n7-dispatch"' in service_worker assert "unpkg.com" not in html assert "cdn.jsdelivr.net" not in html diff --git a/tools/start_n7web.py b/tools/start_n7web.py index af025e4..15e6e07 100644 --- a/tools/start_n7web.py +++ b/tools/start_n7web.py @@ -8,6 +8,7 @@ import json import os import shlex import shutil +import socket import subprocess import sys import time @@ -183,6 +184,41 @@ def _display_command(command: Sequence[str]) -> str: return shlex.join(command) +def _probe_host(bind_address: str) -> str: + address = str(bind_address).strip() + if address in {"", "0.0.0.0"}: + return "127.0.0.1" + if address == "::": + return "::1" + return address + + +def _tcp_port_is_listening(bind_address: str, port: int) -> bool: + try: + with socket.create_connection((_probe_host(bind_address), port), timeout=0.25): + return True + except OSError: + return False + + +def _ensure_service_ports_available(args: argparse.Namespace) -> None: + endpoints = ( + ("main N7Web service", args.host, args.port), + ("LAN mode-card service", args.scene_bind, args.scene_port), + ) + occupied = [ + f"{name} ({_probe_host(host)}:{port})" + for name, host, port in endpoints + if _tcp_port_is_listening(host, port) + ] + if occupied: + joined = ", ".join(occupied) + raise RuntimeError( + f"service port already in use: {joined}; stop the previous N7Web " + "instance before starting another" + ) + + def _resolve_npm() -> Path: executable = shutil.which("npm.cmd" if os.name == "nt" else "npm") if not executable: @@ -314,13 +350,13 @@ def run(argv: Sequence[str] | None = None) -> int: if args.dry_run: return 0 - - if npm is not None: - try: + try: + _ensure_service_ports_available(args) + if npm is not None: _build_web_console(npm) - except RuntimeError as exc: - print(f"startup failed: {exc}", file=sys.stderr, flush=True) - return 1 + except RuntimeError as exc: + print(f"startup failed: {exc}", file=sys.stderr, flush=True) + return 1 children: list[ChildProcess] = [] interrupted = False