from __future__ import annotations import json import os from pathlib import Path import tempfile import unittest from unittest.mock import MagicMock, patch from cognitive_agent.application.model_plan_adapter import ModelPlanAdapter from cognitive_agent.application.native_can_intent import merge_native_can_intent from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService from cognitive_agent.application.self_extension import ( SelfExtensionService, SimulationGateway, ) HARDWARE_DISABLED_ENV = { "COGNITIVE_AGENT_CH1_HIL_ENABLED": "false", "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false", } def _empty_intent() -> dict: return { "actions": [], "domain_actions": [], "native_can_actions": [], "deferred_requests": [], "model_requests": [], "preset_actions": [], "preset_events": [], "ambiguities": [], "detected_domains": [], "decision_summary": [], "confidence": 0.0, } class Ch1DefaultTargetTests(unittest.TestCase): def setUp(self) -> None: self.environment = patch.dict(os.environ, HARDWARE_DISABLED_ENV, clear=False) self.environment.start() self.addCleanup(self.environment.stop) self.temporary_directory = tempfile.TemporaryDirectory() self.addCleanup(self.temporary_directory.cleanup) self.service = SeatDemoService(REPOSITORY_ROOT) self.service.artifact_root = Path(self.temporary_directory.name) def test_local_requests_use_driver_console_defaults_and_one_ch1_operation(self) -> None: cases = ( ( "车窗开启一半", "native_window", "window.set_position", "front_left", {"position_pct": 50}, ), ( "打开车窗", "native_window", "window.set_motion", "front_left", {"motion": "open"}, ), ( "空调调到21度", "native_climate", "climate.set_temperature", "driver", {"temperature_c": 21}, ), ( "把风口左右30%上下40%", "native_outlet", "outlet.set_position_percent", "driver_left", {"horizontal_pct": 30, "vertical_pct": 40}, ), ( "打开座椅加热", "native_seat", "seat.set_driver_heating_native", "driver", {"state": "on", "level": 2}, ), ( "打开座椅通风", "native_seat", "seat.set_driver_ventilation_native", "driver", {"state": "on", "level": 2}, ), ) for index, (utterance, config_key, capability, target, parameters) in enumerate( cases ): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-ch1-default-local-{index}", execution_mode="dry_run", ) self.assertEqual(result["status"], "succeeded") self.assertEqual(set(result["function_configs"]), {config_key}) self.assertNotIn("seat", result["function_configs"]) operations = result["function_configs"][config_key]["spec"][ "operations" ] self.assertEqual(len(operations), 1) self.assertEqual(operations[0]["capability"], capability) self.assertEqual(operations[0]["target"]["zone"], target) self.assertEqual(operations[0]["parameters"], parameters) commands = result["command_plans"][config_key]["spec"]["commands"] self.assertEqual(len(commands), 1) self.assertEqual(commands[0]["bus"], "body_can") summaries = " ".join(result["decision_summary"]) self.assertNotIn("副驾座椅功能单域", summaries) self.assertNotIn("目标：副驾座椅", summaries) def test_explicit_vehicle_targets_take_priority_over_defaults(self) -> None: cases = ( ("右前车窗开启一半", "native_window", "front_right", {"position_pct": 50}), ("副驾空调调到21度", "native_climate", "passenger", {"temperature_c": 21}), ) for index, (utterance, config_key, target, parameters) in enumerate(cases): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-ch1-explicit-target-{index}", ) operation = result["function_configs"][config_key]["spec"][ "operations" ][0] self.assertEqual(operation["target"]["zone"], target) self.assertEqual(operation["parameters"], parameters) passenger = self.service.process( "打开副驾驶座椅加热", provider="local", conversation_id="conv-ch1-explicit-passenger-seat", ) self.assertEqual(set(passenger["function_configs"]), {"seat"}) self.assertNotIn("native_seat", passenger["function_configs"]) self.assertEqual( passenger["function_configs"]["seat"]["spec"]["operations"][0][ "target" ]["zone"], "front_right", ) def test_driver_and_passenger_window_aliases_map_to_the_correct_side(self) -> None: cases = ( ("打开主驾车窗", "front_left", "window.set_motion", {"motion": "open"}), ("打开主驾驶的车窗", "front_left", "window.set_motion", {"motion": "open"}), ("打开副驾车窗", "front_right", "window.set_motion", {"motion": "open"}), ("打开副驾驶的车窗", "front_right", "window.set_motion", {"motion": "open"}), ("把副驾车窗开到35%", "front_right", "window.set_position", {"position_pct": 35}), ) for index, (utterance, target, capability, parameters) in enumerate(cases): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-window-occupant-alias-{index}", execution_mode="dry_run", ) self.assertEqual(result["status"], "succeeded") self.assertEqual(set(result["function_configs"]), {"native_window"}) operation = result["function_configs"]["native_window"]["spec"][ "operations" ][0] self.assertEqual(operation["target"]["zone"], target) self.assertEqual(operation["capability"], capability) self.assertEqual(operation["parameters"], parameters) def test_window_target_does_not_leak_from_a_later_seat_subject(self) -> None: parsed = merge_native_can_intent({}, "打开副驾车窗和主驾座椅通风") windows = [ action for action in parsed["native_can_actions"] if action["capability"] == "window.set_motion" ] self.assertEqual(len(windows), 1) self.assertEqual(windows[0]["target"], "front_right") self.assertEqual(windows[0]["parameters"], {"motion": "open"}) def test_group_or_non_driver_seat_scope_is_not_treated_as_unscoped(self) -> None: for index, utterance in enumerate( ("打开前排座椅加热", "打开后排座椅加热", "打开所有座椅加热") ): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-ch1-non-driver-scope-{index}", ) self.assertEqual(result["status"], "needs_confirmation") self.assertEqual(result["function_configs"], {}) self.assertNotIn("native_seat", result["command_plans"]) def test_unresolved_unscoped_seat_request_never_falls_back_to_passenger_ch0(self) -> None: parsed = _empty_intent() parsed["actions"] = [ { "action_type": "thermal", "target": "passenger", "function": "heating", "state": "on", } ] parsed["model_requests"] = [ { "domain": "seat", "capability": "seat.set_heating", "target_scope": "passenger", "operation": "set", "parameters": [{"name": "state", "value": "on"}], "reason": "model guessed passenger", } ] result = merge_native_can_intent(parsed, "座椅调舒服一点") self.assertFalse(result["actions"]) self.assertFalse(result["native_can_actions"]) self.assertFalse(result["model_requests"]) self.assertTrue( any("不会转入副驾 CH0" in item["question"] for item in result["ambiguities"]) ) def test_unverified_driver_absolute_motion_is_not_degraded_to_a_timed_pulse(self) -> None: for index, utterance in enumerate( ("座椅前后移动到50%", "靠背向后移动70度", "坐盆抬到最高") ): with self.subTest(utterance=utterance): result = self.service.process( utterance, provider="local", conversation_id=f"conv-ch1-unsupported-absolute-{index}", ) self.assertEqual(result["status"], "needs_confirmation") self.assertEqual(result["function_configs"], {}) timed = self.service.process( "座椅向前移动2秒", provider="local", conversation_id="conv-ch1-verified-timed-axis", ) self.assertEqual(timed["status"], "succeeded") operation = timed["function_configs"]["native_seat"]["spec"][ "operations" ][0] self.assertEqual(operation["capability"], "seat.set_driver_axis_native") self.assertEqual( operation["parameters"], {"axis": "horizontal", "direction": "forward", "duration_ms": 2000}, ) def test_model_adapter_preserves_every_nonempty_model_target(self) -> None: window = ModelPlanAdapter._apply_ch1_default_target( { "domain": "window", "capability": "window.set_position", "target_scope": "all", }, "车窗开启一半", ) climate = ModelPlanAdapter._apply_ch1_default_target( { "domain": "climate", "capability": "climate.set_temperature", "target_scope": "vehicle", }, "空调调到21度", ) seat = ModelPlanAdapter._apply_ch1_default_target( { "domain": "seat", "capability": "seat.set_heating", "target_scope": "passenger", }, "打开座椅加热", ) explicit_passenger = ModelPlanAdapter._apply_ch1_default_target( { "domain": "climate", "capability": "climate.set_temperature", "target_scope": "passenger", }, "副驾空调调到21度", ) self.assertEqual(window["target_scope"], "all") self.assertEqual(climate["target_scope"], "vehicle") self.assertEqual(seat["capability"], "seat.set_heating") self.assertEqual(seat["target_scope"], "passenger") self.assertEqual(explicit_passenger["target_scope"], "passenger") def test_local_rules_skip_gpt_and_apply_safe_ch1_defaults(self) -> None: cases = ( ( "车窗开启一半", { "domain": "window", "capability": "window.set_position", "target_scope": "all", "operation": "set", "parameters": [{"name": "position_pct", "value": 100}], "reason": "model chose every window", }, "native_window", "front_left", {"position_pct": 50}, ), ( "空调调到21度", { "domain": "climate", "capability": "climate.set_temperature", "target_scope": "vehicle", "operation": "set", "parameters": [{"name": "temperature_c", "value": 21}], "reason": "model chose the whole vehicle", }, "native_climate", "driver", {"temperature_c": 21}, ), ( "打开座椅加热", { "domain": "seat", "capability": "seat.set_heating", "target_scope": "passenger", "operation": "set", "parameters": [ {"name": "state", "value": "on"}, {"name": "level", "value": 3}, ], "reason": "model guessed the legacy passenger seat", }, "native_seat", "driver", {"state": "on", "level": 2}, ), ) self.service.openai.api_key = "offline-test-key" self.service.openai.model = "gpt-5.6-sol" for index, (utterance, request, config_key, target, parameters) in enumerate( cases ): model_output = { "detected_domains": [request["domain"]], "requests": [request], "ambiguities": [], "decision_summary": ["GPT model output"], "confidence": 0.9, } with self.subTest(utterance=utterance): with patch.object( self.service.openai, "interpret", return_value=model_output ) as interpret: result = self.service.process( utterance, provider="openai", conversation_id=f"conv-ch1-default-openai-{index}", ) interpret.assert_not_called() self.assertEqual(result["provider_used"], "local_rules") self.assertEqual(result["status"], "succeeded") self.assertEqual(set(result["function_configs"]), {config_key}) self.assertNotIn("seat", result["function_configs"]) operation = result["function_configs"][config_key]["spec"][ "operations" ][0] self.assertEqual(operation["target"]["zone"], target) self.assertEqual(operation["parameters"], parameters) self.assertEqual( len(result["command_plans"][config_key]["spec"]["commands"]), 1, ) def test_self_extension_vehicle_endpoint_uses_the_same_default_contract(self) -> None: extension = SelfExtensionService(SimulationGateway()) extension.vehicle_provider = "local" extension.vehicle_execution_mode = "dry_run" extension.vehicle_id = "denza_n7" cases = ( ("车窗开启一半", "window", "front_left", {"position_pct": 50}), ("空调调到21度", "climate", "driver", {"temperature_c": 21}), ("打开座椅加热", "seat", "driver", {"state": "on", "level": 2}), ) for index, (command, domain, target, parameters) in enumerate(cases): with self.subTest(command=command): result = extension.execute_vehicle( { "command": command, "conversation_id": f"conv-extension-ch1-default-{index}", }, self.service, hardware_access_bypass=True, ) self.assertTrue(result["success"]) self.assertEqual(result["status"], "succeeded") self.assertEqual(result["provider_used"], "local_rules") self.assertEqual(len(result["operation_results"]), 1) operation = result["operation_results"][0] self.assertEqual(operation["domain"], domain) self.assertEqual(operation["telemetry"]["target"], target) self.assertEqual( operation["telemetry"]["requested_parameters"], parameters ) def test_self_extension_gpt_prompt_limits_confirmation_to_toy_peripherals(self) -> None: response_plan = { "summary": "ok", "needs_confirmation": False, "question": "", "steps": [], } response = MagicMock() response.__enter__.return_value = response response.read.return_value = json.dumps( {"output_text": json.dumps(response_plan, ensure_ascii=False)}, ensure_ascii=False, ).encode("utf-8") with patch.dict(os.environ, {"OPENAI_STREAM": "false"}, clear=False): extension = SelfExtensionService(SimulationGateway()) with patch( "cognitive_agent.application.self_extension.urlopen", return_value=response, ) as open_request: extension._request_gpt_plan( "车窗开启一半，空调调到21度", [], [{"label": "车窗"}, {"label": "空调"}], "offline-test-key", ) request_payload = json.loads(open_request.call_args.args[0].data.decode("utf-8")) prompt = request_payload["input"][0]["content"][0]["text"] self.assertIn("多台可匹配的外设玩具", prompt) self.assertIn("此追问规则不适用于", prompt) self.assertIn("window front-left position 50", prompt) self.assertIn("ac temp driver 21", prompt) self.assertIn("seat driver", prompt) self.assertIn("绝不能转入副驾/CH0", prompt) if __name__ == "__main__": unittest.main()