from __future__ import annotations import json from pathlib import Path from cognitive_agent.application.mode_profile import ModeProfileRegistry from cognitive_agent.application.mode_scene_catalog import ( ModeSceneCatalogPublisher, build_mode_scene_catalog, ) from cognitive_agent.application.mode_schedule import ModeSchedule from cognitive_agent.application.seat_service import REPOSITORY_ROOT def _write_office_preset(path: Path) -> None: path.parent.mkdir(parents=True) path.write_text( """ id: office name: 办公模式 description: N7 办公场景。 icon: briefcase scope: vehicle_ids: [denza_n7] operations: - domain: climate capability: climate.set_temperature target_scope: passenger operation: set parameters: temperature_c: 24 """.strip() + "\n", encoding="utf-8", ) def test_publisher_tracks_create_override_and_delete(tmp_path: Path) -> None: preset_root = tmp_path / "presets" user_root = tmp_path / "user" _write_office_preset( preset_root / "vehicles" / "denza_n7" / "preset" / "office.yaml" ) registry = ModeProfileRegistry(preset_root, user_root=user_root) catalog_path = tmp_path / "scenses" / "demo-scenes.json" publisher = ModeSceneCatalogPublisher( catalog_path, "denza_n7", server_id="test-n7", ) registry.add_change_listener(publisher.publish) initial = json.loads(catalog_path.read_text(encoding="utf-8")) assert initial["service"] == "n7-scenes" assert initial["schema_version"] == 1 assert initial["server_id"] == "test-n7" assert initial["scene_count"] == 1 assert initial["scenes"][0]["origin"] == "preset" assert initial["scenes"][0]["actions"] == [ { "type": "climate_temperature", "at_ms": 0, "domain": "climate", "capability": "climate.set_temperature", "target_scope": "passenger", "operation": "set", "parameters": {"temperature_c": 24}, "zone": "passenger", "celsius": 24, } ] created = registry.save_mode( "阅读模式", [ { "domain": "ambient_light", "capability": "ambient_light.set_illumination", "target_scope": "vehicle", "parameters": {"color_hex": "#E8F3FF", "brightness_pct": 40}, } ], vehicle_id="denza_n7", ) after_create = json.loads(catalog_path.read_text(encoding="utf-8")) created_scene = next( scene for scene in after_create["scenes"] if scene["id"] == created.id ) assert after_create["scene_count"] == 2 assert created_scene["origin"] == "user" assert created_scene["deletable"] is True assert created_scene["actions"][0]["parameters"]["brightness_pct"] == 40 fan_mode = registry.save_mode( "微风模式", [ { "domain": "climate", "capability": "climate.set_fan_level", "target_scope": "vehicle", "parameters": {"level": 2}, } ], vehicle_id="denza_n7", ) after_fan_create = json.loads(catalog_path.read_text(encoding="utf-8")) fan_scene = next( scene for scene in after_fan_create["scenes"] if scene["id"] == fan_mode.id ) assert fan_scene["actions"][0]["type"] == "climate_fan" assert fan_scene["actions"][0]["level"] == 2 registry.save_mode( "专注办公模式", [ { "domain": "climate", "capability": "climate.set_temperature", "target_scope": "passenger", "parameters": {"temperature_c": 26}, } ], mode_id="office", vehicle_id="denza_n7", ) after_override = json.loads(catalog_path.read_text(encoding="utf-8")) office = next(scene for scene in after_override["scenes"] if scene["id"] == "office") assert office["label"] == "专注办公模式" assert office["origin"] == "customized" assert office["preset_override"] is True assert office["actions"][0]["celsius"] == 26 registry.delete_user_mode(created.id, "denza_n7") registry.delete_user_mode(fan_mode.id, "denza_n7") registry.delete_user_mode("office", "denza_n7") after_delete = json.loads(catalog_path.read_text(encoding="utf-8")) assert [scene["id"] for scene in after_delete["scenes"]] == ["office"] assert after_delete["scenes"][0]["label"] == "办公模式" assert after_delete["scenes"][0]["origin"] == "preset" def test_repository_presets_are_exported_without_internal_command_flows() -> None: registry = ModeProfileRegistry(REPOSITORY_ROOT / "config" / "modes") catalog = build_mode_scene_catalog( registry, "denza_n7", generated_at="2026-09-09T00:00:00Z", ) assert catalog["scene_count"] >= 13 assert len({scene["id"] for scene in catalog["scenes"]}) == catalog["scene_count"] by_id = {scene["id"]: scene for scene in catalog["scenes"]} assert by_id["office"]["operation_count"] == len(registry.get_profile("office", "denza_n7").operations) assert by_id["rocking_horse"]["behavior"] == "cyclic" assert by_id["rocking_horse"]["timeline"]["repeat"] == "until_stopped_or_timeout" encoded = json.dumps(catalog, ensure_ascii=False) assert "command_flow" not in encoded assert "source_request_id" not in encoded def _schedule() -> dict: return ModeSchedule.from_dict({ "schema": "cognitive-agent-mode-schedule/v1", "repeat_count": 2, "phases": [ {"id": "warm", "name": "暖舱", "duration_ms": 1000, "operations": [ {"domain": "climate", "capability": "climate.set_temperature", "target_scope": "vehicle", "operation": "set", "parameters": {"temperature_c": 28}}, ]}, {"id": "settle", "name": "舒适", "duration_ms": 2000, "operations": [ {"domain": "climate", "capability": "climate.set_temperature", "target_scope": "vehicle", "operation": "set", "parameters": {"temperature_c": 24}}, {"domain": "seat", "capability": "seat.set_backrest_recline", "target_scope": "passenger", "operation": "set", "parameters": {"position_pct": 45}}, ]}, ], }).to_dict() def test_schedule_catalog_preserves_phases_but_never_exports_flat_executable_actions(tmp_path: Path) -> None: registry = ModeProfileRegistry(tmp_path / "presets", user_root=tmp_path / "user") schedule = _schedule() profile = registry.save_mode( "分阶段舒适模式", schedule["phases"][0]["operations"], vehicle_id="denza_n7", schedule=ModeSchedule.from_dict(schedule), command_flow={"source_request_id": "req-private", "raw_can": "private-transport-bytes"}, ) catalog = build_mode_scene_catalog(registry, "denza_n7") scene = next(item for item in catalog["scenes"] if item["id"] == profile.id) assert scene["schedule"] == schedule assert scene["behavior"] == "cyclic" assert scene["operation_count"] == 3 assert scene["domains"] == ["climate", "seat"] assert scene["actions"] == [], "legacy clients must not execute all timeline stages at once" assert scene["execution"] == { "kind": "server_schedule", "mode_id": profile.id, "client_actions_supported": False, "requires_authentication": True, } assert "服务端时间轴" in scene["execution_notice"] assert scene["preview_actions_scope"] == "single_cycle" preview = scene["preview_actions"] assert [action["phase_id"] for action in preview] == ["warm", "settle", "settle"] assert [action["at_ms"] for action in preview] == [0, 1000, 1000] assert [action["phase_duration_ms"] for action in preview] == [1000, 2000, 2000] assert [action["celsius"] for action in preview if action["type"] == "climate_temperature"] == [28, 24] assert preview[-1]["target_scope"] == "passenger" assert preview[-1]["parameters"] == {"position_pct": 45} assert ModeSchedule.from_dict(scene["schedule"]).total_duration_ms == 6000 encoded = json.dumps(catalog) assert "source_request_id" not in encoded assert "private-transport-bytes" not in encoded assert "command_flow" not in encoded def test_schedule_publisher_updates_after_save_reload_duration_change_and_delete(tmp_path: Path) -> None: registry = ModeProfileRegistry(tmp_path / "presets", user_root=tmp_path / "user") publisher = ModeSceneCatalogPublisher(tmp_path / "catalog.json", "denza_n7") registry.add_change_listener(publisher.publish) assert publisher.snapshot()["scene_count"] == 0 schedule = _schedule() created = registry.save_mode( "动态释压模式", schedule["phases"][0]["operations"], vehicle_id="denza_n7", schedule=ModeSchedule.from_dict(schedule), ) first = publisher.snapshot() assert first["scenes"][0]["schedule"] == schedule registry.reload() assert publisher.snapshot() == first schedule["phases"][1]["duration_ms"] = 5000 registry.save_mode( created.name, schedule["phases"][0]["operations"], vehicle_id="denza_n7", mode_id=created.id, schedule=ModeSchedule.from_dict(schedule), ) changed = publisher.snapshot() assert changed["catalog_revision"] != first["catalog_revision"] assert changed["scenes"][0]["profile_revision"] != first["scenes"][0]["profile_revision"] assert changed["scenes"][0]["schedule"]["phases"][1]["duration_ms"] == 5000 assert changed["scenes"][0]["actions"] == [] assert json.loads(publisher.path.read_text(encoding="utf-8")) == changed registry.delete_user_mode(created.id, "denza_n7") assert publisher.snapshot()["scene_count"] == 0 assert publisher.snapshot()["scenes"] == []