from __future__ import annotations import json from pathlib import Path import shutil import subprocess import unittest from cognitive_agent.application.conversation_state import ConversationStateStore REPOSITORY_ROOT = Path(__file__).resolve().parents[2] APP_SOURCE_PATH = REPOSITORY_ROOT / "apps" / "web-console" / "app.js" EXTENSION_SOURCE_PATH = ( REPOSITORY_ROOT / "apps" / "web-console" / "self-extension.js" ) def function_source(source: str, signature: str, next_signature: str) -> str: start = source.index(signature) end = source.index(next_signature, start) return source[start:end] class FrontendVehicleStateSyncContractTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.app_source = APP_SOURCE_PATH.read_text(encoding="utf-8") cls.extension_source = EXTENSION_SOURCE_PATH.read_text(encoding="utf-8") def test_both_modules_use_the_same_browser_conversation(self) -> None: key = "cognitive-agent.conversation-id" self.assertIn( f'sessionStorage.getItem("{key}")', self.app_source, ) self.assertIn( f'sessionStorage.setItem("{key}", conversationId)', self.app_source, ) self.assertIn( f'const CONVERSATION_KEY = "{key}"', self.extension_source, ) self.assertIn( "body: JSON.stringify({ command, conversation_id: sharedConversationId() })", self.extension_source, ) def test_shared_twin_theme_tracks_persistent_day_and_night_mode(self) -> None: self.assertIn("function applyTwinSceneTheme(", self.app_source) self.assertIn("function currentTwinThemeRole()", self.app_source) self.assertIn("function currentTwinThemePalette()", self.app_source) self.assertIn("canvas.dataset.themeRole = role", self.app_source) self.assertIn( 'canvas.dataset.themeAccent = `#${palette.accent.toString(16).padStart(6, "0")}`', self.app_source, ) for night_theme_token in ( "accent: 0x7b70f5", "sceneBackground: 0x111426", "floor: 0x1a2038", "vehicleBody: 0x729bc9", "vehicleOutline: 0x73b8ff", "seatShell: 0x426fbd", "glass: 0x4d8ed2", "lamp: 0x4385cf", ): self.assertIn(night_theme_token, self.app_source) for day_theme_token in ( "accent: 0x2f80ed", "sceneBackground: 0xeff6ff", "floor: 0xe5f0fd", "vehicleBody: 0x91c6f5", "vehicleOutline: 0x2879d8", "seatShell: 0x2b76c9", "glass: 0x4b9bea", "lamp: 0x2e7fd4", ): self.assertIn(day_theme_token, self.app_source) self.assertIn('"cognitive-agent:theme-change"', self.app_source) self.assertIn("applyTwinReferenceVehicleTheme(twin.referenceVehicle, palette)", self.app_source) self.assertNotIn("accent: 0x0e7773", self.app_source) self.assertNotIn("cognitive-agent:demo-role-change", self.app_source) def test_shared_twin_ground_uses_theme_radial_alpha_texture(self) -> None: self.assertIn("function createTwinGroundTexture(", self.app_source) self.assertIn("new THREE.PlaneGeometry(30, 30)", self.app_source) self.assertIn( "surfaceGradient.addColorStop(0, rgba(floorColor, 0.96))", self.app_source, ) self.assertIn( "surfaceGradient.addColorStop(1, rgba(floorColor, 0))", self.app_source, ) self.assertIn("const outerRadius = size * 0.48", self.app_source) self.assertIn( "const gridStrong = new THREE.Color(palette.gridStrong)", self.app_source, ) self.assertIn("const gridSoft = new THREE.Color(palette.gridSoft)", self.app_source) self.assertIn( "gridContext.strokeStyle = rgba(major ? gridStrong : gridSoft, major ? 0.38 : 0.24)", self.app_source, ) self.assertIn("gridContext.lineWidth = major ? 0.9 : 0.45", self.app_source) self.assertIn('gridContext.globalCompositeOperation = "destination-in"', self.app_source) self.assertIn("applyTwinGroundTheme(twin, palette)", self.app_source) self.assertNotIn("new THREE.GridHelper(10, 20", self.app_source) def test_lan_http_origin_does_not_require_crypto_random_uuid(self) -> None: id_source = function_source( self.app_source, "function createOpaqueId(prefix) {", "\nfunction createConversationId", ) self.assertIn("globalThis.crypto?.randomUUID", id_source) self.assertIn("globalThis.crypto?.getRandomValues", id_source) self.assertIn("Math.random()", id_source) self.assertIn('return createOpaqueId("conv")', self.app_source) self.assertIn('createOpaqueId("hwc")', self.app_source) def test_extension_accepts_only_explicit_physical_ch1_state(self) -> None: authority_source = function_source( self.extension_source, " function isAuthoritativeVehicleStatePayload(payload) {", "\n function renderAuthoritativeVehicleState", ) self.assertIn( 'payload.conversation_id !== sharedConversationId()', authority_source ) self.assertIn( "containsAuthoritativePhysicalCh1State(payload.actuator_state)", authority_source, ) self.assertIn( "containsAuthoritativePhysicalCh1State(payload.domain_state)", authority_source, ) self.assertNotIn("physical_feedback_confirmed", authority_source) summary_source = function_source( self.extension_source, " function renderAuthoritativeVehicleState(payload) {", "\n function setMode", ) self.assertNotIn( "items.length === 1 && payload.physical_feedback_confirmed", summary_source, ) for state_path in ( "ventilation?.feedback_source", "heating?.feedback_source", "climate?.feedback_source", "roof?.feedback_source", "frontLeftWindow?.feedback_source", ): self.assertIn(f'{state_path} === "physical_can_ch1"', summary_source) def test_orchestration_renders_only_the_authoritative_ch1_patch(self) -> None: apply_source = function_source( self.app_source, "function applyAuthoritativeVehicleState(payload) {", '\nwindow.addEventListener("cognitive-agent:extension-vehicle-result"', ) self.assertIn( "const actuatorPatch = authoritativeCh1StatePatch(payload.actuator_state)", apply_source, ) self.assertIn( "const domainPatch = authoritativeCh1StatePatch(payload.domain_state)", apply_source, ) self.assertIn("renderSeatState(actuatorPatch)", apply_source) self.assertIn("renderDomainState(domainPatch)", apply_source) self.assertIn("if (actuatorPatch?.driver)", apply_source) self.assertNotIn("renderSeatState(payload.actuator_state)", apply_source) self.assertNotIn("renderDomainState(payload.domain_state)", apply_source) render_source = function_source( self.app_source, "function renderResult(result) {", "\nasync function submitRequest", ) ch1_branch = function_source( render_source, " if (ch1HardwareResult) {", " updateTwinExplicitAngles", ) self.assertIn("applyAuthoritativeVehicleState(result)", ch1_branch) self.assertIn("} else {", ch1_branch) self.assertIn("renderSeatState(result.actuator_state)", ch1_branch) self.assertIn("renderDomainState(result.domain_state)", ch1_branch) def test_authoritative_patch_accepts_wired_and_wireless_without_promoting_siblings(self) -> None: node = shutil.which("node") if node is None: self.skipTest("Node.js is required to execute the frontend patch helper") source_helper = function_source( self.app_source, "function isAuthoritativeCh1Source(value) {", "\nfunction renderCabinLiveStatus", ) patch_source = function_source( self.app_source, "function authoritativeCh1StatePatch(value) {", "\nfunction isAuthoritativeVehicleStatePayload", ) state = { "climate": { "feedback_source": "physical_can_ch1", "power": "on", }, "roof": {"shade_position_pct": 100}, "ambient_light": { "feedback_source": "wireless_mqtt_ch1", "power": "on", }, "window": { "positions": {"front_left": {"position_pct": 50}} }, } script = ( 'const AUTHORITATIVE_CH1_SOURCES = new Set(["physical_can_ch1", "wireless_mqtt_ch1"]);\n' f"{source_helper}\n" f"{patch_source}\n" f"const input = {json.dumps(state)};\n" "process.stdout.write(JSON.stringify(authoritativeCh1StatePatch(input)));" ) completed = subprocess.run( [node, "-e", script], check=True, capture_output=True, text=True, ) self.assertEqual( json.loads(completed.stdout), { "climate": state["climate"], "ambient_light": state["ambient_light"], }, ) def test_dual_channel_runtime_polls_each_seat_without_aliasing(self) -> None: passenger_source = function_source( self.app_source, "async function synchronizePassengerSeatState() {", "\nasync function synchronizeDriverSeatState()", ) driver_source = function_source( self.app_source, "async function synchronizeDriverSeatState() {", "\nasync function synchronizePhysicalSeatState()", ) combined_source = function_source( self.app_source, "async function synchronizePhysicalSeatState() {", "\nasync function loadAuthStatus()", ) self.assertIn('requestedTransport === "wireless_mqtt"', passenger_source) self.assertIn("selectedVehicleTransport()?.state_configured", passenger_source) self.assertIn("control_transport: requestedTransport", passenger_source) self.assertIn("requestedTransport !== state.controlTransport", passenger_source) self.assertIn('payload.source === "wireless_mqtt_ch0"', passenger_source) self.assertIn("/api/hardware/seat-state", passenger_source) self.assertIn("selectedVehicleTransport()?.state_configured", driver_source) self.assertIn("control_transport: requestedTransport", driver_source) self.assertIn("requestedTransport !== state.controlTransport", driver_source) self.assertIn("/api/hardware/vehicle-state", driver_source) self.assertIn('state.controlTransport === "wireless_mqtt"', combined_source) self.assertIn("state.passengerStateSyncPromise", combined_source) self.assertLess( combined_source.index("synchronizePassengerSeatState()"), combined_source.index("synchronizeDriverSeatState()"), ) render_source = function_source( self.app_source, "function renderSeatState(actuatorState) {", "\nfunction renderDomainState", ) self.assertNotIn("passenger: mergedActuatorState.driver", render_source) def test_state_polling_surfaces_structured_403_reason(self) -> None: passenger_source = function_source( self.app_source, "async function synchronizePassengerSeatState() {", "\nasync function synchronizeDriverSeatState()", ) driver_source = function_source( self.app_source, "async function synchronizeDriverSeatState() {", "\nasync function synchronizePhysicalSeatState()", ) live_status_source = function_source( self.app_source, "function renderCabinLiveStatus(", "\nfunction refreshIcons", ) telemetry_source = function_source( self.app_source, "function recordVehicleStateTelemetry(payload) {", "\nfunction applyAuthoritativeVehicleState", ) self.assertIn("throw await apiResponseError(response)", passenger_source) self.assertIn("throw await apiResponseError(response)", driver_source) self.assertIn("errorStatus: Number(error.status) || null", driver_source) self.assertIn("errorReason: error.serverMessage", driver_source) self.assertIn("snapshot?.errorStatus === 403", live_status_source) self.assertIn("`${transportLabel} · 403：${reason}`", live_status_source) self.assertIn('telemetryState.textContent = "403 拒绝"', live_status_source) self.assertIn('setSummarySystem("403 · 状态读取被拒绝", "warning")', live_status_source) self.assertIn("errorStatus: null", telemetry_source) self.assertIn('errorReason: ""', telemetry_source) def test_persistent_linux_monitor_uses_backend_selected_fast_polling(self) -> None: interval_source = function_source( self.app_source, "function physicalStatePollIntervalMs() {", "\nfunction refreshIcons", ) self.assertIn("selectedVehicleTransport()?.state_sync?.poll_interval_ms", interval_source) self.assertIn("Math.max(250", interval_source) self.assertIn("DEFAULT_PHYSICAL_STATE_POLL_INTERVAL_MS = 5000", self.app_source) self.assertIn("}, physicalStatePollIntervalMs());", self.app_source) self.assertIn("restartPhysicalStatePolling();", self.app_source) def test_transport_switch_clears_stale_vehicle_values_before_new_feedback(self) -> None: select_source = function_source( self.app_source, "function selectControlTransport(transport) {", "\nfunction renderHardwareAccess", ) reset_source = function_source( self.app_source, "function resetVehicleStatusDisplay() {", "\nfunction mergeSeatActuatorState", ) self.assertIn("resetVehicleStatusDisplay();", select_source) self.assertIn('setText("#windowState", "UNKNOWN")', reset_source) self.assertIn('setText("#frontLightState", "UNKNOWN")', reset_source) self.assertIn('setText("#cabinSummaryDoors", "状态未知")', reset_source) self.assertIn('meter.dataset.state = "unknown"', reset_source) def test_vehicle_selection_cannot_replay_unfiltered_state_in_ch1_mode(self) -> None: select_source = function_source( self.app_source, "function selectVehicle(vehicleId, { silent = false } = {}) {", "\nfunction populateVehicleSelector", ) driver_branch = select_source.index("if (isDriverOnlyCh1Runtime())") filtered_snapshot = select_source.index( "state.vehicleStateSnapshot?.actuatorState", driver_branch, ) unfiltered_result = select_source.index("state.result?.actuator_state") self.assertLess(driver_branch, filtered_snapshot) self.assertLess(filtered_snapshot, unfiltered_result) self.assertIn("} else if (state.result?.actuator_state)", select_source) def test_driver_only_ch1_has_no_hidden_browser_lease_gate(self) -> None: submit_source = function_source( self.app_source, "async function submitRequest(event) {", "\nasync function copyArtifact", ) self.assertIn("const driverOnlyCh1 = isDriverOnlyCh1Runtime()", submit_source) self.assertIn("&& !driverOnlyCh1", submit_source) self.assertIn( '&& state.controlTransport !== "wireless_mqtt"', submit_source, ) startup_source = self.app_source[ self.app_source.index('document.addEventListener("DOMContentLoaded"') : ] self.assertIn("if (!isDriverOnlyCh1Runtime())", startup_source) self.assertIn("loadHardwareAccessStatus", startup_source) def test_state_results_flow_in_both_directions(self) -> None: self.assertIn( 'window.addEventListener("cognitive-agent:extension-vehicle-result"', self.app_source, ) self.assertIn( 'window.addEventListener("cognitive-agent:result"', self.extension_source, ) class PhysicalMarkerInvalidationContractTests(unittest.TestCase): def test_logical_door_and_window_actions_replace_stale_physical_roots(self) -> None: store = ConversationStateStore() conversation_id = "conv-frontend-marker-invalidation" store.synchronize_physical_ch1_feedback( conversation_id, [ { "source": "physical_can_ch1", "channel": 1, "frame_id": "0x3B6", "feedback_frame_id": "0x3B6", "valid": True, "verification": { "authoritative": True, "channel": 1, "simulated": False, "feedback_required": True, "feedback_observed": True, "feedback_match": True, "frame_id": "0x3B6", }, "domain_state": { "door": {"locks": {"rear_left": {"state": "lock"}}}, "window": { "positions": { "front_left": { "position_pct": 20, "closed": False, } } }, }, } ], ) physical = store.snapshot(conversation_id)["domain_devices"] self.assertEqual(physical["door"]["feedback_source"], "physical_can_ch1") self.assertEqual(physical["window"]["feedback_source"], "physical_can_ch1") updated = store.apply_successful_actions( conversation_id, "logical projection must not look physical", [ { "domain": "door", "capability": "door.set_rear_lock", "target": "rear", "parameters": {"state": "unlock"}, }, { "domain": "window", "capability": "window.set_position", "target": "front_left", "parameters": {"position_pct": 60}, }, ], [], )["domain_devices"] self.assertEqual(updated["door"]["feedback_source"], "virtual") self.assertEqual(updated["window"]["feedback_source"], "virtual") if __name__ == "__main__": unittest.main()