From 6b9ee0f349744794116d3e408b52e3d1c829639c Mon Sep 17 00:00:00 2001
From: Miayui <774713819@qq.com>
Date: Tue, 22 Sep 2026 13:01:46 +0800
Subject: [PATCH] fix(scenes): restore cinema seat controls and ordered
romantic lighting
---
apps/web-console/README.md | 20 +++
apps/web-console/app.js | 54 +++++--
apps/web-console/index.html | 2 +-
apps/web-console/service-worker.js | 4 +-
.../application/assistant_capabilities.py | 12 +-
.../application/codex_assistant.py | 28 +++-
.../application/n7_seat_preset.py | 69 +++++++++
.../application/romantic_lighting.py | 109 ++++++++++++++
.../application/scene_composer.py | 7 +-
.../application/seat_service.py | 96 +++++++++++--
src/cognitive_agent/server.py | 44 +++++-
tests/e2e/test_seat_web_api.py | 49 ++++++-
.../test_scene_peripheral_mode_flow.py | 110 ++++++++++++--
.../unit/test_assistant_frontend_contract.py | 2 +-
.../test_energy_monitor_frontend_contract.py | 2 +-
.../test_frontend_performance_contract.py | 4 +-
tests/unit/test_frontend_theme_contract.py | 2 +-
tests/unit/test_mode_gallery_grouping.py | 39 +++++
tests/unit/test_mode_scene_ingress.py | 93 ++++++++++--
tests/unit/test_n7_seat_preset.py | 88 ++++++++++++
tests/unit/test_romantic_lighting.py | 134 ++++++++++++++++++
tests/unit/test_scene_composer_fast_path.py | 8 +-
tests/unit/test_scene_composer_schema.py | 13 +-
tests/unit/test_vue_frontend_contract.py | 2 +-
24 files changed, 908 insertions(+), 83 deletions(-)
create mode 100644 src/cognitive_agent/application/n7_seat_preset.py
create mode 100644 src/cognitive_agent/application/romantic_lighting.py
create mode 100644 tests/unit/test_n7_seat_preset.py
create mode 100644 tests/unit/test_romantic_lighting.py
diff --git a/apps/web-console/README.md b/apps/web-console/README.md
index 2086de0..45a4856 100755
--- a/apps/web-console/README.md
+++ b/apps/web-console/README.md
@@ -1,5 +1,25 @@
# Web Console
+## 浪漫相关模式的外设灯光
+
+浪漫、约会、纪念日及浪漫影院等模式使用当前持久授权的红外目录：打开灯笼、蜡烛和灯带，
+所有具有红色指令的外设灯光统一设置为红色。每台设备均先发送开启指令，再发送红色指令；
+开启下发失败时终止后续序列。缺少开启能力的设备不能只发送颜色指令，缺少红色能力时
+不以其他颜色替代。新模式生成、车端卡片导入及已有卡片的执行入口均使用该规则，
+原生座椅、空调等场景动作保持原有定义。
+
+## 车端影院模式的副驾姿态
+
+从 `from_n7/scenes` 同步的 `local_scene.seat_preset` 会在执行时按 N7 的 M0–M3
+标定转换为副驾前后、坐盆、靠背、腿托四项位置控制，沿用现有座椅安全校验和执行器。
+例如 `[246, 112, 405, 342]` 脉冲转换为当前百分比协议的 `[51, 30, 44, 53]`。
+原始脉冲保留在卡片中，已保存的影院模式及继承该预设的本地模式无需重新上传。
+不完整、越界或错误座位的预设会被拒绝；带延时的预设继续交由车端处理。
+视频启动等车端专属动作仍走 `transto_n7/ctl`，不会重复转发已由 PC 处理的四轴姿态。
+
+模式卡片区分完整成功、部分处理、指令已提交待车端确认及模拟校验，消息入队不会显示为
+车辆执行完成。
+
## 座椅固件开发工作台
“开发指令”支持输入“增加动态木马功能”，也可点击位于位置控制、座椅通风、座椅加热之后的
diff --git a/apps/web-console/app.js b/apps/web-console/app.js
index c5d7847..d3fd4dc 100755
--- a/apps/web-console/app.js
+++ b/apps/web-console/app.js
@@ -7049,12 +7049,16 @@ function setModeExecutionState(modeProfileId, modeState = "idle") {
? "已停止"
: active && modeState === "succeeded"
? "已完成"
+ : active && modeState === "submitted"
+ ? "待车端确认"
+ : active && modeState === "partial"
+ ? "部分处理"
: active && modeState === "failed"
? "执行失败"
: "点击执行";
}
});
- if (modeProfileId && ["succeeded", "failed", "started", "ended", "stopped"].includes(modeState)) {
+ if (modeProfileId && ["succeeded", "failed", "started", "ended", "stopped", "submitted", "partial"].includes(modeState)) {
state.modeRunTimer = window.setTimeout(() => setModeExecutionState(null), 2800);
}
}
@@ -7073,6 +7077,35 @@ function hasExecutableModeCommandFlow(mode) {
|| (plans && typeof plans === "object" && Object.keys(plans).length > 0);
}
+function savedModeExecutionView(mode, result) {
+ const name = mode?.name || "功能模式";
+ if (modeScheduleExecutionView(result)?.state === "processing") {
+ return { state: "started", title: `${name}时间轴已启动`, toast: "时间轴已启动，可输入“停止动态模式”提前结束" };
+ }
+ if (!["succeeded", "partially_succeeded"].includes(result.status)) {
+ return { state: "failed", title: `${name}执行未完成`, toast: `${name}未完成，请查看执行结果` };
+ }
+ const preview = result.execution_mode === "dry_run";
+ const waiting = !preview && executionVerificationStatus(result) === "submitted_unverified";
+ if (result.status === "partially_succeeded") {
+ return {
+ state: "partial",
+ title: `${name}${preview ? "模拟校验部分通过" : waiting ? "部分指令已提交，待车端确认" : "部分完成"}`,
+ toast: preview
+ ? `${name}模拟校验部分通过，仍有未由 PC 执行的动作`
+ : `${name}仅部分处理成功，请查看各项指令和车端转发状态`,
+ };
+ }
+ if (waiting) {
+ return { state: "submitted", title: `${name}指令已提交，待车端确认`, toast: "指令已提交，尚未确认车辆执行完成" };
+ }
+ return {
+ state: "succeeded",
+ title: `${name}${preview ? "模拟校验完成" : "执行完成"}`,
+ toast: `${name}${preview ? "模拟校验完成" : "已完成保存的控制指令流"}`,
+ };
+}
+
async function executeSavedModeProfile(mode, { command = "" } = {}) {
if (!hasExecutableModeCommandFlow(mode)) return false;
const modeCommand = String(command || `进入${mode.name || mode.id}`).trim();
@@ -7125,14 +7158,9 @@ async function executeSavedModeProfile(mode, { command = "" } = {}) {
);
if (!payload) throw new Error("模式卡片未返回执行结果");
renderResult(payload);
- const scheduleRunning = modeScheduleExecutionView(payload)?.state === "processing";
- const completed = ["succeeded", "partially_succeeded"].includes(payload.status);
- $("#resultTitle").textContent = scheduleRunning
- ? `${mode.name}时间轴已启动`
- : completed
- ? `${mode.name}执行完成`
- : `${mode.name}执行未完成`;
- setModeExecutionState(mode.id, scheduleRunning ? "started" : completed ? "succeeded" : "failed");
+ const executionView = savedModeExecutionView(mode, payload);
+ $("#resultTitle").textContent = executionView.title;
+ setModeExecutionState(mode.id, executionView.state);
if (assistantModeHandle && assistant?.completeModeExecution) {
assistant.completeModeExecution(assistantModeHandle, mode, payload, {
command: modeCommand,
@@ -7141,11 +7169,7 @@ async function executeSavedModeProfile(mode, { command = "" } = {}) {
assistant?.announceModeExecution?.(mode, payload, { command: modeCommand });
}
if (state.executionMode === "hardware") await synchronizePhysicalSeatState();
- showToast(scheduleRunning
- ? "时间轴已启动，可输入“停止动态模式”提前结束"
- : completed
- ? `${mode.name}已直接执行保存的控制指令流`
- : `${mode.name}未完成，请查看执行结果`);
+ showToast(executionView.toast);
} catch (error) {
setModeExecutionState(mode.id, "failed");
setRunStatus("failed", "MODE FAILED");
@@ -8926,6 +8950,8 @@ function renderResult(result) {
? `${result.mode_profile?.name ? `${result.mode_profile.name} · ` : ""}${scheduleExecution.title}`
: peripheralRoute && executionSucceeded
? `${result.peripheral?.name || "第三方外设"} · ${result.peripheral?.action || "动作已执行"}`
+ : savedModeExecution && result.mode_profile
+ ? savedModeExecutionView(result.mode_profile, result).title
: partiallySucceeded
? `${domains.length} 域场景部分完成`
: executionSucceeded
diff --git a/apps/web-console/index.html b/apps/web-console/index.html
index 96469f6..b2d1249 100755
--- a/apps/web-console/index.html
+++ b/apps/web-console/index.html
@@ -39,7 +39,7 @@
-
+
diff --git a/apps/web-console/service-worker.js b/apps/web-console/service-worker.js
index acc4729..d5ee166 100644
--- a/apps/web-console/service-worker.js
+++ b/apps/web-console/service-worker.js
@@ -1,4 +1,4 @@
-const CACHE_VERSION = "cognitive-agent-pad-v335-firmware-rocking-horse";
+const CACHE_VERSION = "cognitive-agent-pad-v336-seat-preset";
const SHELL_CACHE = `${CACHE_VERSION}-shell`;
const RUNTIME_CACHE = `${CACHE_VERSION}-runtime`;
const APP_SHELL = [
@@ -6,7 +6,7 @@ const APP_SHELL = [
"/index.html",
"/styles.css?v=dual-seat-24-dialog-size",
"/network.js?v=android-pad-2",
- "/app.js?v=dual-seat-147-video-media",
+ "/app.js?v=dual-seat-148-seat-preset",
"/gesture-control.js?v=android-pad-9",
"/pwa.js?v=android-pad-3",
"/console-v2.css?v=delivery-state-69",
diff --git a/src/cognitive_agent/application/assistant_capabilities.py b/src/cognitive_agent/application/assistant_capabilities.py
index c3483fc..7f558ed 100644
--- a/src/cognitive_agent/application/assistant_capabilities.py
+++ b/src/cognitive_agent/application/assistant_capabilities.py
@@ -1760,6 +1760,9 @@ class AssistantCapabilityCatalog:
peripheral_actions.append(
{
"ref": peripheral_capability_ref(raw),
+ "device_key": hashlib.sha256(
+ str(raw.get("device_id") or device_name).encode("utf-8")
+ ).hexdigest()[:16],
"device_label": device_name,
"device_type": device_type,
"object_names": _safe_string_list(raw.get("object_names"), limit=12),
@@ -2443,7 +2446,12 @@ class AssistantCapabilityCatalog:
continue
try:
profile = self._vehicle.mode_profiles.get_profile(mode_id, vehicle_id)
- requests = self._vehicle.mode_profiles.requests_for_profiles([profile])
+ resolve_requests = getattr(self._vehicle, "mode_profile_execution_requests", None)
+ requests = (
+ resolve_requests(profile, vehicle_id)
+ if callable(resolve_requests)
+ else self._vehicle.mode_profiles.requests_for_profiles([profile])
+ )
except Exception: # noqa: BLE001 - a stale user mode is not executable
requests = []
operation_ids = [
@@ -2491,7 +2499,7 @@ class AssistantCapabilityCatalog:
"aliases": _safe_string_list(raw.get("aliases"), limit=16),
"description": _safe_text(raw.get("description"), limit=240),
"domains": _safe_string_list(raw.get("domains"), limit=16),
- "operation_count": int(raw.get("operation_count") or 0),
+ "operation_count": len(operation_ids),
"origin": _safe_text(raw.get("origin"), limit=32),
"operation_ids": operation_ids,
"operations": operations,
diff --git a/src/cognitive_agent/application/codex_assistant.py b/src/cognitive_agent/application/codex_assistant.py
index 85ce95a..a33e59d 100644
--- a/src/cognitive_agent/application/codex_assistant.py
+++ b/src/cognitive_agent/application/codex_assistant.py
@@ -34,6 +34,11 @@ from cognitive_agent.application.mode_creation_seat_policy import (
)
from cognitive_agent.application.mode_schedule import ModeSchedule
from cognitive_agent.application.mode_schedule_service import public_schedule_status
+from cognitive_agent.application.mode_scene_ingress import MAX_MODE_SCENE_ACTIONS
+from cognitive_agent.application.romantic_lighting import (
+ romantic_light_override_refs,
+ romantic_scene,
+)
from cognitive_agent.application.scene_composer import (
SCENE_COMPOSER_INSTRUCTIONS,
SCENE_PLANNER_VERSION,
@@ -1517,6 +1522,17 @@ class CodexAssistantService:
requests = normalized.get("requests")
if not isinstance(requests, list):
raise SceneCompositionError("scene plan requests must be an array")
+ semantics = scene_context.get("scene_semantics")
+ if isinstance(semantics, Mapping) and romantic_scene(semantics.get("mode_name")):
+ replaced_refs = romantic_light_override_refs(scene_context.get("peripheral_actions", []))
+ requests[:] = [request for request in requests if not (
+ isinstance(request, Mapping) and request.get("capability") == "infrared.emit"
+ and any(isinstance(parameter, Mapping)
+ and parameter.get("name") == "capability_ref"
+ and isinstance(parameter.get("value"), str)
+ and parameter.get("value") in replaced_refs
+ for parameter in request.get("parameters", []))
+ )]
existing_refs = {
str(parameter.get("value") or "").strip()
for request in requests
@@ -4317,13 +4333,17 @@ class CodexAssistantService:
secret=secret,
)
actions: list[dict[str, object]] = []
+ action_limit = (
+ MAX_MODE_SCENE_ACTIONS
+ if isinstance(result.get("mode_profile_draft"), Mapping)
+ or isinstance(result.get("mode_profile"), Mapping)
+ else 96 if result.get("schedule") else MAX_PLAN_REQUESTS
+ )
intent_ir = result.get("intent_ir")
if isinstance(intent_ir, dict):
spec = intent_ir.get("spec")
if isinstance(spec, dict) and isinstance(spec.get("actions"), list):
- for item in spec["actions"][
- :96 if result.get("schedule") else MAX_PLAN_REQUESTS
- ]:
+ for item in spec["actions"][:action_limit]:
if not isinstance(item, dict):
continue
safe_action = {
@@ -4351,7 +4371,7 @@ class CodexAssistantService:
infrared_actions: list[dict[str, object]] = []
raw_infrared = result.get("infrared_actions")
if isinstance(raw_infrared, list):
- for item in raw_infrared[:MAX_PLAN_REQUESTS]:
+ for item in raw_infrared[:action_limit]:
if not isinstance(item, dict):
continue
infrared_actions.append(
diff --git a/src/cognitive_agent/application/n7_seat_preset.py b/src/cognitive_agent/application/n7_seat_preset.py
new file mode 100644
index 0000000..4b5441b
--- /dev/null
+++ b/src/cognitive_agent/application/n7_seat_preset.py
@@ -0,0 +1,69 @@
+"""Translate the N7 passenger-seat preset contract into local seat operations."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from cognitive_agent.application.vehicle_profile import VehicleProfile
+
+
+# N7 targets_pulse uses the ECU's M0, M1, M2, M3 ordering.
+N7_SEAT_PRESET_AXES = (
+ "longitudinal", "cushion_tilt", "backrest_recline", "legrest_height",
+)
+N7_SEAT_PRESET_CAPABILITIES = (
+ "seat.set_longitudinal_position", "seat.set_cushion_tilt",
+ "seat.set_backrest_recline", "seat.set_legrest_height",
+)
+
+
+def expand_n7_seat_preset(
+ action: Mapping[str, Any], vehicle: VehicleProfile,
+) -> list[dict[str, Any]] | None:
+ """Expand an immediate preset; retain unrelated/timed actions for N7 routing.
+
+ The existing seat contract accepts whole percentages. Retain the source
+ pulses in the stored card, and round only at this execution boundary.
+ Never round a nonzero pulse target into the ECU's explicit homing command.
+ """
+ if action.get("capability") != "local_scene.seat_preset":
+ return None
+ if (
+ vehicle.id != "denza_n7"
+ or action.get("domain") != "local_vehicle"
+ or action.get("target_scope") not in {"vehicle", "passenger", "front_right"}
+ or action.get("operation") != "set"
+ ):
+ raise ValueError("车端座椅预设仅支持腾势 N7 副驾的绝对位置设置。")
+ parameters = action.get("parameters")
+ if not isinstance(parameters, Mapping) or set(parameters) != {"targets_pulse"}:
+ raise ValueError("车端座椅预设必须且只能包含 targets_pulse 参数。")
+ targets = parameters["targets_pulse"]
+ if not isinstance(targets, list) or len(targets) != 4:
+ raise ValueError("车端座椅预设需要按 M0–M3 顺序提供四个脉冲目标。")
+ operations = []
+ for axis, capability, pulse in zip(
+ N7_SEAT_PRESET_AXES, N7_SEAT_PRESET_CAPABILITIES, targets, strict=True
+ ):
+ motor = vehicle.seat_motors.get(axis)
+ maximum = motor.hall_total_pulses if motor is not None else 0
+ if maximum <= 0:
+ raise ValueError(f"车端座椅预设缺少 {axis} 的有效霍尔标定。")
+ if isinstance(pulse, bool) or not isinstance(pulse, int) or not 0 <= pulse <= maximum:
+ raise ValueError(f"车端座椅预设 {axis} 脉冲目标必须为 0–{maximum} 的整数。")
+ position_pct = round(pulse * 100 / maximum)
+ if 0 < pulse < maximum:
+ position_pct = max(1, min(99, position_pct))
+ operations.append({
+ "domain": "seat",
+ "capability": capability,
+ "target_scope": "passenger",
+ "operation": "set",
+ "parameters": {"position_pct": position_pct},
+ "reason": f"车端副驾预设 {axis}：{pulse}/{maximum} 脉冲转换为 {position_pct}%。",
+ })
+ # A delayed vehicle-owned action must not become an immediate PC movement.
+ if action.get("at_ms", 0) != 0:
+ return None
+ return operations
diff --git a/src/cognitive_agent/application/romantic_lighting.py b/src/cognitive_agent/application/romantic_lighting.py
new file mode 100644
index 0000000..ac56bae
--- /dev/null
+++ b/src/cognitive_agent/application/romantic_lighting.py
@@ -0,0 +1,109 @@
+"""Deterministic romantic lighting from the current authorized IR catalog."""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping, Sequence
+from copy import deepcopy
+from typing import Any
+
+ROMANTIC_SCENE = re.compile(r"浪漫|约会|七夕|情人节|纪念日|romantic|romance|date[ _-]*night|anniversary", re.I)
+_LIGHT = re.compile(r"灯|蜡烛|light|lamp|lantern|candle", re.I)
+_REQUIRED_POWER = re.compile(r"灯笼|蜡烛|灯带|灯条|闪烁灯|氛围灯|lantern|candle|(?:light|led)\s*strip|ambient\s*light", re.I)
+_ON = re.compile(r"^(?:打开|开启|启动|点亮|on|power\s*on)$", re.I)
+_OFF = re.compile(r"关闭|关掉|熄灭|\boff\b", re.I)
+_COLOR = re.compile(r"红色|绿色|蓝色|白色|紫色|黄色|橙色|粉色|\b(?:red|green|blue|white|purple|yellow|orange|pink)\b", re.I)
+_RED = re.compile(r"红色|\bred\b", re.I)
+
+
+def romantic_scene(value: object) -> bool:
+ return ROMANTIC_SCENE.search(str(value or "")) is not None
+
+
+def light_action_names(item: Mapping[str, Any]) -> list[str]:
+ parameters = item.get("parameters")
+ names = [item.get("action_label"), item.get("action")]
+ if isinstance(parameters, Mapping):
+ names.append(parameters.get("action"))
+ for key in ("action_names", "aliases"):
+ if isinstance(item.get(key), list):
+ names.extend(item[key])
+ return [str(name).strip() for name in names if name]
+
+
+def light_power_or_color(item: Mapping[str, Any]) -> bool:
+ names = light_action_names(item)
+ text = " ".join(names)
+ return any(_ON.fullmatch(name) for name in names) or bool(_OFF.search(text) or _COLOR.search(text))
+
+
+def romantic_light_actions(catalog: Sequence[Mapping[str, Any]]) -> list[dict[str, Any]]:
+ """Select every eligible light, with power before red and no color fallback."""
+ if not isinstance(catalog, Sequence) or isinstance(catalog, (str, bytes)):
+ return []
+ grouped: dict[str, list[Mapping[str, Any]]] = {}
+ identities: dict[str, str] = {}
+ for item in catalog:
+ if not isinstance(item, Mapping) or item.get("enabled", True) is False:
+ continue
+ if item.get("status", "learned") != "learned":
+ continue
+ symbolic = bool(item.get("ref"))
+ if not symbolic and (
+ not item.get("card_uid") or not item.get("device_id")
+ or isinstance(item.get("slot"), bool) or not isinstance(item.get("slot"), int)
+ or not 1 <= item["slot"] <= 248
+ ):
+ continue
+ name = str(item.get("device_label") or item.get("device_name") or "")
+ aliases = item.get("object_names", [])
+ aliases = aliases if isinstance(aliases, list) else []
+ identity = " ".join([name, *(str(x) for x in aliases if isinstance(x, str))])
+ if not _LIGHT.search(identity):
+ continue
+ key = str(item.get("device_key") or item.get("device_id") or name).casefold()
+ grouped.setdefault(key, []).append(item)
+ identities[key] = identity
+ selected = []
+ for key, actions in grouped.items():
+ red = next((item for item in actions
+ if _RED.search(" ".join(light_action_names(item)))
+ and not _OFF.search(" ".join(light_action_names(item)))), None)
+ if red is None and not _REQUIRED_POWER.search(identities[key]):
+ continue
+ activation = next((item for item in actions
+ if any(_ON.fullmatch(name) for name in light_action_names(item))
+ and not _OFF.search(" ".join(light_action_names(item)))
+ and not _COLOR.search(" ".join(light_action_names(item)))), None)
+ if activation is None:
+ continue
+ selected.append(deepcopy(dict(activation)))
+ if red is not None:
+ selected.append(deepcopy(dict(red)))
+ return selected
+
+
+def romantic_light_override_refs(catalog: Sequence[Mapping[str, Any]]) -> set[str]:
+ """Power/color refs superseded by the ordered romantic policy."""
+ refs = set()
+ for item in catalog:
+ if not isinstance(item, Mapping):
+ continue
+ identity = str(item.get("device_label") or item.get("device_name") or "")
+ if _LIGHT.search(identity) and light_power_or_color(item) and item.get("ref"):
+ refs.add(str(item["ref"]))
+ return refs
+
+
+def replace_romantic_light_actions(
+ existing: list[dict[str, Any]], required: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+ """Replace old power/color commands for the selected devices, retaining other actions."""
+ def target(item):
+ return str(item.get("target_scope") or (item.get("target") or {}).get("device") or "")
+ devices = {target(item) for item in required}
+ retained = [item for item in existing if not (
+ item.get("capability") == "infrared.emit"
+ and target(item) in devices and light_power_or_color(item)
+ )]
+ return [*deepcopy(retained), *deepcopy(required)]
diff --git a/src/cognitive_agent/application/scene_composer.py b/src/cognitive_agent/application/scene_composer.py
index 35ac470..005d980 100644
--- a/src/cognitive_agent/application/scene_composer.py
+++ b/src/cognitive_agent/application/scene_composer.py
@@ -20,8 +20,9 @@ from cognitive_agent.application.mode_creation_seat_policy import (
normalized_seat_scope,
)
from cognitive_agent.application.mode_schedule import ModeSchedule
+from cognitive_agent.application.romantic_lighting import romantic_light_actions, romantic_scene
-SCENE_PLANNER_VERSION = "n7web-scene-composer/v9"
+SCENE_PLANNER_VERSION = "n7web-scene-composer/v10"
_MAX_OUTPUT_SCHEMA_BYTES = 64 * 1024
_MAX_SCHEDULE_PHASE_REQUESTS = 12
@@ -392,6 +393,10 @@ def scene_required_peripheral_refs(
"""
clean_name = " ".join(str(mode_name or "").strip().split())[:64]
+ if romantic_scene(clean_name):
+ return [str(item["ref"]) for item in romantic_light_actions(
+ context.get("peripheral_actions", [])
+ ) if item.get("ref")]
themed_scene = _REQUIRED_DECORATIVE_LIGHT_SCENE.search(clean_name) is not None
if not themed_scene and not include_default_bundle:
return []
diff --git a/src/cognitive_agent/application/seat_service.py b/src/cognitive_agent/application/seat_service.py
index a9ea5ef..d837d38 100755
--- a/src/cognitive_agent/application/seat_service.py
+++ b/src/cognitive_agent/application/seat_service.py
@@ -60,6 +60,7 @@ from cognitive_agent.application.mode_device_additions import ( # noqa: E402
mode_device_additions,
)
from cognitive_agent.application.mode_profile import ( # noqa: E402
+ ModeOperation,
ModeProfile,
ModeProfileRegistry,
ModeTimeline,
@@ -95,6 +96,12 @@ from cognitive_agent.application.model_plan_adapter import ( # noqa: E402
from cognitive_agent.application.mqtt_ota_seat_executor import ( # noqa: E402
MqttNativeSeatExecutor,
)
+from cognitive_agent.application.n7_seat_preset import expand_n7_seat_preset # noqa: E402
+from cognitive_agent.application.romantic_lighting import ( # noqa: E402
+ romantic_light_actions,
+ romantic_scene,
+ replace_romantic_light_actions,
+)
from cognitive_agent.application.native_can_feedback import ( # noqa: E402
VehicleSafetyPreflightError,
authoritative_physical_ch1_feedback_state,
@@ -975,6 +982,9 @@ class SeatDemoService(ModeScheduleServiceMixin):
selected_devices: set[str] = set()
result: list[dict[str, Any]] = []
+ if romantic_scene(self._n7_scene_identity_text(scene)):
+ return [as_scene_action(item, category="romantic-light", kind="required")
+ for item in romantic_light_actions(raw_catalog)]
color_patterns = self._cinema_preferred_color_patterns(scene)
for category, identity_pattern in (
("flashing-light", N7_FLASHING_LIGHT_PATTERN),
@@ -1121,6 +1131,14 @@ class SeatDemoService(ModeScheduleServiceMixin):
document = deepcopy(dict(scene))
cinema_augmentation = self._augment_n7_cinema_scene(document)
+ if romantic_scene(self._n7_scene_identity_text(document)):
+ document["actions"] = replace_romantic_light_actions(
+ document.get("actions", []), self._n7_cinema_peripheral_actions(document)
+ )
+ document["operation_count"] = len(document["actions"])
+ document["domains"] = list(dict.fromkeys(
+ str(action.get("domain") or "") for action in document["actions"]
+ ))
actions = document.get("actions")
if not isinstance(actions, list) or not actions:
raise ValueError("from_n7/scenes card must contain executable actions")
@@ -1440,10 +1458,16 @@ class SeatDemoService(ModeScheduleServiceMixin):
def _mode_requests_and_n7_transfers(
self,
profile: ModeProfile,
+ vehicle_id: str | None = None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
"""Separate locally understood operations from N7-only capabilities."""
requests = self.mode_profiles.requests_for_profiles([profile])
+ vehicle = (
+ self.vehicle_profiles.get(vehicle_id)
+ if any(op.capability == "local_scene.seat_preset" for op in profile.operations)
+ else None
+ )
if profile.sync_origin != MQTT_MODE_SCENE_INGRESS_TOPIC:
stored_transfer_actions = (
profile.command_flow.get("n7_transfer_actions")
@@ -1473,6 +1497,7 @@ class SeatDemoService(ModeScheduleServiceMixin):
for action in transfer_actions
}
local_requests = []
+ adapted_selectors = set()
for operation, request in zip(
profile.operations,
requests,
@@ -1490,9 +1515,35 @@ class SeatDemoService(ModeScheduleServiceMixin):
operation.operation,
operation.parameters,
)
+ source_action = next((
+ action for action in transfer_actions
+ if self._n7_transfer_operation_selector(
+ action.get("domain"), action.get("capability"),
+ action.get("target_scope"), action.get("operation"),
+ action.get("parameters"),
+ ) == selector
+ ), operation.to_yaml())
+ expanded = (
+ expand_n7_seat_preset(source_action, vehicle) if vehicle is not None else None
+ )
+ if expanded is not None:
+ local_requests.extend(
+ ModeOperation.from_dict(item).to_request(profile.name)
+ for item in expanded
+ )
+ adapted_selectors.add(selector)
+ continue
if selector in transfer_selectors:
continue
local_requests.append(request)
+ transfer_actions = [
+ action for action in transfer_actions
+ if self._n7_transfer_operation_selector(
+ action.get("domain"), action.get("capability"),
+ action.get("target_scope"), action.get("operation"),
+ action.get("parameters"),
+ ) not in adapted_selectors
+ ]
return local_requests, transfer_actions
source_scene = (
profile.command_flow.get("source_scene")
@@ -1517,6 +1568,22 @@ class SeatDemoService(ModeScheduleServiceMixin):
# The server's durable NFC allowlist owns this local TCP path.
# Authorization failures must not be converted into N7 retries.
continue
+ raw_action = (
+ source_actions[index]
+ if index < len(source_actions)
+ and isinstance(source_actions[index], dict)
+ else self._fallback_scene_action(operation, index=index)
+ )
+ # Use the saved operation as authority; source metadata supplies timing only.
+ expanded = expand_n7_seat_preset(
+ {**operation.to_yaml(), "at_ms": raw_action.get("at_ms", 0)}, vehicle
+ ) if vehicle is not None else None
+ if expanded is not None:
+ local_requests.extend(
+ ModeOperation.from_dict(item).to_request(profile.name)
+ for item in expanded
+ )
+ continue
registered = self.model_plan_adapter.capabilities.get(
operation.capability
)
@@ -1532,20 +1599,21 @@ class SeatDemoService(ModeScheduleServiceMixin):
):
local_requests.append(request)
continue
- raw_action = (
- source_actions[index]
- if index < len(source_actions)
- and isinstance(source_actions[index], dict)
- else None
- )
- action = (
- deepcopy(raw_action)
- if raw_action is not None
- else self._fallback_scene_action(operation, index=index)
- )
- transfer_actions.append(action)
+ transfer_actions.append(deepcopy(raw_action))
return local_requests, transfer_actions
+ def mode_profile_execution_requests(
+ self, profile: ModeProfile, vehicle_id: str | None = None,
+ ) -> list[dict[str, Any]]:
+ """Expose the same expanded capability view used when executing a card."""
+ local, transfers = self._mode_requests_and_n7_transfers(profile, vehicle_id)
+ return [
+ *local,
+ *(ModeOperation.from_dict(item).to_request(profile.name) for item in transfers),
+ *(operation.to_request(profile.name) for operation in profile.operations
+ if operation.domain == "self_extension" and operation.capability == "infrared.emit"),
+ ]
+
@staticmethod
def _has_local_mode_peripheral(profile: ModeProfile) -> bool:
return any(
@@ -3783,7 +3851,7 @@ class SeatDemoService(ModeScheduleServiceMixin):
)
if direct_mode_profile is not None:
mode_requests, n7_transfer_actions = (
- self._mode_requests_and_n7_transfers(direct_mode_profile)
+ self._mode_requests_and_n7_transfers(direct_mode_profile, vehicle_profile.id)
)
if mode_creation_scopes is not None:
mode_requests = project_seed_operations(mode_requests, mode_creation_scopes)
@@ -5147,7 +5215,7 @@ class SeatDemoService(ModeScheduleServiceMixin):
matched_profile = matched_modes[0]
mode_requests, n7_transfer_actions = (
- self._mode_requests_and_n7_transfers(matched_profile)
+ self._mode_requests_and_n7_transfers(matched_profile, vehicle_profile.id)
)
selected_seat_scope = self._explicit_mode_seat_selection(
matched_profile, utterance,
diff --git a/src/cognitive_agent/server.py b/src/cognitive_agent/server.py
index bf27ab2..6a6fad5 100755
--- a/src/cognitive_agent/server.py
+++ b/src/cognitive_agent/server.py
@@ -74,6 +74,12 @@ from cognitive_agent.application.mode_scene_catalog import ( # noqa: E402
)
from cognitive_agent.application.pet_cards import PetCardService # noqa: E402
from cognitive_agent.application.mode_schedule_service import public_schedule_status # noqa: E402
+from cognitive_agent.application.romantic_lighting import ( # noqa: E402
+ romantic_light_actions,
+ romantic_scene,
+ replace_romantic_light_actions,
+)
+from cognitive_agent.application.mode_scene_ingress import MAX_MODE_SCENE_ACTIONS # noqa: E402
from cognitive_agent.application.seat_service import ( # noqa: E402
LOCAL_CONTROL_BUDGET_MS,
SeatDemoService,
@@ -1802,8 +1808,27 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
"source_span": source_utterance,
}
)
- if len(actions) > MAX_PLAN_REQUESTS:
- raise ExtensionIntegrationError("模式草稿包含过多设备动作。")
+ # These are validated execution actions, not the model's raw requests.
+ # A saved seat preset expands to four axes, and IR colors need power-on
+ # prerequisites. Keep the model's 20-request gate at the input boundary.
+ if len(actions) > MAX_MODE_SCENE_ACTIONS:
+ message = f"模式草稿展开后包含 {len(actions)} 项动作，超过 {MAX_MODE_SCENE_ACTIONS} 项上限。"
+ failed = deepcopy(native_result)
+ self.service._scrub_failed_scene_preview(failed)
+ failed.update({
+ "status": "validation_failed", "mode_profile_draft": None,
+ "mode_creation_failure": {
+ "code": "MODE_DRAFT_ACTION_LIMIT", "reason": message,
+ "next_step": "请减少模式中的设备动作后重新生成。",
+ },
+ "decision_summary": [message],
+ "validation": [*list(native_result.get("validation") or []), {
+ "severity": "error", "code": "MODE_DRAFT_ACTION_LIMIT",
+ "path": "intent_ir.spec.actions", "message": message,
+ }],
+ })
+ self.service.persist_composed_result(failed)
+ raise ExtensionIntegrationError(message)
spec["actions"] = actions
spec["requests"] = requests
detected_domains = [
@@ -4241,10 +4266,21 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
if isinstance(extension_config, dict)
else []
)
+ catalog = self.extension.list_peripheral_capabilities()
+ if romantic_scene(getattr(profile, "name", "")):
+ required = [{
+ "id": f"romantic-light-{item['slot']}",
+ "domain": "self_extension", "capability": "infrared.emit",
+ "operation": "emit",
+ "target": {
+ "device": item["device_id"], "name": item.get("device_name", ""),
+ "card_uid": item["card_uid"],
+ },
+ "parameters": {"action": item["action"], "slot": item["slot"]},
+ } for item in romantic_light_actions(catalog)]
+ operations = replace_romantic_light_actions(operations, required)
if not operations:
return []
-
- catalog = self.extension.list_peripheral_capabilities()
entries: list[dict[str, object]] = []
for index, operation in enumerate(operations):
if not isinstance(operation, dict) or operation.get("capability") != "infrared.emit":
diff --git a/tests/e2e/test_seat_web_api.py b/tests/e2e/test_seat_web_api.py
index 4530d62..a7456af 100755
--- a/tests/e2e/test_seat_web_api.py
+++ b/tests/e2e/test_seat_web_api.py
@@ -846,6 +846,45 @@ operations:
)
)
+ def test_uploaded_seat_preset_executes_four_local_axes_through_mode_api(self) -> None:
+ scene = {
+ "id": "n7-cinema-seat-test", "name": "影院模式", "label": "影院模式",
+ "aliases": ["影院模式"], "description": "车端副驾预设", "icon": "film",
+ "origin": "headunit", "scope": {"vehicle_ids": ["denza_n7"]},
+ "actions": [{
+ "type": "seat_preset", "at_ms": 0, "domain": "local_vehicle",
+ "capability": "local_scene.seat_preset", "target_scope": "vehicle",
+ "operation": "set", "parameters": {"targets_pulse": [246, 112, 405, 342]},
+ }],
+ }
+ with tempfile.TemporaryDirectory() as directory:
+ registry = ModeProfileRegistry(Path(directory) / "presets", user_root=Path(directory) / "modes")
+ with patch.object(self.service, "mode_profiles", registry), patch.object(
+ self.service._vehicle_mqtt_transport, "publish_n7_control_transfer",
+ side_effect=AssertionError("dry run must not publish"),
+ ):
+ self.service.import_n7_mode_scene(scene)
+ request = urllib.request.Request(
+ f"{self.base_url}/api/mode-profiles/n7-cinema-seat-test/execute",
+ method="POST", headers={"Content-Type": "application/json"},
+ data=json.dumps({
+ "vehicle_id": "denza_n7", "execution_mode": "dry_run",
+ "control_transport": "wireless_mqtt",
+ "client_type": "web_console_mode_card",
+ }).encode("utf-8"),
+ )
+ with self.opener.open(request, timeout=5) as response:
+ result = json.loads(response.read().decode("utf-8"))
+ self.assertEqual(result["status"], "succeeded")
+ self.assertEqual(result.get("n7_transfer_candidates", []), [])
+ operations = result["function_configs"]["seat"]["spec"]["operations"]
+ self.assertEqual(
+ [(op["capability"], op["parameters"]["position_pct"]) for op in operations],
+ [("seat.set_longitudinal_position", 51), ("seat.set_cushion_tilt", 30),
+ ("seat.set_backrest_recline", 44), ("seat.set_legrest_height", 53)],
+ )
+ self.assertTrue(all(op["target"]["instance"] == "passenger" for op in operations))
+
def test_saved_mode_accepts_prefixed_self_extension_flow_key(self) -> None:
with tempfile.TemporaryDirectory() as directory:
extension = SelfExtensionService(
@@ -997,7 +1036,7 @@ operations:
def runtime_url(file_name: str, source_url: str) -> str:
hashed = runtime_assets.get(file_name)
return f"/{hashed}" if hashed else source_url
- with self.opener.open(f"{self.base_url}/app.js?v=dual-seat-147-video-media", timeout=3) as response:
+ with self.opener.open(f"{self.base_url}/app.js?v=dual-seat-148-seat-preset", timeout=3) as response:
app_script = response.read().decode("utf-8")
with self.opener.open(
f"{self.base_url}/gesture-control.js?v=android-pad-9", timeout=3
@@ -1084,7 +1123,7 @@ operations:
self.assertIn("CMD 0x431 · FB 0x388 · CH1", template)
self.assertNotIn("透明车体 · 实体设备", template)
self.assertIn(
- f'src="{runtime_url("app.js", "/app.js?v=dual-seat-147-video-media")}"',
+ f'src="{runtime_url("app.js", "/app.js?v=dual-seat-148-seat-preset")}"',
html,
)
self.assertIn('import "../styles.css";', vite_entry)
@@ -1433,7 +1472,7 @@ operations:
self.assertIn(f'/{runtime_assets[asset_name]}', service_worker)
else:
self.assertIn(
- 'const CACHE_VERSION = "cognitive-agent-pad-v334-video-media"',
+ 'const CACHE_VERSION = "cognitive-agent-pad-v336-seat-preset"',
service_worker,
)
self.assertIn('/console-v2.css?v=delivery-state-69', service_worker)
@@ -1442,7 +1481,7 @@ operations:
self.assertIn('/assistant-chat.js?v=codex-assistant-39-video-media', service_worker)
self.assertIn('/energy-monitor.css?v=energy-monitor-5-dialog-size', service_worker)
self.assertIn('/energy-monitor.js?v=energy-monitor-5-performance', service_worker)
- self.assertIn('"/app.js?v=dual-seat-147-video-media"', service_worker)
+ self.assertIn('"/app.js?v=dual-seat-148-seat-preset"', service_worker)
self.assertIn('/firmware-workbench-entry.css?v=2', service_worker)
self.assertIn('/firmware-workbench-entry.js?v=3', service_worker)
self.assertIn('/firmware-workbench-theme.js?v=1', service_worker)
@@ -1910,7 +1949,7 @@ operations:
) as response:
extension_script = response.read().decode("utf-8")
with self.opener.open(
- f"{self.base_url}/app.js?v=dual-seat-147-video-media", timeout=3
+ f"{self.base_url}/app.js?v=dual-seat-148-seat-preset", timeout=3
) as response:
app_script = response.read().decode("utf-8")
diff --git a/tests/integration/test_scene_peripheral_mode_flow.py b/tests/integration/test_scene_peripheral_mode_flow.py
index 32dddd9..ed2a14d 100644
--- a/tests/integration/test_scene_peripheral_mode_flow.py
+++ b/tests/integration/test_scene_peripheral_mode_flow.py
@@ -125,8 +125,10 @@ def _received_cinema_scene() -> dict[str, object]:
}
+@pytest.mark.parametrize("full_scene", [False, True])
def test_romantic_cinema_template_preserves_cinema_and_adds_authorized_light(
tmp_path: Path,
+ full_scene: bool,
) -> None:
service = SeatDemoService(REPOSITORY_ROOT)
service.artifact_root = tmp_path / "runs"
@@ -138,10 +140,41 @@ def test_romantic_cinema_template_preserves_cinema_and_adds_authorized_light(
SimulationGateway(),
catalog_db_path=tmp_path / "infrared.sqlite3",
)
- extension.sync_peripheral_catalog([_learned_light_power_on()])
received_cinema = _received_cinema_scene()
+ lights = [_learned_light_power_on()]
+ if full_scene:
+ extra = [
+ *(('window', 'window.set_position', target, {"position_pct": 0})
+ for target in ('front_right', 'rear_left', 'rear_right')),
+ ('roof', 'roof.set_shade_position', 'vehicle', {"position_pct": 0}),
+ ('seat', 'seat.set_ventilation', 'passenger', {"state": "off"}),
+ ('climate', 'climate.set_power', 'vehicle', {"state": "on"}),
+ ('climate', 'climate.set_fan_level', 'vehicle', {"level": 2}),
+ ('climate', 'climate.set_outlet_mode', 'vehicle', {"mode": "face"}),
+ ('climate', 'climate.set_cycle_mode', 'vehicle', {"mode": "internal"}),
+ ('climate', 'climate.set_control_mode', 'vehicle', {"mode": "manual"}),
+ ('interior_light', 'interior_light.set_reading_power', 'all', {"state": "off"}),
+ ('ambient_light', 'ambient_light.set_power', 'vehicle', {"state": "on"}),
+ ('ambient_light', 'ambient_light.set_color_index', 'vehicle', {"color_index": 36}),
+ ('ambient_light', 'ambient_light.set_brightness_level', 'vehicle', {"brightness_level": 4}),
+ ]
+ received_cinema["actions"].extend({
+ "type": "control", "at_ms": 0, "domain": domain, "capability": capability,
+ "target_scope": target, "operation": "set", "parameters": parameters,
+ } for domain, capability, target, parameters in extra)
+ received_cinema["operation_count"] = len(received_cinema["actions"])
+ lights = [{
+ **_learned_light_power_on(), "id": f"ir-slot:{device}:{slot}",
+ "device_id": device, "device_name": name, "object_names": [name],
+ "action": action, "action_names": [action], "slot": slot,
+ } for device, name, action, slot in [
+ ("lantern", "灯笼", "打开", 2), ("lantern", "灯笼", "红色", 3),
+ ("strip", "灯带", "打开", 22), ("strip", "灯带", "红色", 23),
+ ("candle", "蜡烛", "打开", 41),
+ ]]
+ extension.sync_peripheral_catalog(lights)
service.import_n7_mode_scene(received_cinema)
- stale_profile = service.mode_profiles.save_mode(
+ stale_profile = None if full_scene else service.mode_profiles.save_mode(
"浪漫影院模式",
[
{
@@ -203,7 +236,7 @@ def test_romantic_cinema_template_preserves_cinema_and_adds_authorized_light(
side_effect=AssertionError("mode creation must not emit infrared"),
):
response = assistant.stream(
- "创建浪漫影院模式",
+ "创建一个浪漫影院模式",
conversation_id="conv-romantic-cinema-template",
principal="romantic-cinema-template-test",
history=[],
@@ -221,8 +254,11 @@ def test_romantic_cinema_template_preserves_cinema_and_adds_authorized_light(
client.start_turn.assert_not_called()
assert result["status"] == "succeeded"
assert result["mode_profile_draft"]["generation_source"] == "scene_template_copy"
- assert result["mode_profile_draft"]["existing_mode_id"] == stale_profile.id
- assert result["mode_profile_draft"]["operation_count"] == 3
+ if stale_profile is not None:
+ assert result["mode_profile_draft"]["existing_mode_id"] == stale_profile.id
+ assert result["mode_profile_draft"]["operation_count"] == (25 if full_scene else 7)
+ safe_result = CodexAssistantService._safe_vehicle_result(result)
+ assert len(safe_result["actions"]) == (25 if full_scene else 7)
native_actions = {
(
@@ -233,21 +269,35 @@ def test_romantic_cinema_template_preserves_cinema_and_adds_authorized_light(
if action["domain"] != "self_extension"
}
assert native_actions == {
- (action["capability"], action["target_scope"]): action["parameters"]
- for action in received_cinema["actions"]
- if action["domain"] in {"window", "climate"}
+ **{
+ (action["capability"], action["target_scope"]): action["parameters"]
+ for action in received_cinema["actions"]
+ if action["domain"] not in {"local_vehicle", "android"}
+ },
+ ("seat.set_longitudinal_position", "passenger"): {"position_pct": 51},
+ ("seat.set_cushion_tilt", "passenger"): {"position_pct": 30},
+ ("seat.set_backrest_recline", "passenger"): {"position_pct": 44},
+ ("seat.set_legrest_height", "passenger"): {"position_pct": 53},
}
- assert result["n7_transfer_candidates"] == received_cinema["actions"][2:]
- assert [action["slot"] for action in result["infrared_actions"]] == [1]
+ video = [action for action in received_cinema["actions"] if action["domain"] == "android"]
+ assert result["n7_transfer_candidates"] == video
+ slots = [action["slot"] for action in result["infrared_actions"]]
+ assert sorted(slots) == ([2, 3, 22, 23, 41] if full_scene else [1])
+ if full_scene:
+ assert slots.index(2) < slots.index(3)
+ assert slots.index(22) < slots.index(23)
draft = result["mode_profile_draft"]
- saved = service.save_mode_profile_card(
- stale_profile.id,
+ arguments = (
draft["name"],
draft["instruction"],
draft["source_request_id"],
draft["vehicle_id"],
)
+ saved = (
+ service.save_mode_profile_card(stale_profile.id, *arguments)
+ if stale_profile is not None else service.create_mode_profile_card(*arguments)
+ )
profile = service.get_mode_profile_card(saved["mode_profile"]["id"], "denza_n7")
saved_selectors = {
(operation.capability, operation.target_scope)
@@ -256,13 +306,45 @@ def test_romantic_cinema_template_preserves_cinema_and_adds_authorized_light(
assert saved_selectors == {
(action["capability"], action["target_scope"])
for action in received_cinema["actions"]
- } | {("infrared.emit", "romance-light")}
+ if action["capability"] != "local_scene.seat_preset"
+ } | {
+ ("seat.set_longitudinal_position", "passenger"),
+ ("seat.set_cushion_tilt", "passenger"),
+ ("seat.set_backrest_recline", "passenger"),
+ ("seat.set_legrest_height", "passenger"),
+ } | {("infrared.emit", item["device_id"]) for item in lights}
_local_requests, transfer_actions = service._mode_requests_and_n7_transfers(profile)
- assert transfer_actions == received_cinema["actions"][2:]
+ assert transfer_actions == video
finally:
service.close()
+def test_expanded_mode_overflow_cannot_leave_a_saveable_partial_draft() -> None:
+ handler = object.__new__(SeatDemoHandler)
+ handler.service = SimpleNamespace(
+ _scrub_failed_scene_preview=SeatDemoService._scrub_failed_scene_preview,
+ persist_composed_result=MagicMock(),
+ )
+ native = {
+ "request_id": "req-expanded-overflow", "status": "succeeded",
+ "execution_mode": "dry_run", "mode_profile_draft": {"operation_count": 256},
+ "intent_ir": {"spec": {"actions": [
+ {"domain": "seat", "capability": "seat.set_ventilation"}
+ ] * 256}},
+ }
+ infrared = {"function_configs": {"self_extension": {"spec": {"operations": [{
+ "target": {"device": "lamp"}, "parameters": {"action": "打开", "slot": 1},
+ }]}}}}
+ with patch.object(handler, "_merge_can_and_peripheral_results", return_value=native):
+ with pytest.raises(ExtensionIntegrationError, match="257.*256"):
+ handler._merge_scene_peripheral_preview("创建浪漫模式", native, infrared, [])
+ failed = handler.service.persist_composed_result.call_args.args[0]
+ assert failed["status"] == "validation_failed"
+ assert failed["mode_profile_draft"] is None
+ assert failed["command_plans"] == {}
+ assert failed["intent_ir"]["spec"]["actions"] == []
+
+
def test_scene_composer_mixed_can_ir_preview_save_and_revalidate(
tmp_path: Path,
) -> None:
diff --git a/tests/unit/test_assistant_frontend_contract.py b/tests/unit/test_assistant_frontend_contract.py
index cfba60f..d60aa8a 100644
--- a/tests/unit/test_assistant_frontend_contract.py
+++ b/tests/unit/test_assistant_frontend_contract.py
@@ -27,7 +27,7 @@ def test_console_contains_ai_assistant_surface():
def test_codex_assistant_assets_are_versioned_together_in_the_service_worker():
service_worker = (ROOT / "apps/web-console/service-worker.js").read_text(encoding="utf-8")
assert (
- 'const CACHE_VERSION = "cognitive-agent-pad-v334-video-media"'
+ 'const CACHE_VERSION = "cognitive-agent-pad-v336-seat-preset"'
in service_worker
)
assert '"/assistant-chat.css?v=codex-assistant-30"' in service_worker
diff --git a/tests/unit/test_energy_monitor_frontend_contract.py b/tests/unit/test_energy_monitor_frontend_contract.py
index e7762d6..a3c6d8c 100644
--- a/tests/unit/test_energy_monitor_frontend_contract.py
+++ b/tests/unit/test_energy_monitor_frontend_contract.py
@@ -41,7 +41,7 @@ def test_energy_monitor_assets_are_versioned_and_cached() -> None:
"/energy-monitor.js?v=energy-monitor-5-performance"
)
assert (
- 'const CACHE_VERSION = "cognitive-agent-pad-v334-video-media"'
+ 'const CACHE_VERSION = "cognitive-agent-pad-v336-seat-preset"'
in service_worker
)
diff --git a/tests/unit/test_frontend_performance_contract.py b/tests/unit/test_frontend_performance_contract.py
index 0ba3f40..791f41e 100644
--- a/tests/unit/test_frontend_performance_contract.py
+++ b/tests/unit/test_frontend_performance_contract.py
@@ -92,12 +92,12 @@ def test_performance_assets_have_a_new_cache_generation() -> None:
worker = source("service-worker.js")
for asset in (
- "/app.js?v=dual-seat-147-video-media",
+ "/app.js?v=dual-seat-148-seat-preset",
"/console-v2.js?v=local-first-79-cinema-status",
"/self-extension.js?v=peripheral-cards-59",
"/energy-monitor.js?v=energy-monitor-5-performance",
):
assert asset in html
assert f'"{asset}"' in worker
- assert 'const CACHE_VERSION = "cognitive-agent-pad-v334-video-media"' in worker
+ assert 'const CACHE_VERSION = "cognitive-agent-pad-v336-seat-preset"' in worker
assert '"/codepen-glass-theme.css?v=creative-cloud-6-performance"' in worker
diff --git a/tests/unit/test_frontend_theme_contract.py b/tests/unit/test_frontend_theme_contract.py
index 37d72d7..c5c7d62 100644
--- a/tests/unit/test_frontend_theme_contract.py
+++ b/tests/unit/test_frontend_theme_contract.py
@@ -54,7 +54,7 @@ def test_theme_asset_versions_are_consistent_with_the_pwa_shell() -> None:
assert source_asset in main
assert f'"{cached_asset}"' in service_worker
assert 'import "./styles/design-system.css"' in main
- assert 'const CACHE_VERSION = "cognitive-agent-pad-v334-video-media"' in service_worker
+ assert 'const CACHE_VERSION = "cognitive-agent-pad-v336-seat-preset"' in service_worker
def test_reference_glass_theme_covers_the_complete_workspace() -> None:
diff --git a/tests/unit/test_mode_gallery_grouping.py b/tests/unit/test_mode_gallery_grouping.py
index 01cb06c..4462adc 100644
--- a/tests/unit/test_mode_gallery_grouping.py
+++ b/tests/unit/test_mode_gallery_grouping.py
@@ -103,6 +103,45 @@ def node_runtime() -> str:
raise AssertionError("Node.js is required to execute the mode gallery grouping contract")
+def test_saved_mode_status_distinguishes_partial_submission_and_completion() -> None:
+ app = (WEB / "app.js").read_text(encoding="utf-8")
+ start = app.index("function savedModeExecutionView(")
+ end = app.index("\nasync function executeSavedModeProfile", start)
+ cases = [
+ {"status": "partially_succeeded", "execution_mode": "hardware"},
+ {"status": "partially_succeeded", "execution_mode": "hardware",
+ "execution_outcome": "submitted_unverified"},
+ {"status": "succeeded", "execution_mode": "hardware",
+ "execution_outcome": "submitted_unverified"},
+ {"status": "succeeded", "execution_mode": "hardware"},
+ {"status": "succeeded", "execution_mode": "dry_run",
+ "execution_outcome": "submitted_unverified"},
+ {"status": "partially_succeeded", "execution_mode": "dry_run"},
+ {"status": "failed", "execution_mode": "hardware"},
+ ]
+ script = (
+ "function modeScheduleExecutionView() { return null; }\n"
+ "function executionVerificationStatus(result) { return result.execution_outcome; }\n"
+ + app[start:end]
+ + "\nconsole.log(JSON.stringify("
+ + json.dumps(cases)
+ + '.map(result => savedModeExecutionView({name: "影院模式"}, result))));'
+ )
+ completed = subprocess.run(
+ [node_runtime(), "-e", script], check=True, capture_output=True,
+ text=True, encoding="utf-8",
+ )
+ result = json.loads(completed.stdout)
+ assert [item["state"] for item in result] == [
+ "partial", "partial", "submitted", "succeeded", "succeeded", "partial", "failed",
+ ]
+ assert [item["title"] for item in result] == [
+ "影院模式部分完成", "影院模式部分指令已提交，待车端确认",
+ "影院模式指令已提交，待车端确认", "影院模式执行完成",
+ "影院模式模拟校验完成", "影院模式模拟校验部分通过", "影院模式执行未完成",
+ ]
+
+
def test_mode_gallery_classifies_profiles_into_exclusive_custom_static_dynamic_groups() -> None:
app = (WEB / "app.js").read_text(encoding="utf-8")
start = app.index("const MODE_GALLERY_GROUPS")
diff --git a/tests/unit/test_mode_scene_ingress.py b/tests/unit/test_mode_scene_ingress.py
index bb3ba34..2698b52 100644
--- a/tests/unit/test_mode_scene_ingress.py
+++ b/tests/unit/test_mode_scene_ingress.py
@@ -12,6 +12,7 @@ from cognitive_agent.application.assistant_capabilities import (
AssistantCapabilityCatalog,
)
from cognitive_agent.application.mode_profile import ModeProfileRegistry
+from cognitive_agent.application.n7_seat_preset import N7_SEAT_PRESET_CAPABILITIES
from cognitive_agent.application.mode_scene_ingress import (
ModeSceneIngressConsumer,
ModeSceneIngressError,
@@ -714,6 +715,8 @@ def test_mixed_n7_scene_compiles_every_local_action_and_preserves_transfer(
) -> None:
service = configured_service(tmp_path)
scene = cinema_scene_card()
+ scene["actions"].append(cinema_video_launch_action())
+ scene["operation_count"] = len(scene["actions"])
service.import_n7_mode_scene(scene)
result = service.execute_mode_profile_card(
@@ -733,7 +736,21 @@ def test_mixed_n7_scene_compiles_every_local_action_and_preserves_transfer(
for config in result["function_configs"].values()
for operation in config["spec"]["operations"]
]
- assert len(compiled_operations) == 16
+ assert len(compiled_operations) == 20
+ positions = {
+ operation["capability"]: operation["parameters"]["position_pct"]
+ for operation in compiled_operations
+ if operation["capability"] in N7_SEAT_PRESET_CAPABILITIES
+ }
+ assert positions == dict(zip(N7_SEAT_PRESET_CAPABILITIES, [51, 30, 44, 53], strict=True))
+ hardware_plan = service.hardware_compiler.compile(result["function_configs"]["seat"])
+ moves = [
+ command["signals"] for command in hardware_plan["spec"]["commands"]
+ if command["message"] == "T2G_Seat_Move_Command_300"
+ ]
+ assert [(move["Motor_Selector"], move["Opcode"], move["Param_A"]) for move in moves] == [
+ (0, 1, 5100), (1, 1, 3000), (2, 1, 4400), (3, 1, 5300),
+ ]
assert all(
operation.get("parse_source") == "saved_mode_command_flow"
for operation in compiled_operations
@@ -743,15 +760,15 @@ def test_mixed_n7_scene_compiles_every_local_action_and_preserves_transfer(
for operation in compiled_operations
if operation["capability"] == "window.set_position"
} == {"front_left", "front_right", "rear_left", "rear_right"}
- transferred = scene["actions"][5]
+ transferred = cinema_video_launch_action()
assert result["n7_transfer_candidates"] == [transferred]
assert result["deferred_requests"] == [
{
- "domain": "local_vehicle",
- "capability": "local_scene.seat_preset",
- "target_scope": "vehicle",
- "desired_change": "set",
- "parameters": {"targets_pulse": [246, 112, 405, 342]},
+ "domain": "android",
+ "capability": "android.launch_app",
+ "target_scope": "headunit",
+ "desired_change": "launch",
+ "parameters": transferred["parameters"],
"reason": "当前 PC 未注册该能力，将由 N7 端继续处理。",
"execution_backend": "n7_vehicle_transfer",
"rejection_code": "PC_CAPABILITY_UNAVAILABLE",
@@ -760,7 +777,7 @@ def test_mixed_n7_scene_compiles_every_local_action_and_preserves_transfer(
]
assert len(compiled_operations) + len(result["n7_transfer_candidates"]) == len(
scene["actions"]
- )
+ ) + 3
text_result = service.process(
"进入Cinema mode模式",
@@ -774,6 +791,10 @@ def test_mixed_n7_scene_compiles_every_local_action_and_preserves_transfer(
assert text_result["mode_profile"] is None
assert [item["id"] for item in text_result["matched_modes"]] == ["cinema"]
assert len(text_result["n7_transfer_candidates"]) == 1
+ assert {
+ operation["capability"]
+ for operation in text_result["function_configs"]["seat"]["spec"]["operations"]
+ } >= set(N7_SEAT_PRESET_CAPABILITIES)
assert service.mode_profiles.match_text(
"氛围灯设置为Cinema mode模式",
vehicle_id="denza_n7",
@@ -786,8 +807,9 @@ def test_mixed_n7_scene_compiles_every_local_action_and_preserves_transfer(
)
assert catalog_mode["assistant_callable"] is True
assert catalog_mode["n7_transfer_operation_ids"] == [
- "local_scene.seat_preset"
+ "android.launch_app"
]
+ assert set(catalog_mode["operation_ids"]) >= set(N7_SEAT_PRESET_CAPABILITIES)
@pytest.mark.parametrize(
@@ -826,6 +848,59 @@ def test_registered_but_invalid_action_is_not_forwarded_to_n7(
assert transfer_actions == []
+def test_existing_saved_cinema_cards_adapt_on_execution_without_reimport(tmp_path: Path) -> None:
+ service = configured_service(tmp_path)
+ scene = cinema_scene_card()
+ scene["actions"].append(cinema_video_launch_action())
+ service.import_n7_mode_scene(scene)
+ imported = service.get_mode_profile_card("cinema", "denza_n7")
+ legacy = service.mode_profiles.save_mode(
+ "旧版浪漫影院模式", [op.to_yaml() for op in imported.operations],
+ vehicle_id="denza_n7", instruction="进入旧版浪漫影院模式",
+ command_flow={
+ "plans": {"saved": {"spec": {"commands": [{"message": "saved"}]}}},
+ "n7_transfer_actions": [scene["actions"][5], cinema_video_launch_action()],
+ },
+ )
+ before = {path: path.read_bytes() for path in (tmp_path / "profiles").rglob("*.yaml")}
+ service.mode_profiles = ModeProfileRegistry(
+ tmp_path / "presets", user_root=tmp_path / "profiles"
+ )
+ publisher = MagicMock(side_effect=AssertionError("preview must not publish"))
+ service._vehicle_mqtt_transport.publish_n7_control_transfer = publisher
+ try:
+ for mode_id in ("cinema", legacy.id):
+ result = service.execute_mode_profile_card(
+ mode_id, execution_mode="dry_run", control_transport="wireless_mqtt"
+ )
+ assert result["status"] == "partially_succeeded"
+ assert result["n7_transfer_candidates"] == [cinema_video_launch_action()]
+ operations = result["function_configs"]["seat"]["spec"]["operations"]
+ assert [op["parameters"]["position_pct"] for op in operations
+ if op["capability"] in N7_SEAT_PRESET_CAPABILITIES] == [51, 30, 44, 53]
+ assert all(path.read_bytes() == value for path, value in before.items())
+ publisher.assert_not_called()
+ finally:
+ service.close()
+
+
+def test_invalid_preset_blocks_the_card_instead_of_transferring_it(tmp_path: Path) -> None:
+ service = configured_service(tmp_path)
+ scene = cinema_scene_card()
+ scene["actions"][5]["parameters"]["targets_pulse"][0] = 9999
+ service.import_n7_mode_scene(scene)
+ publisher = MagicMock(side_effect=AssertionError("invalid motion must not publish"))
+ service._vehicle_mqtt_transport.publish_n7_control_transfer = publisher
+ try:
+ with pytest.raises(ValueError, match="0–480"):
+ service.execute_mode_profile_card(
+ "cinema", execution_mode="hardware", control_transport="wireless_mqtt"
+ )
+ publisher.assert_not_called()
+ finally:
+ service.close()
+
+
def test_frontend_marks_n7_synchronized_cards() -> None:
script = (REPOSITORY_ROOT / "apps" / "web-console" / "app.js").read_text(
encoding="utf-8"
diff --git a/tests/unit/test_n7_seat_preset.py b/tests/unit/test_n7_seat_preset.py
new file mode 100644
index 0000000..f53d408
--- /dev/null
+++ b/tests/unit/test_n7_seat_preset.py
@@ -0,0 +1,88 @@
+from copy import deepcopy
+from dataclasses import replace
+from pathlib import Path
+
+import pytest
+
+from cognitive_agent.application.n7_seat_preset import (
+ N7_SEAT_PRESET_CAPABILITIES,
+ expand_n7_seat_preset,
+)
+from cognitive_agent.application.vehicle_profile import VehicleProfileRegistry
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def preset(targets=None):
+ return {
+ "domain": "local_vehicle", "capability": "local_scene.seat_preset",
+ "target_scope": "vehicle", "operation": "set", "at_ms": 0,
+ "parameters": {"targets_pulse": [246, 112, 405, 342] if targets is None else targets},
+ }
+
+
+@pytest.fixture
+def vehicle():
+ return VehicleProfileRegistry(ROOT / "config" / "vehicles").get("denza_n7")
+
+
+def test_preset_uses_vehicle_calibration_and_keeps_original_pulses(vehicle):
+ action = preset()
+ original = deepcopy(action)
+ operations = expand_n7_seat_preset(action, vehicle)
+ assert [item["capability"] for item in operations] == list(N7_SEAT_PRESET_CAPABILITIES)
+ assert [item["parameters"]["position_pct"] for item in operations] == [51, 30, 44, 53]
+ assert all(item["target_scope"] == "passenger" for item in operations)
+ assert action == original
+ calibrated = replace(vehicle, seat_motors={
+ **vehicle.seat_motors,
+ "longitudinal": replace(vehicle.seat_motors["longitudinal"], hall_total_pulses=500),
+ })
+ assert expand_n7_seat_preset(action, calibrated)[0]["parameters"] == {"position_pct": 49}
+
+
+@pytest.mark.parametrize("targets", [
+ [], [1, 2, 3], [1, 2, 3, 4, 5], "246,112,405,342",
+ [True, 112, 405, 342], [246.5, 112, 405, 342], ["246", 112, 405, 342],
+ [-1, 112, 405, 342], [481, 112, 405, 342], [246, 375, 405, 342],
+ [246, 112, 918, 342], [246, 112, 405, 648],
+])
+def test_invalid_pulses_are_rejected_without_clamping(vehicle, targets):
+ with pytest.raises(ValueError):
+ expand_n7_seat_preset(preset(targets), vehicle)
+
+
+@pytest.mark.parametrize("field,value", [
+ ("target_scope", "driver"), ("domain", "seat"), ("operation", "toggle"),
+ ("parameters", {"targets_pulse": [246, 112, 405, 342], "speed": 100}),
+])
+def test_unrecognized_target_or_contract_cannot_move_a_seat(vehicle, field, value):
+ action = preset()
+ action[field] = value
+ with pytest.raises(ValueError):
+ expand_n7_seat_preset(action, vehicle)
+
+
+def test_zero_and_small_targets_do_not_accidentally_home(vehicle):
+ assert [item["parameters"]["position_pct"] for item in expand_n7_seat_preset(
+ preset([0, 0, 0, 0]), vehicle
+ )] == [0, 0, 0, 0]
+ assert [item["parameters"]["position_pct"] for item in expand_n7_seat_preset(
+ preset([1, 1, 1, 1]), vehicle
+ )] == [1, 1, 1, 1]
+ assert [item["parameters"]["position_pct"] for item in expand_n7_seat_preset(
+ preset([480, 374, 917, 647]), vehicle
+ )] == [100, 100, 100, 100]
+
+
+def test_delayed_actions_keep_the_vehicle_timing(vehicle):
+ action = preset()
+ action["at_ms"] = 3000
+ assert expand_n7_seat_preset(action, vehicle) is None
+
+
+def test_wrong_vehicle_or_missing_calibration_is_rejected(vehicle):
+ with pytest.raises(ValueError):
+ expand_n7_seat_preset(preset(), replace(vehicle, id="other_vehicle"))
+ with pytest.raises(ValueError):
+ expand_n7_seat_preset(preset(), replace(vehicle, seat_motors={}))
diff --git a/tests/unit/test_romantic_lighting.py b/tests/unit/test_romantic_lighting.py
new file mode 100644
index 0000000..e0ccf5b
--- /dev/null
+++ b/tests/unit/test_romantic_lighting.py
@@ -0,0 +1,134 @@
+from copy import deepcopy
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import pytest
+
+from cognitive_agent.application.codex_assistant import CodexAssistantService
+from cognitive_agent.application.romantic_lighting import romantic_light_actions
+from cognitive_agent.application.scene_composer import scene_required_peripheral_refs
+from cognitive_agent.application.mode_profile import ModeProfileRegistry
+from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
+from cognitive_agent.application.self_extension import GatewayError, SelfExtensionService, SimulationGateway
+from cognitive_agent.server import SeatDemoHandler
+
+
+def light(device, name, action, slot, **extra):
+ return {
+ "id": f"ir-slot:{device}:{slot}", "status": "learned", "enabled": True,
+ "device_id": device, "device_name": name, "device_type": "other",
+ "card_uid": f"{int(device[-1]):08X}", "slot": slot, "action": action,
+ "object_names": [name], "action_names": [action], "risk": "low", **extra,
+ }
+
+
+def catalog():
+ return [
+ light("light1", "灯笼", "关闭", 1), light("light1", "灯笼", "打开", 2),
+ light("light1", "灯笼", "切换为红色", 3), light("light1", "灯笼", "切换为蓝色", 5),
+ light("light2", "灯带", "打开", 22), light("light2", "灯带", "设置为红色", 23),
+ light("light3", "蜡烛", "打开", 41), light("light3", "蜡烛", "关闭", 42),
+ light("light4", "球形灯", "打开", 51), light("light4", "球形灯", "设置为红色", 52),
+ ]
+
+
+def test_romantic_lights_all_use_on_then_red_and_preserve_on_only_candles():
+ entries = catalog()
+ entries += [
+ light("light5", "落地灯", "红色", 61), # No known power-on command.
+ light("light6", "投影灯", "打开", 71),
+ light("light6", "投影灯", "蓝色", 72), # Never substitute blue for red.
+ light("light7", "灯笼", "打开", 81, enabled=False),
+ light("light7", "灯笼", "红色", 82, enabled=False),
+ ]
+ before = deepcopy(entries)
+ assert [item["slot"] for item in romantic_light_actions(entries)] == [2, 3, 22, 23, 41, 51, 52]
+ assert entries == before
+
+
+@pytest.mark.parametrize("name", ["浪漫模式", "浪漫影院模式", "约会模式", "Romantic cinema"])
+def test_creation_replaces_old_colors_and_orders_all_required_lights(name):
+ actions = [{
+ "ref": f"ir:{item['slot']}", "device_key": item["device_id"],
+ "device_label": item["device_name"], "action_label": item["action"],
+ "object_names": item["object_names"], "action_names": item["action_names"],
+ } for item in catalog()]
+ context = {"peripheral_actions": actions}
+ required = scene_required_peripheral_refs(name, context)
+ assert required == ["ir:2", "ir:3", "ir:22", "ir:23", "ir:41", "ir:51", "ir:52"]
+ context["scene_semantics"] = {"mode_name": name, "peripheral_policy": {"required_refs": required}}
+ plan = {"requests": [{
+ "domain": "self_extension", "capability": "infrared.emit",
+ "target_scope": ref, "operation": "emit",
+ "parameters": [{"name": "capability_ref", "value": ref}],
+ } for ref in ["ir:5", "ir:3", "ir:2"]], "detected_domains": ["self_extension"]}
+ result = CodexAssistantService._with_required_scene_peripherals(plan, context)
+ assert [item["parameters"][0]["value"] for item in result["requests"]] == required
+
+
+def test_same_named_lights_remain_separate_devices():
+ actions = [{
+ "ref": f"ir:{device}-{action}", "device_key": device,
+ "device_label": "灯笼", "action_label": label,
+ } for device in ["one", "two"] for action, label in [("on", "打开"), ("red", "红色")]]
+ assert [item["ref"] for item in romantic_light_actions(actions)] == [
+ "ir:one-on", "ir:one-red", "ir:two-on", "ir:two-red",
+ ]
+
+
+def test_romantic_import_adds_ordered_lights_without_requiring_cinema_base(tmp_path: Path):
+ service = SeatDemoService(REPOSITORY_ROOT)
+ service.mode_profiles = ModeProfileRegistry(tmp_path / "presets", user_root=tmp_path / "modes")
+ service.set_mode_scene_peripheral_catalog_provider(catalog)
+ climate = {"type": "control", "at_ms": 0, "domain": "climate",
+ "capability": "climate.set_temperature", "target_scope": "vehicle",
+ "operation": "set", "parameters": {"temperature_c": 25}}
+ scene = {"id": "romantic-date", "name": "浪漫约会模式", "label": "浪漫约会模式",
+ "origin": "headunit", "actions": [climate], "scope": {"vehicle_ids": ["denza_n7"]}}
+ try:
+ for _ in range(2):
+ service.import_n7_mode_scene(scene)
+ profile = service.get_mode_profile_card("romantic-date", "denza_n7")
+ actions = profile.command_flow["source_scene"]["actions"]
+ assert actions[0] == climate
+ assert [item["parameters"]["slot"] for item in actions[1:]] == [2, 3, 22, 23, 41, 51, 52]
+ scene = profile.command_flow["source_scene"]
+ assert len(profile.operations) == 8
+ finally:
+ service.close()
+
+
+@pytest.mark.parametrize("fail_power", [False, True])
+def test_saved_romantic_mode_orders_real_dispatch_and_stops_on_power_failure(tmp_path: Path, fail_power):
+ extension = SelfExtensionService(SimulationGateway(), catalog_db_path=tmp_path / "ir.sqlite3")
+ extension.sync_peripheral_catalog(catalog())
+ handler = object.__new__(SeatDemoHandler)
+ handler.extension = extension
+ profile = SimpleNamespace(name="浪漫影院模式", command_flow={"function_configs": {
+ "self_extension": {"spec": {"domain": "self_extension", "operations": [{
+ "capability": "infrared.emit", "target": {"device": "light1", "card_uid": "00000001"},
+ "parameters": {"action": "切换为蓝色", "slot": 5},
+ }]}}
+ }})
+ entries = handler._saved_mode_infrared_entries(profile)
+ capabilities = [entry["capability"] for entry in entries]
+ assert [item["slot"] for item in capabilities] == [2, 3, 22, 23, 41, 51, 52]
+ sent = []
+
+ def send(_command, parameters):
+ sent.append(parameters["slot"])
+ if fail_power:
+ raise GatewayError("power-on transport failed")
+ return {"transport": "tcp", "gateway_acknowledged": True, "request": "mock emit"}
+
+ with patch.object(extension, "execute_gateway", side_effect=send), patch(
+ "cognitive_agent.application.self_extension.time.sleep"
+ ):
+ result = extension.execute_peripheral_plan_request(
+ "执行浪漫影院模式", {"steps": [
+ {"kind": "peripheral", "capability_id": item["id"]} for item in capabilities
+ ]}, capabilities, conversation_id="romantic-order-test", execution_mode="hardware",
+ )
+ assert sent == ([2] if fail_power else [2, 3, 22, 23, 41, 51, 52])
+ assert result["status"] == ("failed" if fail_power else "succeeded")
diff --git a/tests/unit/test_scene_composer_fast_path.py b/tests/unit/test_scene_composer_fast_path.py
index 46cd329..f8a2b0a 100644
--- a/tests/unit/test_scene_composer_fast_path.py
+++ b/tests/unit/test_scene_composer_fast_path.py
@@ -695,8 +695,10 @@ def test_romantic_template_always_adds_authorized_decorative_lights() -> None:
)
assert _peripheral_refs(observed[0]) == [
"ir:flash-open-000000000001",
- "ir:ambient-open-0000000001",
+ "ir:flash-red-0000000000002",
"ir:strip-open-000000000001",
+ "ir:strip-red-0000000000002",
+ "ir:ambient-open-0000000001",
]
assert "ir:flash-close-00000000003" not in _peripheral_refs(observed[0])
@@ -740,8 +742,10 @@ def test_romantic_cinema_template_keeps_native_base_and_adds_lights_without_mode
assert native_requests == template["plan"]["requests"]
assert _peripheral_refs(observed[0]) == [
"ir:flash-open-000000000001",
- "ir:ambient-open-0000000001",
+ "ir:flash-red-0000000000002",
"ir:strip-open-000000000001",
+ "ir:strip-red-0000000000002",
+ "ir:ambient-open-0000000001",
]
diff --git a/tests/unit/test_scene_composer_schema.py b/tests/unit/test_scene_composer_schema.py
index 034c498..857e74a 100644
--- a/tests/unit/test_scene_composer_schema.py
+++ b/tests/unit/test_scene_composer_schema.py
@@ -520,20 +520,23 @@ def test_romantic_and_mid_autumn_require_decorative_light_actions(
context["scene_semantics"] = semantics
schema = build_scene_output_schema(context)
- assert required == [
+ romantic_refs = [
+ "ir:flash-open", "ir:flash-red", "ir:strip-open", "ir:strip-red", "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"] == 18
+ ])
+ assert schema["properties"]["requests"]["maxItems"] == 20 - len(required)
if "中秋" in mode_name:
assert semantics["family"] == "festival_celebration"
diff --git a/tests/unit/test_vue_frontend_contract.py b/tests/unit/test_vue_frontend_contract.py
index 92fd8fe..4261c6e 100644
--- a/tests/unit/test_vue_frontend_contract.py
+++ b/tests/unit/test_vue_frontend_contract.py
@@ -31,7 +31,7 @@ def test_vue_vite_application_is_the_primary_frontend_entry() -> None:
assert package["scripts"]["build"] == "vite build"
assert 'import vue from "@vitejs/plugin-vue"' in vite
assert 'outDir: outputRoot' in vite
- assert 'const CACHE_VERSION = "cognitive-agent-pad-v334-video-media"' in service_worker
+ assert 'const CACHE_VERSION = "cognitive-agent-pad-v336-seat-preset"' in service_worker
assert 'createHash("sha256")' in vite
assert 'runtime-assets.json' in vite
assert 'rewriteRuntimeReferences' in vite