"""Natural-language edits cannot flatten or silently broaden dynamic phases.""" from __future__ import annotations import hashlib from copy import deepcopy from pathlib import Path from unittest.mock import patch import pytest from cognitive_agent.application.mode_profile import ModeProfileRegistry from cognitive_agent.application.mode_schedule import ModeSchedule from cognitive_agent.application.mode_schedule_edit import edit_schedule from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService @pytest.fixture def service(tmp_path): with patch.dict( "os.environ", { "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false", "COGNITIVE_AGENT_WIRED_PASSENGER_CAN_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_MQTT_ENABLED": "false", }, ): instance = SeatDemoService(REPOSITORY_ROOT) instance.artifact_root = tmp_path / "runs" instance.mode_profiles = ModeProfileRegistry( REPOSITORY_ROOT / "config" / "modes", user_root=tmp_path / "user-modes", ) try: with patch.object( instance.hardware_execution, "reserve", side_effect=AssertionError("schedule edit reached hardware"), ): yield instance finally: instance.close() def _build(service): template = service.scene_templates.get_template("boarding-warmup") result = service.preview_native_can( "构建一个上车预热模式", provider="local", conversation_id="conv-schedule-edit", execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=template.plan(), ) assert result["status"] == "succeeded", result.get("validation") return result def _edit(service, built, change): draft = built["mode_profile_draft"] reference = {key: draft[key] for key in ("name", "source_request_id", "vehicle_id")} return service.preview_native_can( f"把上车预热模式的{change}", provider="local", conversation_id="conv-schedule-edit", execution_mode="hardware", vehicle_id="denza_n7", mode_profile_draft_context=reference, ) def _parameters(schedule, index, capability, target="vehicle"): return next( op["parameters"] for op in schedule["phases"][index]["operations"] if op["capability"] == capability and op["target_scope"] == target ) @pytest.mark.parametrize( "change", ( "第二阶段空调温度调整到26度", "第2个阶段的空调温度变成26度", "舒适恒温阶段空调温度改为26度", ), ) def test_draft_temperature_edit_preserves_first_phase_and_timing(service, change): built = _build(service) original = deepcopy(built["schedule"]) revised = _edit(service, built, change) assert revised["status"] == "succeeded", revised.get("mode_creation_failure") expected = deepcopy(original) _parameters(expected, 1, "climate.set_temperature")["temperature_c"] = 26 assert revised["schedule"] == expected assert built["schedule"] == original assert revised["mode_profile_draft"]["schedule"] == expected assert revised["mode_profile_draft"]["draft_revision"] == 2 assert revised["mode_profile_draft"]["revises_source_request_id"] == built["request_id"] assert revised["execution_mode"] == "dry_run" assert revised["hardware_execution_allowed"] is False @pytest.mark.parametrize( ("change", "expected_hold", "expected_repeat"), ( ("第二阶段持续时间调整为3分钟", 180000, 1), ("第二阶段保持30秒", 30000, 1), ("第二阶段时长改成2.5分钟", 150000, 1), ("循环次数改成3次", 300000, 3), ("重复两次", 300000, 2), ), ) def test_draft_time_and_repeat_edits_keep_operations( service, change, expected_hold, expected_repeat ): built = _build(service) revised = _edit(service, built, change) assert revised["status"] == "succeeded", revised.get("mode_creation_failure") assert revised["schedule"]["phases"][1]["duration_ms"] == expected_hold assert revised["schedule"]["repeat_count"] == expected_repeat for index in (0, 1): assert ( revised["schedule"]["phases"][index]["operations"] == built["schedule"]["phases"][index]["operations"] ) def test_all_phases_temperature_edit_is_explicit_and_preserves_all_other_operations(service): built = _build(service) revised = _edit(service, built, "全部阶段空调温度调整到25度") assert revised["status"] == "succeeded", revised.get("mode_creation_failure") expected = deepcopy(built["schedule"]) for index in (0, 1): _parameters(expected, index, "climate.set_temperature")["temperature_c"] = 25 assert revised["schedule"] == expected @pytest.mark.parametrize( "change", ( "空调温度调整到25度", "第三阶段空调温度调整到25度", "第二阶段持续16分钟", "循环21次", "重复1.5次", "持续30秒", "第二阶段主驾靠背调整到90%", "第二阶段空调温度调整到99度", "第二阶段打开后备箱", "第二阶段空调温度调整到26度并打开后备箱", ), ) def test_ambiguous_unsafe_or_unbounded_edit_leaves_original_draft_unchanged(service, change): built = _build(service) original = deepcopy(built["schedule"]) revised = _edit(service, built, change) assert revised["status"] in {"needs_confirmation", "validation_failed"} assert not revised.get("mode_profile_draft") assert revised["hardware_execution_allowed"] is False source = service._load_mode_profile_draft_reference( { key: built["mode_profile_draft"][key] for key in ("name", "source_request_id", "vehicle_id") }, conversation_id="conv-schedule-edit", mode_name="上车预热模式", vehicle_id="denza_n7", )[1] assert source["schedule"] == original def test_saved_schedule_edit_revalidates_and_persists_timeline_not_flat_actions(service): built = _build(service) draft = built["mode_profile_draft"] saved = service.create_mode_profile_card( draft["name"], draft["instruction"], draft["source_request_id"], draft["vehicle_id"] ) mode_id = saved["mode_profile"]["id"] profile = service.mode_profiles.get_profile(mode_id, "denza_n7") path = Path(profile.source) before = hashlib.sha256(path.read_bytes()).hexdigest() rejected = service.preview_native_can( "把上车预热模式的空调温度调整到26度", provider="local", conversation_id="conv-saved-schedule-edit", execution_mode="hardware", vehicle_id="denza_n7", ) assert rejected["status"] == "needs_confirmation" assert hashlib.sha256(path.read_bytes()).hexdigest() == before revised = service.preview_native_can( "把上车预热模式的第二阶段空调温度调整到26度", provider="local", conversation_id="conv-saved-schedule-edit", execution_mode="hardware", vehicle_id="denza_n7", ) assert revised["status"] == "succeeded", revised.get("validation") assert revised["orchestration_route"] == "mode_profile_update" assert revised["hardware_execution_allowed"] is False actual = service.mode_profiles.get_profile(mode_id, "denza_n7").schedule.to_dict() expected = deepcopy(built["schedule"]) _parameters(expected, 1, "climate.set_temperature")["temperature_c"] = 26 assert actual == expected assert revised["schedule"] == expected assert revised["mode_profile"]["behavior"] == "cyclic" def test_combined_phase_action_hold_then_repeat_revision_preserves_timeline(service): built = _build(service) revised = _edit(service, built, "第二阶段空调温度调整到26度，持续3分钟") assert revised["status"] == "succeeded", revised.get("mode_creation_failure") repeated = _edit(service, revised, "重复两次") assert repeated["status"] == "succeeded", repeated.get("mode_creation_failure") expected = deepcopy(built["schedule"]) _parameters(expected, 1, "climate.set_temperature")["temperature_c"] = 26 expected["phases"][1]["duration_ms"] = 180000 expected["repeat_count"] = 2 assert repeated["schedule"] == expected assert repeated["mode_profile_draft"]["draft_revision"] == 3 draft = repeated["mode_profile_draft"] saved = service.create_mode_profile_card( draft["name"], draft["instruction"], draft["source_request_id"], draft["vehicle_id"] ) assert ( service.mode_profiles.get_profile( saved["mode_profile"]["id"], "denza_n7" ).schedule.to_dict() == expected ) def test_all_phases_is_not_mistaken_for_all_vehicle_zones(service): source = service.scene_templates.get_template("boarding-warmup").plan()["schedule"] for phase in source["phases"]: for operation in phase["operations"]: if operation["capability"] == "climate.set_temperature": operation["target_scope"] = "passenger" schedule = ModeSchedule.from_dict(source) updated = edit_schedule( schedule, "全部阶段空调温度调整到25度", parse_operations=lambda _: [ { "domain": "climate", "capability": "climate.set_temperature", "target_scope": "vehicle", "operation": "set", "parameters": {"temperature_c": 25}, } ], target_is_explicit=service._mode_update_target_is_explicit, ) for index in (0, 1): assert _parameters(updated.to_dict(), index, "climate.set_temperature", "passenger") == { "temperature_c": 25 } assert not any( op["capability"] == "climate.set_temperature" and op["target_scope"] == "vehicle" for op in updated.phases[index].operations ) def test_phase_motor_duration_cannot_become_a_hold_duration(service): built = _build(service) result = _edit(service, built, "第二阶段副驾靠背向后移动3s") assert result["status"] in {"needs_confirmation", "validation_failed"} assert not result.get("mode_profile_draft") def test_stage_selector_does_not_authorize_delayed_action_semantics(service): built = _build(service) result = _edit(service, built, "第二阶段30秒后空调温度调整到26度") assert result["status"] == "needs_confirmation" assert not result.get("mode_profile_draft")