vee/Cognitive_Agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

添加NFC绑定红外设备方式，融合车端设备和扩展设备统一控制

Browse Source
This commit is contained in:
minghanchen
2026-09-01 11:04:59 +08:00
parent 6d75944c0f
commit ca26259e63
14 changed files with 3694 additions and 316 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/e2e/test_seat_web_api.py
+595 -15
View File
@@ -33,6 +33,7 @@ class SeatWebApiTests(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
cls.temporary_directory = tempfile.TemporaryDirectory()
cls.extension_temporary_directory = tempfile.TemporaryDirectory()
# Web regression tests must never inherit a developer's live bench
# enablement, especially when local debug mode intentionally skips the
# per-request approval token.
@@ -50,6 +51,13 @@ class SeatWebApiTests(unittest.TestCase):
cls.service = service
cls.original_service = SeatDemoHandler.service
SeatDemoHandler.service = service
cls.original_extension = SeatDemoHandler.extension
SeatDemoHandler.extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(
cls.extension_temporary_directory.name, "gscu-ir-catalog.sqlite3"
),
)
cls.original_auth = SeatDemoHandler.auth
SeatDemoHandler.auth = AuthRuntime(
AuthConfig(
@@ -79,7 +87,9 @@ class SeatWebApiTests(unittest.TestCase):
cls.server.server_close()
cls.thread.join(timeout=2)
SeatDemoHandler.service = cls.original_service
SeatDemoHandler.extension = cls.original_extension
SeatDemoHandler.auth = cls.original_auth
cls.extension_temporary_directory.cleanup()
cls.temporary_directory.cleanup()
def setUp(self) -> None:
@@ -341,19 +351,19 @@ operations:
) as response:
gesture_script = response.read().decode("utf-8")
with self.opener.open(
f"{self.base_url}/console-v2.js?v=task-outcome-47", timeout=3
f"{self.base_url}/console-v2.js?v=delivery-state-49", timeout=3
) as response:
console_script = response.read().decode("utf-8")
with self.opener.open(
f"{self.base_url}/console-v2.css?v=sync-layout-41", timeout=3
f"{self.base_url}/console-v2.css?v=delivery-state-43", timeout=3
) as response:
console_styles = response.read().decode("utf-8")
with self.opener.open(
f"{self.base_url}/self-extension.js?v=gscu-tcp-37", timeout=3
f"{self.base_url}/self-extension.js?v=gscu-tcp-40", timeout=3
) as response:
extension_script = response.read().decode("utf-8")
with self.opener.open(
f"{self.base_url}/self-extension.css?v=gscu-tcp-37", timeout=3
f"{self.base_url}/self-extension.css?v=gscu-tcp-39", timeout=3
) as response:
extension_styles = response.read().decode("utf-8").replace("\r\n", "\n")
with self.opener.open(
@@ -388,13 +398,13 @@ operations:
self.assertIn('href="/styles.css?v=dual-seat-16"', html)
self.assertIn('src="/gesture-control.js?v=android-pad-9"', html)
self.assertNotIn('class="vehicle-system-hud"', html)
self.assertIn('src="/console-v2.js?v=task-outcome-47"', html)
self.assertIn('href="/console-v2.css?v=sync-layout-41"', html)
self.assertIn('src="/console-v2.js?v=delivery-state-49"', html)
self.assertIn('href="/console-v2.css?v=delivery-state-43"', html)
self.assertIn('<body class="console-v2">', html)
self.assertIn('id="v2BootScreen"', html)
self.assertIn('正在加载车机控制与实时状态', html)
self.assertIn('src="/self-extension.js?v=gscu-tcp-37"', html)
self.assertIn('href="/self-extension.css?v=gscu-tcp-37"', html)
self.assertIn('src="/self-extension.js?v=gscu-tcp-40"', html)
self.assertIn('href="/self-extension.css?v=gscu-tcp-39"', html)
self.assertNotIn("self-calibration", html)
self.assertNotIn("self-repair", html)
self.assertNotIn("self-evolution", html)
@@ -639,7 +649,7 @@ operations:
self.assertIn("CAN 指令已提交，等待车辆状态确认", console_script)
self.assertIn('canvas.addEventListener("webglcontextlost"', app_script)
self.assertIn('"cognitive-agent:module-change"', app_script)
self.assertIn('const CACHE_VERSION = "cognitive-agent-pad-v214"', service_worker)
self.assertIn('const CACHE_VERSION = "cognitive-agent-pad-v218"', service_worker)
self.assertIn('"/app.js?v=dual-seat-107"', service_worker)
self.assertIn('"/assets/pets/self-extension-cat.png?v=3"', service_worker)
self.assertIn(".then(() => self.skipWaiting())", service_worker)
@@ -1013,7 +1023,12 @@ operations:
def test_self_extension_gateway_uses_unified_authenticated_service(self) -> None:
original_extension = SeatDemoHandler.extension
SeatDemoHandler.extension = SelfExtensionService(SimulationGateway())
SeatDemoHandler.extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(
self.extension_temporary_directory.name, "gateway-auth.sqlite3"
),
)
try:
with self.opener.open(
f"{self.base_url}/api/extension/health", timeout=3
@@ -1039,7 +1054,12 @@ operations:
def test_self_extension_tcp_configuration_endpoint_updates_runtime(self) -> None:
original_extension = SeatDemoHandler.extension
extension = SelfExtensionService(SimulationGateway())
extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(
self.extension_temporary_directory.name, "tcp-config.sqlite3"
),
)
SeatDemoHandler.extension = extension
try:
with tempfile.TemporaryDirectory() as directory:
@@ -1093,11 +1113,38 @@ operations:
with self.opener.open(request, timeout=3) as response:
self.assertEqual(response.status, 200)
structured_request = urllib.request.Request(
f"{self.base_url}/api/extension/gscu/slots/5",
method="PUT",
headers={"Content-Type": "application/json"},
data=json.dumps(
{
"device_id": "toy-light-001",
"object_names": ["闪烁灯", "闪灯玩具"],
"action_names": ["切换为蓝色", "设置为蓝色"],
"source": "tcp",
},
ensure_ascii=False,
).encode("utf-8"),
)
with self.opener.open(structured_request, timeout=3) as response:
structured = json.loads(response.read().decode("utf-8"))
self.assertEqual(structured["slot"]["device_id"], "toy-light-001")
self.assertEqual(
structured["slot"]["object_names"], ["闪烁灯", "闪灯玩具"]
)
self.assertEqual(
structured["slot"]["action_names"],
["切换为蓝色", "设置为蓝色"],
)
with self.opener.open(
f"{self.base_url}/api/extension/gscu/slots", timeout=3
) as response:
catalog = json.loads(response.read().decode("utf-8"))
self.assertEqual([item["slot"] for item in catalog["slots"]], [3, 12])
self.assertEqual(
[item["slot"] for item in catalog["slots"]], [3, 5, 12]
)
delete_request = urllib.request.Request(
f"{self.base_url}/api/extension/gscu/slots/3",
@@ -1117,7 +1164,7 @@ operations:
with self.opener.open(erase_request, timeout=3) as response:
erased = json.loads(response.read().decode("utf-8"))
self.assertTrue(erased["catalog_cleared"])
self.assertEqual(erased["catalog_removed_count"], 1)
self.assertEqual(erased["catalog_removed_count"], 2)
with self.opener.open(
f"{self.base_url}/api/extension/gscu/slots", timeout=3
@@ -1130,7 +1177,12 @@ operations:
def test_main_request_routes_live_nfc_peripheral_to_gscu(self) -> None:
original_extension = SeatDemoHandler.extension
original_nfc = SeatDemoHandler.nfc
SeatDemoHandler.extension = SelfExtensionService(SimulationGateway())
SeatDemoHandler.extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(
self.extension_temporary_directory.name, "main-nfc-route.sqlite3"
),
)
SeatDemoHandler.nfc = SimpleNamespace(
snapshot=lambda: {
"revision": 1,
@@ -1197,10 +1249,538 @@ operations:
self.assertEqual(operation["domain"], "self_extension")
self.assertTrue(operation["telemetry"]["gateway_acknowledged"])
def test_main_request_combines_explicit_ir_clause_with_existing_can_parser(self) -> None:
original_extension = SeatDemoHandler.extension
original_nfc = SeatDemoHandler.nfc
extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(
self.extension_temporary_directory.name, "mixed-nfc-can.sqlite3"
),
)
SeatDemoHandler.extension = extension
SeatDemoHandler.nfc = SimpleNamespace(
snapshot=lambda: {
"revision": 1,
"reader": {"online": True},
"card": {"action": "present", "uid": "2CC8B103"},
}
)
catalog_request = urllib.request.Request(
f"{self.base_url}/api/extension/display-sync",
method="POST",
headers={"Content-Type": "application/json"},
data=json.dumps(
{
"peripherals": [
{
"id": "toy-001-light-on",
"status": "learned",
"enabled": True,
"device_id": "toy-001",
"device_name": "闪灯玩具",
"object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
"card_uid": "2CC8B103",
"action": "开启",
"action_names": ["打开", "开启", "启动"],
"risk": "low",
"slot": 1,
},
{
"id": "toy-001-light-red",
"status": "learned",
"enabled": True,
"device_id": "toy-001",
"device_name": "闪灯玩具",
"object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
"card_uid": "2CC8B103",
"action": "切换到红色",
"action_names": ["切换到红色", "切换为红色", "设置为红色"],
"risk": "low",
"slot": 3,
},
]
},
ensure_ascii=False,
).encode("utf-8"),
)
request = urllib.request.Request(
f"{self.base_url}/api/requests",
method="POST",
headers={
"Content-Type": "application/json",
"Accept": "application/x-ndjson",
},
data=json.dumps(
{
"utterance": "打开闪烁灯，颜色切换到红色，打开空调",
"provider": "local",
"conversation_id": "conv-mixed-nfc-can",
"execution_mode": "dry_run",
"client_type": "web_console",
},
ensure_ascii=False,
).encode("utf-8"),
)
try:
with self.opener.open(catalog_request, timeout=3) as response:
synced = json.loads(response.read().decode("utf-8"))
with self.opener.open(request, timeout=3) as response:
events = [
json.loads(line)
for line in response.read().decode("utf-8").splitlines()
if line.strip()
]
result = events[-1]["payload"]
finally:
SeatDemoHandler.extension = original_extension
SeatDemoHandler.nfc = original_nfc
self.assertEqual(synced["learned_count"], 2)
self.assertEqual(events[-1]["type"], "result")
self.assertTrue(any(item["type"] == "validated_commands" for item in events))
validated_payloads = [
item["payload"]
for item in events
if item["type"] == "validated_commands"
]
self.assertTrue(validated_payloads)
self.assertTrue(
all(
"self_extension" in payload["function_configs"]
for payload in validated_payloads
)
)
self.assertEqual(result["status"], "succeeded")
self.assertEqual(result["orchestration_route"], "unified_can_infrared")
self.assertEqual(
[item["slot"] for item in result["infrared_actions"]], [1, 3]
)
self.assertEqual(
result["infrared_actions"][0]["parse_source"], "local_rule"
)
self.assertIn("self_extension", result["function_configs"])
self.assertTrue(
any(key != "self_extension" for key in result["function_configs"])
)
self.assertTrue(
any(
item.get("domain") == "self_extension"
for item in result["operation_results"]
)
)
def test_hardware_can_failure_keeps_infrared_in_validated_command_event(
self,
) -> None:
original_extension = SeatDemoHandler.extension
original_nfc = SeatDemoHandler.nfc
extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(
self.extension_temporary_directory.name,
"mixed-hardware-validation.sqlite3",
),
)
extension.sync_peripheral_catalog(
[
{
"id": "ir-slot:tony-light-001:1",
"status": "learned",
"enabled": True,
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
"card_uid": "2CC8B103",
"action": "打开",
"action_names": ["打开", "开启", "启动"],
"risk": "low",
"slot": 1,
},
{
"id": "ir-slot:tony-light-001:4",
"status": "learned",
"enabled": True,
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
"card_uid": "2CC8B103",
"action": "切换到绿色",
"action_names": ["切换到绿色", "切换为绿色", "设置为绿色"],
"risk": "low",
"slot": 4,
},
]
)
SeatDemoHandler.extension = extension
SeatDemoHandler.nfc = SimpleNamespace(
snapshot=lambda: {
"reader": {"online": True},
"card": {"action": "present", "uid": "2CC8B103"},
}
)
request = urllib.request.Request(
f"{self.base_url}/api/requests",
method="POST",
headers={
"Content-Type": "application/json",
"Accept": "application/x-ndjson",
},
data=json.dumps(
{
"utterance": (
"打开空调，温度设置为24度，风速调整为3档，"
"打开遮阳帘，开启闪光灯，颜色设置为绿色"
),
"provider": "local",
"conversation_id": "conv-hardware-mixed-validation",
"execution_mode": "hardware",
"control_transport": "wired_can",
"client_type": "web_console",
},
ensure_ascii=False,
).encode("utf-8"),
)
def fail_can_after_validation(*_args, **kwargs):
callback = kwargs.get("_on_commands_validated")
self.assertTrue(callable(callback))
callback(
{
"request_id": "req-can-validation-only",
"utterance": "打开空调，温度设置为24度，风速调整为3档，打开遮阳帘",
"status": "processing",
"execution_mode": "hardware",
"control_transport": "wired_can",
"provider_used": "local_rules",
"decision_summary": [],
"function_configs": {
"native_climate": {
"spec": {
"operations": [
{
"id": "vehicle_set_power_1",
"capability": "climate.set_power",
"target": {"zone": "vehicle"},
"parameters": {"state": "on"},
"parse_source": "local_rule",
}
]
}
}
},
"command_plans": {
"native_climate": {"spec": {"commands": []}}
},
"operation_results": [],
"execution_report": None,
}
)
raise CanGatewayError("simulated CAN transport failure")
try:
with patch.object(
self.service,
"process",
side_effect=fail_can_after_validation,
):
with self.opener.open(request, timeout=3) as response:
events = [
json.loads(line)
for line in response.read().decode("utf-8").splitlines()
if line.strip()
]
finally:
SeatDemoHandler.extension = original_extension
SeatDemoHandler.nfc = original_nfc
self.assertEqual(events[-1]["type"], "error")
self.assertEqual(events[-1]["error"], "can_transport_failed")
validated = next(
item["payload"]
for item in events
if item["type"] == "validated_commands"
)
self.assertEqual(validated["status"], "processing")
self.assertEqual(
set(validated["function_configs"]),
{"native_climate", "self_extension"},
)
self.assertEqual(
[item["slot"] for item in validated["infrared_actions"]],
[1, 4],
)
def test_disabled_nfc_ir_command_does_not_discard_allowed_can_and_ir(
self,
) -> None:
original_extension = SeatDemoHandler.extension
original_nfc = SeatDemoHandler.nfc
extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(
self.extension_temporary_directory.name,
"mixed-disabled-nfc-command.sqlite3",
),
)
extension.sync_peripheral_catalog(
[
{
"id": "ir-slot:tony-light-001:1",
"status": "learned",
"enabled": True,
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
"card_uid": "2CC8B103",
"action": "打开",
"action_names": ["打开", "开启", "启动"],
"risk": "low",
"slot": 1,
},
{
"id": "ir-slot:tony-light-001:5",
"status": "learned",
"enabled": False,
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
"card_uid": "2CC8B103",
"action": "切换到蓝色",
"action_names": ["切换到蓝色", "切换为蓝色", "设置为蓝色"],
"risk": "low",
"slot": 5,
},
]
)
SeatDemoHandler.extension = extension
SeatDemoHandler.nfc = SimpleNamespace(
snapshot=lambda: {
"revision": 1,
"reader": {"online": True},
"card": {"action": "present", "uid": "2CC8B103"},
}
)
request = urllib.request.Request(
f"{self.base_url}/api/requests",
method="POST",
headers={
"Content-Type": "application/json",
"Accept": "application/x-ndjson",
},
data=json.dumps(
{
"utterance": (
"打开空调，温度设置为24度，风速调整为3档，"
"开启闪光灯，颜色设置为蓝色"
),
"provider": "openai",
"conversation_id": "conv-disabled-nfc-command",
"execution_mode": "dry_run",
"client_type": "web_console",
},
ensure_ascii=False,
).encode("utf-8"),
)
try:
with patch.object(
extension,
"plan",
side_effect=AssertionError(
"a disabled exact command must not be sent to the cloud"
),
):
with self.opener.open(request, timeout=3) as response:
events = [
json.loads(line)
for line in response.read().decode("utf-8").splitlines()
if line.strip()
]
result = events[-1]["payload"]
finally:
SeatDemoHandler.extension = original_extension
SeatDemoHandler.nfc = original_nfc
self.assertEqual(events[-1]["type"], "result")
self.assertEqual(result["status"], "partially_succeeded")
self.assertEqual(result["orchestration_route"], "unified_can_infrared")
self.assertEqual(
[item["slot"] for item in result["infrared_actions"]], [1]
)
self.assertEqual(
[item["slot"] for item in result["denied_peripheral_actions"]],
[5],
)
self.assertEqual(
result["denied_peripheral_actions"][0]["reason"],
"nfc_command_disabled",
)
debug_commands = [
command["signals"]["debug_command"]
for plan in result["command_plans"].values()
for command in plan["spec"].get("commands", [])
if command.get("signals", {}).get("debug_command")
]
self.assertEqual(
debug_commands,
["ac on", "ac temp driver 24", "ac fan 3"],
)
infrared_slots = [
command["signals"]["slot"]
for command in result["command_plans"]["self_extension"]["spec"][
"commands"
]
]
self.assertEqual(infrared_slots, [1])
def test_generic_vehicle_light_text_never_routes_to_ir_alias(self) -> None:
original_extension = SeatDemoHandler.extension
original_nfc = SeatDemoHandler.nfc
extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(
self.extension_temporary_directory.name, "vehicle-light-not-ir.sqlite3"
),
)
extension.sync_peripheral_catalog(
[
{
"id": "toy-001-light-on",
"status": "learned",
"device_id": "toy-001",
"device_name": "玩具 001",
"device_aliases": ["氛围灯"],
"card_uid": "2CC8B103",
"action": "开启闪灯",
"aliases": ["打开"],
"risk": "low",
"slot": 1,
}
]
)
SeatDemoHandler.extension = extension
SeatDemoHandler.nfc = SimpleNamespace(
snapshot=lambda: {
"reader": {"online": True},
"card": {"action": "present", "uid": "2CC8B103"},
}
)
request = urllib.request.Request(
f"{self.base_url}/api/requests",
method="POST",
headers={"Content-Type": "application/json"},
data=json.dumps(
{
"utterance": "打开氛围灯，空调24度",
"provider": "local",
"conversation_id": "conv-vehicle-light-not-ir",
"execution_mode": "dry_run",
"client_type": "web_console",
},
ensure_ascii=False,
).encode("utf-8"),
)
try:
with self.opener.open(request, timeout=3) as response:
result = json.loads(response.read().decode("utf-8"))
finally:
SeatDemoHandler.extension = original_extension
SeatDemoHandler.nfc = original_nfc
self.assertNotIn("infrared_actions", result)
self.assertNotEqual(result.get("orchestration_route"), "unified_can_infrared")
self.assertNotIn("self_extension", result["function_configs"])
self.assertTrue(result["function_configs"])
def test_mixed_preview_reports_cloud_selected_infrared_capability(self) -> None:
original_extension = SeatDemoHandler.extension
original_nfc = SeatDemoHandler.nfc
extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(
self.extension_temporary_directory.name,
"mixed-cloud-nfc-can.sqlite3",
),
)
extension.sync_peripheral_catalog(
[
{
"id": "toy-001-light-red",
"status": "learned",
"enabled": True,
"device_id": "toy-001",
"device_name": "闪灯玩具",
"object_names": ["闪灯玩具", "闪烁灯"],
"card_uid": "2CC8B103",
"action": "切换到红色",
"action_names": ["切换到红色", "切换为红色"],
"slot": 3,
}
]
)
SeatDemoHandler.extension = extension
SeatDemoHandler.nfc = SimpleNamespace(
snapshot=lambda: {
"reader": {"online": True},
"card": {"action": "present", "uid": "2CC8B103"},
}
)
request = urllib.request.Request(
f"{self.base_url}/api/requests/preview",
method="POST",
headers={
"Content-Type": "application/json",
"Accept": "application/x-ndjson",
},
data=json.dumps(
{
"utterance": "把闪烁灯调成喜庆颜色，打开空调",
"provider": "openai",
"conversation_id": "conv-mixed-cloud-ir",
"execution_mode": "dry_run",
"client_type": "web_console",
},
ensure_ascii=False,
).encode("utf-8"),
)
plan_result = {
"model": "gpt-5.6-luna",
"summary": "喜庆颜色对应已学习的红色动作。",
"needs_confirmation": False,
"steps": [
{"kind": "peripheral", "capability_id": "toy-001-light-red"}
],
}
try:
with patch.object(extension, "plan", return_value=plan_result) as planner:
with self.opener.open(request, timeout=3) as response:
events = [
json.loads(line)
for line in response.read().decode("utf-8").splitlines()
if line.strip()
]
result = events[-1]["payload"]
finally:
SeatDemoHandler.extension = original_extension
SeatDemoHandler.nfc = original_nfc
planner.assert_called_once()
self.assertEqual(result["orchestration_route"], "unified_can_infrared")
self.assertIn("gpt:gpt-5.6-luna", result["provider_used"])
self.assertEqual(
result["infrared_actions"][0]["parse_source"], "cloud_model"
)
self.assertFalse(result["supplement_required"])
def test_main_request_executes_gpt_nfc_sequence_without_vehicle_can(self) -> None:
original_extension = SeatDemoHandler.extension
original_nfc = SeatDemoHandler.nfc
extension = SelfExtensionService(SimulationGateway())
extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(
self.extension_temporary_directory.name, "main-nfc-sequence.sqlite3"
),
)
SeatDemoHandler.extension = extension
SeatDemoHandler.nfc = SimpleNamespace(
snapshot=lambda: {
tests/unit/test_nfc_frontend_contract.py
+53 -16
View File
@@ -10,27 +10,33 @@ def source(name: str) -> str:
return (WEB / name).read_text(encoding="utf-8")
def test_unified_console_exposes_nfc_discovery_and_learning_flow() -> None:
def test_unified_console_exposes_device_id_based_nfc_authorization() -> None:
script = source("self-extension.js")
styles = source("self-extension.css")
html = source("index.html")
assert 'data-open-nfc-learning' in script
assert 'data-open-nfc-binding' in script
assert '靠近读卡器后' in script
assert '感应玩具 NFC' in script
assert 'NFC · DEVICE AUTHORIZATION' in script
assert 'NFC 设备授权' in script
assert 'api("nfc/status")' in script
assert 'function createToyFromNfc()' in script
assert 'function bindCardToDevice(device, cardUid)' in script
assert 'function infraredDeviceGroups()' in script
assert 'function configuredCapabilities(deviceId = null)' in script
assert 'data-nfc-device-select' in script
assert 'data-nfc-capability-enabled' in script
assert 'data-nfc-select-all' in script
assert '全部命令已默认选中' in script
assert 'function createToyFromNfc()' not in script
assert 'data-nfc-learn-capability' not in script
assert 'data-nfc-relearn-capability' not in script
assert 'data-nfc-continue-learning' not in script
assert 'function importLegacyProfiles(value)' in script
assert 'N7WEB_REQUEST_IR_TOY_PROFILES' in script
assert 'IR_TOY_LEGACY_PROFILES' in script
assert 'value.legacy_profiles' in script
assert 'value.legacy_bridge_addresses' in script
assert 'cardUid: normalizeCardUid(cardUid)' in script
assert 'data-nfc-run-capability' in script
assert 'data-nfc-learn-capability' in script
assert 'data-nfc-relearn-capability' in script
assert '重新学习' in script
assert 'device.cardUid = normalizedUid' in script
assert 'const targetSlot = relearning ? Number(capability.slot) : allocateSlot()' in script
assert 'sendGatewayCommand("read", { slot: learning.draft.slot })' in script
assert 'function infraredSignalFromResponse(value)' in script
@@ -38,19 +44,24 @@ def test_unified_console_exposes_nfc_discovery_and_learning_flow() -> None:
assert '学习指令已下发，但物理槽位回读为空' in script
assert 'signalSource: "n7web"' in script
assert 'if (capability.signalSource === "n7web") return' in script
assert '建立档案并开始学习' in script
assert 'const activeCatalog = sensedCardUid' in script
assert 'normalizeCardUid(item.card_uid) === sensedCardUid' in script
assert '&& item.status === "learned"' in script
assert 'slot: capability.slot' in script
assert 'device_aliases: device?.aliases || []' in script
assert 'async function hydratePeripheralCatalog()' in script
assert 'api("peripherals")' in script
assert 'signalSource: "backend_catalog"' in script
assert '当前未感应到外设' in script
assert '将已绑定且已学习的 NFC 外设靠近读卡器后显示能力' in script
assert '将已绑定设备 ID 的 NFC 卡靠近读卡器后显示命令' in script
assert '.v2-nfc-inline' in styles
assert '.v2-nfc-dialog' in styles
assert '.v2-nfc-capability-grid' in styles
assert 'self-extension.js?v=gscu-tcp-37' in html
assert 'self-extension.css?v=gscu-tcp-37' in html
assert '.v2-nfc-command-list' in styles
assert '.v2-nfc-command-option' in styles
assert '.v2-nfc-select-all' in styles
assert 'self-extension.js?v=gscu-tcp-40' in html
assert 'self-extension.css?v=gscu-tcp-39' in html
def test_unified_console_exposes_gscu_tcp_configuration_and_actions() -> None:
@@ -59,7 +70,10 @@ def test_unified_console_exposes_gscu_tcp_configuration_and_actions() -> None:
assert 'id="extensionGscuTcpHost"' in script
assert 'id="extensionGscuTcpPort"' in script
assert 'id="extensionGscuCommandName"' in script
assert 'id="extensionGscuDeviceId"' in script
assert 'id="extensionGscuObjectNames"' in script
assert 'id="extensionGscuActionNames"' in script
assert 'id="extensionGscuTcpResultDetails"' in script
assert 'id="extensionGscuNamedCommands"' in script
assert 'data-open-gscu-tcp' in script
assert 'api("gscu/config"' in script
@@ -71,7 +85,12 @@ def test_unified_console_exposes_gscu_tcp_configuration_and_actions() -> None:
assert "学习指令已下发，但槽位回读失败" in script
assert "确认擦除 GSCU1B 中的全部红外码吗" in script
assert "GSCU1B 设备中的该红外码不会被擦除" in script
assert "function upsertNamedInfraredCommand(slot, name" in script
assert "async function upsertNamedInfraredCommand(" in script
assert "device_id: normalizedDeviceId" in script
assert "object_names: normalizedObjects" in script
assert "action_names: normalizedActions" in script
assert "object_names: objectNames" in script
assert "action_names: actionNames" in script
assert "button.disabled = true;" in script
assert "buttons.forEach((item) => { item.disabled = true; });" not in script
for action in (
@@ -92,6 +111,10 @@ def test_unified_console_exposes_gscu_tcp_configuration_and_actions() -> None:
assert ".v2-gscu-config-form" in styles
assert "#extensionGscuTcpResult" in styles
assert "#extensionGscuNamedCommands" in styles
assert ".v2-gscu-device-group" in styles
assert ".v2-gscu-device-command-grid" in styles
assert ".v2-gscu-result" in styles
assert "max-height: min(42vh, 360px)" in styles
def test_server_exposes_nfc_snapshot_through_same_origin_extension_api() -> None:
@@ -99,7 +122,21 @@ def test_server_exposes_nfc_snapshot_through_same_origin_extension_api() -> None
assert 'nfc = NfcReaderService()' in server
assert 'path == "/api/extension/nfc/status"' in server
assert 'path == "/api/extension/peripherals"' in server
assert '"nfc": self.nfc.snapshot()' in server
assert 'SeatDemoHandler.nfc.start()' in server
assert 'SeatDemoHandler.nfc.close()' in server
assert "frame-src 'self' http://*:3001" in server
def test_main_parsed_control_panel_renders_validated_infrared_operations() -> None:
console = source("console-v2.js")
styles = source("console-v2.css")
html = source("index.html")
assert 'capability === "infrared.emit"' in console
assert 'channel: "IR"' in console
assert 'frameLabel: "GSCU1B TCP"' in console
assert "已解析控制指令" in console
assert 'li[data-can-channel="IR"]' in styles
assert "console-v2.js?v=delivery-state-49" in html
tests/unit/test_requested_frontend_enhancements.py
+106 -1
View File
@@ -200,7 +200,9 @@ def test_left_sidebar_exposes_collapsed_parsed_can_command_panel() -> None:
assert "function parsedCanCommands(result = {})" in console
assert "getParsedCanCommands: (result) => parsedCanCommands(result)" in console
assert "const configurations = result.function_configs || {};" in console
assert "const plans = result.command_plans || {};" in console
assert "if (!configurationEntries.length && result.function_config)" in console
assert 'deliveryLabel: submitted ? "下发" : "未下发"' in console
assert 'data-delivery-state="${escapeHtml(item.deliveryState)}"' in console
assert 'const command = (category, text, channel = "CAN1", frame = "")' in console
assert 'data-can-channel="${escapeHtml(item.channel)}"' in console
assert 'class="v2-can-command-channel"' in console
@@ -243,6 +245,8 @@ def test_left_sidebar_exposes_collapsed_parsed_can_command_panel() -> None:
assert ".v2-can-command-panel[open]" in styles
assert ".v2-can-command-list li code" in styles
assert ".v2-can-command-channel" in styles
assert ".v2-can-command-delivery" in styles
assert '.v2-can-command-delivery[data-delivery-state="submitted"]' in styles
assert 'li[data-can-channel="CAN0"]' in styles
assert ".v2-sidebar-collapsed .v2-can-command-panel" in styles
@@ -306,16 +310,117 @@ def test_parsed_can_command_panel_combines_can0_seat_and_can1_vehicle_commands()
"command": "seat passenger backrest position 36",
"frame": "0x300 · 01 12 <SEQ_LO> <SEQ_HI> 10 0E 30 75",
"parseSource": "local_rule",
"deliveryState": "not_submitted",
"deliveryLabel": "未下发",
"deliveryTransport": "有线 CAN",
},
{
"channel": "CAN1",
"category": "空调温度",
"command": "ac temp driver 24",
"parseSource": "cloud_supplement",
"deliveryState": "not_submitted",
"deliveryLabel": "未下发",
"deliveryTransport": "有线 CAN",
},
]
def test_parsed_command_delivery_is_independent_of_transport_and_target() -> None:
node = shutil.which("node")
if node is None:
return
console = source("console-v2.js")
start = console.index(" const normalizedCanTarget")
end = console.index("\n\n function renderParsedCanCommands", start)
helpers = console[start:end]
base = {
"function_configs": {
"native_climate": {
"spec": {
"operations": [
{
"id": "climate-power",
"capability": "climate.set_power",
"target": {"zone": "vehicle"},
"parameters": {"state": "on"},
"parse_source": "local_rule",
}
]
}
}
}
}
samples = [
{**base, "execution_mode": "dry_run", "control_transport": "wired_can"},
{
**base,
"execution_mode": "dry_run",
"control_transport": "wireless_mqtt",
},
{
**base,
"execution_mode": "hardware",
"control_transport": "wired_can",
"execution_report": {
"spec": {
"results": [
{
"ref": "climate-power",
"status": "succeeded",
"telemetry": {"sent_frames": [{"frame_id": "0x1DE"}]},
}
]
}
},
},
{
**base,
"execution_mode": "hardware",
"control_transport": "wireless_mqtt",
"execution_report": {
"spec": {
"results": [
{
"ref": "climate-power",
"status": "succeeded",
"telemetry": {"delivery_status": "queued"},
}
]
}
},
},
]
script = (
f"{helpers}\n"
f"const samples = {json.dumps(samples)};\n"
"process.stdout.write(JSON.stringify(samples.map((sample) => "
"parsedCanCommands(sample))));"
)
completed = subprocess.run(
[node, "-e", script],
check=True,
capture_output=True,
text=True,
)
parsed = json.loads(completed.stdout)
assert [len(commands) for commands in parsed] == [1, 1, 1, 1]
assert [commands[0]["command"] for commands in parsed] == ["ac on"] * 4
assert [commands[0]["deliveryLabel"] for commands in parsed] == [
"未下发",
"未下发",
"下发",
"下发",
]
assert [commands[0]["deliveryTransport"] for commands in parsed] == [
"有线 CAN",
"无线 CAN",
"有线 CAN",
"无线 CAN",
]
def test_mode_editor_renders_readable_commands_with_expandable_mqtt_sequences() -> None:
app = source("app.js")
styles = source("styles.css")
tests/unit/test_self_extension.py
+554
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import json
import os
import socket
import sqlite3
import threading
import time
import unittest
@@ -22,6 +23,7 @@ from cognitive_agent.application.self_extension import (
active_peripheral_plan_catalog,
build_gateway_command,
create_gateway,
partition_active_peripheral_intent,
resolve_active_peripheral_command,
validate_slot,
)
@@ -78,6 +80,216 @@ class SelfExtensionTests(unittest.TestCase):
with self.assertRaises(GatewayError):
build_gateway_command("erase", {"confirm": "wrong"})
def test_learned_peripheral_whitelist_is_durable_and_normalized(self) -> None:
with TemporaryDirectory() as directory:
database_path = Path(directory, "infrared-slots.sqlite3")
service = SelfExtensionService(
SimulationGateway(), catalog_db_path=database_path
)
synced = service.sync_peripheral_catalog(
[
{
"id": "toy-001-light-on",
"status": "learned",
"enabled": True,
"device_id": "toy-001",
"device_name": "玩具 001",
"device_aliases": ["闪灯玩具", "闪灯玩具"],
"card_uid": "2c:c8:b1:03",
"action": "开启闪灯",
"aliases": ["打开闪灯", "点亮"],
"risk": "low",
"slot": 1,
}
]
)
self.assertEqual(synced["count"], 1)
with sqlite3.connect(database_path) as connection:
connection.execute(
"UPDATE infrared_slots SET device_id = '', "
"object_names_json = '[]', action_names_json = '[]'"
)
reloaded = SelfExtensionService(
SimulationGateway(), catalog_db_path=database_path
).list_peripheral_capabilities()
self.assertEqual(len(reloaded), 1)
self.assertEqual(reloaded[0]["card_uid"], "2CC8B103")
self.assertEqual(reloaded[0]["device_aliases"], ["闪灯玩具"])
self.assertEqual(reloaded[0]["object_names"], ["玩具 001", "闪灯玩具"])
self.assertEqual(
reloaded[0]["action_names"], ["开启闪灯", "打开闪灯", "点亮"]
)
self.assertEqual(reloaded[0]["slot"], 1)
slot = service.list_infrared_slots()["slots"][0]
self.assertEqual(slot["device_id"], "toy-001")
self.assertEqual(slot["object_names"], reloaded[0]["object_names"])
self.assertEqual(slot["action_names"], reloaded[0]["action_names"])
def test_partition_claims_explicit_ir_clause_and_preserves_can_residual(self) -> None:
catalog = [
{
"id": "toy-001-light-on",
"status": "learned",
"enabled": True,
"device_id": "toy-001",
"device_name": "玩具 001",
"device_aliases": ["闪灯玩具", "氛围灯"],
"card_uid": "2CC8B103",
"action": "开启闪灯",
"aliases": ["打开闪灯", "点亮"],
"risk": "low",
"slot": 1,
}
]
present = {"card": {"action": "present", "uid": "2CC8B103"}}
partition = partition_active_peripheral_intent(
"打开闪灯玩具，同时把空调调到24度，风量2档",
catalog,
present,
)
self.assertEqual(
[item["capability"]["id"] for item in partition["peripheral_actions"]],
["toy-001-light-on"],
)
self.assertEqual(partition["residual_utterance"], "把空调调到24度，风量2档")
self.assertEqual(partition["unresolved_peripheral_clauses"], [])
def test_partition_inherits_explicit_device_for_followup_ir_action(self) -> None:
catalog = [
{
"id": capability_id,
"status": "learned",
"enabled": True,
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
"card_uid": "2CC8B103",
"action": action,
"action_names": aliases,
"slot": slot,
}
for capability_id, action, aliases, slot in (
("light-on", "开启", ["打开", "开启", "启动"], 1),
(
"light-red",
"切换到红色",
["切换到红色", "切换为红色", "设置为红色"],
3,
),
("light-off", "关闭", ["关闭", "关掉", "熄灭"], 2),
)
]
partition = partition_active_peripheral_intent(
"打开闪烁灯，颜色切换到红色，打开空调",
catalog,
{"card": {"uid": "2CC8B103"}},
)
self.assertEqual(
[item["capability"]["id"] for item in partition["peripheral_actions"]],
["light-on", "light-red"],
)
self.assertEqual(
[item["clause"] for item in partition["peripheral_actions"]],
["打开闪烁灯", "颜色切换到红色"],
)
self.assertEqual(partition["residual_utterance"], "打开空调")
self.assertEqual(partition["unresolved_peripheral_clauses"], [])
vehicle_breaks_context = partition_active_peripheral_intent(
"打开闪烁灯，打开空调，然后关闭",
catalog,
{"card": {"uid": "2CC8B103"}},
)
self.assertEqual(
[
item["capability"]["id"]
for item in vehicle_breaks_context["peripheral_actions"]
],
["light-on"],
)
self.assertEqual(
vehicle_breaks_context["residual_utterance"], "打开空调，关闭"
)
def test_partition_filters_disabled_followup_without_losing_allowed_work(
self,
) -> None:
catalog = [
{
"id": capability_id,
"status": "learned",
"enabled": enabled,
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"object_names": ["闪灯玩具", "闪烁灯", "闪光灯"],
"card_uid": "2CC8B103",
"action": action,
"action_names": aliases,
"slot": slot,
}
for capability_id, action, aliases, slot, enabled in (
("light-on", "开启", ["打开", "开启", "启动"], 1, True),
(
"light-blue",
"切换到蓝色",
["切换到蓝色", "切换为蓝色", "设置为蓝色"],
5,
False,
),
)
]
partition = partition_active_peripheral_intent(
"打开空调，温度设置为24度，风速调整为3档，开启闪光灯，颜色设置为蓝色",
catalog,
{"card": {"uid": "2CC8B103"}},
)
self.assertEqual(
[item["capability"]["id"] for item in partition["peripheral_actions"]],
["light-on"],
)
self.assertEqual(
partition["residual_utterance"],
"打开空调，温度设置为24度，风速调整为3档",
)
self.assertEqual(partition["unresolved_peripheral_clauses"], [])
self.assertEqual(
[
(item["clause"], item["capability"]["id"], item["reason"])
for item in partition["denied_peripheral_actions"]
],
[("颜色设置为蓝色", "light-blue", "nfc_command_disabled")],
)
def test_partition_never_claims_generic_vehicle_light_request(self) -> None:
catalog = [
{
"id": "toy-001-light-on",
"status": "learned",
"device_id": "toy-001",
"device_name": "玩具 001",
"device_aliases": ["闪灯玩具", "氛围灯"],
"card_uid": "2CC8B103",
"action": "开启闪灯",
"slot": 1,
}
]
partition = partition_active_peripheral_intent(
"打开氛围灯，空调24度",
catalog,
{"card": {"uid": "2CC8B103"}},
)
self.assertEqual(partition["peripheral_actions"], [])
self.assertEqual(partition["residual_utterance"], "打开氛围灯，空调24度")
def test_infrared_slot_catalog_is_durable_sorted_and_device_safe(self) -> None:
with TemporaryDirectory() as directory:
database_path = Path(directory, "infrared-slots.sqlite3")
@@ -111,6 +323,127 @@ class SelfExtensionTests(unittest.TestCase):
[12],
)
def test_infrared_slot_catalog_migrates_legacy_schema_without_data_loss(self) -> None:
with TemporaryDirectory() as directory:
database_path = Path(directory, "infrared-slots.sqlite3")
with sqlite3.connect(database_path) as connection:
connection.execute(
"""
CREATE TABLE infrared_slots (
slot INTEGER PRIMARY KEY,
name TEXT NOT NULL,
source TEXT NOT NULL,
learned_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
"""
)
connection.execute(
"INSERT INTO infrared_slots VALUES (?, ?, ?, ?, ?)",
(1, "打开闪烁灯", "tcp", "before", "before"),
)
service = SelfExtensionService(
SimulationGateway(), catalog_db_path=database_path
)
slots = service.list_infrared_slots()["slots"]
self.assertEqual(slots[0]["name"], "打开闪烁灯")
self.assertEqual(slots[0]["device_id"], "")
self.assertEqual(slots[0]["object_names"], [])
self.assertEqual(slots[0]["action_names"], [])
with sqlite3.connect(database_path) as connection:
columns = {
row[1]
for row in connection.execute(
"PRAGMA table_info(infrared_slots)"
).fetchall()
}
self.assertIn("object_names_json", columns)
self.assertIn("action_names_json", columns)
self.assertIn("device_id", columns)
def test_structured_slot_names_persist_and_update_bound_whitelist(self) -> None:
with TemporaryDirectory() as directory:
database_path = Path(directory, "infrared-slots.sqlite3")
service = SelfExtensionService(
SimulationGateway(), catalog_db_path=database_path
)
service.sync_peripheral_catalog(
[
{
"id": "toy-light-on",
"status": "learned",
"device_id": "toy-light",
"device_name": "灯光玩具",
"card_uid": "2CC8B103",
"action": "开启灯光",
"slot": 1,
}
]
)
saved = service.save_infrared_slot(
1,
{
"device_id": "toy-light",
"object_names": "闪烁灯，闪光灯、闪灯玩具",
"action_names": ["打开", "开启", "启动"],
"source": "tcp",
},
)["slot"]
self.assertEqual(saved["device_id"], "toy-light")
self.assertEqual(saved["object_names"], ["闪烁灯", "闪光灯", "闪灯玩具"])
self.assertEqual(saved["action_names"], ["打开", "开启", "启动"])
reloaded = SelfExtensionService(
SimulationGateway(), catalog_db_path=database_path
)
self.assertEqual(
reloaded.list_infrared_slots()["slots"][0]["object_names"],
["闪烁灯", "闪光灯", "闪灯玩具"],
)
whitelist = reloaded.list_peripheral_capabilities()
self.assertEqual(whitelist[0]["object_names"], saved["object_names"])
self.assertEqual(whitelist[0]["action_names"], saved["action_names"])
resolved = resolve_active_peripheral_command(
"打开闪灯玩具",
whitelist,
{"card": {"uid": "2CC8B103"}},
)
self.assertEqual(resolved["id"], "toy-light-on")
changed = service.save_infrared_slot(
1,
{
"device_id": "toy-light-v2",
"object_names": ["新版闪烁灯"],
"action_names": ["打开"],
"source": "tcp",
},
)["slot"]
self.assertEqual(changed["device_id"], "toy-light-v2")
self.assertEqual(service.list_peripheral_capabilities(), [])
def test_structured_slot_rejects_invalid_or_missing_device_id(self) -> None:
with TemporaryDirectory() as directory:
service = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(directory, "infrared-slots.sqlite3"),
)
for device_id in (None, "含 空格", "-bad-prefix"):
with self.subTest(device_id=device_id), self.assertRaisesRegex(
GatewayError, "设备 ID"
):
service.save_infrared_slot(
1,
{
"device_id": device_id,
"object_names": ["闪烁灯"],
"action_names": ["打开"],
},
)
def test_erase_all_clears_catalog_only_after_valid_device_submission(self) -> None:
with TemporaryDirectory() as directory:
service = SelfExtensionService(
@@ -374,6 +707,47 @@ class SelfExtensionTests(unittest.TestCase):
)
)
def test_structured_color_actions_require_a_unique_result_value(self) -> None:
present = {"card": {"uid": "2CC8B103"}}
catalog = [
{
"id": capability_id,
"status": "learned",
"device_name": "灯光玩具",
"object_names": ["闪烁灯", "闪灯玩具"],
"card_uid": "2CC8B103",
"action": color,
"action_names": [f"设置为{color}", f"切换到{color}", f"切换为{color}"],
"slot": slot,
}
for capability_id, color, slot in (
("blue", "蓝色", 5),
("red", "红色", 6),
)
]
resolved = resolve_active_peripheral_command(
"闪烁灯切换为蓝色", catalog, present
)
self.assertEqual(resolved["id"], "blue")
self.assertIsNone(
resolve_active_peripheral_command(
"闪烁灯切换为蓝色",
[
{**item, "enabled": item["id"] != "blue"}
for item in catalog
],
present,
)
)
ambiguous = [
{**item, "action_names": ["切换为"]}
for item in catalog
]
with self.assertRaisesRegex(ExtensionIntegrationError, "同时匹配多个"):
resolve_active_peripheral_command("闪烁灯切换为", ambiguous, present)
def test_timed_phrase_selects_live_nfc_catalog_for_gpt_plan(self) -> None:
catalog = [
{
@@ -410,6 +784,38 @@ class SelfExtensionTests(unittest.TestCase):
[],
)
def test_exact_infrared_sequence_uses_local_rules_without_cloud(self) -> None:
service = SelfExtensionService(SimulationGateway())
catalog = [
{
"id": capability_id,
"status": "learned",
"device_name": "第三方氛围灯",
"action": action,
"aliases": aliases,
"slot": slot,
}
for capability_id, action, aliases, slot in (
("on", "开启灯光", ["打开"], 1),
("off", "关闭灯光", ["关掉"], 2),
)
]
with patch.dict(os.environ, {"OPENAI_API_KEY": ""}, clear=False):
result = service.plan(
{
"utterance": "打开第三方氛围灯，5秒后关闭第三方氛围灯",
"peripherals": catalog,
"vehicle_devices": [],
}
)
self.assertEqual(result["provider_used"], "infrared_local_rules")
self.assertEqual(
[step["kind"] for step in result["steps"]],
["peripheral", "delay", "peripheral"],
)
def test_main_peripheral_result_uses_gscu_without_vehicle_can(self) -> None:
service = SelfExtensionService(SimulationGateway())
@@ -502,6 +908,151 @@ class SelfExtensionTests(unittest.TestCase):
sleep.assert_any_call(5.0)
self.assertNotIn(0.5, [call.args[0] for call in sleep.call_args_list])
def test_failed_first_emit_retains_complete_compiled_infrared_plan(self) -> None:
service = SelfExtensionService(SimulationGateway())
catalog = [
{
"id": capability_id,
"status": "learned",
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"card_uid": "2CC8B103",
"action": action,
"slot": slot,
}
for capability_id, action, slot in (
("light-on", "打开", 1),
("light-green", "切换到绿色", 4),
)
]
plan_result = {
"provider_used": "infrared_local_rules",
"summary": "打开闪光灯并切换为绿色",
"steps": [
{"kind": "peripheral", "capability_id": "light-on"},
{"kind": "peripheral", "capability_id": "light-green"},
],
}
with patch.object(
service,
"execute_gateway",
side_effect=GatewayError("simulated infrared transport failure"),
) as execute_gateway:
result = service.execute_peripheral_plan_request(
"开启闪光灯，颜色设置为绿色",
plan_result,
catalog,
conversation_id="conv-failed-first-infrared-emit",
execution_mode="hardware",
)
execute_gateway.assert_called_once_with("emit", {"slot": 1})
self.assertEqual(result["status"], "failed")
self.assertEqual(result["sequence"]["action_count"], 2)
self.assertEqual(result["sequence"]["emit_count"], 0)
self.assertEqual(
[
item["parameters"]["slot"]
for item in result["function_config"]["spec"]["operations"]
],
[1, 4],
)
self.assertEqual(
[
item["signals"]["slot"]
for item in result["command_plan"]["spec"]["commands"]
],
[1, 4],
)
self.assertEqual(
[item["status"] for item in result["operation_results"]],
["failed"],
)
def test_submitted_emit_without_ack_continues_complete_infrared_plan(self) -> None:
service = SelfExtensionService(SimulationGateway())
catalog = [
{
"id": capability_id,
"status": "learned",
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"card_uid": "2CC8B103",
"action": action,
"slot": slot,
}
for capability_id, action, slot in (
("light-on", "打开", 1),
("light-green", "切换到绿色", 4),
)
]
def submit_without_ack(_action, payload):
command = build_gateway_command("emit", payload)
return {
"success": True,
"transport": "tcp",
"request": command,
"device_response": {
"success": True,
"delivery_status": "submitted",
"transport_submitted": True,
"gateway_acknowledged": False,
},
"gateway_acknowledged": False,
}
with (
patch.object(
service,
"execute_gateway",
side_effect=submit_without_ack,
) as execute_gateway,
patch("cognitive_agent.application.self_extension.time.sleep"),
):
result = service.execute_peripheral_plan_request(
"开启闪光灯，颜色设置为绿色",
{
"provider_used": "infrared_local_rules",
"summary": "打开闪光灯并切换为绿色",
"steps": [
{"kind": "peripheral", "capability_id": "light-on"},
{"kind": "peripheral", "capability_id": "light-green"},
],
},
catalog,
conversation_id="conv-submitted-unacknowledged-infrared",
execution_mode="hardware",
)
self.assertEqual(execute_gateway.call_count, 2)
self.assertEqual(result["status"], "succeeded")
self.assertEqual(result["sequence"]["emit_count"], 2)
self.assertEqual(
[item["telemetry"]["slot"] for item in result["operation_results"]],
[1, 4],
)
self.assertEqual(
[
item["telemetry"]["delivery_status"]
for item in result["operation_results"]
],
["submitted", "submitted"],
)
self.assertTrue(
all(
item["telemetry"]["transport_submitted"]
for item in result["operation_results"]
)
)
self.assertTrue(
all(
item["telemetry"]["gateway_acknowledged"] is False
for item in result["operation_results"]
)
)
def test_color_sequence_never_cycles_the_power_off_capability(self) -> None:
service = SelfExtensionService(SimulationGateway())
catalog = [
@@ -645,6 +1196,9 @@ class SelfExtensionTests(unittest.TestCase):
) -> str:
self.assertEqual(peripherals[0]["id"], "peripheral_1")
self.assertNotIn(opaque_id, [item["id"] for item in peripherals])
self.assertNotIn("slot", peripherals[0])
self.assertNotIn("card_uid", peripherals[0])
self.assertNotIn("transport", peripherals[0])
return json.dumps(
{
"summary": "切换红灯",
tests/unit/test_two_stage_can_parsing.py
+148
View File
@@ -4,6 +4,7 @@ from pathlib import Path
from unittest.mock import patch
from cognitive_agent.application.seat_service import SeatDemoService
from cognitive_agent.server import SeatDemoHandler
ROOT = Path(__file__).resolve().parents[2]
@@ -187,7 +188,154 @@ def test_frontend_two_stage_contract_and_source_badges() -> None:
assert 'class="v2-can-command-source"' in console
assert 'item.parseSource === "direct_control" ? "设备控件"' in console
assert 'item.parseSource === "local_rule" ? "本地规则" : "云端模型"' in console
assert 'item.deliveryLabel' in console
assert 'data-delivery-state="${escapeHtml(item.deliveryState)}"' in console
assert '"cognitive-agent:validated-can-commands"' in console
assert '.v2-can-command-source[data-parse-source="local_rule"]' in styles
assert '.v2-can-command-delivery[data-delivery-state="submitted"]' in styles
assert '"/api/can/commands"' in server
assert '"type": "validated_commands"' in server
def test_cloud_supplement_preserves_validated_infrared_preview() -> None:
infrared_operation = {
"ref": "toy-001-light-on",
"domain": "self_extension",
"status": "succeeded",
}
preview = {
"request_id": "req-preview-ir",
"status": "succeeded",
"orchestration_route": "unified_can_infrared",
"function_configs": {
"self_extension": {
"spec": {
"operations": [
{
"capability": "infrared.emit",
"parse_source": "local_rule",
}
]
}
}
},
"command_plans": {"self_extension": {"spec": {"commands": []}}},
"decision_summary": ["红外本地规则已完成"],
"trace_events": [],
"validation": [],
"parsed_control_signatures": [],
"execution_report": {"spec": {"results": [infrared_operation]}},
"operation_results": [infrared_operation],
"infrared_actions": [{"capability_id": "toy-001-light-on", "slot": 1}],
"peripheral": {"id": "toy-001", "name": "玩具 001"},
"control_transports": {"vehicle": "wired_can", "infrared": "gscu_tcp"},
}
supplement = {
"request_id": "req-cloud-ir",
"status": "no_action",
"provider_used": "openai:test",
"function_configs": {},
"command_plans": {},
"decision_summary": [],
"trace_events": [],
"validation": [],
"parsed_control_signatures": [],
"execution_report": None,
"operation_results": [],
}
merged = SeatDemoService._merge_native_can_phases(preview, supplement)
assert merged["orchestration_route"] == "unified_can_infrared"
assert merged["infrared_actions"][0]["slot"] == 1
assert merged["peripheral"]["id"] == "toy-001"
assert merged["operation_results"] == [infrared_operation]
assert "local_self_extension" in merged["function_configs"]
def test_unified_result_preserves_cloud_infrared_attribution() -> None:
can_result = {
"status": "succeeded",
"provider_used": "local_rules",
"function_configs": {},
"command_plans": {},
"operation_results": [],
"decision_summary": [],
}
infrared_result = {
"status": "succeeded",
"provider_used": "gpt:gpt-5.6-luna",
"control_transport": "gscu_dry_run",
"function_configs": {},
"command_plans": {},
"operation_results": [],
"decision_summary": [],
}
entries = [
{
"index": 0,
"clause": "把闪烁灯调成喜庆颜色",
"capability": {
"id": "light-red",
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"action": "切换到红色",
"slot": 3,
"parse_source": "cloud_model",
},
}
]
merged = SeatDemoHandler._merge_can_and_peripheral_results(
"把闪烁灯调成喜庆颜色，打开空调",
can_result,
infrared_result,
entries,
)
assert merged["provider_used"] == "local_rules+gpt:gpt-5.6-luna"
assert merged["infrared_actions"][0]["parse_source"] == "cloud_model"
def test_nfc_command_denial_keeps_allowed_result_and_reports_partial() -> None:
result = {
"status": "succeeded",
"success": True,
"decision_summary": ["已解析并验证 4 条允许的控制命令。"],
"execution_report": {"spec": {"status": "succeeded"}},
}
filtered = SeatDemoHandler._apply_nfc_command_denials(
result,
[
{
"index": 4,
"clause": "颜色设置为蓝色",
"reason": "nfc_command_disabled",
"capability": {
"id": "light-blue",
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"action": "切换到蓝色",
"slot": 5,
},
}
],
)
assert filtered["status"] == "partially_succeeded"
assert filtered["success"] is False
assert filtered["authorization_filtered"] is True
assert filtered["execution_report"]["spec"]["status"] == "partially_succeeded"
assert filtered["denied_peripheral_actions"] == [
{
"index": 4,
"clause": "颜色设置为蓝色",
"capability_id": "light-blue",
"device_id": "tony-light-001",
"device_name": "闪灯玩具",
"action": "切换到蓝色",
"slot": 5,
"reason": "nfc_command_disabled",
}
]