from __future__ import annotations import copy import json from collections.abc import Mapping import pytest from jsonschema import Draft202012Validator from cognitive_agent.application.assistant_capabilities import AssistantCapabilityCatalog from cognitive_agent.application.scene_composer import ( build_scene_output_schema, normalize_scene_output, scene_has_explicit_control_object, scene_required_native_fallbacks, scene_required_peripheral_refs, scene_semantic_guidance, ) from cognitive_agent.application.seat_service import SeatDemoService def _context() -> dict[str, object]: return { "schema_version": "n7web-scene-capabilities/v1", "vehicle": {"id": "denza_n7", "name": "腾势 N7"}, "capabilities": [ { "id": "climate.set_temperature", "domain": "climate", "target_scopes": ["driver", "vehicle"], "parameters": [ { "name": "temperature_c", "type": "number", "minimum": 18, "maximum": 32, "required": True, } ], }, { "id": "roof.set_shade_position", "domain": "roof", "target_scopes": ["vehicle"], "parameters": [ { "name": "position_pct", "type": "integer", "minimum": 0, "maximum": 100, "required": True, } ], }, { "id": "ambient_light.set_mode", "domain": "ambient_light", "target_scopes": ["vehicle"], "parameters": [ { "name": "mode", "type": "string", "allowed_values": ["cinema", "rest", "off"], "required": True, } ], }, { "id": "seat.set_longitudinal_position", "domain": "seat", "target_scopes": ["passenger"], "parameters": [ { "name": "position_pct", "type": "integer", "minimum": 0, "maximum": 100, "required": False, }, { "name": "distance_mm", "type": "number", "minimum": 0, "maximum": 215, "required": False, "description": "Relative movement from the current position.", }, ], }, { "id": "seat.set_backrest_recline", "domain": "seat", "target_scopes": ["passenger"], "parameters": [ { "name": "angle_deg", "type": "number", "minimum": 95, "maximum": 163, "required": False, }, { "name": "angle_delta_deg", "type": "number", "minimum": 0, "maximum": 68, "required": False, }, ], }, ], } def _output(request: Mapping[str, object]) -> dict[str, object]: return { "requests": [copy.deepcopy(dict(request))], "ambiguities": [], "confidence": 0.9, } def _request( capability: str, target_scope: str, name: str, value: object, ) -> dict[str, object]: return { "capability": capability, "target_scope": target_scope, "parameters": {name: value}, } def _branch(schema: Mapping[str, object], capability: str) -> Mapping[str, object]: request_schema = schema["properties"]["requests"] branches = request_schema["items"]["anyOf"] return next( item for item in branches if item["properties"]["capability"]["enum"] == [capability] ) def test_schema_builds_one_capability_specific_request_branch() -> None: schema = build_scene_output_schema(_context()) Draft202012Validator.check_schema(schema) request_items = schema["properties"]["requests"]["items"] branches = request_items["anyOf"] assert len(branches) == 5 temperature = _branch(schema, "climate.set_temperature") assert temperature["properties"]["target_scope"]["enum"] == ["driver", "vehicle"] value_schema = temperature["properties"]["parameters"]["anyOf"][0]["properties"][ "temperature_c" ] assert value_schema == {"type": "number", "minimum": 18, "maximum": 32} ambient_mode = _branch(schema, "ambient_light.set_mode") mode_value = ambient_mode["properties"]["parameters"]["anyOf"][0]["properties"][ "mode" ] assert mode_value == {"type": "string", "enum": ["cinema", "rest", "off"]} def test_schema_avoids_unsupported_structured_output_composition_keywords() -> None: schema = build_scene_output_schema(_context()) def keys(value: object) -> set[str]: if isinstance(value, Mapping): return set(value) | { nested_key for nested in value.values() for nested_key in keys(nested) } if isinstance(value, list): return {nested_key for nested in value for nested_key in keys(nested)} return set() assert not keys(schema) & { "allOf", "contains", "dependentRequired", "dependentSchemas", "else", "if", "maxContains", "minContains", "not", "then", } assert "anyOf" in keys(schema) def test_relative_scene_parameters_are_not_exposed_by_the_schema() -> None: schema_text = json.dumps(build_scene_output_schema(_context()), ensure_ascii=False) assert "distance_mm" not in schema_text assert "angle_delta_deg" not in schema_text assert "position_pct" in schema_text assert "angle_deg" in schema_text @pytest.mark.parametrize( "scene_request", ( _request("climate.set_temperature", "vehicle", "temperature_c", 24.5), _request("roof.set_shade_position", "vehicle", "position_pct", 100), _request("ambient_light.set_mode", "vehicle", "mode", "cinema"), _request("seat.set_longitudinal_position", "passenger", "position_pct", 80), _request("seat.set_backrest_recline", "passenger", "angle_deg", 125), ), ) def test_schema_accepts_only_valid_capability_bound_requests( scene_request: Mapping[str, object], ) -> None: schema = build_scene_output_schema(_context()) assert Draft202012Validator(schema).is_valid(_output(scene_request)) @pytest.mark.parametrize( "scene_request", ( # Target and parameter names valid elsewhere in the catalog must not be # freely combined with a different capability. _request("roof.set_shade_position", "driver", "position_pct", 100), _request("climate.set_temperature", "vehicle", "position_pct", 24), _request("seat.set_longitudinal_position", "passenger", "distance_mm", 100), _request("seat.set_backrest_recline", "passenger", "angle_delta_deg", 20), # Scalar type, enum and numeric bounds come from that exact parameter. _request("climate.set_temperature", "vehicle", "temperature_c", "24"), _request("climate.set_temperature", "vehicle", "temperature_c", 17.9), _request("roof.set_shade_position", "vehicle", "position_pct", 101), _request("ambient_light.set_mode", "vehicle", "mode", "party"), ), ) def test_schema_rejects_cross_capability_relative_or_out_of_contract_values( scene_request: Mapping[str, object], ) -> None: schema = build_scene_output_schema(_context()) assert not Draft202012Validator(schema).is_valid(_output(scene_request)) def test_schema_requires_exact_required_parameters_and_rejects_mutually_exclusive_forms() -> None: schema = build_scene_output_schema(_context()) validator = Draft202012Validator(schema) missing = { "capability": "climate.set_temperature", "target_scope": "vehicle", "parameters": {}, } conflicting = { "capability": "seat.set_backrest_recline", "target_scope": "passenger", "parameters": {"position_pct": 50, "angle_deg": 125}, } assert not validator.is_valid(_output(missing)) assert not validator.is_valid(_output(conflicting)) def test_schema_optional_parameter_never_substitutes_for_a_required_parameter() -> None: context = _context() context["capabilities"].append( { "id": "seat.set_heating", "domain": "seat", "target_scopes": ["driver"], "parameters": [ { "name": "state", "type": "string", "allowed_values": ["on", "off"], "required": True, }, { "name": "level", "type": "integer", "minimum": 1, "maximum": 3, "required": False, }, ], } ) validator = Draft202012Validator(build_scene_output_schema(context)) assert not validator.is_valid( _output( { "capability": "seat.set_heating", "target_scope": "driver", "parameters": {"level": 2}, } ) ) assert validator.is_valid( _output( { "capability": "seat.set_heating", "target_scope": "driver", "parameters": {"state": "on"}, } ) ) assert validator.is_valid( _output( { "capability": "seat.set_heating", "target_scope": "driver", "parameters": {"state": "on", "level": 2}, } ) ) def test_normalize_remains_compatible_with_a_schema_valid_response() -> None: context = _context() raw = _output(_request("climate.set_temperature", "driver", "temperature_c", 24)) assert Draft202012Validator(build_scene_output_schema(context)).is_valid(raw) normalized = normalize_scene_output(raw, context) assert normalized["detected_domains"] == ["climate"] assert normalized["requests"] == [ { "domain": "climate", "capability": "climate.set_temperature", "target_scope": "driver", "operation": "set", "duration_ms": None, "parameters": [{"name": "temperature_c", "value": 24}], "reason": "Scene Composer 候选动作，等待应用安全链复核。", } ] def test_normalize_rejects_duplicate_parameter_names_even_without_schema_enforcement() -> None: raw = _output(_request("climate.set_temperature", "vehicle", "temperature_c", 24)) raw["requests"][0]["parameters"] = [ {"name": "temperature_c", "value": 24}, {"name": "temperature_c", "value": 25}, ] with pytest.raises(ValueError, match="invalid parameter"): normalize_scene_output(raw, _context()) def test_production_scene_schema_remains_below_app_server_limit() -> None: catalog = AssistantCapabilityCatalog(SeatDemoService()) schema = build_scene_output_schema(catalog.scene_composition_context()) encoded = json.dumps(schema, ensure_ascii=False, separators=(",", ":")).encode("utf-8") assert len(encoded) < 64 * 1024 def test_schema_and_normalizer_accept_only_opaque_peripheral_refs() -> None: context = _context() context["peripheral_actions"] = [ { "ref": "ir:0123456789abcdef01234567", "device_label": "氛围灯带", "action_label": "设置为红色", } ] schema = build_scene_output_schema(context) raw = { "requests": [{"capability_ref": "ir:0123456789abcdef01234567"}], "ambiguities": [], "confidence": 0.91, } assert Draft202012Validator(schema).is_valid(raw) assert not Draft202012Validator(schema).is_valid( { **raw, "requests": [ { "capability_ref": "ir:invented", "slot": 7, "card_uid": "2CC8B103", } ], } ) normalized = normalize_scene_output(raw, context) assert normalized["requests"] == [ { "domain": "self_extension", "capability": "infrared.emit", "target_scope": "ir:0123456789abcdef01234567", "operation": "emit", "duration_ms": None, "parameters": [ { "name": "capability_ref", "value": "ir:0123456789abcdef01234567", } ], "reason": ( "Scene Composer selected a server-issued learned-device reference; " "runtime binding resolution is still required." ), } ] def test_schema_remains_bounded_with_full_peripheral_catalog() -> None: context = _context() context["peripheral_actions"] = [ {"ref": f"ir:{index:024x}"} for index in range(248) ] schema = build_scene_output_schema(context) encoded = json.dumps(schema, ensure_ascii=False, separators=(",", ":")).encode() assert len(encoded) < 64 * 1024 peripheral_branch = schema["properties"]["requests"]["items"]["anyOf"][-1] assert len( peripheral_branch["properties"]["capability_ref"]["enum"] ) == 248 def test_learning_mode_is_grounded_as_free_semantic_composition() -> None: context = _context() guidance = scene_semantic_guidance("学习模式") context["scene_semantics"] = guidance schema = build_scene_output_schema(context) validator = Draft202012Validator(schema) valid = _output( _request( "climate.set_temperature", "vehicle", "temperature_c", 24, ) ) assert guidance["grounded"] is True assert guidance["family"] == "focused_learning" assert scene_semantic_guidance("灯光主题模式")["family"] == ( "lighting_ambience" ) assert "interior_light.set_reading_power" in guidance[ "preferred_capability_ids" ] assert schema["properties"]["requests"]["minItems"] == 1 assert schema["properties"]["ambiguities"]["maxItems"] == 0 assert validator.is_valid(valid) assert not validator.is_valid( {"requests": [], "ambiguities": [], "confidence": 0.4} ) assert not validator.is_valid( {**valid, "ambiguities": ["不知道学习指什么"]} ) with pytest.raises(ValueError, match="at least one request"): normalize_scene_output( {"requests": [], "ambiguities": [], "confidence": 0.4}, context, ) @pytest.mark.parametrize("mode_name", ("浪漫模式", "中秋欢快模式")) def test_romantic_and_mid_autumn_require_decorative_light_actions( mode_name: str, ) -> None: context = _context() context["peripheral_actions"] = [ { "ref": "ir:flash-open", "device_label": "闪烁灯", "object_names": ["闪烁灯", "闪光灯"], "action_label": "打开", "action_names": ["打开", "开启"], }, { "ref": "ir:flash-red", "device_label": "闪烁灯", "object_names": ["闪烁灯", "闪光灯"], "action_label": "切换为红色", "action_names": ["切换为红色", "设置为红色"], }, { "ref": "ir:flash-close", "device_label": "闪烁灯", "object_names": ["闪烁灯", "闪光灯"], "action_label": "关闭", "action_names": ["关闭", "熄灭"], }, { "ref": "ir:strip-open", "device_label": "灯带", "object_names": ["灯带", "LED 灯条"], "action_label": "打开", "action_names": ["打开", "开启"], }, { "ref": "ir:strip-red", "device_label": "灯带", "object_names": ["灯带", "LED 灯条"], "action_label": "切换为红色", "action_names": ["切换为红色", "设置为红色"], }, { "ref": "ir:ambient-open", "device_label": "外设氛围灯", "object_names": ["外设氛围灯", "氛围灯"], "action_label": "打开", "action_names": ["打开", "开启"], }, ] required = scene_required_peripheral_refs(mode_name, context) semantics = scene_semantic_guidance(mode_name) semantics["peripheral_policy"] = {"required_refs": required} context["scene_semantics"] = semantics schema = build_scene_output_schema(context) romantic_refs = [ "ir:flash-open", "ir:strip-open", "ir:ambient-open", ] assert required == (romantic_refs if "浪漫" in mode_name else [ "ir:flash-open", "ir:strip-open", ]) assert scene_required_peripheral_refs( mode_name, context, include_default_bundle=True, ) == (romantic_refs if "浪漫" in mode_name else [ "ir:flash-open", "ir:ambient-open", "ir:strip-open", ]) assert schema["properties"]["requests"]["maxItems"] == 20 - len(required) if "中秋" in mode_name: assert semantics["family"] == "festival_celebration" @pytest.mark.parametrize( ("mode_name", "expected_shade"), (("浪漫模式", 0), ("中秋欢快模式", 100)), ) def test_required_native_fallbacks_cover_climate_ambient_and_shade( mode_name: str, expected_shade: int, ) -> None: context = _context() context["capabilities"].append( { "id": "ambient_light.set_power", "domain": "ambient_light", "target_scopes": ["vehicle"], "parameters": [ { "name": "state", "type": "string", "allowed_values": ["on", "off"], "required": True, } ], } ) fallbacks = scene_required_native_fallbacks(mode_name, context) assert [item["domain"] for item in fallbacks] == [ "climate", "ambient_light", "roof", ] assert fallbacks[0]["capability"] == "climate.set_temperature" assert fallbacks[1]["capability"] == "ambient_light.set_power" assert fallbacks[2]["capability"] == "roof.set_shade_position" assert fallbacks[2]["parameters"] == [ {"name": "position_pct", "value": expected_shade} ] def test_name_only_scene_default_bundle_covers_all_required_native_controls() -> None: service = SeatDemoService() catalog = AssistantCapabilityCatalog(service) context = catalog.scene_composition_context(catalog.snapshot()) fallbacks = scene_required_native_fallbacks( "星夜模式", context, include_default_bundle=True, ) semantics = scene_semantic_guidance("星夜模式") semantics["native_fallback_requests"] = fallbacks semantics["peripheral_policy"] = {"required_refs": []} context["scene_semantics"] = semantics assert [(item["capability"], item["target_scope"]) for item in fallbacks] == [ ("climate.set_power", "vehicle"), ("climate.set_temperature", "vehicle"), ("climate.set_fan_level", "vehicle"), ("climate.set_cycle_mode", "vehicle"), ("window.set_position", "front_left"), ("window.set_position", "front_right"), ("window.set_position", "rear_left"), ("window.set_position", "rear_right"), ("roof.set_shade_position", "vehicle"), ("seat.set_ventilation", "passenger"), ("seat.set_backrest_recline", "passenger"), ("seat.set_legrest_height", "passenger"), ("ambient_light.set_power", "vehicle"), ] assert build_scene_output_schema(context)["properties"]["requests"]["maxItems"] == 7 @pytest.mark.parametrize( ("message", "expected"), ( ("创建浪漫中秋模式", False), ("创建浪漫中秋模式，空调温度设置为25度", True), ("创建灯光主题模式", True), ("创建主驾浪漫模式", True), ("创建节日模式", False), ), ) def test_scene_detects_whether_user_named_a_concrete_control_object( message: str, expected: bool, ) -> None: assert scene_has_explicit_control_object(message, _context()) is expected