diff --git a/apps/web-console/app.js b/apps/web-console/app.js
index 41726d4..ba413f9 100755
--- a/apps/web-console/app.js
+++ b/apps/web-console/app.js
@@ -8310,20 +8310,69 @@ window.CognitiveAgentConsole = {
? draft.trim()
: String(draft?.instruction || "").trim();
if (!instruction) throw new Error("功能模式草稿缺少可试运行的指令。");
- const { payload } = await requestNaturalLanguageControl(instruction, {
- executionMode: "dry_run",
- clientType: "assistant_mode_trial",
- provider: "local",
- previewOnly: false,
- });
+ let payload;
+ if (draft && typeof draft === "object" && draft.source_request_id) {
+ const response = await apiRequest("/api/mode-profile-drafts/trial", {
+ method: "POST",
+ headers: { "Content-Type": "application/json" },
+ body: JSON.stringify({
+ name: draft.name,
+ instruction,
+ source_request_id: draft.source_request_id,
+ vehicle_id: draft.vehicle_id || state.vehicleId || null,
+ }),
+ timeoutMs: 30 * 1000,
+ retries: 0,
+ });
+ payload = await response.json();
+ if (response.status === 401) {
+ window.location.replace("/login");
+ return null;
+ }
+ if (!response.ok) {
+ throw new Error(payload.message || payload.error || `HTTP ${response.status}`);
+ }
+ } else {
+ ({ payload } = await requestNaturalLanguageControl(instruction, {
+ executionMode: "dry_run",
+ clientType: "assistant_mode_trial",
+ provider: "local",
+ previewOnly: false,
+ }));
+ }
if (!payload) throw new Error("模拟试运行没有返回结果。");
const actualOperationCount = Object.values(payload.function_configs || {})
.reduce((total, config) => total + (config?.spec?.operations?.length || 0), 0);
const expectedOperationCount = Number(
typeof draft === "object" ? draft?.operation_count : 0,
);
+ const actionSignature = (action) => {
+ const target = action?.target && typeof action.target === "object"
+ ? action.target.instance || action.target.zone || action.target.device || ""
+ : action?.target || "";
+ const parameters = action?.parameters && typeof action.parameters === "object"
+ ? Object.fromEntries(Object.entries(action.parameters).sort(([left], [right]) => left.localeCompare(right)))
+ : {};
+ return JSON.stringify([
+ String(action?.domain || ""),
+ String(action?.capability || ""),
+ String(target),
+ parameters,
+ ]);
+ };
+ const expectedSignatures = Array.isArray(draft?.actions)
+ ? draft.actions.map(actionSignature).sort()
+ : [];
+ const actualActions = Array.isArray(payload?.intent_ir?.spec?.actions)
+ ? payload.intent_ir.spec.actions
+ : [];
+ const actualSignatures = actualActions.map(actionSignature).sort();
+ const sameActions = expectedSignatures.length === 0
+ || (expectedSignatures.length === actualSignatures.length
+ && expectedSignatures.every((signature, index) => signature === actualSignatures[index]));
if (payload.status !== "succeeded"
- || (expectedOperationCount > 0 && actualOperationCount < expectedOperationCount)) {
+ || (expectedOperationCount > 0 && actualOperationCount < expectedOperationCount)
+ || !sameActions) {
throw new Error("试运行未完整覆盖草稿动作，请重新生成模式预览。");
}
return payload;
diff --git a/apps/web-console/assistant-chat.js b/apps/web-console/assistant-chat.js
index fca929f..484bf66 100644
--- a/apps/web-console/assistant-chat.js
+++ b/apps/web-console/assistant-chat.js
@@ -271,7 +271,7 @@
followLatest();
}
- function appendModeCreatedState(message, profile) {
+ function appendModeCreatedState(message, profile, { updated = false } = {}) {
if (!message?.body || !profile?.id) return;
const selector = `[data-assistant-mode-created="${CSS.escape(String(profile.id))}"]`;
if ($(selector, message.body)) return;
@@ -286,7 +286,7 @@
const copy = document.createElement("span");
copy.className = "assistant-mode-created-copy";
const label = document.createElement("small");
- label.textContent = "已添加到「我的模式」";
+ label.textContent = updated ? "已更新「我的模式」" : "已添加到「我的模式」";
const name = document.createElement("strong");
name.textContent = String(profile.name || "新功能模式");
copy.append(label, name);
@@ -396,9 +396,16 @@
icon = "panels-top-left";
order = instructionOrder("车窗");
} else if (domain === "seat") {
- label = ["passenger", "front_right"].includes(zone) ? "副驾座椅" : "主驾座椅";
+ const seatLabel = ["passenger", "front_right"].includes(zone) ? "副驾" : "主驾";
+ const seatCapabilityLabels = {
+ "seat.set_longitudinal_position": "座椅前后位置",
+ "seat.set_cushion_tilt": "坐垫角度",
+ "seat.set_backrest_recline": "靠背角度",
+ "seat.set_legrest_height": "腿托高度",
+ };
+ label = `${seatLabel}${seatCapabilityLabels[capability] || "座椅"}`;
value = params.position_pct != null
- ? `${params.position_pct}%`
+ ? `${params.position_pct}%${params.angle_deg != null ? `（${params.angle_deg}°）` : ""}`
: motionLabel(params.state || params.mode_id);
icon = "armchair";
order = instructionOrder("座椅", "副驾", "主驾");
@@ -408,8 +415,12 @@
icon = "thermometer";
order = instructionOrder("空调", "温度");
} else if (domain === "climate") {
- label = "空调";
- value = params.level != null ? `${params.level}档` : motionLabel(params.state);
+ const cycleLabels = { internal: "内循环", external: "外循环", auto: "自动循环" };
+ label = capability === "climate.set_fan_level" ? "空调风量"
+ : capability === "climate.set_cycle_mode" ? "空调循环" : "空调";
+ value = params.level != null ? `${params.level}档`
+ : params.mode != null ? (cycleLabels[String(params.mode)] || String(params.mode))
+ : motionLabel(params.state);
icon = "fan";
order = instructionOrder("空调", "风量");
} else if (capability === "interior_light.set_reading_power") {
@@ -433,9 +444,15 @@
card.classList.add("is-busy");
status.hidden = false;
status.className = "assistant-mode-draft-status working";
- status.textContent = "正在保存到「我的模式」…";
+ const existingModeId = String(draft.existing_mode_id || "").trim();
+ status.textContent = existingModeId
+ ? "正在将验证结果保存为模式更新…"
+ : "正在保存到「我的模式」…";
try {
- const response = await window.CognitiveAgentNetwork.request("/api/mode-profiles", {
+ const endpoint = existingModeId
+ ? `/api/mode-profiles/${encodeURIComponent(existingModeId)}`
+ : "/api/mode-profiles";
+ const response = await window.CognitiveAgentNetwork.request(endpoint, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
@@ -456,10 +473,10 @@
card.classList.remove("is-busy");
card.classList.add("is-saved");
status.className = "assistant-mode-draft-status success";
- status.textContent = "模式已保存";
+ status.textContent = existingModeId ? "模式已更新" : "模式已保存";
card.querySelector(".assistant-mode-draft-actions")?.setAttribute("hidden", "");
await window.CognitiveAgentConsole?.refreshModeProfiles?.(payload.mode_profile?.id);
- appendModeCreatedState(message, payload.mode_profile);
+ appendModeCreatedState(message, payload.mode_profile, { updated: Boolean(existingModeId) });
persistMessages();
} catch (error) {
card.classList.remove("is-busy");
@@ -469,14 +486,14 @@
}
}
- async function tryModeProfileDraft(card, draft, buttons, status) {
+ async function tryModeProfileDraft(card, draft, actions, buttons, status) {
buttons.forEach((button) => { button.disabled = true; });
card.classList.add("is-busy");
status.hidden = false;
status.className = "assistant-mode-draft-status working";
status.textContent = "正在进行模拟试运行…";
try {
- const result = await window.CognitiveAgentConsole?.tryModeDraft?.(draft);
+ const result = await window.CognitiveAgentConsole?.tryModeDraft?.({ ...draft, actions });
if (!result || result.status !== "succeeded") throw new Error("试运行未通过，请检查动作设置。");
status.className = "assistant-mode-draft-status success";
status.textContent = "试运行通过 · 尚未保存";
@@ -501,7 +518,9 @@
const heading = document.createElement("header");
const headingCopy = document.createElement("span");
const eyebrow = document.createElement("small");
- eyebrow.textContent = "已为你生成功能模式";
+ eyebrow.textContent = draft.existing_mode_id
+ ? "已载入现有功能模式"
+ : "已为你生成功能模式";
const title = document.createElement("strong");
title.textContent = `「${draft.name}」`;
headingCopy.append(eyebrow, title);
@@ -530,7 +549,9 @@
const save = document.createElement("button");
save.type = "button";
save.className = "assistant-mode-draft-save";
- save.innerHTML = '保存为功能模式';
+ save.innerHTML = draft.existing_mode_id
+ ? '确认更新此模式'
+ : '保存为功能模式';
const trial = document.createElement("button");
trial.type = "button";
trial.className = "assistant-mode-draft-trial";
@@ -545,7 +566,7 @@
void saveModeProfileDraft(message, card, draft, buttons, status);
});
trial.addEventListener("click", () => {
- void tryModeProfileDraft(card, draft, buttons, status);
+ void tryModeProfileDraft(card, draft, actions, buttons, status);
});
card.append(heading, list, controls, status);
diff --git a/apps/web-console/index.html b/apps/web-console/index.html
index d34a6ab..28895d7 100755
--- a/apps/web-console/index.html
+++ b/apps/web-console/index.html
@@ -20,13 +20,13 @@
-
+
-
+
diff --git a/apps/web-console/service-worker.js b/apps/web-console/service-worker.js
index aeb5f23..fde7aaf 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-v269";
+const CACHE_VERSION = "cognitive-agent-pad-v271";
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-23",
"/network.js?v=android-pad-2",
- "/app.js?v=dual-seat-126",
+ "/app.js?v=dual-seat-127",
"/gesture-control.js?v=android-pad-9",
"/pwa.js?v=android-pad-3",
"/console-v2.css?v=delivery-state-65",
@@ -17,7 +17,7 @@ const APP_SHELL = [
"/speech-input.js?v=sensevoice-12",
"/speech-audio-worklet.js?v=sensevoice-5",
"/assistant-chat.css?v=codex-assistant-22",
- "/assistant-chat.js?v=codex-assistant-20",
+ "/assistant-chat.js?v=codex-assistant-22",
"/energy-monitor.css?v=energy-monitor-4",
"/energy-monitor.js?v=energy-monitor-4",
"/assets/pets/self-extension-cat.png?v=3",
diff --git a/deploy/server/Dockerfile b/deploy/server/Dockerfile
index 944ea8a..76dbacc 100755
--- a/deploy/server/Dockerfile
+++ b/deploy/server/Dockerfile
@@ -42,6 +42,7 @@ RUN node --version \
/app/artifacts/server \
/app/artifacts/tmp \
/app/artifacts/self-repair \
+ /app/artifacts/mode-profiles \
&& chown -R 10001:10001 /var/lib/codex /app/artifacts /home/cognitive-agent
USER 10001:10001
diff --git a/deploy/server/README.md b/deploy/server/README.md
index a785b9a..69818ec 100755
--- a/deploy/server/README.md
+++ b/deploy/server/README.md
@@ -16,11 +16,15 @@ The server layout is:
├── artifacts/user-presets/
├── artifacts/server/ # GSCU infrared catalog and server-side state
├── artifacts/tmp/ # Runtime gateway and calibration state
-└── artifacts/self-repair/
+├── artifacts/self-repair/
+└── artifacts/mode-profiles/ # User-created modes for all vehicles
```
-Vehicle-specific user-created function modes are persisted from
-`config/modes/vehicles/denza_n7/user/` into the otherwise read-only container.
+User-created function modes are persisted from the ignored host path
+`artifacts/mode-profiles/` into the service's separate writable user-mode
+registry at `/app/artifacts/mode-profiles/`, including common and future
+vehicle-specific modes.
+Preset files under `/app/config/modes` remain read-only and are never modified.
Build and start the service:
@@ -32,21 +36,21 @@ mkdir -p \
artifacts/server \
artifacts/tmp \
artifacts/self-repair \
- config/modes/vehicles/denza_n7/user
+ artifacts/mode-profiles
chown -R 10001:10001 \
artifacts/runs \
artifacts/user-presets \
artifacts/server \
artifacts/tmp \
artifacts/self-repair \
- config/modes/vehicles/denza_n7/user
+ artifacts/mode-profiles
chmod 0700 \
artifacts/runs \
artifacts/user-presets \
artifacts/server \
artifacts/tmp \
artifacts/self-repair \
- config/modes/vehicles/denza_n7/user
+ artifacts/mode-profiles
docker compose -f deploy/server/compose.yaml up -d --build
docker compose -f deploy/server/compose.yaml ps
```
diff --git a/deploy/server/compose.yaml b/deploy/server/compose.yaml
index 94842dd..2f1f64d 100755
--- a/deploy/server/compose.yaml
+++ b/deploy/server/compose.yaml
@@ -39,7 +39,7 @@ services:
- ../../artifacts/tmp:/app/artifacts/tmp
- ../../artifacts/self-repair:/app/artifacts/self-repair
- ../../artifacts/models:/app/artifacts/models:ro
- - ../../config/modes/vehicles/denza_n7/user:/app/config/modes/vehicles/denza_n7/user
+ - ../../artifacts/mode-profiles:/app/artifacts/mode-profiles
- codex_home:/var/lib/codex
read_only: true
tmpfs:
diff --git a/src/cognitive_agent/agent/prompts/codex_cockpit_assistant.md b/src/cognitive_agent/agent/prompts/codex_cockpit_assistant.md
index 87158f2..bc34513 100644
--- a/src/cognitive_agent/agent/prompts/codex_cockpit_assistant.md
+++ b/src/cognitive_agent/agent/prompts/codex_cockpit_assistant.md
@@ -6,6 +6,7 @@
- 只有 `n7_vehicle.execute_cockpit_request` 可以产生车辆、座椅、座舱模式或已授权外设的副作用。该工具不接受模型生成的控制文本或原始 CAN；它只执行服务端绑定到本回合令牌的原始需求或服务端规范化的舒适性动作。禁止用 shell、文件修改、原始 CAN 帧或其他工具绕过它。
- 对目录中的原子设备控制，调用工具时同时提交结构化 `plan`：只能逐字使用本回合能力目录中 `assistant_callable=true` 的 capability、target scope 和参数契约。对于目录中已有的座舱模式或当前 NFC 卡已授权的红外外设，省略 `plan`，由应用按动态注册表解析。结构化计划只是候选数据，应用会再次执行原始文本证据校验、Schema 校验和安全校验；本地确定性解析与计划重叠时，以本地结果为准。
+- 用户仅给出一个尚未登记的功能模式名称时，应结合该场景的通常含义，从本回合能力目录中选择少量、明确且相互协调的 `assistant_callable=true` 原子能力，提交一个结构化 `plan`。每项动作必须是可重复的绝对 `set` 状态，不得使用相对增减、时长动作、动态/循环模式、保存或应用操作；同一能力与目标不得重复。若任何动作或参数有歧义，就不要提交残缺计划。不得引用目录外能力，不得生成原始 CAN，不得声称已经保存或真实执行；应用会把整组计划原子校验为 dry-run 动作预览，用户确认后才会保存。已有模式名称不要重新推测动作，应省略 `plan` 复用工程配置。
- 用户明确要求控制车辆，或直接表达当前座舱不适并期待处理时，必须调用该工具；不要只给建议，也不要要求用户先选择“问答/车控”模式。
- “车上有点热”“有点热”“太热了”等当前不适，在没有否定、假设或纯知识提问时，直接调用一次工具；具体规范动作由服务端决定。
- “如果车上有点热应该怎么办”“为什么暴晒后车里会热”“车上有点热，但不要调整设备”等是假设、知识问题或否定请求，不得执行。
diff --git a/src/cognitive_agent/application/assistant_capabilities.py b/src/cognitive_agent/application/assistant_capabilities.py
index 667dfd4..0c16474 100644
--- a/src/cognitive_agent/application/assistant_capabilities.py
+++ b/src/cognitive_agent/application/assistant_capabilities.py
@@ -850,7 +850,14 @@ class AssistantCapabilityCatalog:
}
)
compact_modes = [
- {"id": item.get("id"), "name": item.get("name"), "aliases": item.get("aliases")}
+ {
+ "id": item.get("id"),
+ "name": item.get("name"),
+ "aliases": item.get("aliases"),
+ "description": item.get("description"),
+ "domains": item.get("domains"),
+ "operation_ids": item.get("operation_ids"),
+ }
for item in catalog.get("modes", [])
if isinstance(item, Mapping)
]
diff --git a/src/cognitive_agent/application/assistant_plan_guard.py b/src/cognitive_agent/application/assistant_plan_guard.py
index dc47189..4918b1f 100644
--- a/src/cognitive_agent/application/assistant_plan_guard.py
+++ b/src/cognitive_agent/application/assistant_plan_guard.py
@@ -30,6 +30,47 @@ MAX_PLAN_REQUESTS = 12
MAX_PLAN_PARAMETERS = 12
_OPERATIONS = frozenset({"activate", "deactivate", "increase", "decrease", "set", "save", "apply"})
+_SCENE_COMPOSABLE_CAPABILITIES = frozenset(
+ {
+ "seat.set_longitudinal_position",
+ "seat.set_cushion_tilt",
+ "seat.set_backrest_recline",
+ "seat.set_legrest_height",
+ "seat.set_legrest_extension",
+ "seat.set_ventilation",
+ "seat.set_heating",
+ "climate.set_power",
+ "climate.set_temperature",
+ "climate.set_fan_level",
+ "climate.set_cycle_mode",
+ "climate.set_control_mode",
+ "climate.set_outlet_mode",
+ "climate.set_front_defrost",
+ "roof.set_shade_position",
+ "roof.set_transmittance",
+ "roof.set_automatic_dimming",
+ "window.set_position",
+ "interior_light.set_indoor_power",
+ "interior_light.set_reading_power",
+ "interior_light.set_wall_power",
+ "ambient_light.set_power",
+ "ambient_light.set_illumination",
+ "ambient_light.set_mode",
+ "ambient_light.set_color_index",
+ "ambient_light.set_brightness_level",
+ "outlet.set_switch_state",
+ "outlet.set_position_percent",
+ }
+)
+# Reusable scenes are replayed through the native compiler after the user saves
+# them. Keep scene-only parameter bounds aligned with that compiler even when
+# a broader virtual capability contract accepts additional values. In
+# particular, fan level 0 is a virtual shorthand for "off", while the native
+# operation is explicitly 1-7; changing it to HVAC power-off here would alter
+# the model's proposed intent.
+_SCENE_PARAMETER_BOUNDS: dict[tuple[str, str], tuple[float | None, float | None]] = {
+ ("climate.set_fan_level", "level"): (1, 7),
+}
_ROOT_FIELDS = frozenset(
{"detected_domains", "requests", "ambiguities", "decision_summary", "confidence"}
)
@@ -404,6 +445,123 @@ def _registered_capabilities(
return registered
+def _registered_capability_specs(
+ capability_context: Mapping[str, Any],
+ executable_capabilities: AbstractSet[str],
+) -> dict[str, tuple[str, Mapping[str, Any]]]:
+ registered: dict[str, tuple[str, Mapping[str, Any]]] = {}
+ domains = capability_context.get("domains", [])
+ if not isinstance(domains, Sequence) or isinstance(
+ domains,
+ (str, bytes, bytearray),
+ ):
+ return registered
+ for raw_domain in domains:
+ if not isinstance(raw_domain, Mapping):
+ continue
+ domain = _safe_text(raw_domain.get("domain"), limit=64)
+ capabilities = raw_domain.get("capabilities", [])
+ if not isinstance(capabilities, Sequence) or isinstance(
+ capabilities,
+ (str, bytes, bytearray),
+ ):
+ continue
+ for raw_capability in capabilities:
+ if not isinstance(raw_capability, Mapping):
+ continue
+ capability = _safe_text(raw_capability.get("name"), limit=128)
+ if capability and domain and capability in executable_capabilities:
+ registered[capability] = (domain, raw_capability)
+ return registered
+
+
+def _scene_contract_error(
+ capability: str,
+ target_scope: str,
+ parameters: Sequence[Mapping[str, Any]],
+ specification: Mapping[str, Any],
+) -> str | None:
+ """Validate a scene request before adapter/native routing can drop it."""
+
+ target_scopes = specification.get("target_scopes", [])
+ if (
+ not isinstance(target_scopes, Sequence)
+ or isinstance(target_scopes, (str, bytes, bytearray))
+ or target_scope not in {str(item) for item in target_scopes}
+ ):
+ return f"能力 {capability} 不支持目标范围 {target_scope!r}"
+
+ raw_specs = specification.get("parameters", [])
+ if not isinstance(raw_specs, Sequence) or isinstance(
+ raw_specs,
+ (str, bytes, bytearray),
+ ):
+ return f"能力 {capability} 的参数契约不可用"
+ specs = {
+ str(item.get("name")): item
+ for item in raw_specs
+ if isinstance(item, Mapping) and str(item.get("name") or "").strip()
+ }
+ values = {str(item["name"]): item.get("value") for item in parameters}
+ unknown = sorted(set(values) - set(specs))
+ if unknown:
+ return f"能力 {capability} 不支持参数 {', '.join(unknown)}"
+ relative = sorted(
+ name
+ for name in values
+ if name in {"distance_mm", "angle_delta_deg"} or "delta" in name
+ )
+ if relative:
+ return f"功能模式不接受相对参数 {', '.join(relative)}"
+ if capability.startswith("seat.set_") and len(
+ {
+ name
+ for name in ("position_pct", "angle_deg")
+ if values.get(name) is not None
+ }
+ ) > 1:
+ return "座椅绝对位置只能使用 position_pct 或 angle_deg 中的一种表示"
+
+ for name, spec in specs.items():
+ value = values.get(name)
+ if spec.get("required") and value is None:
+ return f"能力 {capability} 缺少必填参数 {name}"
+ if value is None:
+ continue
+ expected_type = str(spec.get("type") or "")
+ numeric = isinstance(value, (int, float)) and not isinstance(value, bool)
+ type_valid = {
+ "number": numeric,
+ "integer": numeric and float(value).is_integer(),
+ "string": isinstance(value, str),
+ "boolean": isinstance(value, bool),
+ }.get(expected_type, False)
+ if not type_valid:
+ return f"参数 {name} 的类型必须是 {expected_type}"
+ minimum = spec.get("minimum")
+ maximum = spec.get("maximum")
+ scene_bounds = _SCENE_PARAMETER_BOUNDS.get((capability, name))
+ if scene_bounds is not None:
+ minimum, maximum = scene_bounds
+ if numeric and (
+ (minimum is not None and float(value) < float(minimum))
+ or (maximum is not None and float(value) > float(maximum))
+ ):
+ return (
+ f"参数 {name} 超出能力 {capability} 的允许范围 "
+ f"{minimum if minimum is not None else '-'}-"
+ f"{maximum if maximum is not None else '+'}"
+ )
+ allowed = spec.get("allowed_values")
+ if (
+ isinstance(allowed, Sequence)
+ and not isinstance(allowed, (str, bytes, bytearray))
+ and value not in allowed
+ ):
+ return f"参数 {name} 不在能力 {capability} 的允许值内"
+ return None
+
+
def _relevant_text(capability: str, text: str) -> str:
"""Narrow semantic checks to clauses mentioning this capability's subject."""
@@ -920,12 +1078,16 @@ def guard_assistant_plan(
executable_capabilities: AbstractSet[str],
fallback_intents: Sequence[Mapping[str, Any]] = (),
trusted_local: Mapping[str, Any] | None = None,
+ allow_scene_composition: bool = False,
) -> GuardedAssistantPlan:
"""Authorize a Codex plan against the server-owned unresolved text.
Rejected capability requests are omitted and represented as adapter-style
- ambiguities. Structurally malformed plans raise ``ValueError`` so callers
- can fail the tool call without invoking any executor.
+ ambiguities. Structurally malformed plans raise ``ValueError`` so callers
+ can fail the tool call without invoking any executor. ``allow_scene_composition``
+ is reserved for a name-only mode-creation preview: it relaxes textual value
+ evidence, but never registry membership, schema/range validation, dry-run,
+ or the user's explicit save confirmation.
"""
if not isinstance(raw_plan, Mapping) or set(raw_plan) != _ROOT_FIELDS:
@@ -948,6 +1110,10 @@ def guard_assistant_plan(
capability_context,
executable_capabilities,
)
+ registered_specs = _registered_capability_specs(
+ capability_context,
+ executable_capabilities,
+ )
parser_claims: set[str] = set()
for intent in fallback_intents:
if isinstance(intent, Mapping):
@@ -957,6 +1123,21 @@ def guard_assistant_plan(
accepted: list[dict[str, Any]] = []
rejected: list[dict[str, str]] = []
+ raw_ambiguities = raw_plan.get("ambiguities", [])
+ if allow_scene_composition:
+ if not isinstance(raw_ambiguities, Sequence) or isinstance(
+ raw_ambiguities,
+ (str, bytes, bytearray),
+ ):
+ raise ValueError("assistant scene plan ambiguities must be an array")
+ if raw_ambiguities:
+ rejected.append(
+ {
+ "field": "ambiguities",
+ "question": "场景候选计划仍有未解决歧义，已拒绝生成部分模式。",
+ }
+ )
+ scene_keys: set[tuple[str, str, str]] = set()
for index, raw_request in enumerate(requests):
if not isinstance(raw_request, Mapping) or set(raw_request) != _REQUEST_FIELDS:
raise ValueError(f"assistant plan request {index} is invalid")
@@ -974,11 +1155,30 @@ def guard_assistant_plan(
}
)
continue
+ if allow_scene_composition and capability not in _SCENE_COMPOSABLE_CAPABILITIES:
+ rejected.append(
+ {
+ "field": f"requests[{index}].capability",
+ "question": "该能力不是可重复功能模式允许组合的绝对状态能力。",
+ }
+ )
+ continue
+ if allow_scene_composition and operation != "set":
+ rejected.append(
+ {
+ "field": f"requests[{index}].operation",
+ "question": "功能模式只接受可重复的绝对 set 操作。",
+ }
+ )
+ continue
request_text = _relevant_text(capability, text)
- if non_execution or not _has_explicit_evidence(
- capability,
- request_text,
- parser_claims,
+ if non_execution or (
+ not allow_scene_composition
+ and not _has_explicit_evidence(
+ capability,
+ request_text,
+ parser_claims,
+ )
):
rejected.append(
{
@@ -987,7 +1187,10 @@ def guard_assistant_plan(
}
)
continue
- if not _target_is_evidenced(capability, target_scope, request_text):
+ if (
+ not allow_scene_composition
+ and not _target_is_evidenced(capability, target_scope, request_text)
+ ):
rejected.append(
{
"field": f"requests[{index}].target_scope",
@@ -1009,6 +1212,37 @@ def guard_assistant_plan(
names = [str(item["name"]) for item in normalized_parameters]
if len(names) != len(set(names)):
raise ValueError(f"assistant plan request {index} repeats a parameter")
+ if allow_scene_composition:
+ registered_spec = registered_specs.get(capability)
+ contract_error = (
+ _scene_contract_error(
+ capability,
+ target_scope,
+ normalized_parameters,
+ registered_spec[1],
+ )
+ if registered_spec is not None
+ else "能力契约不可用"
+ )
+ if contract_error is not None:
+ rejected.append(
+ {
+ "field": f"requests[{index}]",
+ "question": f"{contract_error}，已拒绝生成该功能模式。",
+ }
+ )
+ continue
+
+ scene_key = (domain, capability, target_scope)
+ if allow_scene_composition and scene_key in scene_keys:
+ rejected.append(
+ {
+ "field": f"requests[{index}]",
+ "question": "功能模式包含重复的能力和目标，已拒绝生成部分模式。",
+ }
+ )
+ continue
+ scene_keys.add(scene_key)
local_key = (domain, capability, target_scope)
if local_key in local_keys or (
@@ -1025,12 +1259,24 @@ def guard_assistant_plan(
or not 0 <= duration <= 600_000
):
raise ValueError(f"assistant plan request {index} has invalid duration_ms")
- semantic_error = _semantic_parameter_error(
- capability,
- operation,
- normalized_parameters,
- duration,
- request_text,
+ if allow_scene_composition and duration is not None:
+ rejected.append(
+ {
+ "field": f"requests[{index}].duration_ms",
+ "question": "功能模式只接受无时长的稳定目标状态。",
+ }
+ )
+ continue
+ semantic_error = (
+ None
+ if allow_scene_composition
+ else _semantic_parameter_error(
+ capability,
+ operation,
+ normalized_parameters,
+ duration,
+ request_text,
+ )
)
if semantic_error is not None:
rejected.append(
@@ -1052,6 +1298,11 @@ def guard_assistant_plan(
}
)
+ if allow_scene_composition and rejected:
+ # A reusable mode must be atomic. Never offer a draft that silently
+ # dropped one hallucinated, conflicting, relative, or transient step.
+ accepted = []
+
summaries = raw_plan.get("decision_summary", [])
if not isinstance(summaries, Sequence) or isinstance(summaries, (str, bytes, bytearray)):
summaries = []
diff --git a/src/cognitive_agent/application/codex_assistant.py b/src/cognitive_agent/application/codex_assistant.py
index 87949c6..72d1b19 100644
--- a/src/cognitive_agent/application/codex_assistant.py
+++ b/src/cognitive_agent/application/codex_assistant.py
@@ -19,6 +19,7 @@ from pathlib import Path
from typing import Protocol
LOGGER = logging.getLogger(__name__)
+INTERNAL_SCENE_BUILDER_MARKER = "根据该场景名称使用当前可控原子能力"
class CodexAssistantError(RuntimeError):
@@ -99,6 +100,7 @@ class _VehicleTurnContext:
capability_queries: int = 0
action_succeeded: bool | None = None
action_state: str | None = None
+ action_result: Mapping[str, object] | None = None
@dataclass(frozen=True)
@@ -139,9 +141,9 @@ _MODE_PROFILE_CREATE_PATTERN = re.compile(
r"(?:构建|创建|新建|新增|生成|添加)(?:一下)?"
r"(?:一个|个|一套|套)?"
r"(?:名为|叫做|叫)?[“\"'「『]?"
- r"[^，。,.!！:：\s“”\"'「」『』]{1,64}?模式[”\"'」』]?"
+ r"(?P[^，。,.!！:：\s“”\"'「」『』]{1,64}?模式)[”\"'」』]?"
r"(?:配置)?(?:并(?:添加|加入)(?:到)?功能模式(?:栏)?)?"
- r"(?:[，,:：].{1,800})?[。.!！]?$"
+ r"(?:[，,:：](?P.{1,800}))?[。.!！]?$"
)
_HOT_DISCOMFORT_PATTERN = re.compile(r"热|闷热|好热|太晒")
_COLD_DISCOMFORT_PATTERN = re.compile(r"冷|好冷")
@@ -164,12 +166,114 @@ def _normalize_instruction(value: object) -> str:
return re.sub(r"[\s，。！？!?、；;：:]", "", str(value).lower())
+def _mode_profile_create_parts(message: str) -> tuple[str, str] | None:
+ """Return the requested scene name and optional explicit settings."""
+
+ normalized = "".join(str(message).strip().split())
+ match = _MODE_PROFILE_CREATE_PATTERN.fullmatch(normalized)
+ if match is None:
+ return None
+ return (
+ str(match.group("name") or "").strip(),
+ str(match.group("instruction") or "").strip("，,:：。.!！"),
+ )
+
+
+def is_mode_profile_create_request(message: str) -> bool:
+ """Return whether the original user text is a bounded mode-create request."""
+
+ return _mode_profile_create_parts(message) is not None
+
+
+def _catalog_exact_mode_match_count(
+ catalog: Mapping[str, object],
+ requested_name: str,
+) -> int | None:
+ """Count exact visible mode matches; ``None`` means catalog unavailable."""
+
+ needle = "".join(str(requested_name).casefold().split())
+ source_status = catalog.get("source_status")
+ if "modes" not in catalog or (
+ isinstance(source_status, Mapping)
+ and source_status.get("modes") != "ready"
+ ):
+ return None
+ matched_ids: set[str] = set()
+ raw_modes = catalog.get("modes", [])
+ if not isinstance(raw_modes, Sequence) or isinstance(
+ raw_modes,
+ (str, bytes, bytearray),
+ ):
+ return None
+ for raw_mode in raw_modes:
+ if not isinstance(raw_mode, Mapping):
+ continue
+ candidates = [raw_mode.get("id"), raw_mode.get("name")]
+ aliases = raw_mode.get("aliases", [])
+ if isinstance(aliases, Sequence) and not isinstance(
+ aliases,
+ (str, bytes, bytearray),
+ ):
+ candidates.extend(aliases)
+ if needle in {
+ "".join(str(candidate or "").casefold().split())
+ for candidate in candidates
+ if str(candidate or "").strip()
+ }:
+ matched_ids.add(str(raw_mode.get("id") or ""))
+ matched_ids.discard("")
+ return len(matched_ids)
+
+
def _vehicle_result_succeeded(result: Mapping[str, object]) -> bool:
"""Only a terminal, fully successful service result completes a tool call."""
return str(result.get("status") or "").lower() == "succeeded"
+def _mode_profile_result_succeeded(result: Mapping[str, object]) -> bool:
+ """Require a complete, non-physical mode preview before claiming success."""
+
+ if (
+ not _vehicle_result_succeeded(result)
+ or str(result.get("execution_mode") or "").lower() != "dry_run"
+ or result.get("hardware_execution_allowed") is not False
+ or result.get("orchestration_route") != "mode_profile_create_preview"
+ ):
+ return False
+ draft = result.get("mode_profile_draft")
+ if not isinstance(draft, Mapping):
+ return False
+ operation_count = draft.get("operation_count")
+ if (
+ not isinstance(operation_count, int)
+ or isinstance(operation_count, bool)
+ or operation_count <= 0
+ ):
+ return False
+ intent_ir = result.get("intent_ir")
+ spec = intent_ir.get("spec") if isinstance(intent_ir, Mapping) else None
+ actions = spec.get("actions") if isinstance(spec, Mapping) else None
+ if not isinstance(actions, Sequence) or isinstance(
+ actions,
+ (str, bytes, bytearray),
+ ):
+ return False
+ if len(actions) != operation_count or not all(
+ isinstance(action, Mapping) for action in actions
+ ):
+ return False
+ validation = result.get("validation", [])
+ return not (
+ isinstance(validation, Sequence)
+ and not isinstance(validation, (str, bytes, bytearray))
+ and any(
+ isinstance(issue, Mapping) and issue.get("severity") == "error"
+ for issue in validation
+ )
+ )
+
+
def _vehicle_verification_status(result: Mapping[str, object]) -> str:
"""Normalize the service's execution evidence without inventing feedback."""
@@ -231,7 +335,7 @@ def _authorize_vehicle_request(
):
return _VehicleAuthorization(None, "knowledge_or_non_vehicle_request")
- if _MODE_PROFILE_CREATE_PATTERN.fullmatch(normalized):
+ if is_mode_profile_create_request(normalized):
# Named-mode creation is a bounded configuration operation. The
# vehicle service parses the same original sentence again and always
# validates the generated scene in dry-run before offering it to the
@@ -373,7 +477,27 @@ class CodexAssistantService:
self._fast_implicit_controls and authorization.reason in _FAST_COMFORT_REASONS
)
mode_profile_preview = authorization.reason == "explicit_mode_profile_create"
- direct_vehicle_control = fast_comfort_control or mode_profile_preview
+ mode_parts = _mode_profile_create_parts(message) if mode_profile_preview else None
+ exact_mode_matches = (
+ _catalog_exact_mode_match_count(catalog, mode_parts[0])
+ if mode_parts
+ else None
+ )
+ direct_mode_profile_preview = bool(
+ mode_parts
+ and (
+ mode_parts[1]
+ # Ambiguous or unavailable mode catalogs fail closed in the
+ # local builder. Only a known-zero match is safe to hand to
+ # semantic scene composition.
+ or exact_mode_matches != 0
+ )
+ )
+ # Explicit device settings and repository-backed scenes remain on the
+ # zero-model-latency path. A genuinely new name-only scene needs Codex
+ # to compose a candidate plan from the live capability catalog; the
+ # vehicle service still limits it to dry-run and requires confirmation.
+ direct_vehicle_control = fast_comfort_control or direct_mode_profile_preview
route = (
"capability_catalog"
if authoritative_answer
@@ -385,7 +509,7 @@ class CodexAssistantService:
"n7web-capability-registry"
if authoritative_answer
else "n7web-mode-builder"
- if mode_profile_preview
+ if direct_mode_profile_preview
else "n7web-control-router"
if fast_comfort_control
else "codex-app-server"
@@ -414,6 +538,8 @@ class CodexAssistantService:
"正在读取工程能力目录……"
if authoritative_answer
else "正在构建并校验功能模式……"
+ if direct_mode_profile_preview
+ else "Codex 正在理解场景并组合工程能力……"
if mode_profile_preview
else "正在快速处理座舱舒适需求……"
if fast_comfort_control
@@ -542,6 +668,8 @@ class CodexAssistantService:
safe_result = self._safe_vehicle_result(result)
action_state, successful, completion_text = _vehicle_action_state(result)
if mode_profile_preview:
+ successful = _mode_profile_result_succeeded(result)
+ action_state = "simulated" if successful else "failed"
completion_text = (
"功能模式预览已生成" if successful else "功能模式预览生成失败"
)
@@ -632,6 +760,14 @@ class CodexAssistantService:
reason = clean(failure.get("reason"))
next_step = clean(failure.get("next_step"))
if reason and next_step:
+ if (
+ INTERNAL_SCENE_BUILDER_MARKER in reason
+ or INTERNAL_SCENE_BUILDER_MARKER in next_step
+ ):
+ return (
+ "未能根据场景名称生成通过工程能力校验的座舱动作",
+ "请稍后重试；若该场景尚未登记，可补充希望联动的设备状态",
+ )
return reason, next_step
unresolved = result.get("unresolved_clauses")
@@ -644,6 +780,8 @@ class CodexAssistantService:
clean(item, limit=120)
for item in unresolved[:3]
if clean(item, limit=120)
+ and INTERNAL_SCENE_BUILDER_MARKER
+ not in clean(item, limit=120)
)
)
if clauses:
@@ -716,6 +854,40 @@ class CodexAssistantService:
if isinstance(draft, Mapping)
else "新功能模式"
)
+ if isinstance(draft, Mapping) and draft.get("existing_mode_id"):
+ raw_operation_count = draft.get("operation_count")
+ operation_count = (
+ raw_operation_count
+ if isinstance(raw_operation_count, int)
+ and not isinstance(raw_operation_count, bool)
+ and raw_operation_count >= 0
+ else 0
+ )
+ if draft.get("generation_source") == "registered_mode_profile":
+ return (
+ f"工程中已有“{name}”，已按当前车型载入预览并验证"
+ f"{operation_count}项设备动作。可以先试运行，确认后更新现有模式。"
+ )
+ return (
+ f"已按你提供的设置为现有“{name}”生成预览并验证"
+ f"{operation_count}项设备动作。可以先试运行，确认后更新现有模式。"
+ )
+ if (
+ isinstance(draft, Mapping)
+ and draft.get("generation_source") == "codex_scene_composition"
+ ):
+ raw_operation_count = draft.get("operation_count")
+ operation_count = (
+ raw_operation_count
+ if isinstance(raw_operation_count, int)
+ and not isinstance(raw_operation_count, bool)
+ and raw_operation_count > 0
+ else 0
+ )
+ return (
+ f"已理解“{name}”的场景意图，并从当前工程能力中组合、验证了"
+ f"{operation_count}项设备动作。确认预览后可保存为功能模式，也可以先试运行。"
+ )
return f"已为你生成“{name}”预览。确认动作后可保存为功能模式，也可以先试运行。"
@staticmethod
@@ -765,6 +937,9 @@ class CodexAssistantService:
query_capabilities,
)
vehicle_control_expected = tool_context.authorization.instruction is not None
+ mode_profile_expected = (
+ tool_context.authorization.reason == "explicit_mode_profile_create"
+ )
with self._tool_contexts_lock:
self._tool_contexts[request_token] = tool_context
@@ -938,8 +1113,6 @@ class CodexAssistantService:
raise CodexAssistantError(str(detail or f"Codex turn {status}"))
if not final_answer:
final_answer = streamed_answer.strip()
- if not final_answer:
- raise CodexAssistantError("Codex produced an empty response")
if vehicle_control_expected:
if not tool_context.called:
tool_context.action_succeeded = False
@@ -955,9 +1128,25 @@ class CodexAssistantService:
}
)
elif tool_context.action_succeeded is not True:
- final_answer = "车辆控制未完成，受控执行链未返回成功结果。"
+ final_answer = (
+ self._mode_profile_preview_answer(
+ tool_context.action_result or {},
+ False,
+ )
+ if mode_profile_expected
+ else "车辆控制未完成，受控执行链未返回成功结果。"
+ )
+ elif mode_profile_expected:
+ final_answer = self._mode_profile_preview_answer(
+ tool_context.action_result or {},
+ True,
+ )
elif tool_context.action_state == "submitted":
final_answer = "控制指令已提交，正在等待车辆状态确认。"
+ elif not final_answer and tool_context.action_state == "simulated":
+ final_answer = "车辆控制模拟编排已完成。"
+ elif not final_answer and tool_context.action_state == "verified":
+ final_answer = "车辆控制已完成并通过状态确认。"
if (
emit(
{
@@ -969,6 +1158,8 @@ class CodexAssistantService:
is False
):
return {"cancelled": True, "answer": final_answer}
+ if not final_answer:
+ raise CodexAssistantError("Codex produced an empty response")
work_succeeded = (
tool_context.called and tool_context.action_succeeded is True
if vehicle_control_expected
@@ -980,7 +1171,11 @@ class CodexAssistantService:
"message_id": message_id,
"success": work_succeeded,
"display_text": (
- "车辆控制模拟编排已完成"
+ "功能模式预览已生成"
+ if mode_profile_expected and work_succeeded
+ else "功能模式预览生成失败"
+ if mode_profile_expected
+ else "车辆控制模拟编排已完成"
if vehicle_control_expected
and work_succeeded
and tool_context.action_state == "simulated"
@@ -1106,6 +1301,15 @@ class CodexAssistantService:
secret=request_token,
)
action_state, successful, completion_text = _vehicle_action_state(result)
+ if context.authorization.reason == "explicit_mode_profile_create":
+ successful = _mode_profile_result_succeeded(result)
+ action_state = "simulated" if successful else "failed"
+ completion_text = (
+ "功能模式预览已生成"
+ if successful
+ else "功能模式预览生成失败"
+ )
+ context.action_result = result
context.action_succeeded = successful
context.action_state = action_state
context.events.put(
@@ -1158,7 +1362,17 @@ class CodexAssistantService:
if "decision_summary" in normalized_plan:
normalized_plan["decision_summary"] = ["Codex 结构化计划已提交应用安全链复核。"]
if "ambiguities" in normalized_plan:
- normalized_plan["ambiguities"] = []
+ normalized_plan["ambiguities"] = (
+ [
+ {
+ "field": "assistant_plan",
+ "question": "Codex 候选计划包含未解决歧义。",
+ }
+ ]
+ if isinstance(normalized_plan["ambiguities"], list)
+ and normalized_plan["ambiguities"]
+ else []
+ )
requests = normalized_plan.get("requests")
if isinstance(requests, list):
for request in requests:
@@ -1482,6 +1696,9 @@ class CodexAssistantService:
"vehicle_id",
"domains",
"operation_count",
+ "existing_mode_id",
+ "existing_mode_revision",
+ "generation_source",
)
if key in mode_profile_draft
}
@@ -1507,6 +1724,7 @@ class CodexAssistantService:
)
for item in unresolved_items
if isinstance(item, str)
+ and INTERNAL_SCENE_BUILDER_MARKER not in item
],
}
peripheral = result.get("peripheral")
diff --git a/src/cognitive_agent/application/mode_profile.py b/src/cognitive_agent/application/mode_profile.py
index dee6d4f..f17a2f1 100644
--- a/src/cognitive_agent/application/mode_profile.py
+++ b/src/cognitive_agent/application/mode_profile.py
@@ -7,6 +7,7 @@ import json
import re
from dataclasses import dataclass
from pathlib import Path
+from threading import RLock
from typing import Any
import yaml
@@ -419,26 +420,28 @@ class ModeProfileRegistry:
self.user_root = Path(user_root) if user_root is not None else self.profile_root
self.session_user_profiles = user_root is not None
self._profiles: list[ModeProfile] = []
+ self._write_lock = RLock()
self.reload()
def reload(self) -> None:
- profiles: list[ModeProfile] = []
- if self.profile_root.exists():
- for path in sorted(self.profile_root.rglob("*.yaml")):
- if not path.is_file():
- continue
- if self.session_user_profiles and _is_user_profile_path(
- path, self.profile_root
- ):
- continue
- profiles.append(ModeProfile.from_yaml(path, self.profile_root))
- if self.session_user_profiles and self.user_root.exists():
- profiles.extend(
- ModeProfile.from_yaml(path, self.user_root)
- for path in sorted(self.user_root.rglob("*.yaml"))
- if path.is_file()
- )
- self._profiles = profiles
+ with self._write_lock:
+ profiles: list[ModeProfile] = []
+ if self.profile_root.exists():
+ for path in sorted(self.profile_root.rglob("*.yaml")):
+ if not path.is_file():
+ continue
+ if self.session_user_profiles and _is_user_profile_path(
+ path, self.profile_root
+ ):
+ continue
+ profiles.append(ModeProfile.from_yaml(path, self.profile_root))
+ if self.session_user_profiles and self.user_root.exists():
+ profiles.extend(
+ ModeProfile.from_yaml(path, self.user_root)
+ for path in sorted(self.user_root.rglob("*.yaml"))
+ if path.is_file()
+ )
+ self._profiles = profiles
def profiles(self) -> list[ModeProfile]:
return list(self._profiles)
@@ -505,6 +508,61 @@ class ModeProfileRegistry:
matches.append(selected[profile.id])
return matches
+ def resolve_exact_name(
+ self,
+ value: str,
+ vehicle_id: str | None = None,
+ ) -> ModeProfile | None:
+ """Resolve one visible mode by an exact ID, name, or alias.
+
+ Scene creation must not use the substring semantics of :meth:`match_text`:
+ a name can contain another mode alias and accidentally combine two profiles.
+ Returning ``None`` for both no match and an ambiguous alias keeps that path
+ fail-closed so callers can choose an explicit fallback without combining modes.
+ """
+
+ needle = _normalize_match_text(str(value).strip())
+ if not needle:
+ return None
+ matches = [
+ profile
+ for profile in self._profiles_for_vehicle(vehicle_id)
+ if needle
+ in {
+ _normalize_match_text(str(candidate))
+ for candidate in (profile.id, profile.name, *profile.aliases)
+ if str(candidate).strip()
+ }
+ ]
+ return matches[0] if len(matches) == 1 else None
+
+ @staticmethod
+ def profile_revision(profile: ModeProfile) -> str:
+ """Return a stable semantic revision for optimistic mode updates."""
+
+ document = {
+ "id": profile.id,
+ "name": profile.name,
+ "aliases": list(profile.aliases),
+ "description": profile.description,
+ "icon": profile.icon,
+ "scope": {
+ "common": profile.common,
+ "vehicle_ids": list(profile.vehicle_ids),
+ },
+ "operations": [operation.to_yaml() for operation in profile.operations],
+ "timeline": profile.timeline.to_yaml() if profile.timeline is not None else None,
+ "instruction": profile.instruction,
+ "command_flow": profile.command_flow,
+ }
+ canonical = json.dumps(
+ document,
+ ensure_ascii=False,
+ sort_keys=True,
+ separators=(",", ":"),
+ )
+ return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16]
+
@staticmethod
def requests_for_profiles(profiles: list[ModeProfile]) -> list[dict[str, Any]]:
return [
@@ -526,6 +584,8 @@ class ModeProfileRegistry:
instruction: str = "",
command_flow: dict[str, Any] | None = None,
timeline: ModeTimeline | None = None,
+ expected_revision: str | None = None,
+ create_only: bool = False,
) -> ModeProfile:
clean_name = _clean_name(name)
canonical_operations = [
@@ -566,20 +626,64 @@ class ModeProfileRegistry:
if vehicle_id
else self.user_root / "common" / "user" / f"{profile_id}.yaml"
)
- path.parent.mkdir(parents=True, exist_ok=True)
- temporary = path.with_suffix(path.suffix + ".tmp")
- temporary.write_text(
- yaml.safe_dump(document, allow_unicode=True, sort_keys=False),
- encoding="utf-8",
- )
- temporary.replace(path)
- self.reload()
- return ModeProfile.from_yaml(path, self.user_root)
+ with self._write_lock:
+ # Refresh inside the same critical section as the compare-and-swap.
+ # This makes concurrent assistant saves one-winner within the server
+ # process instead of letting two stale previews overwrite each other.
+ self.reload()
+ visible_profiles = self._profiles_for_vehicle(vehicle_id)
+ current = next(
+ (profile for profile in visible_profiles if profile.id == profile_id),
+ None,
+ )
+ normalized_name = _normalize_match_text(clean_name)
+ name_collision = any(
+ profile.id != profile_id
+ and normalized_name
+ in {
+ _normalize_match_text(str(candidate))
+ for candidate in (profile.id, profile.name, *profile.aliases)
+ if str(candidate).strip()
+ }
+ for profile in visible_profiles
+ )
+ if name_collision:
+ raise ValueError("该模式名称已被其他模式使用。")
+ if create_only and current is not None:
+ raise ValueError("该模式已由其他会话创建，请刷新后重试。")
+ if expected_revision is not None:
+ expected = str(expected_revision).strip()
+ if (
+ not expected
+ or current is None
+ or self.profile_revision(current) != expected
+ ):
+ raise ValueError(
+ "功能模式已被其他会话更新，请重新生成预览后再保存。"
+ )
+
+ path.parent.mkdir(parents=True, exist_ok=True)
+ temporary = path.with_suffix(path.suffix + ".tmp")
+ temporary.write_text(
+ yaml.safe_dump(document, allow_unicode=True, sort_keys=False),
+ encoding="utf-8",
+ )
+ temporary.replace(path)
+ self.reload()
+ return ModeProfile.from_yaml(path, self.user_root)
def delete_user_mode(
self,
mode_id: str,
vehicle_id: str | None = None,
+ ) -> ModeProfile:
+ with self._write_lock:
+ return self._delete_user_mode_locked(mode_id, vehicle_id)
+
+ def _delete_user_mode_locked(
+ self,
+ mode_id: str,
+ vehicle_id: str | None = None,
) -> ModeProfile:
clean_id = str(mode_id).strip()
if not clean_id:
@@ -623,6 +727,14 @@ class ModeProfileRegistry:
self,
profile: ModeProfile,
operations: list[dict[str, Any] | ModeOperation],
+ ) -> ModeProfile:
+ with self._write_lock:
+ return self._update_mode_locked(profile, operations)
+
+ def _update_mode_locked(
+ self,
+ profile: ModeProfile,
+ operations: list[dict[str, Any] | ModeOperation],
) -> ModeProfile:
canonical_updates = _canonical_operations(operations)
if not canonical_updates:
@@ -642,6 +754,10 @@ class ModeProfileRegistry:
return ModeProfile.from_yaml(path, self._root_for_path(path))
def restore_initial_mode(self, profile: ModeProfile) -> ModeProfile:
+ with self._write_lock:
+ return self._restore_initial_mode_locked(profile)
+
+ def _restore_initial_mode_locked(self, profile: ModeProfile) -> ModeProfile:
path = Path(profile.source)
backup_path = _initial_backup_path(path)
if not backup_path.is_file():
diff --git a/src/cognitive_agent/application/seat_service.py b/src/cognitive_agent/application/seat_service.py
index 91dc494..1c3135e 100755
--- a/src/cognitive_agent/application/seat_service.py
+++ b/src/cognitive_agent/application/seat_service.py
@@ -8,6 +8,7 @@ import re
import sys
import time
import uuid
+from collections import Counter
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from copy import deepcopy
@@ -728,6 +729,90 @@ class SeatDemoService:
vehicle_id,
)
+ def trial_mode_profile_draft(
+ self,
+ name: str,
+ instruction: str,
+ source_request_id: str,
+ vehicle_id: str | None = None,
+ ) -> dict[str, Any]:
+ """Replay the server-owned dry-run evidence for one assistant draft."""
+
+ clean_name = str(name).strip()
+ clean_instruction = str(instruction).strip()
+ clean_request_id = str(source_request_id).strip()
+ resolved_vehicle_id = vehicle_id or self.vehicle_profiles.default().id
+ if not clean_name or not clean_instruction:
+ raise ValueError("功能模式草稿名称和指令不能为空。")
+ if not MODE_EDITOR_REQUEST_ID_PATTERN.fullmatch(clean_request_id):
+ raise ValueError("功能模式草稿来源请求编号无效。")
+
+ result_path = self.artifact_root / clean_request_id / "result.json"
+ if not result_path.is_file():
+ raise ValueError("未找到功能模式草稿，请重新生成。")
+ try:
+ source_result = json.loads(result_path.read_text(encoding="utf-8"))
+ except (OSError, json.JSONDecodeError) as exc:
+ raise ValueError("功能模式草稿结果无法读取，请重新生成。") from exc
+ if not isinstance(source_result, dict) or source_result.get(
+ "request_id"
+ ) != clean_request_id:
+ raise ValueError("功能模式草稿来源结果不完整，请重新生成。")
+ if (
+ source_result.get("status") != "succeeded"
+ or source_result.get("execution_mode") != "dry_run"
+ ):
+ raise ValueError("功能模式草稿没有通过完整的模拟校验。")
+
+ draft = source_result.get("mode_profile_draft")
+ if not isinstance(draft, dict):
+ raise ValueError("来源请求不是可试运行的功能模式草稿。")
+ if str(draft.get("source_request_id") or "").strip() != clean_request_id:
+ raise ValueError("功能模式草稿来源请求编号不匹配，请重新生成。")
+ if str(draft.get("name") or "").strip() != clean_name:
+ raise ValueError("功能模式草稿名称不匹配，请重新生成。")
+ if str(draft.get("instruction") or "").strip() != clean_instruction:
+ raise ValueError("功能模式草稿指令不匹配，请重新生成。")
+ if str(draft.get("vehicle_id") or "").strip() != resolved_vehicle_id:
+ raise ValueError("功能模式草稿不属于当前车型。")
+
+ intent_ir = source_result.get("intent_ir")
+ spec = intent_ir.get("spec") if isinstance(intent_ir, dict) else None
+ actions = spec.get("actions") if isinstance(spec, dict) else None
+ function_configs = source_result.get("function_configs")
+ operation_count = draft.get("operation_count")
+ if (
+ not isinstance(actions, list)
+ or not actions
+ or not isinstance(function_configs, dict)
+ or not function_configs
+ or not isinstance(operation_count, int)
+ or isinstance(operation_count, bool)
+ or len(actions) != operation_count
+ ):
+ raise ValueError("功能模式草稿动作证据不完整，请重新生成。")
+ validation = source_result.get("validation", [])
+ validation = validation if isinstance(validation, list) else []
+ if any(
+ isinstance(issue, dict) and issue.get("severity") == "error"
+ for issue in validation
+ ):
+ raise ValueError("功能模式草稿未通过安全校验，请重新生成。")
+
+ return {
+ "request_id": clean_request_id,
+ "status": "succeeded",
+ "execution_mode": "dry_run",
+ "hardware_execution_allowed": False,
+ "orchestration_route": "mode_profile_draft_trial",
+ "mode_profile_draft": deepcopy(draft),
+ "intent_ir": {"spec": {"actions": deepcopy(actions)}},
+ "function_configs": deepcopy(function_configs),
+ "execution_report": deepcopy(source_result.get("execution_report")),
+ "validation": deepcopy(validation),
+ "trial_source": "server_validated_preview_artifact",
+ }
+
def _persist_mode_profile_card(
self,
mode_id: str | None,
@@ -797,11 +882,18 @@ class SeatDemoService:
if source_vehicle_id and source_vehicle_id != resolved_vehicle_id:
raise ValueError("控制指令流不属于当前车型，请重新生成。")
source_draft = source_result.get("mode_profile_draft")
+ draft_existing_mode_revision = ""
if isinstance(source_draft, dict):
draft_request_id = str(source_draft.get("source_request_id") or "").strip()
draft_vehicle_id = str(source_draft.get("vehicle_id") or "").strip()
draft_instruction = str(source_draft.get("instruction") or "").strip()
draft_name = str(source_draft.get("name") or "").strip()
+ draft_existing_mode_id = str(
+ source_draft.get("existing_mode_id") or ""
+ ).strip()
+ draft_existing_mode_revision = str(
+ source_draft.get("existing_mode_revision") or ""
+ ).strip()
if draft_request_id != clean_request_id:
raise ValueError("功能模式草稿与控制指令流不匹配，请重新生成。")
if draft_vehicle_id and draft_vehicle_id != resolved_vehicle_id:
@@ -810,6 +902,19 @@ class SeatDemoService:
raise ValueError("自然语言指令已修改，请重新生成模式预览。")
if draft_name and draft_name != clean_name:
raise ValueError("功能模式名称已修改，请重新生成模式预览。")
+ if clean_mode_id and not draft_existing_mode_id:
+ raise ValueError("新功能模式预览不能通过已有模式的更新接口保存。")
+ if draft_existing_mode_id and draft_existing_mode_id != clean_mode_id:
+ raise ValueError("该预览属于已有功能模式，请通过对应模式的更新接口保存。")
+ if draft_existing_mode_id and not draft_existing_mode_revision:
+ raise ValueError("功能模式预览版本过旧，请重新生成后再更新。")
+ if (
+ draft_existing_mode_revision
+ and profile is not None
+ and draft_existing_mode_revision
+ != self.mode_profiles.profile_revision(profile)
+ ):
+ raise ValueError("功能模式已被其他会话更新，请重新生成预览后再保存。")
recorded_instruction = str(source_result.get("utterance", "")).strip()
if not recorded_instruction:
source_spans = {
@@ -920,6 +1025,8 @@ class SeatDemoService:
instruction=clean_instruction,
command_flow=command_flow,
timeline=profile.timeline if profile is not None else None,
+ expected_revision=draft_existing_mode_revision or None,
+ create_only=not bool(clean_mode_id),
)
public_profile = next(
(
@@ -1422,15 +1529,7 @@ class SeatDemoService:
"conversation_id must start with conv- and contain only lowercase letters, numbers, or hyphens"
)
vehicle_profile = self.vehicle_profiles.get(vehicle_id)
- control_plane_request = any(
- matcher(utterance) is not None
- for matcher in (
- self._parse_mode_profile_save_name,
- self._parse_mode_profile_create_name,
- self._parse_mode_profile_restore_name,
- self._parse_mode_profile_update,
- )
- )
+ control_plane_request = self.is_mode_profile_configuration_request(utterance)
if control_plane_request:
result = self.process(
utterance,
@@ -2070,6 +2169,8 @@ class SeatDemoService:
direct_mode_profile_id: str | None = None,
speech_understanding: dict[str, Any] | None = None,
assistant_plan: dict[str, Any] | None = None,
+ allow_fuzzy_mode_profile_match: bool = True,
+ allow_scene_composition: bool = False,
) -> dict[str, Any]:
control_transport = self._validate_control_transport(control_transport)
utterance = utterance.strip()
@@ -2151,7 +2252,11 @@ class SeatDemoService:
[direct_mode_profile]
if direct_mode_profile is not None
else []
- if direct_can_batch is not None or speech_understanding is not None
+ if (
+ direct_can_batch is not None
+ or speech_understanding is not None
+ or not allow_fuzzy_mode_profile_match
+ )
else self.mode_profiles.match_text(
utterance,
vehicle_id=vehicle_profile.id,
@@ -2325,6 +2430,7 @@ class SeatDemoService:
local_preview=local_preview,
model_parse_source=model_parse_source,
assistant_plan=assistant_plan,
+ allow_scene_composition=allow_scene_composition,
)
dynamic_mode_stop = self._is_dynamic_mode_stop_request(utterance)
if dynamic_mode_stop:
@@ -3422,6 +3528,21 @@ class SeatDemoService:
)
return generic.group("name").strip() if generic is not None else None
+ @classmethod
+ def is_mode_profile_configuration_request(cls, utterance: str) -> bool:
+ """Identify mode configuration text that must never reach live executors."""
+
+ clean_utterance = str(utterance).strip()
+ return any(
+ matcher(clean_utterance) is not None
+ for matcher in (
+ cls._parse_mode_profile_save_name,
+ cls._parse_mode_profile_create_name,
+ cls._parse_mode_profile_restore_name,
+ cls._parse_mode_profile_update,
+ )
+ )
+
@staticmethod
def _parse_mode_profile_create_name(utterance: str) -> str | None:
explicit = CREATE_MODE_PROFILE_NAME.search(utterance)
@@ -4378,6 +4499,109 @@ class SeatDemoService:
"issues": issues,
}
+ @staticmethod
+ def _scene_plan_signatures(
+ assistant_plan: dict[str, Any],
+ ) -> Counter[tuple[str, str, str]]:
+ signatures: Counter[tuple[str, str, str]] = Counter()
+ requests = assistant_plan.get("requests", [])
+ if not isinstance(requests, list):
+ return signatures
+ for request in requests:
+ if not isinstance(request, dict):
+ continue
+ domain = str(request.get("domain") or "")
+ capability = str(request.get("capability") or "")
+ target = str(request.get("target_scope") or "")
+ if domain == "seat" and target == "front_right":
+ target = "passenger"
+ if domain and capability and target:
+ signatures[(domain, capability, target)] += 1
+ return signatures
+
+ @staticmethod
+ def _scene_action_signatures(
+ actions: list[dict[str, Any]],
+ ) -> Counter[tuple[str, str, str]]:
+ signatures: Counter[tuple[str, str, str]] = Counter()
+ for action in actions:
+ domain = str(action.get("domain") or "")
+ capability = str(action.get("capability") or "")
+ raw_target = action.get("target")
+ target = (
+ str(raw_target.get("instance") or raw_target.get("zone") or "")
+ if isinstance(raw_target, dict)
+ else str(action.get("target_scope") or raw_target or "")
+ )
+ if domain == "seat" and target == "front_right":
+ target = "passenger"
+ if domain and capability and target:
+ signatures[(domain, capability, target)] += 1
+ return signatures
+
+ @staticmethod
+ def _scene_parameters_preserved(
+ assistant_plan: dict[str, Any],
+ actions: list[dict[str, Any]],
+ ) -> bool:
+ actual_parameters: dict[tuple[str, str, str], dict[str, Any]] = {}
+ for action in actions:
+ domain = str(action.get("domain") or "")
+ capability = str(action.get("capability") or "")
+ raw_target = action.get("target")
+ target = (
+ str(raw_target.get("instance") or raw_target.get("zone") or "")
+ if isinstance(raw_target, dict)
+ else str(action.get("target_scope") or raw_target or "")
+ )
+ if domain == "seat" and target == "front_right":
+ target = "passenger"
+ parameters = action.get("parameters")
+ if domain and capability and target and isinstance(parameters, dict):
+ actual_parameters[(domain, capability, target)] = parameters
+
+ requests = assistant_plan.get("requests", [])
+ if not isinstance(requests, list):
+ return False
+ for request in requests:
+ if not isinstance(request, dict):
+ return False
+ domain = str(request.get("domain") or "")
+ capability = str(request.get("capability") or "")
+ target = str(request.get("target_scope") or "")
+ if domain == "seat" and target == "front_right":
+ target = "passenger"
+ parameters = request.get("parameters")
+ if not isinstance(parameters, list):
+ return False
+ expected = {
+ str(parameter.get("name") or ""): parameter.get("value")
+ for parameter in parameters
+ if isinstance(parameter, dict)
+ }
+ actual = actual_parameters.get((domain, capability, target))
+ if actual is None or any(actual.get(name) != value for name, value in expected.items()):
+ return False
+ return True
+
+ @staticmethod
+ def _scrub_failed_scene_preview(result: dict[str, Any]) -> None:
+ failed_intent = result.get("intent_ir")
+ failed_spec = (
+ failed_intent.get("spec") if isinstance(failed_intent, dict) else None
+ )
+ if isinstance(failed_spec, dict):
+ failed_spec["actions"] = []
+ failed_spec["requests"] = []
+ result["deferred_requests"] = []
+ result["model_requests"] = []
+ result["function_config"] = None
+ result["function_configs"] = {}
+ result["command_plan"] = None
+ result["command_plans"] = {}
+ result["scene_config"] = None
+ result["execution_report"] = None
+
def _process_mode_profile_create(
self,
utterance: str,
@@ -4405,9 +4629,25 @@ class SeatDemoService:
if create_match is not None
else ""
)
- source_utterance = requested_instruction or (
- f"进入{mode_name}，请根据该场景名称使用当前可控原子能力构建并执行场景"
+ existing_profile = self.mode_profiles.resolve_exact_name(
+ mode_name,
+ vehicle_id=vehicle_profile.id,
)
+ source_profile = existing_profile if not requested_instruction else None
+ draft_mode_name = existing_profile.name if existing_profile is not None else mode_name
+ generation_source = (
+ "registered_mode_profile"
+ if source_profile is not None
+ else "codex_scene_composition"
+ if assistant_plan is not None and not requested_instruction
+ else "explicit_scene_requirements"
+ )
+ # A name-only request may already describe a repository-backed scene.
+ # Reuse that vehicle-scoped profile as trusted structured input instead
+ # of appending a model prompt to the user's utterance. The latter was
+ # being parsed as a second control clause on the assistant's local fast
+ # path, turning a valid nine-action mode into an incomplete draft.
+ source_utterance = requested_instruction or f"进入{mode_name}"
source_result = self._process_once(
source_utterance,
provider,
@@ -4417,7 +4657,20 @@ class SeatDemoService:
vehicle_profile.id,
simulation_fault,
control_transport=control_transport,
+ direct_mode_profile_id=(
+ source_profile.id if source_profile is not None else None
+ ),
assistant_plan=assistant_plan,
+ # ``source_utterance`` is synthesized for a name-only, unknown
+ # scene. It must not fall back to substring matching (for example
+ # ``我的小憩模式`` -> ``小憩模式``), otherwise the exact resolver's
+ # ambiguity and collision checks can be bypassed.
+ allow_fuzzy_mode_profile_match=bool(requested_instruction),
+ allow_scene_composition=(
+ assistant_plan is not None
+ and source_profile is None
+ and not requested_instruction
+ ),
)
draft_context = self._context_for_vehicle(
self.conversations.snapshot(conversation_id),
@@ -4434,9 +4687,12 @@ class SeatDemoService:
clause
for clause, _parsed, status in local_clause_results
if status != "complete"
- and not self.mode_profiles.match_text(
- clause,
- vehicle_id=vehicle_profile.id,
+ and not (
+ (requested_instruction or source_profile is not None)
+ and self.mode_profiles.match_text(
+ clause,
+ vehicle_id=vehicle_profile.id,
+ )
)
]
if provider == "local" and assistant_plan is None and unresolved_draft_clauses:
@@ -4468,7 +4724,49 @@ class SeatDemoService:
]
self._persist(source_result)
return source_result
+ source_intent = source_result.get("intent_ir")
+ source_spec = (
+ source_intent.get("spec") if isinstance(source_intent, dict) else None
+ )
+ source_actions = (
+ source_spec.get("actions", []) if isinstance(source_spec, dict) else []
+ )
+ typed_source_actions = [
+ action for action in source_actions if isinstance(action, dict)
+ ] if isinstance(source_actions, list) else []
+ if (
+ source_result.get("status") == "succeeded"
+ and generation_source == "codex_scene_composition"
+ and assistant_plan is not None
+ and (
+ self._scene_plan_signatures(assistant_plan)
+ != self._scene_action_signatures(typed_source_actions)
+ or not self._scene_parameters_preserved(
+ assistant_plan,
+ typed_source_actions,
+ )
+ )
+ ):
+ source_result["status"] = "validation_failed"
+ source_result["validation"] = [
+ *list(source_result.get("validation", [])),
+ {
+ "severity": "error",
+ "code": "MODE_PROFILE_SCENE_PLAN_INCOMPLETE",
+ "path": "intent_ir.spec.actions",
+ "message": (
+ "Codex 场景候选动作未全部进入当前车型的可复用执行链，"
+ "已拒绝生成部分功能模式。"
+ ),
+ },
+ ]
+
if source_result.get("status") != "succeeded":
+ if generation_source == "codex_scene_composition":
+ # Final fail-closed barrier for errors discovered after the
+ # plan guard (adapter, schema, compiler, or safety validation).
+ # A failed reusable scene must not expose a trial-able subset.
+ self._scrub_failed_scene_preview(source_result)
source_result["mode_creation_failure"] = {
"code": "MODE_PROFILE_ACTION_VALIDATION_FAILED",
"reason": "场景动作未通过车辆能力或安全校验",
@@ -4528,12 +4826,23 @@ class SeatDemoService:
)
source_request_id = str(source_result.get("request_id") or "")
source_result["mode_profile_draft"] = {
- "name": mode_name,
+ "name": draft_mode_name,
"instruction": source_utterance,
"source_request_id": source_request_id,
"vehicle_id": vehicle_profile.id,
"domains": domains,
"operation_count": len(reusable_actions),
+ **(
+ {
+ "existing_mode_id": existing_profile.id,
+ "existing_mode_revision": (
+ self.mode_profiles.profile_revision(existing_profile)
+ ),
+ }
+ if existing_profile is not None
+ else {}
+ ),
+ "generation_source": generation_source,
}
source_result["mode_creation"] = {
"requested_utterance": utterance,
@@ -4542,12 +4851,34 @@ class SeatDemoService:
"source_status": source_result.get("status"),
"source_provider": source_result.get("provider_used"),
"operation_count": len(reusable_actions),
+ **(
+ {
+ "existing_mode_id": existing_profile.id,
+ "existing_mode_revision": (
+ self.mode_profiles.profile_revision(existing_profile)
+ ),
+ }
+ if existing_profile is not None
+ else {}
+ ),
+ "generation_source": generation_source,
}
source_result["orchestration_route"] = "mode_profile_create_preview"
source_result["decision_summary"] = [
- f"已构建并验证“{mode_name}”场景。",
+ (
+ f"已从当前车型的工程模式目录载入并验证“{source_profile.name}”的"
+ f" {len(reusable_actions)} 项设备动作。"
+ if source_profile is not None
+ else f"Codex 已根据场景语义组合并验证“{draft_mode_name}”的工程能力。"
+ if generation_source == "codex_scene_composition"
+ else f"已构建并验证“{mode_name}”场景。"
+ ),
*list(source_result.get("decision_summary", [])),
- "请确认动作后保存为功能模式，或先进行一次模拟试运行。",
+ (
+ "可先进行一次模拟试运行，确认动作后更新现有功能模式。"
+ if existing_profile is not None
+ else "请确认动作后保存为功能模式，或先进行一次模拟试运行。"
+ ),
]
# Overwrite the persisted source result with the draft metadata. The
# mode-card endpoint later re-reads this server-owned artifact and
@@ -6236,6 +6567,7 @@ class SeatDemoService:
local_preview: bool = False,
model_parse_source: str = "cloud_model",
assistant_plan: dict[str, Any] | None = None,
+ allow_scene_composition: bool = False,
) -> tuple[dict[str, Any], str, str | None]:
vehicle_profile = vehicle_profile or self.vehicle_profiles.default()
clause_results = self._local_clause_results(utterance, context)
@@ -6282,6 +6614,7 @@ class SeatDemoService:
if status != "complete"
],
trusted_local=trusted_local,
+ allow_scene_composition=allow_scene_composition,
)
model = self.model_plan_adapter.adapt(
guarded.plan,
@@ -6289,6 +6622,65 @@ class SeatDemoService:
vehicle_profile,
utterance=fallback_utterance,
)
+ if allow_scene_composition:
+ promoted_native: list[dict[str, Any]] = []
+ remaining_domain_actions: list[dict[str, Any]] = []
+ for action in model.get("domain_actions", []):
+ if self.native_can_runtime.supports(action):
+ promoted = dict(action)
+ promoted["execution_backend"] = "native_can"
+ promoted_native.append(promoted)
+ else:
+ remaining_domain_actions.append(action)
+ remaining_deferred: list[dict[str, Any]] = []
+ for deferred in model.get("deferred_requests", []):
+ candidate = {
+ "domain": str(deferred.get("domain") or ""),
+ "capability": str(deferred.get("capability") or ""),
+ "target": str(
+ deferred.get("target_scope")
+ or deferred.get("target")
+ or ""
+ ),
+ "operation": str(deferred.get("desired_change") or "set"),
+ "parameters": dict(deferred.get("parameters") or {}),
+ "reason": "Codex 场景候选动作已通过工程能力契约校验。",
+ "execution_backend": "native_can",
+ }
+ if self.native_can_runtime.supports(candidate):
+ promoted_native.append(candidate)
+ else:
+ remaining_deferred.append(deferred)
+ model.setdefault("ambiguities", []).append(
+ {
+ "field": "assistant_plan",
+ "question": (
+ f"能力 {candidate['capability']} 无法进入当前车型的"
+ "可复用执行链。"
+ ),
+ }
+ )
+ model["native_can_actions"] = [
+ *model.get("native_can_actions", []),
+ *promoted_native,
+ ]
+ model["domain_actions"] = remaining_domain_actions
+ model["deferred_requests"] = remaining_deferred
+ if allow_scene_composition and model.get("ambiguities"):
+ # Adapter validation is the second half of the scene gate: it
+ # owns targets, required parameters, types and ranges. If any
+ # candidate fails there, discard every sibling action so the
+ # reusable draft can never become a silently partial scene.
+ for key in (
+ "actions",
+ "domain_actions",
+ "native_can_actions",
+ "deferred_requests",
+ "preset_actions",
+ "preset_events",
+ "model_requests",
+ ):
+ model[key] = []
model = self._mark_control_parse_source(
model,
model_parse_source,
diff --git a/src/cognitive_agent/server.py b/src/cognitive_agent/server.py
index 1aeff3d..caf311e 100755
--- a/src/cognitive_agent/server.py
+++ b/src/cognitive_agent/server.py
@@ -486,6 +486,9 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
if path.startswith("/api/self-repair/"):
self._handle_self_repair_post(path)
return
+ if path == "/api/mode-profile-drafts/trial":
+ self._handle_mode_profile_draft_trial()
+ return
if path == "/api/mode-profiles" or path.startswith("/api/mode-profiles/"):
self._handle_mode_profile_post(path)
return
@@ -1201,6 +1204,29 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
"""Route IR-only or mixed IR/CAN requests without changing CAN parsing."""
utterance = str(common_arguments.get("utterance") or "").strip()
+ if self.service.is_mode_profile_configuration_request(utterance):
+ # Creating, saving, updating, or restoring a reusable mode is
+ # configuration, never a live-control command. Bypass the
+ # NFC/peripheral executor so a learned device name embedded in the
+ # mode sentence cannot claim it before SeatDemoService handles it.
+ mode_arguments = {
+ **common_arguments,
+ "execution_mode": "dry_run",
+ }
+ callback = (
+ on_commands_validated
+ if callable(on_commands_validated)
+ else None
+ )
+ if path == "/api/requests/preview":
+ return self.service.preview_native_can(
+ **mode_arguments,
+ on_commands_validated=callback,
+ )
+ return self.service.process(
+ **mode_arguments,
+ _on_commands_validated=callback,
+ )
catalog = self.extension.list_peripheral_capabilities()
nfc_snapshot = self.nfc.snapshot()
partition = partition_active_peripheral_intent(
@@ -2003,6 +2029,22 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
HTTPStatus.BAD_REQUEST,
)
+ def _handle_mode_profile_draft_trial(self) -> None:
+ try:
+ payload = self._read_json_body(max_bytes=16 * 1024)
+ result = self.service.trial_mode_profile_draft(
+ str(payload.get("name", "")),
+ str(payload.get("instruction", "")),
+ str(payload.get("source_request_id", "")),
+ str(payload.get("vehicle_id", "")) or None,
+ )
+ self._json(result, HTTPStatus.OK)
+ except (ValueError, json.JSONDecodeError) as exc:
+ self._json(
+ {"error": "invalid_request", "message": str(exc)},
+ HTTPStatus.BAD_REQUEST,
+ )
+
def _saved_mode_infrared_entries(
self,
profile: object,
diff --git a/tests/e2e/test_assistant_web_api.py b/tests/e2e/test_assistant_web_api.py
index 768fcc3..c6baf7f 100644
--- a/tests/e2e/test_assistant_web_api.py
+++ b/tests/e2e/test_assistant_web_api.py
@@ -178,6 +178,7 @@ class FakeVehicleService:
def __init__(self) -> None:
self.calls: list[dict[str, object]] = []
self.mode_save_calls: list[dict[str, object]] = []
+ self.mode_trial_calls: list[dict[str, object]] = []
self.fail = False
def preview_native_can(self, **kwargs: object) -> dict[str, object]:
@@ -286,6 +287,7 @@ class FakeVehicleService:
"conversation_id": kwargs["conversation_id"],
"status": "succeeded",
"execution_mode": "dry_run",
+ "hardware_execution_allowed": False,
"supplement_required": False,
"orchestration_route": "mode_profile_create_preview",
"mode_profile_draft": {
@@ -361,6 +363,31 @@ class FakeVehicleService:
"source_request_id": source_request_id,
}
+ def trial_mode_profile_draft(
+ self,
+ name: str,
+ instruction: str,
+ source_request_id: str,
+ vehicle_id: str | None = None,
+ ) -> dict[str, object]:
+ self.mode_trial_calls.append(
+ {
+ "name": name,
+ "instruction": instruction,
+ "source_request_id": source_request_id,
+ "vehicle_id": vehicle_id,
+ }
+ )
+ return {
+ "request_id": source_request_id,
+ "status": "succeeded",
+ "execution_mode": "dry_run",
+ "hardware_execution_allowed": False,
+ "orchestration_route": "mode_profile_draft_trial",
+ "intent_ir": {"spec": {"actions": []}},
+ "function_configs": {},
+ }
+
def _auth_disabled() -> AuthRuntime:
return AuthRuntime(AuthConfig(False, "", "", b"", 3600, 5, 300, 120, False))
@@ -402,6 +429,7 @@ class TestCodexAssistantWebApi:
def setup_method(self) -> None:
AssistantApiHandler.service.calls.clear()
AssistantApiHandler.service.mode_save_calls.clear()
+ AssistantApiHandler.service.mode_trial_calls.clear()
AssistantApiHandler.service.fail = False
FAKE_CODEX.available = True
FAKE_CODEX.turn_calls.clear()
@@ -636,6 +664,32 @@ class TestCodexAssistantWebApi:
assert "预览" in events[-1]["answer"]
assert "已添加" not in events[-1]["answer"]
+ def test_mode_draft_trial_reuses_the_server_owned_preview(self) -> None:
+ status, content_type, payload = self._request_json(
+ "/api/mode-profile-drafts/trial",
+ {
+ "name": "午休模式",
+ "instruction": "进入午休模式",
+ "source_request_id": "req-api-mode-trial",
+ "vehicle_id": "denza_n7",
+ },
+ )
+
+ assert status == 200
+ assert content_type == "application/json"
+ assert payload["status"] == "succeeded"
+ assert payload["execution_mode"] == "dry_run"
+ assert payload["hardware_execution_allowed"] is False
+ assert payload["orchestration_route"] == "mode_profile_draft_trial"
+ assert AssistantApiHandler.service.mode_trial_calls == [
+ {
+ "name": "午休模式",
+ "instruction": "进入午休模式",
+ "source_request_id": "req-api-mode-trial",
+ "vehicle_id": "denza_n7",
+ }
+ ]
+
def test_exact_movie_mode_request_streams_a_nine_action_preview(self) -> None:
events = self.post_assistant(
{
diff --git a/tests/e2e/test_seat_web_api.py b/tests/e2e/test_seat_web_api.py
index aeb651d..676e350 100755
--- a/tests/e2e/test_seat_web_api.py
+++ b/tests/e2e/test_seat_web_api.py
@@ -19,6 +19,7 @@ from cognitive_agent.application.self_extension import (
GatewayError,
SelfExtensionService,
SimulationGateway,
+ partition_active_peripheral_intent,
)
from cognitive_agent.auth import AuthConfig, AuthRuntime, hash_password
from cognitive_agent.infrastructure.can.access import HardwareAccessController
@@ -620,7 +621,7 @@ operations:
def test_static_console_and_runtime_endpoint(self) -> None:
with self.opener.open(f"{self.base_url}/", timeout=3) as response:
html = response.read().decode("utf-8")
- with self.opener.open(f"{self.base_url}/app.js?v=dual-seat-126", timeout=3) as response:
+ with self.opener.open(f"{self.base_url}/app.js?v=dual-seat-127", 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
@@ -655,7 +656,7 @@ operations:
) as response:
extension_styles = response.read().decode("utf-8").replace("\r\n", "\n")
with self.opener.open(
- f"{self.base_url}/assistant-chat.js?v=codex-assistant-20", timeout=3
+ f"{self.base_url}/assistant-chat.js?v=codex-assistant-22", timeout=3
) as response:
assistant_script = response.read().decode("utf-8")
with self.opener.open(
@@ -696,7 +697,7 @@ operations:
self.assertIn("CMD 0x431 · FB 0x41A · CH1", html)
self.assertIn("CMD 0x431 · FB 0x388 · CH1", html)
self.assertNotIn("透明车体 · 实体设备", html)
- self.assertIn('src="/app.js?v=dual-seat-126"', html)
+ self.assertIn('src="/app.js?v=dual-seat-127"', html)
self.assertIn('href="/styles.css?v=dual-seat-23"', html)
self.assertIn('src="/gesture-control.js?v=android-pad-9"', html)
self.assertNotIn('class="vehicle-system-hud"', html)
@@ -718,7 +719,7 @@ operations:
self.assertIn('aria-label="发送消息"', html)
self.assertIn('id="assistantTranscript"', html)
self.assertIn('href="/assistant-chat.css?v=codex-assistant-22"', html)
- self.assertIn('src="/assistant-chat.js?v=codex-assistant-20"', html)
+ self.assertIn('src="/assistant-chat.js?v=codex-assistant-22"', html)
self.assertIn('href="/energy-monitor.css?v=energy-monitor-4"', html)
self.assertIn('src="/energy-monitor.js?v=energy-monitor-4"', html)
self.assertNotIn('"admin"', console_script)
@@ -991,14 +992,14 @@ 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-v269"', service_worker)
+ self.assertIn('const CACHE_VERSION = "cognitive-agent-pad-v271"', service_worker)
self.assertIn('/console-v2.css?v=delivery-state-65', service_worker)
self.assertIn('/console-v2.js?v=delivery-state-62', service_worker)
self.assertIn('/assistant-chat.css?v=codex-assistant-22', service_worker)
- self.assertIn('/assistant-chat.js?v=codex-assistant-20', service_worker)
+ self.assertIn('/assistant-chat.js?v=codex-assistant-22', service_worker)
self.assertIn('/energy-monitor.css?v=energy-monitor-4', service_worker)
self.assertIn('/energy-monitor.js?v=energy-monitor-4', service_worker)
- self.assertIn('"/app.js?v=dual-seat-126"', service_worker)
+ self.assertIn('"/app.js?v=dual-seat-127"', service_worker)
self.assertIn('"/speech-input.js?v=sensevoice-12"', service_worker)
self.assertIn('"/speech-input.css?v=sensevoice-7"', service_worker)
self.assertIn('"/speech-audio-worklet.js?v=sensevoice-5"', service_worker)
@@ -1455,7 +1456,7 @@ operations:
) as response:
extension_script = response.read().decode("utf-8")
with self.opener.open(
- f"{self.base_url}/app.js?v=dual-seat-126", timeout=3
+ f"{self.base_url}/app.js?v=dual-seat-127", timeout=3
) as response:
app_script = response.read().decode("utf-8")
@@ -1814,6 +1815,92 @@ operations:
self.assertEqual(operation["domain"], "self_extension")
self.assertTrue(operation["telemetry"]["gateway_acknowledged"])
+ def test_mode_creation_cannot_be_claimed_by_live_nfc_peripheral(self) -> None:
+ original_extension = SeatDemoHandler.extension
+ original_nfc = SeatDemoHandler.nfc
+ with tempfile.TemporaryDirectory() as directory:
+ gateway = SimulationGateway()
+ extension = SelfExtensionService(
+ gateway,
+ catalog_db_path=Path(directory, "mode-create-peripheral.sqlite3"),
+ )
+ extension.sync_peripheral_catalog(
+ [
+ {
+ "id": "ir-slot:flashing-toy:1",
+ "status": "learned",
+ "enabled": True,
+ "device_id": "flashing-toy",
+ "device_name": "闪灯玩具",
+ "object_names": ["闪灯玩具"],
+ "card_uid": "2CC8B103",
+ "action": "打开",
+ "action_names": ["打开", "开启"],
+ "risk": "low",
+ "slot": 1,
+ }
+ ]
+ )
+ nfc_snapshot = {
+ "revision": 1,
+ "reader": {"online": True},
+ "card": {"action": "present", "uid": "2CC8B103"},
+ }
+ utterance = "创建一个打开闪灯玩具模式"
+ partition = partition_active_peripheral_intent(
+ utterance,
+ extension.list_peripheral_capabilities(),
+ nfc_snapshot,
+ )
+ self.assertTrue(partition["peripheral_actions"])
+
+ SeatDemoHandler.extension = extension
+ SeatDemoHandler.nfc = SimpleNamespace(snapshot=lambda: nfc_snapshot)
+ request = urllib.request.Request(
+ f"{self.base_url}/api/requests/preview",
+ method="POST",
+ headers={"Content-Type": "application/json"},
+ data=json.dumps(
+ {
+ "utterance": utterance,
+ "provider": "local",
+ "conversation_id": "conv-mode-create-peripheral-boundary",
+ "execution_mode": "hardware",
+ "client_type": "web_console_assistant",
+ },
+ ensure_ascii=False,
+ ).encode("utf-8"),
+ )
+ try:
+ with patch.object(gateway, "send", wraps=gateway.send) as send:
+ with self.opener.open(request, timeout=3) as response:
+ result = json.loads(response.read().decode("utf-8"))
+ self.assertEqual(send.call_count, 0)
+ finally:
+ extension.close()
+ SeatDemoHandler.extension = original_extension
+ SeatDemoHandler.nfc = original_nfc
+
+ self.assertEqual(result["execution_mode"], "dry_run")
+ self.assertFalse(result["hardware_execution_allowed"])
+ self.assertNotEqual(result.get("orchestration_route"), "nfc_peripheral")
+ self.assertNotIn("peripheral", result)
+
+ def test_all_mode_configuration_phrases_are_control_plane_only(self) -> None:
+ for utterance in (
+ "创建一个打开闪灯玩具模式",
+ "固化为打开闪灯玩具模式",
+ "修改打开闪灯玩具模式，空调调整到24度",
+ "恢复打开闪灯玩具模式初始配置",
+ ):
+ with self.subTest(utterance=utterance):
+ self.assertTrue(
+ self.service.is_mode_profile_configuration_request(utterance)
+ )
+ self.assertFalse(
+ self.service.is_mode_profile_configuration_request("进入小憩模式")
+ )
+
def test_speech_routes_exact_asr_text_to_learned_peripheral_and_ui_config(
self,
) -> None:
diff --git a/tests/integration/test_named_mode_inference_regression.py b/tests/integration/test_named_mode_inference_regression.py
new file mode 100644
index 0000000..1ed46b4
--- /dev/null
+++ b/tests/integration/test_named_mode_inference_regression.py
@@ -0,0 +1,1086 @@
+from __future__ import annotations
+
+import json
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from threading import Barrier
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from cognitive_agent.application.codex_assistant import CodexAssistantService
+from cognitive_agent.application.assistant_capabilities import AssistantCapabilityCatalog
+from cognitive_agent.application.mode_profile import ModeProfileRegistry
+from cognitive_agent.application.seat_service import (
+ REPOSITORY_ROOT,
+ SeatDemoService,
+)
+
+NAP_REQUEST = "构建一个小憩模式"
+INTERNAL_SCENE_PLACEHOLDER = "请根据该场景名称使用当前可控原子能力构建并执行场景"
+EXPECTED_NAP_CAPABILITIES = {
+ "seat.set_longitudinal_position",
+ "seat.set_cushion_tilt",
+ "seat.set_backrest_recline",
+ "seat.set_legrest_height",
+ "climate.set_temperature",
+ "climate.set_fan_level",
+ "climate.set_cycle_mode",
+ "roof.set_shade_position",
+ "ambient_light.set_power",
+}
+
+
+def _scene_request(
+ domain: str,
+ capability: str,
+ target_scope: str,
+ parameters: dict[str, object],
+) -> dict[str, object]:
+ return {
+ "domain": domain,
+ "capability": capability,
+ "target_scope": target_scope,
+ "operation": "set",
+ "duration_ms": None,
+ "parameters": [
+ {"name": name, "value": value}
+ for name, value in parameters.items()
+ ],
+ "reason": "根据午休场景组合当前工程已登记能力。",
+ }
+
+
+LUNCH_BREAK_SCENE_PLAN = {
+ "detected_domains": ["seat", "climate", "roof", "ambient_light"],
+ "requests": [
+ _scene_request(
+ "seat",
+ "seat.set_backrest_recline",
+ "passenger",
+ {"position_pct": 70},
+ ),
+ _scene_request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": 23},
+ ),
+ _scene_request(
+ "climate",
+ "climate.set_fan_level",
+ "vehicle",
+ {"level": 1},
+ ),
+ _scene_request(
+ "roof",
+ "roof.set_shade_position",
+ "vehicle",
+ {"position_pct": 0},
+ ),
+ _scene_request(
+ "ambient_light",
+ "ambient_light.set_power",
+ "vehicle",
+ {"state": "off"},
+ ),
+ ],
+ "ambiguities": [],
+ "decision_summary": ["组合安静、遮光且温度适宜的午休场景。"],
+ "confidence": 0.92,
+}
+
+
+@pytest.fixture
+def vehicle_service(tmp_path: Path) -> SeatDemoService:
+ service = SeatDemoService(REPOSITORY_ROOT)
+ service.artifact_root = tmp_path / "runs"
+ service.mode_profiles = ModeProfileRegistry(
+ REPOSITORY_ROOT / "config" / "modes",
+ user_root=tmp_path / "user-modes",
+ )
+ return service
+
+
+def _build_nap_mode(service: SeatDemoService, conversation_id: str) -> dict[str, object]:
+ with patch.object(
+ service.openai,
+ "interpret",
+ side_effect=AssertionError(
+ "a repository-backed named mode must not depend on a cloud model"
+ ),
+ ):
+ return service.preview_native_can(
+ NAP_REQUEST,
+ provider="local",
+ conversation_id=conversation_id,
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ )
+
+
+def _intent_actions(result: dict[str, object]) -> list[dict[str, object]]:
+ intent_ir = result.get("intent_ir")
+ assert isinstance(intent_ir, dict)
+ spec = intent_ir.get("spec")
+ assert isinstance(spec, dict)
+ actions = spec.get("actions")
+ assert isinstance(actions, list)
+ assert all(isinstance(action, dict) for action in actions)
+ return actions
+
+
+def test_name_only_nap_request_reuses_the_verified_repository_profile(
+ vehicle_service: SeatDemoService,
+ tmp_path: Path,
+) -> None:
+ preset = vehicle_service.mode_profiles.get_profile("nap", "denza_n7")
+
+ result = _build_nap_mode(vehicle_service, "conv-nap-name-inference")
+
+ assert result["status"] == "succeeded"
+ assert result["execution_mode"] == "dry_run"
+ assert result["hardware_execution_allowed"] is False
+ assert result["orchestration_route"] == "mode_profile_create_preview"
+ assert result["mode_profile"] is None
+ draft = result["mode_profile_draft"]
+ assert isinstance(draft, dict)
+ assert draft["name"] == "小憩模式"
+ assert draft["operation_count"] == len(preset.operations) == 9
+ assert set(draft["domains"]) == {"seat", "climate", "roof", "ambient_light"}
+ assert draft["existing_mode_id"] == "nap"
+ assert isinstance(draft["existing_mode_revision"], str)
+ assert len(draft["existing_mode_revision"]) == 16
+ assert draft["generation_source"] == "registered_mode_profile"
+
+ actions = _intent_actions(result)
+ assert {action["capability"] for action in actions} == {
+ operation.capability for operation in preset.operations
+ } == EXPECTED_NAP_CAPABILITIES
+ expected_targets_and_parameters = {
+ operation.capability: (
+ "passenger"
+ if operation.domain == "seat" and operation.target_scope == "front_right"
+ else operation.target_scope,
+ operation.parameters,
+ )
+ for operation in preset.operations
+ }
+ actual_targets_and_parameters = {
+ action["capability"]: (
+ action["target"].get("instance") or action["target"].get("zone"),
+ action["parameters"],
+ )
+ for action in actions
+ }
+ assert actual_targets_and_parameters == expected_targets_and_parameters
+ assert not list((tmp_path / "user-modes").rglob("*.yaml"))
+
+ public_result = json.dumps(result, ensure_ascii=False, default=str)
+ assert INTERNAL_SCENE_PLACEHOLDER not in public_result
+
+
+def test_inferred_nap_actions_are_bounded_by_the_runtime_capability_catalog(
+ vehicle_service: SeatDemoService,
+) -> None:
+ result = _build_nap_mode(vehicle_service, "conv-nap-capability-boundary")
+ actions = _intent_actions(result)
+ catalog = {
+ capability["name"]: capability
+ for capability in vehicle_service.runtime_info()["capabilities"]
+ }
+
+ for action in actions:
+ capability = catalog[action["capability"]]
+ assert capability.get("execution_support") != "observational_only"
+ assert capability.get("physical_control_supported", True) is not False
+ assert action["domain"] in {"seat", "climate", "roof", "ambient_light"}
+
+ assert "door.set_vehicle_lock" not in {
+ action["capability"] for action in actions
+ }
+ assert all("media" not in str(action["capability"]) for action in actions)
+
+
+def test_codex_composes_an_unregistered_name_only_scene_from_runtime_capabilities(
+ vehicle_service: SeatDemoService,
+ tmp_path: Path,
+) -> None:
+ with patch.object(
+ vehicle_service.openai,
+ "interpret",
+ side_effect=AssertionError("the embedded Codex plan must not invoke legacy GPT"),
+ ):
+ result = vehicle_service.preview_native_can(
+ "构建一个午休模式",
+ provider="local",
+ conversation_id="conv-codex-scene-composition",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ assistant_plan=LUNCH_BREAK_SCENE_PLAN,
+ )
+
+ draft = result["mode_profile_draft"]
+ assert result["status"] == "succeeded"
+ assert result["execution_mode"] == "dry_run"
+ assert result["hardware_execution_allowed"] is False
+ assert result["provider_used"] == "local_rules+codex_app_server"
+ assert result["orchestration_route"] == "mode_profile_create_preview"
+ assert draft["name"] == "午休模式"
+ assert draft["operation_count"] == 5
+ assert draft["generation_source"] == "codex_scene_composition"
+ assert "existing_mode_id" not in draft
+ assert {action["capability"] for action in _intent_actions(result)} == {
+ "seat.set_backrest_recline",
+ "climate.set_temperature",
+ "climate.set_fan_level",
+ "roof.set_shade_position",
+ "ambient_light.set_power",
+ }
+ assert not list((tmp_path / "user-modes").rglob("*.yaml"))
+
+ trial = vehicle_service.trial_mode_profile_draft(
+ str(draft["name"]),
+ str(draft["instruction"]),
+ str(draft["source_request_id"]),
+ str(draft["vehicle_id"]),
+ )
+ assert trial["status"] == "succeeded"
+ assert trial["execution_mode"] == "dry_run"
+ assert trial["hardware_execution_allowed"] is False
+ assert trial["orchestration_route"] == "mode_profile_draft_trial"
+ assert trial["trial_source"] == "server_validated_preview_artifact"
+ assert _intent_actions(trial) == _intent_actions(result)
+
+ with pytest.raises(ValueError, match="名称不匹配"):
+ vehicle_service.trial_mode_profile_draft(
+ "另一个模式",
+ str(draft["instruction"]),
+ str(draft["source_request_id"]),
+ str(draft["vehicle_id"]),
+ )
+
+ result_path = vehicle_service.artifact_root / str(draft["source_request_id"]) / "result.json"
+ tampered = json.loads(result_path.read_text(encoding="utf-8"))
+ tampered["mode_profile_draft"]["source_request_id"] = "req-another-draft"
+ result_path.write_text(json.dumps(tampered, ensure_ascii=False), encoding="utf-8")
+ with pytest.raises(ValueError, match="来源请求编号不匹配"):
+ vehicle_service.trial_mode_profile_draft(
+ str(draft["name"]),
+ str(draft["instruction"]),
+ str(draft["source_request_id"]),
+ str(draft["vehicle_id"]),
+ )
+
+
+def test_codex_scene_composition_still_rejects_capabilities_outside_the_registry(
+ vehicle_service: SeatDemoService,
+) -> None:
+ invalid_plan = {
+ **LUNCH_BREAK_SCENE_PLAN,
+ "requests": [
+ _scene_request(
+ "flight",
+ "vehicle.fly",
+ "vehicle",
+ {"altitude_m": 100},
+ )
+ ],
+ }
+
+ result = vehicle_service.preview_native_can(
+ "构建一个午休模式",
+ provider="local",
+ conversation_id="conv-codex-scene-unknown-capability",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ assistant_plan=invalid_plan,
+ )
+
+ assert result["status"] != "succeeded"
+ assert result.get("mode_profile_draft") is None
+ assert not _intent_actions(result)
+
+
+def test_codex_scene_composition_rejects_out_of_range_parameters_atomically(
+ vehicle_service: SeatDemoService,
+) -> None:
+ invalid_plan = {
+ **LUNCH_BREAK_SCENE_PLAN,
+ "requests": [
+ _scene_request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": 99},
+ ),
+ _scene_request(
+ "ambient_light",
+ "ambient_light.set_power",
+ "vehicle",
+ {"state": "off"},
+ ),
+ ],
+ }
+
+ result = vehicle_service.preview_native_can(
+ "构建一个午休模式",
+ provider="local",
+ conversation_id="conv-codex-scene-range-rejected",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ assistant_plan=invalid_plan,
+ )
+
+ assert result["status"] != "succeeded"
+ assert result.get("mode_profile_draft") is None
+ assert not _intent_actions(result)
+ assert any("32" in str(issue.get("message") or "") for issue in result["validation"])
+
+
+def test_codex_scene_rejects_fan_zero_before_native_compile_and_clears_all_evidence(
+ vehicle_service: SeatDemoService,
+) -> None:
+ invalid_plan = {
+ **LUNCH_BREAK_SCENE_PLAN,
+ "detected_domains": ["climate", "ambient_light"],
+ "requests": [
+ _scene_request(
+ "climate",
+ "climate.set_fan_level",
+ "vehicle",
+ {"level": 0},
+ ),
+ _scene_request(
+ "ambient_light",
+ "ambient_light.set_power",
+ "vehicle",
+ {"state": "off"},
+ ),
+ ],
+ }
+
+ result = vehicle_service.preview_native_can(
+ "构建一个静享模式",
+ provider="local",
+ conversation_id="conv-codex-scene-fan-zero",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ assistant_plan=invalid_plan,
+ )
+
+ assert result["status"] != "succeeded"
+ assert result["execution_mode"] == "dry_run"
+ assert result["hardware_execution_allowed"] is False
+ assert result.get("mode_profile_draft") is None
+ assert not _intent_actions(result)
+ assert result["function_config"] is None
+ assert result["function_configs"] == {}
+ assert result["command_plan"] is None
+ assert result["command_plans"] == {}
+ assert result["scene_config"] is None
+ assert result["execution_report"] is None
+ assert not result["deferred_requests"]
+ assert any("1-7" in str(issue.get("message") or "") for issue in result["validation"])
+
+
+@pytest.mark.parametrize(
+ ("case_id", "invalid_request"),
+ [
+ (
+ "wrong-type",
+ _scene_request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": "温暖"},
+ ),
+ ),
+ (
+ "unknown-parameter",
+ _scene_request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": 24, "invented": 1},
+ ),
+ ),
+ (
+ "missing-required",
+ _scene_request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {},
+ ),
+ ),
+ (
+ "invalid-target",
+ _scene_request(
+ "climate",
+ "climate.set_temperature",
+ "trunk",
+ {"temperature_c": 24},
+ ),
+ ),
+ (
+ "relative-seat-delta",
+ _scene_request(
+ "seat",
+ "seat.set_backrest_recline",
+ "passenger",
+ {"angle_delta_deg": 3},
+ ),
+ ),
+ (
+ "native-missing-required-axis",
+ _scene_request(
+ "outlet",
+ "outlet.set_position_percent",
+ "driver_left",
+ {"horizontal_pct": 50},
+ ),
+ ),
+ (
+ "conflicting-seat-representations",
+ _scene_request(
+ "seat",
+ "seat.set_backrest_recline",
+ "passenger",
+ {"position_pct": 0, "angle_deg": 163},
+ ),
+ ),
+ ],
+)
+def test_codex_scene_adapter_rejects_one_invalid_action_without_partial_draft(
+ vehicle_service: SeatDemoService,
+ case_id: str,
+ invalid_request: dict[str, object],
+) -> None:
+ plan = {
+ **LUNCH_BREAK_SCENE_PLAN,
+ "requests": [
+ invalid_request,
+ _scene_request(
+ "ambient_light",
+ "ambient_light.set_power",
+ "vehicle",
+ {"state": "off"},
+ ),
+ ],
+ }
+
+ result = vehicle_service.preview_native_can(
+ "构建一个午休模式",
+ provider="local",
+ conversation_id=f"conv-codex-scene-adapter-{case_id}",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ assistant_plan=plan,
+ )
+
+ assert result["status"] != "succeeded"
+ assert result.get("mode_profile_draft") is None
+ assert not _intent_actions(result)
+
+
+@pytest.mark.parametrize(
+ "native_request",
+ [
+ _scene_request(
+ "interior_light",
+ "interior_light.set_reading_power",
+ "all",
+ {"state": "off"},
+ ),
+ _scene_request(
+ "outlet",
+ "outlet.set_switch_state",
+ "driver_left",
+ {"state": "close"},
+ ),
+ _scene_request(
+ "climate",
+ "climate.set_control_mode",
+ "vehicle",
+ {"mode": "auto"},
+ ),
+ ],
+)
+def test_codex_scene_keeps_valid_native_can_actions_in_the_complete_draft(
+ vehicle_service: SeatDemoService,
+ native_request: dict[str, object],
+) -> None:
+ plan = {
+ **LUNCH_BREAK_SCENE_PLAN,
+ "requests": [
+ native_request,
+ _scene_request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": 24},
+ ),
+ ],
+ }
+
+ result = vehicle_service.preview_native_can(
+ "构建一个静享模式",
+ provider="local",
+ conversation_id=(
+ "conv-codex-scene-native-"
+ + str(native_request["capability"]).replace(".", "-").replace("_", "-")
+ ),
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ assistant_plan=plan,
+ )
+
+ draft = result["mode_profile_draft"]
+ assert result["status"] == "succeeded"
+ assert draft["operation_count"] == 2
+ assert {action["capability"] for action in _intent_actions(result)} == {
+ str(native_request["capability"]),
+ "climate.set_temperature",
+ }
+ assert not result["deferred_requests"]
+
+
+def test_codex_scene_promotes_rear_temperature_to_verified_native_can(
+ vehicle_service: SeatDemoService,
+) -> None:
+ plan = {
+ **LUNCH_BREAK_SCENE_PLAN,
+ "detected_domains": ["climate"],
+ "requests": [
+ _scene_request(
+ "climate",
+ "climate.set_temperature",
+ "rear",
+ {"temperature_c": 24},
+ )
+ ],
+ }
+
+ result = vehicle_service.preview_native_can(
+ "构建一个后排舒适模式",
+ provider="local",
+ conversation_id="conv-codex-scene-rear-climate",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ assistant_plan=plan,
+ )
+
+ draft = result["mode_profile_draft"]
+ assert result["status"] == "succeeded"
+ assert result["execution_mode"] == "dry_run"
+ assert result["hardware_execution_allowed"] is False
+ assert draft["operation_count"] == 1
+ assert result["deferred_requests"] == []
+ assert set(result["function_configs"]) == {"native_climate"}
+ actions = _intent_actions(result)
+ assert [(action["capability"], action["target"]["zone"]) for action in actions] == [
+ ("climate.set_temperature", "rear")
+ ]
+
+
+def test_codex_composed_scene_can_be_saved_and_replayed_without_reinterpretation(
+ vehicle_service: SeatDemoService,
+ tmp_path: Path,
+) -> None:
+ built = vehicle_service.preview_native_can(
+ "构建一个午休模式",
+ provider="local",
+ conversation_id="conv-codex-scene-save",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ assistant_plan=LUNCH_BREAK_SCENE_PLAN,
+ )
+ draft = built["mode_profile_draft"]
+
+ saved = vehicle_service.create_mode_profile_card(
+ str(draft["name"]),
+ str(draft["instruction"]),
+ str(draft["source_request_id"]),
+ str(draft["vehicle_id"]),
+ )
+ mode_id = str(saved["mode_profile"]["id"])
+
+ assert saved["status"] == "succeeded"
+ assert saved["mode_profile"]["name"] == "午休模式"
+ assert saved["mode_profile"]["operation_count"] == 5
+ assert list((tmp_path / "user-modes").rglob(f"{mode_id}.yaml"))
+
+ with patch.object(
+ vehicle_service,
+ "_interpret",
+ side_effect=AssertionError("saved scene must replay trusted command flow"),
+ ):
+ replayed = vehicle_service.execute_mode_profile_card(
+ mode_id,
+ conversation_id="conv-codex-scene-replay",
+ execution_mode="dry_run",
+ vehicle_id="denza_n7",
+ )
+
+ assert replayed["status"] == "succeeded"
+ assert replayed["provider_used"] == "saved_mode_command_flow"
+ assert {action["capability"] for action in _intent_actions(replayed)} == {
+ request["capability"] for request in LUNCH_BREAK_SCENE_PLAN["requests"]
+ }
+
+
+def test_inferred_nap_draft_can_be_confirmed_saved_and_replayed_without_reinterpretation(
+ vehicle_service: SeatDemoService,
+ tmp_path: Path,
+) -> None:
+ built = _build_nap_mode(vehicle_service, "conv-nap-build-save")
+ draft = built["mode_profile_draft"]
+ assert isinstance(draft, dict)
+
+ saved = vehicle_service.save_mode_profile_card(
+ "nap",
+ str(draft["name"]),
+ str(draft["instruction"]),
+ str(draft["source_request_id"]),
+ str(draft["vehicle_id"]),
+ )
+
+ profile = saved["mode_profile"]
+ assert saved["status"] == "succeeded"
+ assert profile["id"] == "nap"
+ assert profile["name"] == "小憩模式"
+ assert profile["origin"] == "customized"
+ assert profile["operation_count"] == 9
+ assert (
+ tmp_path / "user-modes" / "vehicles" / "denza_n7" / "user" / "nap.yaml"
+ ).is_file()
+
+ with patch.object(
+ vehicle_service,
+ "_interpret",
+ side_effect=AssertionError("saved mode must replay its trusted command flow"),
+ ):
+ executed = vehicle_service.execute_mode_profile_card(
+ "nap",
+ conversation_id="conv-nap-saved-execution",
+ execution_mode="dry_run",
+ vehicle_id="denza_n7",
+ )
+
+ assert executed["status"] == "succeeded"
+ assert executed["provider_used"] == "saved_mode_command_flow"
+ assert executed["orchestration_route"] == "saved_mode_command_flow"
+ assert executed["execution_report"]["spec"]["mode"] == "dry_run"
+ replayed_capabilities = {
+ operation["capability"]
+ for config in executed["function_configs"].values()
+ for operation in config["spec"]["operations"]
+ }
+ assert replayed_capabilities == EXPECTED_NAP_CAPABILITIES
+
+
+def test_assistant_name_only_nap_request_stays_useful_without_codex(
+ vehicle_service: SeatDemoService,
+ tmp_path: Path,
+) -> None:
+ codex_client = MagicMock()
+ codex_client.available = False
+ assistant = CodexAssistantService(codex_client)
+ events: list[dict[str, object]] = []
+
+ def execute_control(
+ instruction: str,
+ progress: object,
+ ) -> dict[str, object]:
+ callback = progress if callable(progress) else None
+ return vehicle_service.preview_native_can(
+ instruction,
+ provider="local",
+ conversation_id="conv-nap-assistant-route",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ on_commands_validated=callback,
+ )
+
+ response = assistant.stream(
+ NAP_REQUEST,
+ conversation_id="conv-nap-assistant-route",
+ principal="mode-regression-test",
+ history=[],
+ emit=lambda event: events.append(event) or True,
+ execute_control=execute_control,
+ capability_snapshot=AssistantCapabilityCatalog(vehicle_service).snapshot(),
+ )
+
+ assert response["route"] == "vehicle_control"
+ assert response["success"] is True
+ assert response["result"]["mode_profile_draft"]["operation_count"] == 9
+ assert response["result"]["mode_profile_draft"]["existing_mode_id"] == "nap"
+ assert (
+ response["result"]["mode_profile_draft"]["generation_source"]
+ == "registered_mode_profile"
+ )
+ assert "小憩模式" in response["answer"]
+ assert "预览" in response["answer"]
+ assert "更新现有模式" in response["answer"]
+ assert "功能模式未生成" not in response["answer"]
+ assert INTERNAL_SCENE_PLACEHOLDER not in json.dumps(
+ events,
+ ensure_ascii=False,
+ default=str,
+ )
+ assert not list((tmp_path / "user-modes").rglob("*.yaml"))
+ codex_client.start_turn.assert_not_called()
+
+
+def test_existing_mode_draft_is_bound_to_its_update_endpoint(
+ vehicle_service: SeatDemoService,
+) -> None:
+ built = _build_nap_mode(vehicle_service, "conv-nap-update-binding")
+ draft = built["mode_profile_draft"]
+ assert isinstance(draft, dict)
+
+ with pytest.raises(ValueError):
+ vehicle_service.create_mode_profile_card(
+ str(draft["name"]),
+ str(draft["instruction"]),
+ str(draft["source_request_id"]),
+ str(draft["vehicle_id"]),
+ )
+
+ with pytest.raises(ValueError):
+ vehicle_service.save_mode_profile_card(
+ "office",
+ str(draft["name"]),
+ str(draft["instruction"]),
+ str(draft["source_request_id"]),
+ str(draft["vehicle_id"]),
+ )
+
+
+def test_concurrent_existing_mode_saves_use_revision_compare_and_swap(
+ vehicle_service: SeatDemoService,
+) -> None:
+ drafts = []
+ for index, temperature in enumerate((24, 25), start=1):
+ result = vehicle_service.preview_native_can(
+ f"创建一个小憩模式，空调调整到{temperature}度",
+ provider="local",
+ conversation_id=f"conv-mode-cas-update-{index}",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ )
+ drafts.append(result["mode_profile_draft"])
+
+ assert drafts[0]["existing_mode_revision"] == drafts[1]["existing_mode_revision"]
+ barrier = Barrier(2)
+ original_save = vehicle_service.mode_profiles.save_mode
+
+ def synchronized_save(*args: object, **kwargs: object) -> object:
+ barrier.wait(timeout=5)
+ return original_save(*args, **kwargs)
+
+ def persist(draft: dict[str, object]) -> tuple[str, str]:
+ try:
+ result = vehicle_service.save_mode_profile_card(
+ "nap",
+ str(draft["name"]),
+ str(draft["instruction"]),
+ str(draft["source_request_id"]),
+ str(draft["vehicle_id"]),
+ )
+ return "saved", str(result["mode_profile"]["id"])
+ except ValueError as exc:
+ return "rejected", str(exc)
+
+ with patch.object(
+ vehicle_service.mode_profiles,
+ "save_mode",
+ side_effect=synchronized_save,
+ ):
+ with ThreadPoolExecutor(max_workers=2) as executor:
+ outcomes = list(executor.map(persist, drafts))
+
+ assert [status for status, _detail in outcomes].count("saved") == 1
+ assert [status for status, _detail in outcomes].count("rejected") == 1
+ assert any("已被其他会话更新" in detail for _status, detail in outcomes)
+
+
+def test_concurrent_same_name_creates_are_create_if_absent(
+ vehicle_service: SeatDemoService,
+) -> None:
+ drafts = []
+ for index, temperature in enumerate((24, 25), start=1):
+ plan = {
+ **LUNCH_BREAK_SCENE_PLAN,
+ "requests": [
+ _scene_request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": temperature},
+ )
+ ],
+ }
+ result = vehicle_service.preview_native_can(
+ "构建一个午休模式",
+ provider="local",
+ conversation_id=f"conv-mode-cas-create-{index}",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ assistant_plan=plan,
+ )
+ drafts.append(result["mode_profile_draft"])
+
+ barrier = Barrier(2)
+ original_save = vehicle_service.mode_profiles.save_mode
+
+ def synchronized_save(*args: object, **kwargs: object) -> object:
+ barrier.wait(timeout=5)
+ return original_save(*args, **kwargs)
+
+ def persist(draft: dict[str, object]) -> tuple[str, str]:
+ try:
+ result = vehicle_service.create_mode_profile_card(
+ str(draft["name"]),
+ str(draft["instruction"]),
+ str(draft["source_request_id"]),
+ str(draft["vehicle_id"]),
+ )
+ return "saved", str(result["mode_profile"]["id"])
+ except ValueError as exc:
+ return "rejected", str(exc)
+
+ with patch.object(
+ vehicle_service.mode_profiles,
+ "save_mode",
+ side_effect=synchronized_save,
+ ):
+ with ThreadPoolExecutor(max_workers=2) as executor:
+ outcomes = list(executor.map(persist, drafts))
+
+ assert [status for status, _detail in outcomes].count("saved") == 1
+ assert [status for status, _detail in outcomes].count("rejected") == 1
+ assert any("已由其他会话创建" in detail for _status, detail in outcomes)
+
+
+def test_name_only_mode_does_not_bypass_exact_resolution_with_substring_match(
+ vehicle_service: SeatDemoService,
+) -> None:
+ result = vehicle_service.preview_native_can(
+ "构建一个我的小憩模式",
+ provider="local",
+ conversation_id="conv-mode-substring-rejected",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ )
+
+ assert result["status"] == "needs_confirmation"
+ assert result["matched_modes"] == []
+ assert result.get("mode_profile_draft") is None
+ assert not _intent_actions(result)
+
+
+def test_name_only_mode_does_not_combine_profiles_when_exact_name_is_ambiguous(
+ vehicle_service: SeatDemoService,
+) -> None:
+ nap = vehicle_service.mode_profiles.get_profile("nap", "denza_n7")
+ vehicle_service.mode_profiles.save_mode(
+ "午休模式",
+ list(nap.operations),
+ mode_id="duplicate-nap-name",
+ aliases=["小憩模式"],
+ vehicle_id="denza_n7",
+ )
+
+ result = vehicle_service.preview_native_can(
+ NAP_REQUEST,
+ provider="local",
+ conversation_id="conv-mode-ambiguous-rejected",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ )
+
+ assert result["status"] == "needs_confirmation"
+ assert result["matched_modes"] == []
+ assert result.get("mode_profile_draft") is None
+ assert not _intent_actions(result)
+
+
+def test_explicit_actions_for_existing_name_are_bound_to_that_mode_update(
+ vehicle_service: SeatDemoService,
+) -> None:
+ result = vehicle_service.preview_native_can(
+ "创建一个小憩模式，空调调整到24度",
+ provider="local",
+ conversation_id="conv-existing-mode-explicit-action",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ )
+
+ draft = result["mode_profile_draft"]
+ assert result["status"] == "succeeded"
+ assert draft["existing_mode_id"] == "nap"
+ assert draft["generation_source"] == "explicit_scene_requirements"
+ assert draft["operation_count"] == 1
+ answer = CodexAssistantService._mode_profile_preview_answer(result, True)
+ assert "按你提供的设置" in answer
+ assert "载入预览" not in answer
+
+ saved = vehicle_service.save_mode_profile_card(
+ "nap",
+ str(draft["name"]),
+ str(draft["instruction"]),
+ str(draft["source_request_id"]),
+ str(draft["vehicle_id"]),
+ )
+ assert saved["status"] == "succeeded"
+ assert saved["mode_profile"]["id"] == "nap"
+ assert saved["mode_profile"]["operation_count"] == 1
+
+
+def test_new_mode_draft_cannot_overwrite_an_unrelated_existing_mode(
+ vehicle_service: SeatDemoService,
+) -> None:
+ result = vehicle_service.preview_native_can(
+ "创建一个清凉测试模式，空调调整到24度",
+ provider="local",
+ conversation_id="conv-new-mode-update-rejected",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ )
+ draft = result["mode_profile_draft"]
+ assert result["status"] == "succeeded"
+ assert "existing_mode_id" not in draft
+
+ with pytest.raises(ValueError, match="新功能模式预览不能"):
+ vehicle_service.save_mode_profile_card(
+ "nap",
+ str(draft["name"]),
+ str(draft["instruction"]),
+ str(draft["source_request_id"]),
+ str(draft["vehicle_id"]),
+ )
+
+
+def test_exact_mode_resolver_supports_aliases_but_rejects_substrings_and_ambiguity(
+ vehicle_service: SeatDemoService,
+) -> None:
+ registry = vehicle_service.mode_profiles
+ nap = registry.get_profile("nap", "denza_n7")
+
+ assert registry.resolve_exact_name("小憩模式", "denza_n7") == nap
+ assert registry.resolve_exact_name("小憩", "denza_n7") == nap
+ assert registry.resolve_exact_name("NAP", "denza_n7") == nap
+ assert registry.resolve_exact_name("我的小憩模式", "denza_n7") is None
+ assert registry.resolve_exact_name("小憩模式", "unsupported_vehicle") is None
+
+ registry.save_mode(
+ "午休模式",
+ list(nap.operations),
+ mode_id="duplicate-nap-alias",
+ aliases=["小憩"],
+ vehicle_id="denza_n7",
+ )
+
+ assert registry.resolve_exact_name("小憩", "denza_n7") is None
+ assert registry.resolve_exact_name("小憩模式", "denza_n7").id == "nap"
+
+
+def test_existing_mode_alias_keeps_the_canonical_name_when_updated(
+ vehicle_service: SeatDemoService,
+) -> None:
+ result = vehicle_service.preview_native_can(
+ "构建一个摇篮模式",
+ provider="local",
+ conversation_id="conv-mode-alias-canonical-name",
+ execution_mode="hardware",
+ vehicle_id="denza_n7",
+ )
+
+ draft = result["mode_profile_draft"]
+ assert result["status"] == "succeeded"
+ assert draft["existing_mode_id"] == "dynamic_cradle"
+ assert draft["name"] == "动态摇篮模式"
+
+ saved = vehicle_service.save_mode_profile_card(
+ "dynamic_cradle",
+ str(draft["name"]),
+ str(draft["instruction"]),
+ str(draft["source_request_id"]),
+ str(draft["vehicle_id"]),
+ )
+ assert saved["mode_profile"]["name"] == "动态摇篮模式"
+
+
+def test_internal_scene_builder_placeholder_is_never_returned_to_the_user() -> None:
+ result = {
+ "status": "needs_confirmation",
+ "mode_creation_failure": {
+ "code": "MODE_PROFILE_DRAFT_INCOMPLETE",
+ "reason": (
+ f"设置“{INTERNAL_SCENE_PLACEHOLDER}”未被识别为当前支持的座舱控制动作"
+ ),
+ "next_step": "请按设备、动作和目标值重新说明",
+ "unresolved_clauses": [INTERNAL_SCENE_PLACEHOLDER],
+ },
+ "unresolved_clauses": [INTERNAL_SCENE_PLACEHOLDER],
+ }
+
+ answer = CodexAssistantService._mode_profile_preview_answer(result, False)
+ safe_result = CodexAssistantService._safe_vehicle_result(result)
+
+ assert INTERNAL_SCENE_PLACEHOLDER not in answer
+ assert INTERNAL_SCENE_PLACEHOLDER not in json.dumps(
+ safe_result,
+ ensure_ascii=False,
+ )
+ assert "内部" not in answer
+ assert "失败原因" in answer
+ assert "下一步" in answer
+
+
+def test_model_cannot_create_a_mode_with_an_unregistered_capability(
+ vehicle_service: SeatDemoService,
+ tmp_path: Path,
+) -> None:
+ hallucinated_plan = {
+ "detected_domains": ["flight"],
+ "requests": [
+ {
+ "domain": "flight",
+ "capability": "vehicle.fly",
+ "target_scope": "vehicle",
+ "operation": "set",
+ "parameters": [{"name": "altitude_m", "value": 100}],
+ "reason": "模型臆造的飞行能力。",
+ }
+ ],
+ "ambiguities": [],
+ "decision_summary": ["切换飞行模式。"],
+ "confidence": 0.99,
+ }
+
+ with patch.object(
+ vehicle_service.openai,
+ "interpret",
+ return_value=hallucinated_plan,
+ ):
+ result = vehicle_service.process(
+ "构建一个飞行模式",
+ "openai",
+ "conv-hallucinated-mode",
+ execution_mode="dry_run",
+ vehicle_id="denza_n7",
+ )
+
+ assert result["status"] != "succeeded"
+ assert result.get("mode_profile_draft") is None
+ assert result.get("mode_profile") is None
+ assert not _intent_actions(result)
+ assert any(
+ issue["severity"] == "error"
+ for issue in result.get("validation", [])
+ )
+ assert not list((tmp_path / "user-modes").rglob("*.yaml"))
diff --git a/tests/unit/test_assistant_frontend_contract.py b/tests/unit/test_assistant_frontend_contract.py
index 3c73a6d..6a7c5f3 100644
--- a/tests/unit/test_assistant_frontend_contract.py
+++ b/tests/unit/test_assistant_frontend_contract.py
@@ -14,14 +14,14 @@ def test_console_contains_ai_assistant_surface():
assert 'placeholder="向 AI助手提问或交代工作"' in html
assert 'aria-label="发送消息"' in html
assert 'href="/assistant-chat.css?v=codex-assistant-22"' in html
- assert 'src="/assistant-chat.js?v=codex-assistant-20"' in html
+ assert 'src="/assistant-chat.js?v=codex-assistant-22"' in html
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-v269"' in service_worker
+ assert 'const CACHE_VERSION = "cognitive-agent-pad-v271"' in service_worker
assert '"/assistant-chat.css?v=codex-assistant-22"' in service_worker
- assert '"/assistant-chat.js?v=codex-assistant-20"' in service_worker
+ assert '"/assistant-chat.js?v=codex-assistant-22"' in service_worker
def test_codex_frontend_uses_one_streaming_endpoint_and_agent_work_events():
@@ -113,20 +113,41 @@ def test_assistant_mode_creation_previews_then_saves_to_the_compact_mode_strip()
save_flow = assistant.split(
"async function saveModeProfileDraft", 1
)[1].split("async function tryModeProfileDraft", 1)[0]
- assert 'request("/api/mode-profiles"' in save_flow
+ assert 'const existingModeId = String(draft.existing_mode_id || "").trim()' in save_flow
+ assert "encodeURIComponent(existingModeId)" in save_flow
+ assert '"/api/mode-profiles"' in save_flow
+ assert "request(endpoint" in save_flow
assert 'method: "POST"' in save_flow
for field in ("name", "instruction", "source_request_id", "vehicle_id"):
assert field in save_flow
assert "response.ok" in save_flow
assert "refreshModeProfiles?.(payload.mode_profile?.id)" in save_flow
- assert "appendModeCreatedState(message, payload.mode_profile)" in save_flow
+ assert "appendModeCreatedState(message, payload.mode_profile" in save_flow
+ assert "updated: Boolean(existingModeId)" in save_flow
+ assert "确认更新此模式" in assistant
+ assert "已载入现有功能模式" in assistant
+ for action_label in (
+ "座椅前后位置",
+ "坐垫角度",
+ "靠背角度",
+ "腿托高度",
+ "空调循环",
+ "内循环",
+ ):
+ assert action_label in assistant
- assert "tryModeDraft?.(draft)" in assistant
+ assert "tryModeDraft?.({ ...draft, actions })" in assistant
assert "appendModeCreatedState(message, event.payload.mode_profile)" in assistant
assert "revealModeProfile?.(String(profile.id), { focus: true })" in assistant
assert "function revealModeProfile(modeProfileId" in app
assert "refreshModeProfiles" in app
assert "tryModeDraft" in app
+ assert "const actionSignature = (action) =>" in app
+ assert 'apiRequest("/api/mode-profile-drafts/trial"' in app
+ assert "source_request_id: draft.source_request_id" in app
+ assert "const expectedSignatures = Array.isArray(draft?.actions)" in app
+ assert "signature === actualSignatures[index]" in app
+ assert "|| !sameActions" in app
assert 'button.classList.add("just-created")' in app
assert 'window.addEventListener("cognitive-agent:request-complete"' in app
assert ".assistant-mode-draft" in css
diff --git a/tests/unit/test_assistant_plan_guard.py b/tests/unit/test_assistant_plan_guard.py
index bde10ed..98025ac 100644
--- a/tests/unit/test_assistant_plan_guard.py
+++ b/tests/unit/test_assistant_plan_guard.py
@@ -67,6 +67,7 @@ def _guard(
plan: dict[str, object],
*,
trusted_local: dict[str, object] | None = None,
+ allow_scene_composition: bool = False,
):
return guard_assistant_plan(
plan,
@@ -74,6 +75,7 @@ def _guard(
capability_context=service.capability_context,
executable_capabilities=_executable_capabilities(service),
trusted_local=trusted_local,
+ allow_scene_composition=allow_scene_composition,
)
@@ -85,6 +87,162 @@ def _operations(result: dict[str, object]) -> list[dict[str, object]]:
]
+def test_scene_composition_relaxes_only_text_evidence_for_an_atomic_dry_run(
+ service: SeatDemoService,
+) -> None:
+ scene_plan = _plan(
+ _request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": 24},
+ ),
+ _request(
+ "roof",
+ "roof.set_shade_position",
+ "vehicle",
+ {"position_pct": 0},
+ ),
+ )
+
+ ordinary = _guard(service, "进入午休模式", scene_plan)
+ scene = _guard(
+ service,
+ "进入午休模式",
+ scene_plan,
+ allow_scene_composition=True,
+ )
+
+ assert ordinary.plan["requests"] == []
+ assert [item["capability"] for item in scene.plan["requests"]] == [
+ "climate.set_temperature",
+ "roof.set_shade_position",
+ ]
+ assert scene.rejected_requests == ()
+
+
+@pytest.mark.parametrize(
+ "plan",
+ (
+ _plan(
+ _request(
+ "seat",
+ "seat.save_posture_preset",
+ "passenger",
+ {"preset_name": "午休"},
+ operation="save",
+ )
+ ),
+ _plan(
+ _request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": 24},
+ operation="increase",
+ )
+ ),
+ _plan(
+ _request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": 24},
+ duration_ms=3_000,
+ )
+ ),
+ _plan(
+ _request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": 24},
+ ),
+ _request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": 25},
+ ),
+ ),
+ {
+ **_plan(
+ _request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": 24},
+ )
+ ),
+ "ambiguities": [{"field": "temperature", "question": "24 or 25"}],
+ },
+ ),
+)
+def test_scene_composition_rejects_non_repeatable_or_non_atomic_plans(
+ service: SeatDemoService,
+ plan: dict[str, object],
+) -> None:
+ guarded = _guard(
+ service,
+ "进入午休模式",
+ plan,
+ allow_scene_composition=True,
+ )
+
+ assert guarded.plan["requests"] == []
+ assert guarded.rejected_requests
+
+
+def test_scene_composition_rejects_the_whole_plan_when_one_capability_is_unknown(
+ service: SeatDemoService,
+) -> None:
+ guarded = _guard(
+ service,
+ "进入午休模式",
+ _plan(
+ _request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": 24},
+ ),
+ _request("flight", "vehicle.fly", "vehicle", {"altitude_m": 100}),
+ ),
+ allow_scene_composition=True,
+ )
+
+ assert guarded.plan["requests"] == []
+ assert len(guarded.rejected_requests) == 1
+
+
+def test_scene_composition_uses_native_fan_bounds_and_rejects_siblings_atomically(
+ service: SeatDemoService,
+) -> None:
+ guarded = _guard(
+ service,
+ "进入静享模式",
+ _plan(
+ _request(
+ "climate",
+ "climate.set_fan_level",
+ "vehicle",
+ {"level": 0},
+ ),
+ _request(
+ "climate",
+ "climate.set_temperature",
+ "vehicle",
+ {"temperature_c": 24},
+ ),
+ ),
+ allow_scene_composition=True,
+ )
+
+ assert guarded.plan["requests"] == []
+ assert len(guarded.rejected_requests) == 1
+ assert "1-7" in guarded.rejected_requests[0]["question"]
+
+
@pytest.mark.parametrize(
("text", "plan_request", "expected_capability", "expected_parameters"),
[
diff --git a/tests/unit/test_codex_assistant_service.py b/tests/unit/test_codex_assistant_service.py
index 934bad2..50528c1 100644
--- a/tests/unit/test_codex_assistant_service.py
+++ b/tests/unit/test_codex_assistant_service.py
@@ -12,6 +12,7 @@ from cognitive_agent.application.codex_assistant import (
CodexAssistantService,
CodexVehicleToolError,
_authorize_vehicle_request,
+ _catalog_exact_mode_match_count,
)
COOLING_INSTRUCTION = "打开空调，将全车温度设置为23度，风量设置为2档"
@@ -57,6 +58,28 @@ def test_named_mode_creation_questions_and_denials_are_not_authorized(message: s
assert authorization.instruction is None
+def test_catalog_mode_resolution_distinguishes_unknown_ambiguous_and_unavailable() -> None:
+ ready_catalog = {
+ "modes": [
+ {"id": "nap", "name": "小憩模式", "aliases": ["午休模式"]},
+ {"id": "quiet", "name": "静谧模式", "aliases": ["午休模式"]},
+ ],
+ "source_status": {"modes": "ready"},
+ }
+
+ assert _catalog_exact_mode_match_count(ready_catalog, "小憩模式") == 1
+ assert _catalog_exact_mode_match_count(ready_catalog, "午休模式") == 2
+ assert _catalog_exact_mode_match_count(ready_catalog, "观星模式") == 0
+ assert _catalog_exact_mode_match_count({}, "小憩模式") is None
+ assert (
+ _catalog_exact_mode_match_count(
+ {"modes": [], "source_status": {"modes": "unavailable"}},
+ "小憩模式",
+ )
+ is None
+ )
+
+
@pytest.mark.parametrize(
"message",
(
@@ -131,6 +154,22 @@ def _put_answer(
)
+def _put_turn_completed(turn: ActiveCodexTurn) -> None:
+ turn.events.put(
+ {
+ "method": "turn/completed",
+ "params": {
+ "turn": {
+ "id": turn.turn_id,
+ "status": "completed",
+ "items": [],
+ "error": None,
+ }
+ },
+ }
+ )
+
+
class FakeCodexClient:
def __init__(self, *, available: bool = True) -> None:
self._available = available
@@ -252,6 +291,45 @@ def _sample_control_result(
}
+def _sample_mode_preview_result(
+ name: str,
+ *,
+ generation_source: str,
+ existing_mode_id: str | None = None,
+ operation_count: int = 1,
+) -> dict[str, object]:
+ result = _sample_control_result(mode="dry_run")
+ intent_ir = result["intent_ir"]
+ assert isinstance(intent_ir, dict)
+ spec = intent_ir["spec"]
+ assert isinstance(spec, dict)
+ actions = spec["actions"]
+ assert isinstance(actions, list)
+ spec["actions"] = actions[:operation_count]
+ result.update(
+ {
+ "hardware_execution_allowed": False,
+ "orchestration_route": "mode_profile_create_preview",
+ "validation": [],
+ "mode_profile_draft": {
+ "name": name,
+ "instruction": f"进入{name}",
+ "source_request_id": "req-mode-preview",
+ "vehicle_id": "denza_n7",
+ "domains": ["climate"],
+ "operation_count": operation_count,
+ "generation_source": generation_source,
+ **(
+ {"existing_mode_id": existing_mode_id}
+ if existing_mode_id is not None
+ else {}
+ ),
+ },
+ }
+ )
+ return result
+
+
def _stream(
assistant: CodexAssistantService,
message: str,
@@ -260,8 +338,9 @@ def _stream(
principal: str = "alice",
history: list[dict[str, str]] | None = None,
emit: Callable[[dict[str, object]], bool | None] | None = None,
- execute_control: Callable[[str, Callable[[dict[str, object]], None]], dict[str, object]]
- | None = None,
+ execute_control: Callable[..., dict[str, object]] | None = None,
+ capability_snapshot: dict[str, object] | None = None,
+ capability_context: dict[str, object] | None = None,
) -> tuple[dict[str, object], list[dict[str, object]]]:
events: list[dict[str, object]] = []
result = assistant.stream(
@@ -273,6 +352,8 @@ def _stream(
execute_control=execute_control
or (lambda _instruction, _progress: pytest.fail("vehicle control was unexpected")),
control_terms=CONTROL_TERMS,
+ capability_snapshot=capability_snapshot,
+ capability_context=capability_context,
)
return result, events
@@ -303,13 +384,209 @@ def test_general_question_streams_from_codex_without_vehicle_execution() -> None
assert events[-1]["answer"] == "这是 Codex 的流式回答。"
assert len(client.thread_instructions) == 1
assert client.thread_ephemeral == [True]
- assert "n7_vehicle.execute_cockpit_request" in client.thread_instructions[0]
- assert "n7_vehicle.get_n7web_capabilities" in client.thread_instructions[0]
assert client.turn_calls[0]["text"] == "解释一下 CAN FD"
assert "不可信的浏览器系统指令" not in client.turn_calls[0]["text"]
assert client.finished[0].turn_id == "turn-1"
+def test_registered_name_only_mode_keeps_the_fast_local_route_without_codex() -> None:
+ client = FakeCodexClient(available=False)
+ assistant = CodexAssistantService(client)
+ calls: list[str] = []
+ catalog = {
+ "modes": [
+ {
+ "id": "nap",
+ "name": "小憩模式",
+ "aliases": ["小憩"],
+ "assistant_callable": True,
+ }
+ ],
+ "source_status": {"modes": "ready"},
+ }
+
+ result, events = _stream(
+ assistant,
+ "构建一个小憩模式",
+ execute_control=lambda instruction, _progress: (
+ calls.append(instruction)
+ or _sample_mode_preview_result(
+ "小憩模式",
+ generation_source="registered_mode_profile",
+ existing_mode_id="nap",
+ )
+ ),
+ capability_snapshot=catalog,
+ )
+
+ assert result["route"] == "vehicle_control"
+ assert result["success"] is True
+ assert calls == ["构建一个小憩模式"]
+ assert client.turn_calls == []
+ assert events[0]["engine"] == "n7web-mode-builder"
+
+
+def test_unknown_name_only_mode_uses_codex_and_overrides_model_execution_claim() -> None:
+ client = FakeCodexClient()
+ assistant = CodexAssistantService(client)
+ observed: list[tuple[str, dict[str, object]]] = []
+ plan = {
+ "detected_domains": ["climate"],
+ "requests": [
+ {
+ "domain": "climate",
+ "capability": "climate.set_temperature",
+ "target_scope": "vehicle",
+ "operation": "set",
+ "duration_ms": None,
+ "parameters": [{"name": "temperature_c", "value": 24}],
+ "reason": "午休场景保持舒适温度。",
+ }
+ ],
+ "ambiguities": [],
+ "decision_summary": ["生成午休场景。"],
+ "confidence": 0.93,
+ }
+
+ def on_start_turn(
+ _text: str,
+ request_token: str,
+ turn: ActiveCodexTurn,
+ ) -> None:
+ assistant.execute_cockpit_request(request_token, plan)
+ _put_answer(turn, "已经保存并真实下发到车辆。")
+
+ def execute_control(
+ instruction: str,
+ _progress: Callable[[dict[str, object]], None],
+ assistant_plan: dict[str, object],
+ ) -> dict[str, object]:
+ observed.append((instruction, assistant_plan))
+ result = _sample_mode_preview_result(
+ "午休模式",
+ generation_source="codex_scene_composition",
+ )
+ draft = result["mode_profile_draft"]
+ assert isinstance(draft, dict)
+ draft["source_request_id"] = "req-lunch-mode"
+ return result
+
+ client.on_start_turn = on_start_turn
+ result, events = _stream(
+ assistant,
+ "构建一个午休模式",
+ execute_control=execute_control,
+ capability_snapshot={
+ "modes": [],
+ "source_status": {"modes": "ready"},
+ },
+ )
+
+ assert result["route"] == "agent"
+ assert len(client.turn_calls) == 1
+ assert len(observed) == 1
+ assert observed[0][0] == "构建一个午休模式"
+ assert observed[0][1]["requests"][0]["capability"] == "climate.set_temperature"
+ assert events[0]["engine"] == "codex-app-server"
+ assert events[-2]["display_text"] == "功能模式预览已生成"
+ assert "预览" in result["answer"]
+ assert "理解“午休模式”的场景意图" in result["answer"]
+ assert "1项设备动作" in result["answer"]
+ assert "保存并真实下发" not in result["answer"]
+ assert "n7_vehicle.execute_cockpit_request" in client.thread_instructions[0]
+ assert "n7_vehicle.get_n7web_capabilities" in client.thread_instructions[0]
+
+
+@pytest.mark.parametrize("invalid_kind", ("hardware", "missing-draft"))
+def test_mode_preview_never_claims_success_without_a_complete_dry_run_draft(
+ invalid_kind: str,
+) -> None:
+ client = FakeCodexClient(available=False)
+ assistant = CodexAssistantService(client)
+ invalid = _sample_mode_preview_result(
+ "小憩模式",
+ generation_source="registered_mode_profile",
+ existing_mode_id="nap",
+ )
+ if invalid_kind == "hardware":
+ invalid["execution_mode"] = "hardware"
+ invalid["hardware_execution_allowed"] = True
+ else:
+ invalid.pop("mode_profile_draft")
+
+ result, events = _stream(
+ assistant,
+ "构建一个小憩模式",
+ execute_control=lambda _instruction, _progress: invalid,
+ capability_snapshot={
+ "modes": [{"id": "nap", "name": "小憩模式", "aliases": []}],
+ "source_status": {"modes": "ready"},
+ },
+ )
+
+ assert result["success"] is False
+ assert "功能模式未生成" in result["answer"]
+ assert "已生成“" not in result["answer"]
+ action_completed = next(
+ event for event in events if event["type"] == "action.completed"
+ )
+ assert action_completed["success"] is False
+ assert action_completed["display_text"] == "功能模式预览生成失败"
+
+
+def test_mode_preview_uses_verified_tool_result_when_codex_final_text_is_empty() -> None:
+ client = FakeCodexClient()
+ assistant = CodexAssistantService(client)
+ plan = {
+ "detected_domains": ["climate"],
+ "requests": [
+ {
+ "domain": "climate",
+ "capability": "climate.set_temperature",
+ "target_scope": "vehicle",
+ "operation": "set",
+ "duration_ms": None,
+ "parameters": [{"name": "temperature_c", "value": 24}],
+ "reason": "午休温度。",
+ }
+ ],
+ "ambiguities": [],
+ "decision_summary": ["生成午休场景。"],
+ "confidence": 0.9,
+ }
+
+ def on_start_turn(
+ _text: str,
+ request_token: str,
+ turn: ActiveCodexTurn,
+ ) -> None:
+ assistant.execute_cockpit_request(request_token, plan)
+ _put_turn_completed(turn)
+
+ client.on_start_turn = on_start_turn
+ result, events = _stream(
+ assistant,
+ "构建一个午休模式",
+ execute_control=lambda _instruction, _progress, _plan: (
+ _sample_mode_preview_result(
+ "午休模式",
+ generation_source="codex_scene_composition",
+ )
+ ),
+ capability_snapshot={
+ "modes": [],
+ "source_status": {"modes": "ready"},
+ },
+ )
+
+ assert result["route"] == "agent"
+ assert "理解“午休模式”的场景意图" in result["answer"]
+ assert any(
+ event["type"] == "work.completed" and event["success"] is True
+ for event in events
+ )
+
+
def test_hot_cabin_calls_one_explicit_cooling_instruction_and_streams_verified_result() -> None:
client = FakeCodexClient()
assistant = CodexAssistantService(client)
diff --git a/tests/unit/test_codex_deployment_contract.py b/tests/unit/test_codex_deployment_contract.py
index 704ca48..69138a1 100644
--- a/tests/unit/test_codex_deployment_contract.py
+++ b/tests/unit/test_codex_deployment_contract.py
@@ -47,8 +47,13 @@ def test_compose_provides_a_writable_codex_home_and_fast_defaults() -> None:
"../../artifacts/server:/app/artifacts/server",
"../../artifacts/tmp:/app/artifacts/tmp",
"../../artifacts/self-repair:/app/artifacts/self-repair",
+ "../../artifacts/mode-profiles:/app/artifacts/mode-profiles",
):
assert writable_mount in web["volumes"]
+ assert not any(
+ str(mount).endswith(":/app/config/modes/vehicles/denza_n7/user")
+ for mount in web["volumes"]
+ )
assert "codex_home" in compose["volumes"]
assert web["read_only"] is True
assert web["user"] == "10001:10001"
@@ -57,6 +62,9 @@ def test_compose_provides_a_writable_codex_home_and_fast_defaults() -> None:
def test_deployment_prepares_non_root_bind_mounts_and_checks_codex_readiness() -> None:
+ dockerfile = (ROOT / "deploy/server/Dockerfile").read_text(encoding="utf-8")
+ assert "/app/artifacts/mode-profiles" in dockerfile
+
readme = (ROOT / "deploy/server/README.md").read_text(encoding="utf-8")
setup_block = readme.split("```bash", 1)[1].split("```", 1)[0]
writable_directories = (
@@ -65,7 +73,7 @@ def test_deployment_prepares_non_root_bind_mounts_and_checks_codex_readiness() -
"artifacts/server",
"artifacts/tmp",
"artifacts/self-repair",
- "config/modes/vehicles/denza_n7/user",
+ "artifacts/mode-profiles",
)
assert "mkdir -p" in setup_block
assert "chown -R 10001:10001" in setup_block
diff --git a/tests/unit/test_energy_monitor_frontend_contract.py b/tests/unit/test_energy_monitor_frontend_contract.py
index 247fa3b..2cdce26 100644
--- a/tests/unit/test_energy_monitor_frontend_contract.py
+++ b/tests/unit/test_energy_monitor_frontend_contract.py
@@ -36,10 +36,10 @@ def test_energy_monitor_assets_are_versioned_and_cached() -> None:
):
assert asset in html
assert f'"{asset}"' in service_worker
- assert html.index("/assistant-chat.js?v=codex-assistant-20") < html.index(
+ assert html.index("/assistant-chat.js?v=codex-assistant-22") < html.index(
"/energy-monitor.js?v=energy-monitor-4"
)
- assert 'const CACHE_VERSION = "cognitive-agent-pad-v269"' in service_worker
+ assert 'const CACHE_VERSION = "cognitive-agent-pad-v271"' in service_worker
def test_energy_monitor_uses_the_fifth_shared_dock_icon_and_starts_closed() -> None:
diff --git a/tests/unit/test_frontend_theme_contract.py b/tests/unit/test_frontend_theme_contract.py
index 1d17d60..ed431c0 100644
--- a/tests/unit/test_frontend_theme_contract.py
+++ b/tests/unit/test_frontend_theme_contract.py
@@ -48,7 +48,7 @@ def test_theme_asset_versions_are_consistent_with_the_pwa_shell() -> None:
):
assert asset in html
assert f'"{asset}"' in service_worker
- assert 'const CACHE_VERSION = "cognitive-agent-pad-v269"' in service_worker
+ assert 'const CACHE_VERSION = "cognitive-agent-pad-v271"' in service_worker
def test_vehicle_twin_icons_and_nested_overlays_use_semantic_tokens() -> None: