From 33c92d1352dd6ea1efb12d9ed1d5e9c8d00da656 Mon Sep 17 00:00:00 2001
From: minghanchen <17188325+minghanchen@user.noreply.gitee.com>
Date: Mon, 24 Aug 2026 14:15:53 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=90=8E=E7=AB=AF=E5=92=8C?=
=?UTF-8?q?=E5=89=8D=E7=AB=AF?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
apps/web-console/app.js | 195 +++++--
apps/web-console/console-v2.css | 18 +
apps/web-console/console-v2.js | 37 +-
apps/web-console/index.html | 6 +-
apps/web-console/service-worker.js | 8 +-
contracts/schemas/function-config.schema.json | 4 +
docs/CAN1控制指令白名单.md | 74 +++
domains/climate/config/capabilities.yaml | 2 +-
domains/seat/src/seat_domain/generator.py | 12 +
domains/seat/src/seat_domain/intent.py | 19 +
.../tests/integration/test_seat_service.py | 2 +-
domains/seat/tests/unit/test_intent.py | 15 +
.../application/conversation_state.py | 11 +-
.../application/domain_runtime.py | 3 +
.../application/native_can_intent.py | 414 ++++++++++----
.../application/native_can_relative.py | 1 +
.../application/native_can_runtime.py | 74 ++-
.../application/seat_service.py | 535 +++++++++++++++++-
.../infrastructure/can/mqtt_transport.py | 19 +-
src/cognitive_agent/server.py | 241 +++++++-
tests/e2e/test_seat_web_api.py | 144 ++++-
tests/unit/test_can1_semantic_whitelist.py | 351 ++++++++++++
tests/unit/test_ch1_default_targets.py | 38 ++
tests/unit/test_driver_seat_native_can.py | 10 +-
tests/unit/test_mqtt_can_transport.py | 133 ++++-
.../test_requested_frontend_enhancements.py | 4 +
tests/unit/test_two_stage_can_parsing.py | 192 +++++++
27 files changed, 2328 insertions(+), 234 deletions(-)
create mode 100644 docs/CAN1控制指令白名单.md
create mode 100644 tests/unit/test_can1_semantic_whitelist.py
create mode 100644 tests/unit/test_two_stage_can_parsing.py
diff --git a/apps/web-console/app.js b/apps/web-console/app.js
index 72e5c71..4a4f88c 100755
--- a/apps/web-console/app.js
+++ b/apps/web-console/app.js
@@ -7023,14 +7023,14 @@ function renderExecution(result) {
const operationResults = report.spec.results.filter((item) => item.ref !== "seat-controller-firmware");
const succeededOperations = operationResults.filter((item) => item.status === "succeeded");
const failedOperations = operationResults.filter((item) => item.status === "failed");
- const mqttPublishedOperations = operationResults.filter(
- (item) => item.telemetry?.delivery_status === "published",
+ const mqttSubmittedOperations = operationResults.filter(
+ (item) => ["queued", "published"].includes(item.telemetry?.delivery_status),
);
const physicallyConfirmedOperations = operationResults.filter(
(item) => item.telemetry?.physical_effect_confirmed === true,
);
const mqttPublishedUnconfirmed = result.control_transport === "wireless_mqtt"
- && mqttPublishedOperations.length > physicallyConfirmedOperations.length;
+ && mqttSubmittedOperations.length > physicallyConfirmedOperations.length;
const succeededDomains = [...new Set(succeededOperations.map((item) => item.domain))];
const failedDomains = [...new Set(failedOperations.map((item) => item.domain))];
let executionDetail = modeSave
@@ -7041,7 +7041,7 @@ function renderExecution(result) {
? `${executedDomains.join(" / ")} 已完成隔离 HIL 仿真，未访问物理 CAN`
: physical
? result.control_transport === "wireless_mqtt"
- ? `${executedDomains.join(" / ")} 已通过 MQTT 发布 ${mqttPublishedOperations.length || succeededOperations.length} 个白名单控制步骤${mqttPublishedUnconfirmed ? "；车辆效果尚未确认，以状态话题回读为准" : `，并确认 ${physicallyConfirmedOperations.length || succeededOperations.length} 个车辆状态`}`
+ ? `${executedDomains.join(" / ")} 已向 MQTT 提交 ${mqttSubmittedOperations.length || succeededOperations.length} 个白名单控制步骤${mqttPublishedUnconfirmed ? "；车辆效果尚未确认，以状态话题回读为准" : `，并确认 ${physicallyConfirmedOperations.length || succeededOperations.length} 个车辆状态`}`
: selectedVehicleTransport()?.target_policy === "ch1_body_can_driver_only"
? `${executedDomains.join(" / ")} 已通过 CH1 下发，并以车端状态字段确认 ${operationResults.filter((item) => item.status === "succeeded").length} 个步骤`
: `${executedDomains.join(" / ")} 已下发真实 CAN，并以 IAR 固件实际反馈确认 ${operationResults.filter((item) => item.status === "succeeded").length} 个步骤`
@@ -7081,7 +7081,7 @@ function renderExecution(result) {
: presetSave ? "姿态预设已保存"
: simulatedHil ? "模拟台架执行成功"
: physical
- ? mqttPublishedUnconfirmed ? "MQTT 指令已发布" : "真实台架执行成功"
+ ? mqttPublishedUnconfirmed ? "MQTT 指令已提交" : "真实台架执行成功"
: "虚拟执行成功"
: partiallySucceeded ? (physical ? "真实台架部分执行成功" : "部分执行成功")
: simulatedHil ? "模拟故障已触发" : physical ? "真实台架执行失败" : "虚拟执行失败";
@@ -7091,7 +7091,7 @@ function renderExecution(result) {
summary.innerHTML = `${executionTitle}${executionDetail}
`;
$("#resultNavMeta").textContent = reportSucceeded
? mqttPublishedUnconfirmed
- ? `${mqttPublishedOperations.length} 项已发布 · MQTT`
+ ? `${mqttSubmittedOperations.length} 项已提交 · MQTT`
: `${report.spec.results.length} 项完成 · ${simulatedHil ? "SIM-HIL" : physical ? "HIL" : "DRY-RUN"}`
: partiallySucceeded ? `${succeededOperations.length} 项成功 / ${failedOperations.length} 项失败 · 查看明细`
: "执行失败 · 查看回读";
@@ -7251,7 +7251,7 @@ function renderResult(result) {
const feedbackSource = result.execution_report?.metadata?.labels?.seat_feedback;
const mqttPublishedUnconfirmed = result.control_transport === "wireless_mqtt"
&& (result.execution_report?.spec?.results || []).some(
- (item) => item.telemetry?.delivery_status === "published"
+ (item) => ["queued", "published"].includes(item.telemetry?.delivery_status)
&& item.telemetry?.physical_effect_confirmed !== true,
);
if (feedbackSource === "simulated_hardware") {
@@ -7291,7 +7291,7 @@ function renderResult(result) {
: result.scene_config
? `${domains.length} 域场景闭环完成`
: mqttPublishedUnconfirmed
- ? `${domainLabels[domains[0]] || "车舱"}控制指令已发布`
+ ? `${domainLabels[domains[0]] || "车舱"}控制指令已提交`
: `${domainLabels[domains[0]] || "车舱"}单域闭环完成`
: "请求未执行";
const conversationLabel = result.conversation_id?.slice(-8) || "no-context";
@@ -7324,6 +7324,67 @@ function renderResult(result) {
window.dispatchEvent(new CustomEvent("cognitive-agent:result", { detail: result }));
}
+async function postControlPhase(path, body, executionMode, onEvent = null) {
+ const streaming = typeof onEvent === "function";
+ const response = await apiRequest(path, {
+ method: "POST",
+ timeoutMs: requestTimeoutForExecutionMode(executionMode),
+ headers: {
+ "Content-Type": "application/json",
+ ...(streaming ? { Accept: "application/x-ndjson" } : {}),
+ },
+ body: JSON.stringify(body),
+ });
+ if (response.status === 401) {
+ window.location.replace("/login");
+ return null;
+ }
+ if (streaming && response.ok && response.body?.getReader) {
+ const reader = response.body.getReader();
+ const decoder = new TextDecoder();
+ let buffered = "";
+ let resultPayload = null;
+ const handleLine = (line) => {
+ if (!line.trim()) return;
+ const phaseEvent = JSON.parse(line);
+ if (phaseEvent.type === "error") {
+ const requestError = new Error(
+ phaseEvent.message || phaseEvent.error || "流式控制请求失败",
+ );
+ requestError.code = phaseEvent.error;
+ requestError.retryAfter = Number(phaseEvent.retry_after_seconds || 0);
+ throw requestError;
+ }
+ if (phaseEvent.type === "result") {
+ resultPayload = phaseEvent.payload;
+ } else {
+ onEvent(phaseEvent);
+ }
+ };
+ while (true) {
+ const { value, done } = await reader.read();
+ buffered += decoder.decode(value || new Uint8Array(), { stream: !done });
+ const lines = buffered.split("\n");
+ buffered = lines.pop() || "";
+ lines.forEach(handleLine);
+ if (done) break;
+ }
+ if (buffered.trim()) handleLine(buffered);
+ if (!resultPayload) throw new Error("流式控制请求未返回完整结果");
+ return resultPayload;
+ }
+ const payload = await response.json();
+ if (!response.ok) {
+ const requestError = new Error(payload.message || payload.error || `HTTP ${response.status}`);
+ requestError.code = payload.error;
+ requestError.retryAfter = Number(
+ payload.retry_after_seconds || response.headers.get("Retry-After") || 0,
+ );
+ throw requestError;
+ }
+ return payload;
+}
+
async function submitRequest(event) {
event.preventDefault();
const utterance = $("#utterance").value.trim();
@@ -7360,39 +7421,89 @@ async function submitRequest(event) {
setRunStatus("processing", "PROCESSING");
$("#resultTitle").textContent = "正在执行车舱编排";
try {
- const response = await apiRequest("/api/requests", {
- method: "POST",
- timeoutMs: requestTimeoutForExecutionMode(state.executionMode),
- headers: { "Content-Type": "application/json" },
- body: JSON.stringify({
- utterance,
- client_type: "web_console",
- provider: state.provider,
- conversation_id: state.conversationId,
- vehicle_id: state.vehicleId || state.runtime?.default_vehicle_id,
- execution_mode: state.executionMode,
- control_transport: state.controlTransport,
- ...(state.provider === "openai" ? { reasoning_effort: state.reasoningEffort } : {}),
- ...(state.executionMode === "hardware" && !driverOnlyCh1
- ? { hardware_client_id: state.hardwareClientId }
- : {}),
- ...(state.executionMode === "simulated_hardware"
- ? { simulation_fault: $("#simulationFault").value }
- : {}),
- }),
- });
- const payload = await response.json();
- if (response.status === 401) {
- window.location.replace("/login");
- return;
- }
- if (!response.ok) {
- const requestError = new Error(payload.message || payload.error || `HTTP ${response.status}`);
- requestError.code = payload.error;
- requestError.retryAfter = Number(
- payload.retry_after_seconds || response.headers.get("Retry-After") || 0,
+ const requestBody = {
+ utterance,
+ client_type: "web_console",
+ provider: state.provider,
+ conversation_id: state.conversationId,
+ vehicle_id: state.vehicleId || state.runtime?.default_vehicle_id,
+ execution_mode: state.executionMode,
+ control_transport: state.controlTransport,
+ ...(state.provider === "openai" ? { reasoning_effort: state.reasoningEffort } : {}),
+ ...(state.executionMode === "hardware" && !driverOnlyCh1
+ ? { hardware_client_id: state.hardwareClientId }
+ : {}),
+ ...(state.executionMode === "simulated_hardware"
+ ? { simulation_fault: $("#simulationFault").value }
+ : {}),
+ };
+ let payload;
+ let supplementWarning = "";
+ const onControlPhaseEvent = (phaseEvent) => {
+ if (phaseEvent.type !== "validated_commands" || !phaseEvent.payload) return;
+ window.dispatchEvent(new CustomEvent(
+ "cognitive-agent:validated-can-commands",
+ { detail: phaseEvent.payload },
+ ));
+ };
+ if (state.provider === "openai") {
+ const preview = await postControlPhase(
+ "/api/requests/preview",
+ requestBody,
+ state.executionMode,
+ onControlPhaseEvent,
);
- throw requestError;
+ if (!preview) return;
+ renderResult(preview);
+ completedResult = preview;
+ if (preview.supplement_required) {
+ setRunStatus("processing", "CLOUD SUPPLEMENT");
+ $("#resultTitle").textContent = "本地指令已处理，正在补充云端解析";
+ try {
+ payload = await postControlPhase(
+ `/api/requests/${preview.request_id}/supplement`,
+ {
+ client_type: "web_console",
+ provider: state.provider,
+ reasoning_effort: state.reasoningEffort,
+ ...(state.executionMode === "hardware" && !driverOnlyCh1
+ ? { hardware_client_id: state.hardwareClientId }
+ : {}),
+ },
+ state.executionMode,
+ onControlPhaseEvent,
+ );
+ if (!payload) return;
+ } catch (error) {
+ const localCommandCount = Object.values(preview.function_configs || {})
+ .reduce((total, config) => total + (config?.spec?.operations?.length || 0), 0);
+ if (!localCommandCount) throw error;
+ payload = {
+ ...preview,
+ status: "partially_succeeded",
+ supplement_required: false,
+ cloud_supplement_error: {
+ code: error.code || "cloud_supplement_failed",
+ message: error.message,
+ },
+ decision_summary: [
+ ...(preview.decision_summary || []),
+ `云端补充失败：${error.message}；本地规则命令已保留且不会撤销。`,
+ ],
+ };
+ supplementWarning = "本地规则命令已执行；云端补充失败，已保留本地结果";
+ }
+ } else {
+ payload = preview;
+ }
+ } else {
+ payload = await postControlPhase(
+ "/api/requests",
+ requestBody,
+ state.executionMode,
+ onControlPhaseEvent,
+ );
+ if (!payload) return;
}
renderResult(payload);
completedResult = payload;
@@ -7405,7 +7516,7 @@ async function submitRequest(event) {
setModeExecutionState(submittedModeProfileId, completed ? "succeeded" : "failed");
}
const simulated = payload.execution_report?.metadata?.labels?.seat_feedback === "simulated_hardware";
- showToast(completed
+ showToast(supplementWarning || (completed
? (payload.mode_profile
? `${payload.mode_profile.name}已创建并添加到功能模式`
: payload.status === "partially_succeeded"
@@ -7415,7 +7526,7 @@ async function submitRequest(event) {
: payload.execution_report?.spec?.mode === "hil"
? "真实台架执行与反馈确认已完成"
: "可执行动作的 dry-run 已完成")
- : simulated ? "模拟故障已触发，详情见执行结果" : "请求未执行");
+ : simulated ? "模拟故障已触发，详情见执行结果" : "请求未执行"));
} catch (error) {
completedError = error;
if (submittedModeProfileId) setModeExecutionState(submittedModeProfileId, "failed");
diff --git a/apps/web-console/console-v2.css b/apps/web-console/console-v2.css
index f9727d2..7127528 100644
--- a/apps/web-console/console-v2.css
+++ b/apps/web-console/console-v2.css
@@ -7201,6 +7201,24 @@ body.device-power-drawer-open {
text-align: center;
}
+.console-v2.v2-passive-display.v2-interactive-sync .v2-can-command-source {
+ flex: 0 0 auto;
+ padding: 1px 5px;
+ color: #2366a1;
+ background: #e4f0fb;
+ border: 1px solid rgba(35, 102, 161, 0.14);
+ border-radius: 999px;
+ font-size: 7px;
+ font-weight: 850;
+ line-height: 1.5;
+}
+
+.console-v2.v2-passive-display.v2-interactive-sync .v2-can-command-source[data-parse-source="local_rule"] {
+ color: #08766f;
+ background: #d9f3ef;
+ border-color: rgba(8, 118, 111, 0.14);
+}
+
.console-v2.v2-passive-display.v2-interactive-sync .v2-can-command-list li[data-can-channel="CAN0"] {
background: #f7f4ec;
}
diff --git a/apps/web-console/console-v2.js b/apps/web-console/console-v2.js
index 83af78d..4f43d1d 100644
--- a/apps/web-console/console-v2.js
+++ b/apps/web-console/console-v2.js
@@ -21,6 +21,7 @@
};
let reasoningPickerControl = null;
+ let validatedCommandsVisible = false;
function make(tag, className, html) {
const node = document.createElement(tag);
@@ -61,8 +62,13 @@
const target = normalizedCanTarget(operation.target);
const onOff = normalizedOnOff(parameters.state ?? parameters.power);
const hyphenated = (value) => String(value || "").replaceAll("_", "-");
+ const parseSource = String(operation.parse_source || "");
const command = (category, text, channel = "CAN1", frame = "") => ({
- channel, category, command: text, ...(frame ? { frame } : {}),
+ channel,
+ category,
+ command: text,
+ ...(frame ? { frame } : {}),
+ ...(parseSource ? { parseSource } : {}),
});
const hexByte = (value) => Number(value).toString(16).padStart(2, "0").toUpperCase();
const otaCan0Frame = (opcode, selector, argument) => {
@@ -271,7 +277,15 @@
if (item) commands.push(item);
});
});
- return [...new Map(commands.map((item) => [`${item.channel}:${item.command}`, item])).values()];
+ const unique = new Map();
+ commands.forEach((item) => {
+ const key = `${item.channel}:${item.command}`;
+ const previous = unique.get(key);
+ if (!previous || (previous.parseSource !== "local_rule" && item.parseSource === "local_rule")) {
+ unique.set(key, item);
+ }
+ });
+ return [...unique.values()];
}
function renderParsedCanCommands(result, { message = "" } = {}) {
@@ -285,7 +299,11 @@
if (commands.length) {
list.innerHTML = `${commands.map((item) => `
-
- ${escapeHtml(item.channel)}${escapeHtml(item.category)}
+
+ ${escapeHtml(item.channel)}
+ ${item.parseSource ? `${item.parseSource === "local_rule" ? "本地规则" : "云端模型"}` : ""}
+ ${escapeHtml(item.category)}
+
${escapeHtml(item.command)}
${item.frame ? `seat-ota · ${escapeHtml(item.frame)}` : ""}
`).join("")}
`;
@@ -1348,10 +1366,19 @@
});
$("#v2LogFilter")?.addEventListener("change", renderSyncLogs);
$("#requestForm")?.addEventListener("submit", () => {
+ validatedCommandsVisible = false;
renderParsedCanCommands(null, { message: "正在解析自然语言并生成 CAN 控制指令…" });
});
+ window.addEventListener("cognitive-agent:validated-can-commands", (event) => {
+ const result = event.detail || {};
+ validatedCommandsVisible = parsedCanCommands(result).length > 0;
+ renderParsedCanCommands(result, {
+ message: "已完成白名单校验，正在下发控制指令…",
+ });
+ });
window.addEventListener("cognitive-agent:result", (event) => {
const result = event.detail || {};
+ validatedCommandsVisible = parsedCanCommands(result).length > 0;
renderParsedCanCommands(result);
appendSyncLog({
key: result.request_id ? `result:${result.request_id}` : "",
@@ -1375,8 +1402,10 @@
});
renderSyncedDevices();
}
- if (event.detail?.error) {
+ if (event.detail?.error && !validatedCommandsVisible) {
renderParsedCanCommands(null, { message: "本次请求未生成 CAN 控制指令" });
+ }
+ if (event.detail?.error) {
appendSyncLog({ category: "task", title: "编排任务执行失败", detail: event.detail.error.message || "请求失败", status: "error" });
}
});
diff --git a/apps/web-console/index.html b/apps/web-console/index.html
index c2d0429..ed235ec 100755
--- a/apps/web-console/index.html
+++ b/apps/web-console/index.html
@@ -10,17 +10,17 @@
511工程-智能座舱
-
+
-
+
-
+
diff --git a/apps/web-console/service-worker.js b/apps/web-console/service-worker.js
index e22b3ff..3bee611 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-v202";
+const CACHE_VERSION = "cognitive-agent-pad-v204";
const SHELL_CACHE = `${CACHE_VERSION}-shell`;
const RUNTIME_CACHE = `${CACHE_VERSION}-runtime`;
const APP_SHELL = [
@@ -6,11 +6,11 @@ const APP_SHELL = [
"/index.html",
"/styles.css?v=dual-seat-16",
"/network.js?v=android-pad-2",
- "/app.js?v=dual-seat-101",
+ "/app.js?v=dual-seat-103",
"/gesture-control.js?v=android-pad-9",
"/pwa.js?v=android-pad-3",
- "/console-v2.css?v=sync-layout-40",
- "/console-v2.js?v=sync-layout-41",
+ "/console-v2.css?v=sync-layout-41",
+ "/console-v2.js?v=sync-layout-43",
"/self-extension.css?v=sync-layout-26",
"/self-extension.js?v=sync-layout-26",
"/assets/pets/self-extension-cat.png?v=3",
diff --git a/contracts/schemas/function-config.schema.json b/contracts/schemas/function-config.schema.json
index 64d0f99..b56b5dd 100755
--- a/contracts/schemas/function-config.schema.json
+++ b/contracts/schemas/function-config.schema.json
@@ -12,6 +12,10 @@
"capability": {"type": "string", "pattern": "^[a-z][a-z0-9_]*\\.[a-z][a-z0-9_]*$"},
"target": {"$ref": "common.schema.json#/$defs/target"},
"parameters": {"type": "object"},
+ "parse_source": {
+ "type": "string",
+ "enum": ["local_rule", "cloud_model", "cloud_supplement"]
+ },
"motion": {
"type": "object",
"additionalProperties": false,
diff --git a/docs/CAN1控制指令白名单.md b/docs/CAN1控制指令白名单.md
new file mode 100644
index 0000000..4b2e982
--- /dev/null
+++ b/docs/CAN1控制指令白名单.md
@@ -0,0 +1,74 @@
+# CAN1 控制指令白名单
+
+- 适用项目：`Cognitive_Agent`
+- 适用分支：`feature/N7Web`
+- 核对日期：2026-08-24
+- 验证层级：自然语言解析、精确指令解析和 CAN1 白名单编译通过；未在本次核对中发送真实 CAN 帧。
+
+“发送逻辑”列以当前执行器代码为准。有线模式按该顺序直接提交 CAN1 帧；无线 MQTT 模式将每个原始 CAN 帧分别封装为 1 条 `n7/n7-headunit/can/command` 消息（默认 QoS 1）。因此表中“3 帧，间隔 100 ms”在无线模式下表示后端连续发布 3 条 MQTT 消息，每条消息内部仍为 `repeat: 1`、`intervalMs: 0`。MQTT 客户端或 Broker 接受消息只代表已提交，车辆是否完成动作仍以状态话题的反馈帧为准。
+
+## 有效白名单
+
+| 类别 | 有效指令 | 参数约束 | 发送逻辑 |
+| --- | --- | --- | --- |
+| 后排门锁 | `door rear ` | 仅后排门锁 | `0x32B` 经典 CAN，发送 1 帧；反馈 ID `0x407` |
+| 车窗动作 | `window ` | 目标和动作必须使用列出的枚举值 | `0x431` CAN FD+BRS，连续 3 帧，相邻帧间隔 100 ms；反馈 ID `0x388` |
+| 单窗位置 | `window position <0-100>` | 仅支持单个车窗，位置为整数百分比 | 先等待最长 2500 ms 的有效 `0x388`；已在目标 ±3% 内则发送 0 帧。否则用 `0x431` CAN FD+BRS 发送 3 帧启动命令（间隔 100 ms），最长闭环监听 `0x388` 12 s；无论到位或超时，最后强制发送 3 帧停止命令（间隔 100 ms） |
+| 空调电源 | `ac ` | 整车 HVAC 电源 | `0x1DE` 经典 CAN，连续 3 帧，间隔 100 ms；反馈 ID `0x404` |
+| 空调风量 | `ac fan <1-7>` | 只接受 1–7 的整数 | `0x1DE` 经典 CAN，连续 3 帧，间隔 100 ms；反馈 ID `0x404` |
+| 出风模式 | `ac mode ` | 仅接受列出的 7 种模式 | `0x1DE` 经典 CAN，连续 3 帧，间隔 100 ms；反馈 ID `0x404` |
+| 空调温度 | `ac temp <18-32>` | 温度只接受 18–32 的整数 | `0x1DE` 经典 CAN；`driver`/`passenger` 各发 3 帧（间隔 100 ms），`rear` 发 1 帧；反馈 ID `0x404` |
+| 空调控制模式 | `ac control ` | 手动或自动 | `0x1DE` 经典 CAN，连续 3 帧，间隔 100 ms；反馈 ID `0x404` |
+| 空调循环 | `ac cycle ` | 内循环或外循环 | `0x1DE` 经典 CAN，连续 3 帧，间隔 100 ms；反馈 ID `0x404` |
+| 风口百分比 | `outlet percent <0-100> <0-100>` | 依次为左右、上下整数百分比 | `0x4EF` CAN FD+BRS，连续 3 帧，间隔 100 ms；反馈 ID `0x2F5`/`0x37B` |
+| 风口标定坐标 | `outlet <区域> position <左右坐标> <上下坐标>` | 必须同时满足通用有符号范围和对应风口标定范围 | `0x4EF` CAN FD+BRS，连续 3 帧，间隔 100 ms；反馈 ID `0x2F5`/`0x37B` |
+| 主驾座椅通风/加热 | `seat driver ` | `front-left` 可作为主驾输入兼容别名，规范输出仍为 `driver` | `0x431` CAN FD+BRS，连续 3 帧，间隔 100 ms；反馈 ID `0x4FA` |
+| 主驾座椅按摩关闭 | `seat driver massage off` | 仅关闭动作 | `0x4C1` CAN FD+BRS，连续 3 帧，间隔 100 ms；反馈 ID `0x4FA` |
+| 主驾座椅按摩强度 | `seat driver massage strength ` | `1\|2\|3` 也可作为输入兼容别名，规范输出为单词枚举 | `0x4C1` CAN FD+BRS，连续 3 帧，间隔 100 ms；反馈 ID `0x4FA` |
+| 主驾座椅按摩模式 | `seat driver massage mode ` | 仅接受列出的 5 种模式 | `0x4C1` CAN FD+BRS，连续 3 帧，间隔 100 ms；反馈 ID `0x4FA` |
+| 主驾座椅水平调节 | `seat driver horizontal [duration-ms]` | `duration-ms` 为 100–2000；省略时默认 500 ms | `0x4C1` CAN FD+BRS；动作帧数为 `ceil(duration-ms/100)`，间隔 100 ms，最后一帧后额外保持 100 ms；随后强制发送 3 帧 STOP（间隔 100 ms）；反馈 ID `0x494` |
+| 主驾座椅高度调节 | `seat driver height [duration-ms]` | `duration-ms` 为 100–2000；省略时默认 500 ms | `0x4C1` CAN FD+BRS；动作帧数为 `ceil(duration-ms/100)`，间隔 100 ms，最后一帧后额外保持 100 ms；随后强制发送 3 帧 STOP（间隔 100 ms）；反馈 ID `0x494` |
+| 主驾座椅靠背调节 | `seat driver backrest [duration-ms]` | `duration-ms` 为 100–2000；省略时默认 500 ms | `0x4C1` CAN FD+BRS；动作帧数为 `ceil(duration-ms/100)`，间隔 100 ms，最后一帧后额外保持 100 ms；随后强制发送 3 帧 STOP（间隔 100 ms）；反馈 ID `0x494` |
+| 主驾座椅坐盆调节 | `seat driver basin [duration-ms]` | `duration-ms` 为 100–2000；省略时默认 500 ms | `0x4C1` CAN FD+BRS；动作帧数为 `ceil(duration-ms/100)`，间隔 100 ms，最后一帧后额外保持 100 ms；随后强制发送 3 帧 STOP（间隔 100 ms）；反馈 ID `0x494` |
+| 主驾座椅停止 | `seat driver stop` | `stop` 后不允许附加持续时间 | `0x4C1` CAN FD+BRS，直接连续发送 3 帧 STOP，间隔 100 ms；反馈 ID `0x494` |
+| 主驾座椅记忆 | `seat driver position <1-3>` | 需要先由操作员完成 PAD 只读学习 | `0x4C1` CAN FD+BRS；下发前最长等待 2500 ms 的新鲜 `0x494`，且四轴初始化均为真；然后将已学习的 `0x16 → 0x15` 交替序列发送 10 轮，共 20 帧，间隔 50 ms；发送后最长等待 1200 ms 的 `0x494` |
+| 主驾座椅记忆兼容别名 | `seat driver memory recall <1-3>` | 与 `seat driver position <1-3>` 使用同一已学习帧序列 | 发送逻辑与 `seat driver position <1-3>` 完全相同：`0x4C1` CAN FD+BRS，四轴初始化门禁通过后发送 20 帧 `0x16/0x15` 交替序列，间隔 50 ms |
+| 室内灯总开关 | `light indoor ` | 整车室内灯总开关 | `0x4F5` CAN FD+BRS，连续 3 帧，间隔 100 ms；反馈 ID `0x42E` |
+| 全部阅读灯 | `light reading all ` | CAN1 仅验证 `all` 目标 | `0x4F5` CAN FD+BRS，连续 3 帧，间隔 100 ms；反馈 ID `0x42E` |
+| 氛围灯开关 | `light ambient ` | 氛围灯总开关 | `0x3FC` CAN FD+BRS，连续 3 帧，间隔 100 ms；反馈 ID `0x42E` |
+| 氛围灯颜色 | `light ambient color <1-127>` | 车型颜色索引，只接受整数 | `0x3FC` CAN FD+BRS，连续 3 帧，间隔 100 ms；反馈 ID `0x42E` |
+| 氛围灯亮度 | `light ambient brightness <1-5>` | 只接受 1–5 的整数档位 | `0x3FC` CAN FD+BRS，连续 3 帧，间隔 100 ms；反馈 ID `0x42E` |
+| 遮阳帘动作 | `shade ` | 0/50/100 位置与已验证预设动作复用 | `0x431` 经典 CAN；`open/close/half` 先发 1 帧预设命令，最长监听 `0x41A` 15 s，在匹配目标位置但仍在运动时追加 1 帧 STOP 并确认停止；`stop` 发 1 帧 STOP 并等待停止反馈 |
+| 遮阳帘位置 | `shade position <0-100>` | 中间位置使用状态反馈闭环并自动停止 | 0/50/100 按上述预设逻辑发送。其他位置先等待最长 2500 ms 的新鲜 `0x41A`；已在目标 ±2% 内且停止则发 0 帧，已到位但仍运动则发 1 帧 STOP。否则用 `0x431` 经典 CAN 发 1 帧开/关启动命令，最长监听 15 s，并在 `finally` 中强制发 1 帧 STOP，再确认位置与停止状态 |
+
+## 风口标定坐标范围
+
+| 风口 | 左右范围 | 上下范围 |
+| --- | ---: | ---: |
+| `driver-left` | -25–40 | -40–30 |
+| `driver-right` | -40–30 | -40–30 |
+| `passenger-left` | -10–40 | -40–30 |
+| `passenger-right` | -40–5 | -40–30 |
+
+风口原始坐标语法层的通用范围为 -128–126，但只有同时落在上表对应风口实际标定范围内才能通过编译。
+
+## 能力层扩展与非白名单入口
+
+| 指令/能力 | 当前状态 | 发送逻辑 |
+| --- | --- | --- |
+| `ac compressor ` | `climate.set_ac` 在后端能力层可编译，但当前没有对应的直接精确文本解析入口 | 若由受信能力层调用：`0x1DE` 经典 CAN，3 帧，间隔 100 ms，反馈 ID `0x404`；但本项不是直接指令白名单入口 |
+| `door vehicle ` | 可解析，但当前 CH1 实车验证未观察到有效整车门锁动作，因此会被安全拒绝，不属于可下发白名单 | 安全门禁拒绝，发送 0 帧 |
+| `ac defrost ` | 存在旧解析入口，但当前 CAN1 运行时未实现；前除霜应使用 `ac mode defrost` | 编译门禁拒绝，发送 0 帧 |
+
+## CAN0 边界
+
+本文档只描述 CAN1 车身控制白名单。CAN0 只用于副驾座椅，其状态、轴位置、热舒适和动态模式控制必须依据 `seat-ota` 协议独立处理，不得使用本 CAN1 白名单生成或猜测 CAN0 帧。
+
+## 实现与回归测试
+
+- 精确命令和自然语言匹配：`src/cognitive_agent/application/native_can_intent.py`
+- 能力、参数及 CAN 帧编译门禁：`src/cognitive_agent/application/native_can_runtime.py`
+- 风口标定范围：`platforms/t2g_cabin_bench/calibration/body_can.yaml`
+- 完整白名单回归测试：`tests/unit/test_can1_semantic_whitelist.py`
+
+白名单回归测试展开覆盖 119 条有效命令，并验证越界、非整数、非法枚举和超出风口标定范围的指令不能通过白名单门禁。
diff --git a/domains/climate/config/capabilities.yaml b/domains/climate/config/capabilities.yaml
index 55089ec..77578fb 100755
--- a/domains/climate/config/capabilities.yaml
+++ b/domains/climate/config/capabilities.yaml
@@ -22,7 +22,7 @@ spec:
- name: climate.set_temperature
description: Set an absolute HVAC temperature from 18 to 32 degrees Celsius.
target_devices: [hvac]
- target_scopes: [driver, passenger, front_row, rear_row, rear_left, rear_right, vehicle]
+ target_scopes: [driver, passenger, front_row, rear, rear_row, rear_left, rear_right, vehicle]
parameters:
- name: temperature_c
type: number
diff --git a/domains/seat/src/seat_domain/generator.py b/domains/seat/src/seat_domain/generator.py
index b13320c..f4324b9 100755
--- a/domains/seat/src/seat_domain/generator.py
+++ b/domains/seat/src/seat_domain/generator.py
@@ -395,6 +395,9 @@ def build_function_config(
"capability": DYNAMIC_MODE_STOP_CAPABILITY,
"target": TARGETS[action["target"]],
"parameters": {},
+ "parse_source": str(
+ action.get("parse_source", "local_rule")
+ ),
"timeout_ms": 5000,
"failure_policy": "abort",
}
@@ -412,6 +415,9 @@ def build_function_config(
"capability": DYNAMIC_MODE_CAPABILITY,
"target": TARGETS[action["target"]],
"parameters": {"mode_id": int(action["mode_id"])},
+ "parse_source": str(
+ action.get("parse_source", "local_rule")
+ ),
"timeout_ms": 5000,
"failure_policy": "abort",
}
@@ -437,6 +443,9 @@ def build_function_config(
"capability": MOTION_CAPABILITIES[action["axis"]],
"target": TARGETS[action["target"]],
"parameters": parameters,
+ "parse_source": str(
+ action.get("parse_source", "local_rule")
+ ),
"timeout_ms": 8000,
"failure_policy": "abort",
}
@@ -460,6 +469,9 @@ def build_function_config(
"capability": CAPABILITIES[action["function"]],
"target": TARGETS[action["target"]],
"parameters": parameters,
+ "parse_source": str(
+ action.get("parse_source", "local_rule")
+ ),
"timeout_ms": 3000,
"failure_policy": "abort",
}
diff --git a/domains/seat/src/seat_domain/intent.py b/domains/seat/src/seat_domain/intent.py
index f75db38..b273d1e 100755
--- a/domains/seat/src/seat_domain/intent.py
+++ b/domains/seat/src/seat_domain/intent.py
@@ -52,6 +52,15 @@ STRONG_PATTERN = re.compile(r"很|太|非常|特别|难受")
AIRFLOW_CAUSE_PATTERN = re.compile(r"吹|风太大|风吹")
FIRST_PERSON_PATTERN = re.compile(r"我|自己")
VEHICLE_SCOPE_PATTERN = re.compile(r"车上|车内|车里|全车|整车|整个车")
+NON_SEAT_DEVICE_PATTERN = re.compile(
+ r"车窗|车门|门锁|空调|温度|风量|风速|出风模式|"
+ r"内循环|外循环|循环模式|风口|遮阳帘|天幕|阅读灯|"
+ r"室内灯|车内灯|氛围灯"
+)
+SEAT_CONTROL_PATTERN = re.compile(
+ r"座椅|坐椅|座位|靠背|椅背|坐盆|坐垫|滑轨|腿托|脚托|四轴|"
+ r"通风|加热|按摩|记忆|预设"
+)
PASSENGER_ONLY_QUESTION = "当前所有座椅能力仅控制副驾，不支持主驾座椅目标。"
UNSUPPORTED_SEAT_SCOPE_PATTERN = re.compile(
r"后排|第二排|二排|第三排|三排|左前座椅|左前座|左前|"
@@ -799,6 +808,16 @@ def parse_local_intent(
"""Parse common Chinese seat commands without making hardware assumptions."""
text = _normalize_text(utterance)
+ if NON_SEAT_DEVICE_PATTERN.search(text) and not SEAT_CONTROL_PATTERN.search(text):
+ # This parser owns CAN0 passenger-seat semantics only. Occupant words
+ # such as “主驾/副驾” in a window or climate clause are spatial
+ # qualifiers for that device and must not create a seat ambiguity.
+ return {
+ "actions": [],
+ "ambiguities": [],
+ "decision_summary": [],
+ "confidence": 0.0,
+ }
clauses = [part for part in re.split(r"[，,。；;！!？?]", text) if part]
actions: list[dict[str, Any]] = []
motion_actions: list[dict[str, Any]] = []
diff --git a/domains/seat/tests/integration/test_seat_service.py b/domains/seat/tests/integration/test_seat_service.py
index 8e43229..7d47ce7 100755
--- a/domains/seat/tests/integration/test_seat_service.py
+++ b/domains/seat/tests/integration/test_seat_service.py
@@ -311,7 +311,7 @@ class SeatServiceIntegrationTests(unittest.TestCase):
]
self.assertEqual(
commands,
- ["seat driver memory recall 3", "ac temp driver 25", "ac fan 1"],
+ ["seat driver position 3", "ac temp driver 25", "ac fan 1"],
)
def test_model_climate_request_completes_domain_loop(self) -> None:
diff --git a/domains/seat/tests/unit/test_intent.py b/domains/seat/tests/unit/test_intent.py
index f6c2493..5e27594 100755
--- a/domains/seat/tests/unit/test_intent.py
+++ b/domains/seat/tests/unit/test_intent.py
@@ -67,6 +67,21 @@ class LocalIntentParserTests(unittest.TestCase):
self.assertEqual(result["actions"][0]["target"], "passenger")
self.assertEqual(result["ambiguities"], [])
+ def test_non_seat_spatial_qualifiers_do_not_enter_can0_seat_rules(self) -> None:
+ utterances = (
+ "打开主驾车窗",
+ "打开副驾车窗",
+ "把主驾和副驾温度都设为24度",
+ "风量调到2档",
+ "设置主驾出风模式为吹脸",
+ )
+ for utterance in utterances:
+ with self.subTest(utterance=utterance):
+ result = parse_local_intent(utterance)
+
+ self.assertEqual(result["actions"], [])
+ self.assertEqual(result["ambiguities"], [])
+
def test_parses_four_axis_zero_for_passenger(self) -> None:
result = parse_local_intent("把水平、坐盆、靠背、腿托四轴归零")
diff --git a/src/cognitive_agent/application/conversation_state.py b/src/cognitive_agent/application/conversation_state.py
index d726fc8..23427c6 100755
--- a/src/cognitive_agent/application/conversation_state.py
+++ b/src/cognitive_agent/application/conversation_state.py
@@ -567,14 +567,19 @@ class ConversationStateStore:
device = state["domain_devices"][domain]
if domain == "climate":
+ # The trusted CAN command vocabulary uses the concise token
+ # ``rear`` while the conversation state stores that same HVAC
+ # zone as ``rear_row``. Keep the public command stable and
+ # normalize only at the state boundary.
+ zone_target = "rear_row" if target == "rear" else target
if capability == "climate.set_power":
- device["zones"][target]["power"] = parameters["state"]
+ device["zones"][zone_target]["power"] = parameters["state"]
if target == "vehicle":
device["power"] = parameters["state"]
for zone in device["zones"].values():
zone["power"] = parameters["state"]
elif capability == "climate.set_temperature":
- device["zones"][target]["temperature_c"] = parameters[
+ device["zones"][zone_target]["temperature_c"] = parameters[
"temperature_c"
]
if target == "vehicle":
@@ -589,7 +594,7 @@ class ConversationStateStore:
elif capability == "climate.set_front_defrost":
device["front_defrost"] = parameters["state"]
elif capability == "climate.set_outlet_mode":
- device["zones"][target]["outlet_mode"] = parameters["mode"]
+ device["zones"][zone_target]["outlet_mode"] = parameters["mode"]
elif capability == "climate.set_control_mode":
device["control_mode"] = parameters["mode"]
device["feedback_source"] = feedback_source
diff --git a/src/cognitive_agent/application/domain_runtime.py b/src/cognitive_agent/application/domain_runtime.py
index ddbe416..6dfaebe 100755
--- a/src/cognitive_agent/application/domain_runtime.py
+++ b/src/cognitive_agent/application/domain_runtime.py
@@ -78,6 +78,9 @@ class ConfiguredDomainRuntime:
"zone": action["target"],
},
"parameters": dict(action["parameters"]),
+ "parse_source": str(
+ action.get("parse_source", "local_rule")
+ ),
"timeout_ms": 8000 if self.domain == "roof" else 4000,
"failure_policy": "abort",
}
diff --git a/src/cognitive_agent/application/native_can_intent.py b/src/cognitive_agent/application/native_can_intent.py
index 2706d38..6e75dd2 100644
--- a/src/cognitive_agent/application/native_can_intent.py
+++ b/src/cognitive_agent/application/native_can_intent.py
@@ -25,6 +25,16 @@ WINDOW_TARGETS = {
"后左": "rear_left",
"右后": "rear_right",
"后右": "rear_right",
+ # Natural cabin language names the front windows by occupant more often
+ # than by left/right. These aliases belong only to the window domain:
+ # driver = left-front, passenger = right-front.
+ "主驾驶": "front_left",
+ "驾驶位": "front_left",
+ "司机位": "front_left",
+ "主驾": "front_left",
+ "副驾驶": "front_right",
+ "乘客位": "front_right",
+ "副驾": "front_right",
}
WINDOW_GROUP_MEMBERS = {
"all": ["front_left", "front_right", "rear_left", "rear_right"],
@@ -49,6 +59,7 @@ LIGHT_TARGETS = {
"左后": "rear_left",
"右后": "rear_right",
}
+CLAUSE_SEPARATOR = re.compile(r"[，,；;。！？!?\n]+")
# Defaults are intentionally scoped to the driver console and verified CH1
# body-CAN vocabulary. A missing target must never become the legacy
@@ -83,7 +94,8 @@ CH1_DEFAULT_TARGETS = {
_EXPLICIT_TARGET_PATTERNS = {
"window": re.compile(
- r"(?:全部|所有|四个|全车|前排|后排|左前|前左|右前|前右|左后|后左|右后|后右)"
+ r"(?:全部|所有|四个|全车|前排|后排|左前|前左|右前|前右|左后|后左|右后|后右|"
+ r"主驾驶|驾驶位|司机位|主驾|副驾驶|乘客位|副驾)"
r"(?:的)?(?:车窗|窗)|window\s+(?:all|front|rear|front-left|front-right|rear-left|rear-right)",
re.IGNORECASE,
),
@@ -113,6 +125,88 @@ _EXPLICIT_TARGET_PATTERNS = {
}
+def _scoped_text(
+ compact: str,
+ subject_pattern: str,
+ continuation_pattern: str | None = None,
+) -> str:
+ """Keep action words inside clauses that name the controlled device."""
+
+ clauses = [clause for clause in CLAUSE_SEPARATOR.split(compact) if clause]
+ subject_indexes = {
+ index
+ for index, clause in enumerate(clauses)
+ if re.search(subject_pattern, clause, re.IGNORECASE)
+ }
+ selected_indexes = set(subject_indexes)
+ if continuation_pattern:
+ selected_indexes.update(
+ index + 1
+ for index in subject_indexes
+ if index + 1 < len(clauses)
+ and re.search(
+ continuation_pattern,
+ clauses[index + 1],
+ re.IGNORECASE,
+ )
+ )
+ return "，".join(
+ clause
+ for index, clause in enumerate(clauses)
+ if index in selected_indexes
+ )
+
+
+def _window_targets(text: str) -> tuple[list[str], bool]:
+ """Resolve target aliases only when they are attached to a window noun."""
+
+ explicit: list[str] = []
+ broad: str | None = None
+ target_labels = "|".join(
+ re.escape(label)
+ for label in sorted(WINDOW_TARGETS, key=len, reverse=True)
+ )
+ for label, target in WINDOW_TARGETS.items():
+ attached = bool(
+ re.search(
+ rf"{re.escape(label)}"
+ rf"(?:(?:和|与|及|、|以及)(?:{target_labels}))*"
+ rf"(?:的)?(?:两个|两侧)?(?:车窗|窗)",
+ text,
+ )
+ )
+ if not attached:
+ continue
+ if target in WINDOW_GROUP_MEMBERS:
+ broad = broad or target
+ elif target not in explicit:
+ explicit.append(target)
+
+ exclusions = {
+ target
+ for label, target in WINDOW_TARGETS.items()
+ if target not in WINDOW_GROUP_MEMBERS
+ and re.search(
+ rf"{re.escape(label)}(?:车窗)?(?:除外|除去|不包括|不含)",
+ text,
+ )
+ }
+ if broad and exclusions:
+ return (
+ [
+ target
+ for target in WINDOW_GROUP_MEMBERS[broad]
+ if target not in exclusions
+ ],
+ True,
+ )
+ if explicit:
+ return ([target for target in explicit if target not in exclusions], True)
+ if broad:
+ return ([broad], True)
+ return (["front_left"], False)
+
+
def _capability_domain(capability: str) -> str:
return "interior_light" if capability.startswith("interior_light.") else capability.split(".", 1)[0]
@@ -156,6 +250,7 @@ def _action(
"operation": "set",
"reason": reason,
"execution_backend": "native_can",
+ "parse_source": "local_rule",
}
@@ -284,6 +379,8 @@ def _parse_exact_command(text: str) -> list[dict[str, Any]]:
}
if direction not in valid_directions[axis]:
return []
+ if direction == "stop" and duration_token is not None:
+ return []
duration = (
500
if direction == "stop"
@@ -381,17 +478,23 @@ def _parse_chinese_driver_seat_commands(compact: str) -> list[dict[str, Any]]:
actions: list[dict[str, Any]] = []
memory = re.search(
- r"(?:记忆(?:位置|位)?|记忆座椅(?:位置)?|座椅记忆(?:位置)?|位置记忆)"
+ r"(?:"
+ r"(?:记忆(?:位置|位)?|预设(?:位置)?|记忆座椅(?:位置)?|"
+ r"座椅记忆(?:位置)?|位置记忆)"
r"(?:切换到|切换为|调整到|调到|唤醒|调用|恢复到|恢复为|设为|设置为)?"
- r"([1-3])(?:号|档|位)?",
+ r"(?P[1-3一二三])(?:号|档|位)?"
+ r"|(?P[1-3一二三])号?(?:记忆|预设)(?:位置|位)?"
+ r")",
compact,
)
if memory is not None:
+ slot_token = memory.group("after") or memory.group("before")
+ slot = {"一": 1, "二": 2, "三": 3}.get(slot_token, slot_token)
actions.append(_action(
"seat",
- "seat.recall_driver_memory_native",
+ "seat.set_driver_position_native",
"driver",
- {"slot": int(memory[1])},
+ {"slot": int(slot)},
(
"用户明确指定了 CH1 主驾座椅记忆位置。"
if explicit_driver
@@ -449,8 +552,10 @@ def _parse_chinese_driver_seat_commands(compact: str) -> list[dict[str, Any]]:
strength = (
1
if re.search(r"弱|低|1档", compact)
+ else 2
+ if re.search(r"中等|中档|2档", compact)
else 3
- if re.search(r"强|高|3档", compact)
+ if re.search(r"强(?!度)|高|3档", compact)
else 2
)
massage = {"action": "strength", "strength": strength}
@@ -549,103 +654,149 @@ def _parse_chinese_driver_seat_commands(compact: str) -> list[dict[str, Any]]:
def _parse_chinese_commands(text: str) -> list[dict[str, Any]]:
compact = re.sub(r"\s+", "", text)
- actions: list[dict[str, Any]] = _parse_chinese_driver_seat_commands(compact)
+ seat_text = _scoped_text(
+ compact,
+ r"座椅|座位|靠背|椅背|坐盆|坐垫|腿托|脚托|按摩|记忆|预设|"
+ r"通风|加热|前移|后移|向前移动|向后移动",
+ )
+ actions: list[dict[str, Any]] = _parse_chinese_driver_seat_commands(
+ seat_text
+ )
+
+ climate_text = _scoped_text(
+ compact,
+ r"空调|温度|风量|风速|内循环|外循环|循环模式|吹面|吹脸|吹脚|除霜|(?:AC|A/C)",
+ )
if (
- "空调" in compact
+ "空调" in climate_text
or re.search(r"(?:AC|A/C)", text, re.IGNORECASE)
- or re.search(r"温度|风量|风速|内循环|外循环|循环模式", compact)
+ or re.search(
+ r"温度|风量|风速|内循环|外循环|循环模式|吹面|吹脸|吹脚|除霜",
+ climate_text,
+ )
):
- if re.search(r"(?:打开|开启|启动)空调", compact):
+ if re.search(r"(?:打开|开启|启动)空调", climate_text):
actions.append(_action("climate", "climate.set_power", "vehicle", {"state": "on"}, "用户要求打开空调。"))
- elif re.search(r"(?:关闭|关掉|关上)空调", compact):
+ elif re.search(r"(?:关闭|关掉|关上)空调", climate_text):
actions.append(_action("climate", "climate.set_power", "vehicle", {"state": "off"}, "用户要求关闭空调。"))
temperature = re.search(
- r"(?:(主驾|副驾|后排)(?:空调)?(?:温度)?|空调(?:温度)?|温度)(?:调到|设为|设置为|到)?(1[89]|2\d|3[0-2])(?:℃|度)?",
- compact,
+ r"(?:空调(?:温度)?|温度)(?:都)?"
+ r"(?:调整到|调整为|调到|调为|设为|设置为|调至|到|为)?"
+ r"(1[89]|2\d|3[0-2])(?:摄氏)?(?:℃|度)?(?![\d.])",
+ climate_text,
)
if temperature:
- target = {"主驾": "driver", "副驾": "passenger", "后排": "rear"}.get(temperature[1], "driver")
+ if (
+ re.search(r"全车|整车|前排|主副驾", climate_text)
+ or ("主驾" in climate_text and "副驾" in climate_text)
+ ):
+ temperature_targets = ("driver", "passenger")
+ elif re.search(r"副驾|副驾驶|乘客", climate_text):
+ temperature_targets = ("passenger",)
+ elif "后排" in climate_text:
+ temperature_targets = ("rear",)
+ else:
+ temperature_targets = ("driver",)
+ explicit_temperature_target = bool(
+ re.search(
+ r"主驾|副驾|副驾驶|乘客|后排|前排|全车|整车|主副驾",
+ climate_text,
+ )
+ )
reason = (
"用户指定了空调温区和温度。"
- if temperature[1]
+ if explicit_temperature_target
else "用户未指定空调温区，按驾驶台默认选择主驾温区。"
)
- actions.append(_action("climate", "climate.set_temperature", target, {"temperature_c": int(temperature[2])}, reason))
+ actions.extend(
+ _action(
+ "climate",
+ "climate.set_temperature",
+ target,
+ {"temperature_c": int(temperature[1])},
+ reason,
+ )
+ for target in temperature_targets
+ )
- fan = re.search(r"(?:风量|风速)(?:调到|设为|设置为|到)?([1-7])(?:档)?", compact)
+ fan = re.search(
+ r"(?:风量|风速)(?:调整到|调整为|调到|调为|设为|设置为|到)?"
+ r"([1-7])(?:档|级)?(?![\d.])",
+ climate_text,
+ )
if fan:
actions.append(_action("climate", "climate.set_fan_level", "vehicle", {"level": int(fan[1])}, "用户指定了空调风量。"))
- if re.search(r"自动(?:控制|模式)", compact):
+ if re.search(r"自动(?:控制|模式)", climate_text):
actions.append(_action("climate", "climate.set_control_mode", "vehicle", {"mode": "auto"}, "用户指定空调自动控制。"))
- elif re.search(r"手动(?:控制|模式)", compact):
+ elif re.search(r"手动(?:控制|模式)", climate_text):
actions.append(_action("climate", "climate.set_control_mode", "vehicle", {"mode": "manual"}, "用户指定空调手动控制。"))
- if re.search(r"内循环", compact):
+ if re.search(r"内循环", climate_text):
actions.append(_action("climate", "climate.set_cycle_mode", "vehicle", {"mode": "internal"}, "用户指定空调内循环。"))
- elif re.search(r"外循环", compact):
+ elif re.search(r"外循环", climate_text):
actions.append(_action("climate", "climate.set_cycle_mode", "vehicle", {"mode": "external"}, "用户指定空调外循环。"))
- outlet_modes = [
- ("吹面吹脚除霜", "face_feet_defrost"),
- ("吹面除霜", "face_defrost"),
- ("吹脚除霜", "feet_defrost"),
- ("吹面吹脚", "face_feet"),
- ("除霜", "defrost"),
- ("吹脚", "feet"),
- ("吹面", "face"),
- ]
- for label, mode in outlet_modes:
- if label in compact:
- actions.append(_action("climate", "climate.set_outlet_mode", "vehicle", {"mode": mode}, "用户指定了空调出风模式。"))
- break
-
- if "车窗" in compact:
- explicit_targets = []
- for label, target_name in WINDOW_TARGETS.items():
- if target_name in WINDOW_GROUP_MEMBERS or label not in compact:
- continue
- if target_name not in explicit_targets:
- explicit_targets.append(target_name)
- broad_target = next(
- (
- target_name
- for label, target_name in WINDOW_TARGETS.items()
- if target_name in WINDOW_GROUP_MEMBERS and label in compact
- ),
- None,
+ has_face = bool(re.search(r"吹面|吹脸", climate_text))
+ has_feet = "吹脚" in climate_text
+ has_defrost = "除霜" in climate_text
+ outlet_mode = (
+ "face_feet_defrost"
+ if has_face and has_feet and has_defrost
+ else "face_feet"
+ if has_face and has_feet
+ else "face_defrost"
+ if has_face and has_defrost
+ else "feet_defrost"
+ if has_feet and has_defrost
+ else "defrost"
+ if has_defrost
+ else "face"
+ if has_face
+ else "feet"
+ if has_feet
+ else None
)
- exclusions = {
- target_name
- for label, target_name in WINDOW_TARGETS.items()
- if target_name not in WINDOW_GROUP_MEMBERS
- and re.search(
- rf"{re.escape(label)}(?:车窗)?(?:除外|除去|不包括|不含)",
- compact,
+ if outlet_mode:
+ actions.append(
+ _action(
+ "climate",
+ "climate.set_outlet_mode",
+ "vehicle",
+ {"mode": outlet_mode},
+ "用户指定了空调出风模式。",
+ )
)
- }
- if broad_target and exclusions:
- targets = [
- target_name
- for target_name in WINDOW_GROUP_MEMBERS[broad_target]
- if target_name not in exclusions
- ]
- elif explicit_targets:
- targets = [target_name for target_name in explicit_targets if target_name not in exclusions]
- else:
- targets = [broad_target or "front_left"]
+
+ window_text = _scoped_text(
+ compact,
+ r"车窗",
+ r"^(?:主驾|主驾驶|驾驶位|司机位|副驾|副驾驶|乘客位|"
+ r"左前|前左|右前|前右|左后|后左|右后|后右)(?:车窗)?(?:除外|除去|不包括|不含)$",
+ )
+ if window_text:
+ targets, explicit_window_target = _window_targets(window_text)
position = re.search(
- r"(?:位置|开度|调到|设为|设置为|开启到|打开到)(\d{1,3})%",
- compact,
+ r"(?:位置|开度|调整到|调到|设为|设置为|开启到|打开到|开到)(\d{1,3})%",
+ window_text,
+ )
+ half_position = bool(
+ re.search(
+ r"(?:开启|打开|开到|开度)?(?:一半|半开)",
+ window_text,
+ )
)
- half_position = bool(re.search(r"(?:开启|打开|开到|开度)?(?:一半|半开)", compact))
position_pct = int(position[1]) if position else 50 if half_position else None
- if position_pct is not None and all(target not in WINDOW_GROUP_MEMBERS for target in targets) and position_pct <= 100:
+ if (
+ position_pct is not None
+ and all(target not in WINDOW_GROUP_MEMBERS for target in targets)
+ and position_pct <= 100
+ ):
reason = (
"用户指定了单个车窗的绝对开度。"
- if explicit_targets or broad_target
+ if explicit_window_target
else "用户未指定车窗目标，按驾驶台默认选择主驾左前窗。"
)
actions.extend(
@@ -658,15 +809,15 @@ def _parse_chinese_commands(text: str) -> list[dict[str, Any]]:
pass
else:
motion = (
- "stop" if re.search(r"停止|停下|暂停", compact)
- else "close" if re.search(r"关闭|关上|升起|关窗", compact)
- else "open" if re.search(r"打开|开启|降下|开窗", compact)
+ "stop" if re.search(r"停止|停下|暂停", window_text)
+ else "close" if re.search(r"关闭|关上|升起|关窗", window_text)
+ else "open" if re.search(r"打开|开启|降下|开窗", window_text)
else None
)
if motion:
reason = (
"用户指定了车窗目标和动作。"
- if explicit_targets or broad_target
+ if explicit_window_target
else "用户未指定车窗目标，按驾驶台默认选择主驾左前窗。"
)
actions.extend(
@@ -674,17 +825,18 @@ def _parse_chinese_commands(text: str) -> list[dict[str, Any]]:
for target in targets
)
+ door_text = _scoped_text(compact, r"门锁|车门|后门|锁车|解锁车辆")
rear_lock_request = bool(
re.search(
- r"(?:后门|后排车门|后排门锁|二排车门|二排门锁)", compact
+ r"(?:后门|后排车门|后排门锁|二排车门|二排门锁)", door_text
)
)
if rear_lock_request:
state = (
"unlock"
- if re.search(r"解锁|开锁", compact)
+ if re.search(r"解锁|开锁", door_text)
else "lock"
- if re.search(r"上锁|落锁|锁上", compact)
+ if re.search(r"上锁|落锁|锁上", door_text)
else None
)
if state:
@@ -695,59 +847,123 @@ def _parse_chinese_commands(text: str) -> list[dict[str, Any]]:
r"整车(?:门锁|车门)|全车(?:门锁|车门)|(?:所有|全部)(?:车门|门锁)|"
r"(?:车门|门锁)(?:上锁|落锁|锁上|解锁|开锁)|"
r"(?:锁车|整车解锁|全车解锁|解锁车辆)",
- compact,
+ door_text,
)
)
if vehicle_lock_request:
state = (
"unlock"
- if re.search(r"解锁|开锁", compact)
+ if re.search(r"解锁|开锁", door_text)
else "lock"
- if re.search(r"上锁|落锁|锁上|锁车", compact)
+ if re.search(r"上锁|落锁|锁上|锁车", door_text)
else None
)
if state:
actions.append(_action("door", "door.set_vehicle_lock", "vehicle", {"state": state}, "用户指定了整车门锁状态。"))
- if re.search(r"遮阳帘|天幕帘", compact):
- position = re.search(r"(?:位置|开度|调到|设为)(\d{1,3})%", compact)
+ shade_text = _scoped_text(compact, r"遮阳帘|天幕帘")
+ if shade_text:
+ position = re.search(
+ r"(?:位置|开度|调整到|调到|设为|设置为|开到|打开(?:到)?|开启(?:到)?)(\d{1,3})%",
+ shade_text,
+ )
if position and int(position[1]) <= 100:
actions.append(_action("roof", "roof.set_shade_position", "vehicle", {"position_pct": int(position[1])}, "用户指定了遮阳帘开度。"))
else:
motion = (
- "stop" if re.search(r"停止|停下|暂停", compact)
- else "half" if re.search(r"半开|一半", compact)
- else "close" if re.search(r"关闭|关上", compact)
- else "open" if re.search(r"打开|开启", compact)
+ "stop" if re.search(r"停止|停下|暂停", shade_text)
+ else "half" if re.search(r"半开|一半", shade_text)
+ else "close" if re.search(r"关闭|关上", shade_text)
+ else "open" if re.search(r"打开|开启", shade_text)
else None
)
if motion:
actions.append(_action("roof", "roof.set_shade_motion", "vehicle", {"motion": motion}, "用户指定了遮阳帘动作。"))
+ indoor_text = _scoped_text(compact, r"室内灯|车内灯")
for label, capability in (("室内灯", "interior_light.set_indoor_power"), ("车内灯", "interior_light.set_indoor_power")):
- if label in compact:
- state = "off" if re.search(r"关闭|关掉|关上", compact) else "on" if re.search(r"打开|开启", compact) else None
+ if label in indoor_text:
+ state = "off" if re.search(r"关闭|关掉|关上", indoor_text) else "on" if re.search(r"打开|开启", indoor_text) else None
if state:
actions.append(_action("interior_light", capability, "vehicle", {"state": state}, "用户指定了车内灯状态。"))
break
- if "阅读灯" in compact:
- target = next((value for label, value in LIGHT_TARGETS.items() if label in compact), "all")
- state = "off" if re.search(r"关闭|关掉|关上", compact) else "on" if re.search(r"打开|开启", compact) else None
+ reading_text = _scoped_text(compact, r"阅读灯")
+ if reading_text:
+ target = next((value for label, value in LIGHT_TARGETS.items() if label in reading_text), "all")
+ state = "off" if re.search(r"关闭|关掉|关上", reading_text) else "on" if re.search(r"打开|开启", reading_text) else None
if state and target == "all":
actions.append(_action("interior_light", "interior_light.set_reading_power", target, {"state": state}, "用户指定了阅读灯区域和状态。"))
- if "氛围灯" in compact:
- color_match = re.search(r"颜色(?:索引|编号)?(\d{1,3})", compact)
- brightness_match = re.search(r"亮度(?:档|级)?([1-5])", compact)
+ ambient_text = _scoped_text(
+ compact,
+ r"氛围灯",
+ r"^(?:亮度|强度)(?:调整|调节|调|设置|设)?",
+ )
+ if ambient_text:
+ state = (
+ "off"
+ if re.search(r"关闭|关掉|熄灭", ambient_text)
+ else "on"
+ if re.search(r"打开|开启|点亮", ambient_text)
+ else None
+ )
+ if state:
+ actions.append(
+ _action(
+ "ambient_light",
+ "ambient_light.set_power",
+ "vehicle",
+ {"state": state},
+ "用户指定了氛围灯总开关状态。",
+ )
+ )
+ color_match = re.search(
+ r"(?:颜色编号|颜色索引|颜色|色号)"
+ r"(?:调整到|调整为|调到|调为|设置为|设为|为)?"
+ r"(\d{1,3})(?![\d.])",
+ ambient_text,
+ )
+ brightness_match = re.search(
+ r"(?:亮度|强度)(?:调整到|调整为|调到|调为|设置为|设为|为)?"
+ r"([1-5])(?:档|级)?(?![\d.])",
+ ambient_text,
+ )
if color_match and 1 <= int(color_match[1]) <= 127:
actions.append(_action("ambient_light", "ambient_light.set_color_index", "vehicle", {"color_index": int(color_match[1])}, "用户指定了已验证的氛围灯颜色索引。"))
if brightness_match:
actions.append(_action("ambient_light", "ambient_light.set_brightness_level", "vehicle", {"brightness_level": int(brightness_match[1])}, "用户指定了已验证的氛围灯亮度档位。"))
- outlet_target = next((value for label, value in OUTLET_TARGETS.items() if f"{label}风口" in compact), None)
- outlet_position = re.search(r"(?:左右|水平)(\d{1,3})%?.*?(?:上下|垂直)(\d{1,3})%?", compact)
- if "风口" in compact and outlet_position and all(int(value) <= 100 for value in outlet_position.groups()):
+ outlet_text = _scoped_text(compact, r"风口")
+ outlet_target = next((value for label, value in OUTLET_TARGETS.items() if f"{label}风口" in outlet_text), None)
+ calibrated_position = re.search(
+ r"(?:标定坐标|标定位置).*?(?:左右|水平)"
+ r"(-?\d{1,3})(?![\d.]).*?(?:上下|垂直)(-?\d{1,3})(?![\d.])",
+ outlet_text,
+ )
+ outlet_position = re.search(
+ r"(?:左右|水平)(\d{1,3})%?(?![\d.]).*?"
+ r"(?:上下|垂直)(\d{1,3})%?(?![\d.])",
+ outlet_text,
+ )
+ if (
+ outlet_target
+ and calibrated_position
+ and all(-128 <= int(value) <= 126 for value in calibrated_position.groups())
+ ):
+ actions.append(
+ _action(
+ "outlet",
+ "outlet.set_calibrated_position",
+ outlet_target,
+ {
+ "horizontal_value": int(calibrated_position[1]),
+ "vertical_value": int(calibrated_position[2]),
+ },
+ "用户指定了已标定风口的双轴原始坐标。",
+ )
+ )
+ elif outlet_text and outlet_position and all(int(value) <= 100 for value in outlet_position.groups()):
target = outlet_target or "driver_left"
reason = (
"用户指定了已标定风口的双轴百分比位置。"
diff --git a/src/cognitive_agent/application/native_can_relative.py b/src/cognitive_agent/application/native_can_relative.py
index b6e56bf..c553717 100644
--- a/src/cognitive_agent/application/native_can_relative.py
+++ b/src/cognitive_agent/application/native_can_relative.py
@@ -275,6 +275,7 @@ class NativeCanRelativeResolver:
"target": target,
"operation": "set",
"parameters": parameters,
+ "parse_source": "local_rule",
"reason": f"用户要求{label}{verb}{delta}；基于当前状态 {current} 计算绝对目标 {target_value}{'，已按允许范围收敛' if clamped else ''}。",
"relative_resolution": {
"direction": direction,
diff --git a/src/cognitive_agent/application/native_can_runtime.py b/src/cognitive_agent/application/native_can_runtime.py
index c0a9003..a09945f 100644
--- a/src/cognitive_agent/application/native_can_runtime.py
+++ b/src/cognitive_agent/application/native_can_runtime.py
@@ -183,6 +183,9 @@ class NativeCanRuntime:
"zone": target,
},
"parameters": dict(action.get("parameters", {})),
+ "parse_source": str(
+ action.get("parse_source", "local_rule")
+ ),
"timeout_ms": (
20_000
if capability_name
@@ -417,6 +420,8 @@ class NativeCanRuntime:
return f"seat driver massage strength {strength}"
return f"seat driver massage mode {parameters['mode']}"
if capability == "seat.set_driver_axis_native":
+ if parameters["direction"] == "stop":
+ return f"seat driver {parameters['axis']} stop"
return (
f"seat driver {parameters['axis']} {parameters['direction']} "
f"{parameters.get('duration_ms', 500)}"
@@ -1129,15 +1134,30 @@ class NativeCanExecutor:
"feedback ID and a target-value match contract"
)
special_handler = command["signals"].get("special_handler")
- if special_handler == "window_position":
+ feedback_required = self.feedback_required
+ if not feedback_required and special_handler in {
+ None,
+ "shade_preset",
+ }:
+ # Wireless MQTT uses the state topic as an asynchronous
+ # source of truth. Fixed frames and self-terminating
+ # shade presets must not hold the HTTP request open
+ # while waiting for the same state stream synchronously.
+ sent_frames = self._send_command(bus, command)
+ feedback = None
+ elif special_handler == "window_position":
sent_frames, feedback = self._execute_window_position(bus, command)
elif special_handler == "driver_seat_axis":
sent_frames, feedback = self._execute_driver_seat_axis(
- bus, command
+ bus,
+ command,
+ wait_for_feedback=feedback_required,
)
elif special_handler == "driver_seat_memory_recall":
sent_frames, feedback = self._execute_driver_seat_memory(
- bus, command
+ bus,
+ command,
+ wait_for_completion=feedback_required,
)
elif special_handler == "shade_preset":
sent_frames, feedback = self._execute_shade_preset(
@@ -1151,7 +1171,6 @@ class NativeCanExecutor:
sent_frames = self._send_command(bus, command)
feedback = self._wait_feedback(bus, command)
feedback_match = self._feedback_matches(command, feedback)
- feedback_required = self.feedback_required
verified = feedback is not None and feedback_match is True
delivered = bool(sent_frames)
succeeded = verified or (delivered and not feedback_required)
@@ -1167,17 +1186,26 @@ class NativeCanExecutor:
publication_start:
]
)
+ delivery_status = "submitted"
+ if not feedback_required and delivered:
+ delivery_status = (
+ "published"
+ if mqtt_messages
+ and all(
+ message.get("delivery_status") == "published"
+ for message in mqtt_messages
+ )
+ else "queued"
+ )
results.append(
{
"ref": operation["id"],
"domain": function_config["spec"]["domain"],
"status": "succeeded" if succeeded else "failed",
"message": (
- "MQTT 白名单控制帧已发布，并收到匹配的车辆状态反馈。"
- if not feedback_required and feedback_match is True
- else "MQTT 白名单控制帧已发布；已收到状态帧，但目标值尚未确认。"
- if not feedback_required and feedback is not None
- else "MQTT 白名单控制帧已发布；车辆状态反馈尚未确认。"
+ "MQTT 白名单控制帧已由 Broker 确认；车辆效果以状态话题回读为准。"
+ if not feedback_required and delivery_status == "published"
+ else "MQTT 白名单控制帧已进入客户端发布队列；车辆效果以状态话题回读为准。"
if not feedback_required
else "CAN 盒已提交车端控制帧并观察到匹配反馈。"
if feedback_match is True
@@ -1197,11 +1225,7 @@ class NativeCanExecutor:
"feedback_match": feedback_match,
"feedback_state": self._feedback_state(feedback),
"feedback_required": feedback_required,
- "delivery_status": (
- "published"
- if not feedback_required and delivered
- else "submitted"
- ),
+ "delivery_status": delivery_status,
"physical_effect_confirmed": verified,
"control_gate": (
"compiled_action_and_frame_whitelist"
@@ -1260,7 +1284,11 @@ class NativeCanExecutor:
return sent
def _execute_driver_seat_axis(
- self, bus: CanBus, command: dict[str, Any]
+ self,
+ bus: CanBus,
+ command: dict[str, Any],
+ *,
+ wait_for_feedback: bool = True,
) -> tuple[list[CanFrame], CanFrame | None]:
"""Send bounded motion and attempt all three stop frames in a finally path."""
@@ -1313,10 +1341,14 @@ class NativeCanExecutor:
f"{len(stop_errors)}/3 forced-stop frame(s) failed"
)
raise RuntimeError("; ".join(details))
- return sent, self._wait_feedback(bus, command)
+ return sent, self._wait_feedback(bus, command) if wait_for_feedback else None
def _execute_driver_seat_memory(
- self, bus: CanBus, command: dict[str, Any]
+ self,
+ bus: CanBus,
+ command: dict[str, Any],
+ *,
+ wait_for_completion: bool = True,
) -> tuple[list[CanFrame], CanFrame | None]:
"""Require fresh initialization, then replay the learned 16/15 sequence."""
@@ -1366,8 +1398,12 @@ class NativeCanExecutor:
)
if index + 1 < len(sequence):
time.sleep(int(special["period_ms"]) / 1_000)
- feedback = self._wait_for_id(
- bus, 0x494, int(special["post_status_timeout_ms"])
+ feedback = (
+ self._wait_for_id(
+ bus, 0x494, int(special["post_status_timeout_ms"])
+ )
+ if wait_for_completion
+ else None
)
return sent, feedback
diff --git a/src/cognitive_agent/application/seat_service.py b/src/cognitive_agent/application/seat_service.py
index 5a8db50..68bc803 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.abc import Callable
from concurrent.futures import ThreadPoolExecutor
from copy import deepcopy
from datetime import UTC, datetime
@@ -177,7 +178,7 @@ UPDATE_MODE_PROFILE_PREFIX = re.compile(
UPDATE_MODE_PROFILE_BA = re.compile(
r"^(?:请(?:帮我)?|帮我)?(?:把|将)"
r"(?P[^，。,.!！\s]{1,64}?模式)"
- r"(?:的)?(?P.+)$"
+ r"的(?P.+)$"
)
RESTORE_MODE_PROFILE_PREFIX = re.compile(
r"^(?:请(?:帮我)?|帮我)?(?:恢复|还原|重置|初始化)"
@@ -1066,6 +1067,340 @@ class SeatDemoService:
return MqttOtaSeatExecutor(executor, resolved_bus_factory)
return executor
+ def preview_native_can(
+ self,
+ utterance: str,
+ provider: str = "auto",
+ conversation_id: str | None = None,
+ execution_mode: str = "dry_run",
+ approval_token: str | None = None,
+ vehicle_id: str | None = None,
+ simulation_fault: str = "normal",
+ *,
+ hardware_client_id: str | None = None,
+ hardware_access_bypass: bool = True,
+ reasoning_effort: str | None = None,
+ control_transport: str = WIRED_CAN_TRANSPORT,
+ on_commands_validated: Callable[[dict[str, Any]], None] | None = None,
+ ) -> dict[str, Any]:
+ """Validate and execute deterministic controls before hosted parsing."""
+
+ utterance = utterance.strip()
+ if not utterance:
+ raise ValueError("utterance must not be empty")
+ if provider not in {"auto", "openai", "local"}:
+ raise ValueError("provider must be auto, openai, or local")
+ conversation_id = conversation_id or f"conv-{uuid.uuid4().hex}"
+ if not CONVERSATION_ID_PATTERN.fullmatch(conversation_id):
+ raise ValueError(
+ "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,
+ )
+ )
+ if control_plane_request:
+ result = self.process(
+ utterance,
+ provider=provider,
+ conversation_id=conversation_id,
+ execution_mode=execution_mode,
+ approval_token=approval_token,
+ vehicle_id=vehicle_profile.id,
+ simulation_fault=simulation_fault,
+ hardware_client_id=hardware_client_id,
+ hardware_access_bypass=hardware_access_bypass,
+ reasoning_effort=reasoning_effort,
+ control_transport=control_transport,
+ _on_commands_validated=on_commands_validated,
+ )
+ result.update(
+ {
+ "phase": "control_plane_complete",
+ "preview_request_id": result["request_id"],
+ "unresolved_clauses": [],
+ "supplement_required": False,
+ }
+ )
+ self._persist(result)
+ return result
+ context = self._context_for_vehicle(
+ self.conversations.snapshot(conversation_id),
+ vehicle_profile,
+ )
+ clause_results = self._local_clause_results(utterance, context)
+ unresolved_clauses = [
+ clause
+ for clause, _, status in clause_results
+ if status != "complete"
+ ]
+ started = time.perf_counter()
+
+ def publish_local_commands(validated: dict[str, Any]) -> None:
+ if on_commands_validated is None:
+ return
+ event = deepcopy(validated)
+ event.update(
+ {
+ "phase": "deterministic_commands_validated",
+ "provider_requested": provider,
+ "unresolved_clauses": unresolved_clauses,
+ "supplement_required": bool(
+ provider != "local" and unresolved_clauses
+ ),
+ }
+ )
+ on_commands_validated(event)
+
+ result = self.process(
+ utterance,
+ provider="local",
+ conversation_id=conversation_id,
+ execution_mode=execution_mode,
+ approval_token=approval_token,
+ vehicle_id=vehicle_profile.id,
+ simulation_fault=simulation_fault,
+ hardware_client_id=hardware_client_id,
+ hardware_access_bypass=hardware_access_bypass,
+ reasoning_effort=reasoning_effort,
+ control_transport=control_transport,
+ _local_preview=True,
+ _allow_empty_control_result=True,
+ _on_commands_validated=publish_local_commands,
+ )
+ elapsed_ms = round((time.perf_counter() - started) * 1000, 3)
+ supplement_required = bool(
+ provider != "local" and unresolved_clauses
+ )
+ result.update(
+ {
+ "phase": "deterministic_preview",
+ "provider_requested": provider,
+ "preview_request_id": result["request_id"],
+ "preview_elapsed_ms": elapsed_ms,
+ "unresolved_clauses": unresolved_clauses,
+ "supplement_required": supplement_required,
+ }
+ )
+ result["decision_summary"] = [
+ *result.get("decision_summary", []),
+ (
+ "本地规则阶段已完成白名单校验并先行处理 "
+ f"{sum(len((config.get('spec') or {}).get('operations', [])) for config in result.get('function_configs', {}).values())} 条 CAN 控制指令。"
+ ),
+ ]
+ self._persist(result)
+ manifest = {
+ "version": 1,
+ "preview_request_id": result["request_id"],
+ "utterance": utterance,
+ "provider": provider,
+ "conversation_id": conversation_id,
+ "execution_mode": execution_mode,
+ "vehicle_id": vehicle_profile.id,
+ "simulation_fault": simulation_fault,
+ "hardware_client_id": hardware_client_id,
+ "reasoning_effort": reasoning_effort,
+ "control_transport": control_transport,
+ "unresolved_clauses": unresolved_clauses,
+ "preview_control_signatures": result.get(
+ "parsed_control_signatures", []
+ ),
+ }
+ self._atomic_write(
+ self.artifact_root / result["request_id"] / "preview.json",
+ json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
+ )
+ return result
+
+ def supplement_native_can(
+ self,
+ preview_request_id: str,
+ *,
+ provider: str | None = None,
+ approval_token: str | None = None,
+ hardware_client_id: str | None = None,
+ hardware_access_bypass: bool = True,
+ reasoning_effort: str | None = None,
+ on_commands_validated: Callable[[dict[str, Any]], None] | None = None,
+ ) -> dict[str, Any]:
+ """Parse only unresolved clauses and merge them with the local result."""
+
+ if not MODE_EDITOR_REQUEST_ID_PATTERN.fullmatch(preview_request_id):
+ raise ValueError("invalid preview request id")
+ preview_dir = self.artifact_root / preview_request_id
+ try:
+ manifest = json.loads(
+ (preview_dir / "preview.json").read_text(encoding="utf-8")
+ )
+ preview = json.loads(
+ (preview_dir / "result.json").read_text(encoding="utf-8")
+ )
+ except (OSError, json.JSONDecodeError) as exc:
+ raise ValueError("native CAN preview not found") from exc
+ if (
+ not isinstance(manifest, dict)
+ or manifest.get("version") != 1
+ or manifest.get("preview_request_id") != preview_request_id
+ or not isinstance(preview, dict)
+ or preview.get("request_id") != preview_request_id
+ ):
+ raise ValueError("invalid native CAN preview manifest")
+ selected_provider = str(manifest.get("provider") or "auto")
+ if provider is not None and provider != selected_provider:
+ raise ValueError("supplement provider does not match preview")
+ selected_effort = manifest.get("reasoning_effort")
+ if reasoning_effort is not None and reasoning_effort != selected_effort:
+ raise ValueError("supplement reasoning effort does not match preview")
+ expected_client_id = manifest.get("hardware_client_id")
+ if hardware_client_id != expected_client_id:
+ raise ValueError("supplement hardware client does not match preview")
+ unresolved_clauses = [
+ str(item).strip()
+ for item in manifest.get("unresolved_clauses", [])
+ if str(item).strip()
+ ]
+ if not unresolved_clauses:
+ raise ValueError("native CAN preview has no unresolved clauses")
+ excluded_signatures = {
+ tuple(item)
+ for item in manifest.get("preview_control_signatures", [])
+ if isinstance(item, list)
+ }
+
+ def publish_cloud_commands(validated: dict[str, Any]) -> None:
+ if on_commands_validated is None:
+ return
+ event = self._merge_native_can_phases(preview, validated)
+ event.update(
+ {
+ "phase": "supplement_commands_validated",
+ "preview_request_id": preview_request_id,
+ "supplement_of": preview_request_id,
+ "supplement_required": False,
+ }
+ )
+ on_commands_validated(event)
+
+ supplement = self.process(
+ "，".join(unresolved_clauses),
+ provider=selected_provider,
+ conversation_id=str(manifest["conversation_id"]),
+ execution_mode=str(manifest.get("execution_mode") or "dry_run"),
+ approval_token=approval_token,
+ vehicle_id=str(manifest["vehicle_id"]),
+ simulation_fault=str(manifest.get("simulation_fault") or "normal"),
+ hardware_client_id=hardware_client_id,
+ hardware_access_bypass=hardware_access_bypass,
+ reasoning_effort=(
+ str(selected_effort) if selected_effort is not None else None
+ ),
+ control_transport=str(
+ manifest.get("control_transport") or WIRED_CAN_TRANSPORT
+ ),
+ _excluded_control_signatures=excluded_signatures,
+ _model_parse_source="cloud_supplement",
+ _allow_empty_control_result=True,
+ _on_commands_validated=publish_cloud_commands,
+ )
+ merged = self._merge_native_can_phases(preview, supplement)
+ merged.update(
+ {
+ "phase": "two_stage_complete",
+ "utterance": str(manifest["utterance"]),
+ "provider_requested": selected_provider,
+ "preview_request_id": preview_request_id,
+ "supplement_of": preview_request_id,
+ "supplement_request_id": supplement["request_id"],
+ "supplement_required": False,
+ "unresolved_clauses": unresolved_clauses,
+ }
+ )
+ self._persist(merged)
+ return merged
+
+ @staticmethod
+ def _merge_native_can_phases(
+ preview: dict[str, Any],
+ supplement: dict[str, Any],
+ ) -> dict[str, Any]:
+ """Create one durable result without collapsing same-domain phases."""
+
+ merged = deepcopy(supplement)
+
+ def phase_documents(
+ local_documents: dict[str, Any] | None,
+ cloud_documents: dict[str, Any] | None,
+ ) -> dict[str, Any]:
+ return {
+ **{
+ f"local_{key}": deepcopy(value)
+ for key, value in (local_documents or {}).items()
+ },
+ **{
+ f"cloud_{key}": deepcopy(value)
+ for key, value in (cloud_documents or {}).items()
+ },
+ }
+
+ configs = phase_documents(
+ preview.get("function_configs"),
+ supplement.get("function_configs"),
+ )
+ plans = phase_documents(
+ preview.get("command_plans"),
+ supplement.get("command_plans"),
+ )
+ merged["function_configs"] = configs
+ merged["command_plans"] = plans
+ merged["function_config"] = next(iter(configs.values()), None)
+ merged["command_plan"] = next(iter(plans.values()), None)
+ merged["decision_summary"] = [
+ *preview.get("decision_summary", []),
+ *supplement.get("decision_summary", []),
+ ]
+ merged["trace_events"] = [
+ *preview.get("trace_events", []),
+ *supplement.get("trace_events", []),
+ ]
+ merged["validation"] = [
+ *preview.get("validation", []),
+ *supplement.get("validation", []),
+ ]
+ merged["parsed_control_signatures"] = [
+ *preview.get("parsed_control_signatures", []),
+ *supplement.get("parsed_control_signatures", []),
+ ]
+ merged["phase_execution_reports"] = {
+ "local_rule": deepcopy(preview.get("execution_report")),
+ "cloud_supplement": deepcopy(supplement.get("execution_report")),
+ }
+ if supplement.get("execution_report") is None:
+ merged["execution_report"] = deepcopy(preview.get("execution_report"))
+ successful = {"succeeded", "partially_succeeded", "no_action"}
+ preview_ok = preview.get("status") in successful
+ supplement_ok = supplement.get("status") in successful
+ has_any_commands = bool(configs and plans)
+ if preview_ok and supplement_ok:
+ merged["status"] = "succeeded" if has_any_commands else "no_action"
+ elif preview_ok and preview.get("function_configs"):
+ merged["status"] = "partially_succeeded"
+ merged["local_preview"] = {
+ "request_id": preview.get("request_id"),
+ "elapsed_ms": preview.get("preview_elapsed_ms"),
+ "command_count": sum(
+ len((config.get("spec") or {}).get("operations", []))
+ for config in preview.get("function_configs", {}).values()
+ ),
+ }
+ return merged
+
def process(
self,
utterance: str,
@@ -1080,6 +1415,11 @@ class SeatDemoService:
hardware_access_bypass: bool = True,
reasoning_effort: str | None = None,
control_transport: str = WIRED_CAN_TRANSPORT,
+ _local_preview: bool = False,
+ _excluded_control_signatures: set[tuple[Any, ...]] | None = None,
+ _model_parse_source: str = "cloud_model",
+ _allow_empty_control_result: bool = False,
+ _on_commands_validated: Callable[[dict[str, Any]], None] | None = None,
) -> dict[str, Any]:
control_transport = self._validate_control_transport(control_transport)
# Creating a reusable scene is a configuration workflow. Synthesize
@@ -1099,6 +1439,11 @@ class SeatDemoService:
simulation_fault,
reasoning_effort,
control_transport,
+ _local_preview,
+ _excluded_control_signatures,
+ _model_parse_source,
+ _allow_empty_control_result,
+ _on_commands_validated,
)
if hardware_access_bypass or self._driver_only_vehicle_execution_active(
control_transport
@@ -1114,6 +1459,11 @@ class SeatDemoService:
simulation_fault,
reasoning_effort,
control_transport,
+ _local_preview,
+ _excluded_control_signatures,
+ _model_parse_source,
+ _allow_empty_control_result,
+ _on_commands_validated,
)
authorized_client_id = self.hardware_access.ensure_authorized(hardware_client_id)
with self.hardware_execution.reserve():
@@ -1128,6 +1478,11 @@ class SeatDemoService:
simulation_fault,
reasoning_effort,
control_transport,
+ _local_preview,
+ _excluded_control_signatures,
+ _model_parse_source,
+ _allow_empty_control_result,
+ _on_commands_validated,
)
def _process_once(
@@ -1141,6 +1496,11 @@ class SeatDemoService:
simulation_fault: str = "normal",
reasoning_effort: str | None = None,
control_transport: str = WIRED_CAN_TRANSPORT,
+ local_preview: bool = False,
+ excluded_control_signatures: set[tuple[Any, ...]] | None = None,
+ model_parse_source: str = "cloud_model",
+ allow_empty_control_result: bool = False,
+ on_commands_validated: Callable[[dict[str, Any]], None] | None = None,
) -> dict[str, Any]:
control_transport = self._validate_control_transport(control_transport)
utterance = utterance.strip()
@@ -1298,6 +1658,8 @@ class SeatDemoService:
context,
vehicle_profile,
reasoning_effort=reasoning_effort,
+ local_preview=local_preview,
+ model_parse_source=model_parse_source,
)
if self._is_dynamic_mode_stop_request(utterance):
parsed = self._apply_dynamic_mode_stop_request(parsed, utterance)
@@ -1305,6 +1667,7 @@ class SeatDemoService:
parsed = self._apply_mode_profiles(
parsed, matched_modes, context, vehicle_profile
)
+ parsed = self._mark_control_parse_source(parsed, "local_rule")
relative_resolution = (
self.native_can_relative_resolver.resolve(utterance, context)
if self.native_can_relative_resolver.matches(utterance)
@@ -1323,6 +1686,15 @@ class SeatDemoService:
relative_resolution.claimed_controls if relative_resolution else None
),
)
+ if local_preview:
+ parsed["ambiguities"] = []
+ parsed["unresolved_spans"] = []
+ parsed["model_requests"] = []
+ if excluded_control_signatures:
+ parsed = self._filter_excluded_controls(
+ parsed,
+ excluded_control_signatures,
+ )
parsed = apply_language_safety_guards(parsed, utterance)
parsed = self._partition_native_actions(
parsed,
@@ -1335,6 +1707,47 @@ class SeatDemoService:
)
),
)
+ result["parsed_control_signatures"] = [
+ list(self._control_signature(action))
+ for action in [
+ *parsed.get("actions", []),
+ *parsed.get("domain_actions", []),
+ *parsed.get("native_can_actions", []),
+ ]
+ ]
+ has_control = any(
+ parsed.get(key)
+ for key in ("actions", "domain_actions", "native_can_actions")
+ )
+ if allow_empty_control_result and not has_control:
+ result["provider_used"] = provider_used
+ result["decision_summary"] = [
+ *parsed.get("decision_summary", []),
+ "当前阶段没有新增的已验证 CAN 控制指令。",
+ ]
+ result["deferred_requests"] = parsed.get("deferred_requests", [])
+ result["model_requests"] = parsed.get("model_requests", [])
+ result["intent_ir"] = build_intent_ir(
+ parsed,
+ utterance,
+ request_id,
+ provider_used,
+ self.semantic_capability_metadata,
+ )
+ result["validation"] = self.contracts.validate(result["intent_ir"])
+ result["status"] = "no_action"
+ self._trace(
+ traces,
+ request_id,
+ "intent_parse",
+ "succeeded",
+ "当前阶段未产生新的白名单控制动作。",
+ started,
+ {"provider": provider_used, "control_count": 0},
+ domain="orchestration",
+ )
+ self._persist(result)
+ return result
if hardware_request and not control_plane_request:
if (
control_transport == WIRELESS_MQTT_TRANSPORT
@@ -1505,6 +1918,7 @@ class SeatDemoService:
simulation_fault=simulation_fault,
vehicle_profile=vehicle_profile,
control_transport=control_transport,
+ on_commands_validated=on_commands_validated,
)
intent_ir = build_intent_ir(
@@ -1776,6 +2190,8 @@ class SeatDemoService:
self._persist(result)
return result
+ self._emit_validated_commands(result, on_commands_validated)
+
safety_started = time.perf_counter()
self._trace(
traces,
@@ -3395,6 +3811,21 @@ class SeatDemoService:
result["deferred_requests"] = deferred_requests
return result
+ @staticmethod
+ def _emit_validated_commands(
+ result: dict[str, Any],
+ callback: Callable[[dict[str, Any]], None] | None,
+ ) -> None:
+ """Expose fully compiled commands before physical execution begins."""
+
+ if callback is None:
+ return
+ event = deepcopy(result)
+ event["status"] = "processing"
+ event["phase"] = "commands_validated"
+ event["execution_report"] = None
+ callback(event)
+
def _process_orchestrated(
self,
utterance: str,
@@ -3411,6 +3842,7 @@ class SeatDemoService:
simulation_fault: str = "normal",
vehicle_profile: VehicleProfile | None = None,
control_transport: str = WIRED_CAN_TRANSPORT,
+ on_commands_validated: Callable[[dict[str, Any]], None] | None = None,
) -> dict[str, Any]:
"""Run isolated adapters under the selected physical-bus policy."""
vehicle_profile = vehicle_profile or self.vehicle_profiles.default()
@@ -3669,6 +4101,8 @@ class SeatDemoService:
self._persist(result)
return result
+ self._emit_validated_commands(result, on_commands_validated)
+
self._trace(
traces,
request_id,
@@ -3842,7 +4276,7 @@ class SeatDemoService:
"execute",
"succeeded" if execution_succeeded and not report_issues else "failed",
(
- "无线 MQTT 白名单控制帧已发布；车辆实际状态以状态话题回读为准。"
+ "无线 MQTT 白名单控制帧已提交；车辆实际状态以状态话题回读为准。"
if (
hardware_request
and control_transport == WIRELESS_MQTT_TRANSPORT
@@ -4385,16 +4819,29 @@ class SeatDemoService:
)
return merged
- def _interpret(
+ @staticmethod
+ def _mark_control_parse_source(
+ parsed: dict[str, Any],
+ source: str,
+ *,
+ overwrite: bool = False,
+ ) -> dict[str, Any]:
+ """Attach per-control provenance without changing semantic content."""
+
+ marked = deepcopy(parsed)
+ for key in ("actions", "domain_actions", "native_can_actions"):
+ for action in marked.get(key, []):
+ if overwrite or not action.get("parse_source"):
+ action["parse_source"] = source
+ return marked
+
+ def _local_clause_results(
self,
utterance: str,
- provider: str,
context: dict[str, Any],
- vehicle_profile: VehicleProfile | None = None,
- *,
- reasoning_effort: str | None = None,
- ) -> tuple[dict[str, Any], str, str | None]:
- vehicle_profile = vehicle_profile or self.vehicle_profiles.default()
+ ) -> list[tuple[str, dict[str, Any], str]]:
+ """Parse each clause deterministically and preserve inherited seat scope."""
+
clause_results: list[tuple[str, dict[str, Any], str]] = []
def parse_clause(clause: str) -> dict[str, Any]:
@@ -4413,7 +4860,7 @@ class SeatDemoService:
relative.claimed_controls if relative else None
),
)
- return local
+ return self._mark_control_parse_source(local, "local_rule")
def has_seat_domain(parsed: dict[str, Any]) -> bool:
if parsed.get("actions") or "seat" in parsed.get("detected_domains", []):
@@ -4441,12 +4888,6 @@ class SeatDemoService:
if clause_has_seat and explicit_driver != explicit_passenger:
inherited_seat_scope = "driver" if explicit_driver else "passenger"
elif clause_has_seat and inherited_seat_scope:
- # Chinese compound controls commonly state the seat once and
- # then list axes or thermal functions after commas. Carry the
- # last explicit seat scope only into later seat clauses; this
- # prevents a passenger request from silently falling through
- # to the CH1 driver default while leaving climate/door/window
- # clauses untouched.
scope_label = (
"主驾座椅"
if inherited_seat_scope == "driver"
@@ -4457,10 +4898,59 @@ class SeatDemoService:
clause_results.append(
(clause, local, self._local_clause_status(local, clause))
)
+ return clause_results
+
+ @classmethod
+ def _trusted_local_preview(
+ cls,
+ clause_results: list[tuple[str, dict[str, Any], str]],
+ ) -> dict[str, Any]:
+ """Keep deterministic controls while deferring incomplete text to GPT."""
+
+ trusted = cls._merge_parsed_results(
+ [parsed for _, parsed, status in clause_results if status != "unparsed"]
+ )
+ # Ambiguities and unresolved spans belong to the supplement phase. A
+ # deterministic action still goes through the normal schema, compiler,
+ # whitelist and physical execution gates before it can be sent.
+ trusted["ambiguities"] = []
+ trusted["unresolved_spans"] = []
+ trusted["model_requests"] = []
+ return trusted
+
+ @classmethod
+ def _filter_excluded_controls(
+ cls,
+ parsed: dict[str, Any],
+ excluded: set[tuple[Any, ...]],
+ ) -> dict[str, Any]:
+ filtered = deepcopy(parsed)
+ for key in ("actions", "domain_actions", "native_can_actions"):
+ filtered[key] = [
+ action
+ for action in filtered.get(key, [])
+ if cls._control_signature(action) not in excluded
+ ]
+ return filtered
+
+ def _interpret(
+ self,
+ utterance: str,
+ provider: str,
+ context: dict[str, Any],
+ vehicle_profile: VehicleProfile | None = None,
+ *,
+ reasoning_effort: str | None = None,
+ local_preview: bool = False,
+ model_parse_source: str = "cloud_model",
+ ) -> tuple[dict[str, Any], str, str | None]:
+ vehicle_profile = vehicle_profile or self.vehicle_profiles.default()
+ clause_results = self._local_clause_results(utterance, context)
trusted_local = self._merge_parsed_results(
[parsed for _, parsed, status in clause_results if status != "unparsed"]
)
+ preview_local = self._trusted_local_preview(clause_results)
all_local = self._merge_parsed_results(
[parsed for _, parsed, _ in clause_results]
)
@@ -4474,7 +4964,13 @@ class SeatDemoService:
safe_fallback = trusted_local if trusted_controls else all_local
if provider == "local" or not fallback_clauses:
return (
- all_local if provider == "local" else trusted_local,
+ (
+ preview_local
+ if provider == "local" and local_preview
+ else all_local
+ if provider == "local"
+ else trusted_local
+ ),
"local_rules",
None,
)
@@ -4505,6 +5001,11 @@ class SeatDemoService:
vehicle_profile,
utterance=fallback_utterance,
)
+ model = self._mark_control_parse_source(
+ model,
+ model_parse_source,
+ overwrite=True,
+ )
return (
self._merge_local_and_model_intent(trusted_local, model),
f"local_rules+openai:{self.openai.model}",
diff --git a/src/cognitive_agent/infrastructure/can/mqtt_transport.py b/src/cognitive_agent/infrastructure/can/mqtt_transport.py
index a8f8b26..336f2f2 100644
--- a/src/cognitive_agent/infrastructure/can/mqtt_transport.py
+++ b/src/cognitive_agent/infrastructure/can/mqtt_transport.py
@@ -411,6 +411,10 @@ def mqtt_control_payload(
"data": frame.data.hex(" ").upper(),
"repeat": 1,
"intervalMs": 0,
+ # Vehicle state is consumed asynchronously from can/state by the
+ # Cognitive Agent. Do not serialize Android's command worker on a
+ # per-command response wait as well.
+ "expectResponse": False,
"responseTimeoutMs": bounded_timeout_ms,
},
ensure_ascii=False,
@@ -788,14 +792,19 @@ class MqttCanTransport:
)
if int(getattr(result, "rc", 0)) != 0:
raise CanGatewayError(f"MQTT CAN publish failed with code {result.rc}")
- if hasattr(result, "wait_for_publish"):
- published = result.wait_for_publish(timeout=2.0)
- if published is False:
- raise CanGatewayError("MQTT CAN publish confirmation timed out")
+ # Paho already accepted the payload into its network-loop queue when
+ # publish() returns rc == MQTT_ERR_SUCCESS. Waiting here serializes
+ # every repeated CAN frame and previously added up to two seconds per
+ # frame. Observe an already-available acknowledgement without ever
+ # blocking the control pipeline.
+ is_published = getattr(result, "is_published", None)
+ broker_confirmed = bool(is_published()) if callable(is_published) else False
return {
"topic": self.settings.control_topic,
"qos": self.settings.qos,
"retain": False,
+ "delivery_status": "published" if broker_confirmed else "queued",
+ "message_id": getattr(result, "mid", None),
"payload": json.loads(raw_payload),
}
@@ -876,7 +885,7 @@ class MqttCanBus:
@property
def published_messages(self) -> tuple[dict[str, Any], ...]:
- """Return this bus view's confirmed MQTT publications in send order."""
+ """Return this bus view's accepted MQTT publications in send order."""
return tuple(deepcopy(self._published_messages))
diff --git a/src/cognitive_agent/server.py b/src/cognitive_agent/server.py
index bd1cbd6..03805c7 100755
--- a/src/cognitive_agent/server.py
+++ b/src/cognitive_agent/server.py
@@ -362,7 +362,14 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
if path.startswith("/api/mode-profiles/"):
self._handle_mode_profile_post(path)
return
- if path != "/api/requests":
+ supplement_match = re.fullmatch(
+ r"/api/requests/(?Preq-[a-z0-9-]+)/supplement",
+ path,
+ )
+ if (
+ path not in {"/api/requests", "/api/requests/preview"}
+ and not supplement_match
+ ):
self._json({"error": "not_found"}, HTTPStatus.NOT_FOUND)
return
@@ -397,40 +404,82 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
payload = self._read_json_body()
utterance = str(payload.get("utterance", ""))
source = str(payload.get("client_type", "head_unit"))
- self.vehicle_display_sync.update_task(
- status="running",
- utterance=utterance,
- source=source,
+ if utterance:
+ self.vehicle_display_sync.update_task(
+ status="running",
+ utterance=utterance,
+ source=source,
+ )
+ hardware_client_id = (
+ str(payload["hardware_client_id"])
+ if payload.get("hardware_client_id")
+ else None
)
- result = self.service.process(
- utterance,
- str(payload.get("provider", "auto")),
- str(payload["conversation_id"]) if payload.get("conversation_id") else None,
- str(payload.get("execution_mode", "dry_run")),
+ reasoning_effort = (
+ str(payload["reasoning_effort"])
+ if payload.get("reasoning_effort") is not None
+ else None
+ )
+ common_arguments = {
+ "utterance": utterance,
+ "provider": str(payload.get("provider", "auto")),
+ "conversation_id": (
+ str(payload["conversation_id"])
+ if payload.get("conversation_id")
+ else None
+ ),
+ "execution_mode": str(
+ payload.get("execution_mode", "dry_run")
+ ),
# Hardware authorization is a server deployment policy. The
- # browser no longer renders or transmits an approval secret.
- approval_token=os.getenv("COGNITIVE_AGENT_HARDWARE_APPROVAL_TOKEN"),
- vehicle_id=(
+ # browser never receives or transmits this secret.
+ "approval_token": os.getenv(
+ "COGNITIVE_AGENT_HARDWARE_APPROVAL_TOKEN"
+ ),
+ "vehicle_id": (
str(payload["vehicle_id"])
if payload.get("vehicle_id")
else None
),
- simulation_fault=str(payload.get("simulation_fault", "normal")),
- hardware_client_id=(
- str(payload["hardware_client_id"])
- if payload.get("hardware_client_id")
- else None
+ "simulation_fault": str(
+ payload.get("simulation_fault", "normal")
),
- hardware_access_bypass=self._is_unproxied_loopback_request(),
- reasoning_effort=(
- str(payload["reasoning_effort"])
- if payload.get("reasoning_effort") is not None
- else None
- ),
- control_transport=str(
+ "hardware_client_id": hardware_client_id,
+ "hardware_access_bypass": self._is_unproxied_loopback_request(),
+ "reasoning_effort": reasoning_effort,
+ "control_transport": str(
payload.get("control_transport", "wired_can")
),
- )
+ }
+ if "application/x-ndjson" in self.headers.get("Accept", ""):
+ self._stream_control_request(
+ path,
+ supplement_match,
+ payload,
+ common_arguments,
+ source=source,
+ )
+ return
+ if path == "/api/requests/preview":
+ result = self.service.preview_native_can(**common_arguments)
+ elif supplement_match:
+ result = self.service.supplement_native_can(
+ supplement_match.group("request_id"),
+ provider=(
+ str(payload["provider"])
+ if payload.get("provider")
+ else None
+ ),
+ approval_token=os.getenv(
+ "COGNITIVE_AGENT_HARDWARE_APPROVAL_TOKEN"
+ ),
+ hardware_client_id=hardware_client_id,
+ hardware_access_bypass=self._is_unproxied_loopback_request(),
+ reasoning_effort=reasoning_effort,
+ )
+ else:
+ result = self.service.process(**common_arguments)
+ utterance = str(result.get("utterance") or utterance)
self.vehicle_display_sync.update_task(
status=str(result.get("status", "succeeded")),
utterance=utterance,
@@ -489,6 +538,146 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
HTTPStatus.INTERNAL_SERVER_ERROR,
)
+ def _stream_control_request(
+ self,
+ path: str,
+ supplement_match: re.Match[str] | None,
+ payload: dict[str, object],
+ common_arguments: dict[str, object],
+ *,
+ source: str,
+ ) -> None:
+ """Stream validated command plans before physical execution finishes."""
+
+ self.send_response(HTTPStatus.OK)
+ self.send_header("Content-Type", "application/x-ndjson; charset=utf-8")
+ self.send_header("Cache-Control", "no-store")
+ self.send_header("X-Accel-Buffering", "no")
+ self._send_security_headers()
+ self.end_headers()
+ stream_open = True
+
+ def emit(event: dict[str, object]) -> None:
+ nonlocal stream_open
+ if not stream_open:
+ return
+ content = (json.dumps(event, ensure_ascii=False) + "\n").encode(
+ "utf-8"
+ )
+ try:
+ self.wfile.write(content)
+ self.wfile.flush()
+ except (BrokenPipeError, ConnectionResetError):
+ # The submitted control request must continue even if the
+ # browser closes while the physical execution is in progress.
+ stream_open = False
+
+ def on_commands_validated(result: dict[str, object]) -> None:
+ emit({"type": "validated_commands", "payload": result})
+
+ utterance = str(common_arguments.get("utterance") or "")
+ try:
+ if path == "/api/requests/preview":
+ result = self.service.preview_native_can(
+ **common_arguments,
+ on_commands_validated=on_commands_validated,
+ )
+ elif supplement_match:
+ result = self.service.supplement_native_can(
+ supplement_match.group("request_id"),
+ provider=(
+ str(payload["provider"])
+ if payload.get("provider")
+ else None
+ ),
+ approval_token=os.getenv(
+ "COGNITIVE_AGENT_HARDWARE_APPROVAL_TOKEN"
+ ),
+ hardware_client_id=(
+ str(payload["hardware_client_id"])
+ if payload.get("hardware_client_id")
+ else None
+ ),
+ hardware_access_bypass=self._is_unproxied_loopback_request(),
+ reasoning_effort=(
+ str(payload["reasoning_effort"])
+ if payload.get("reasoning_effort") is not None
+ else None
+ ),
+ on_commands_validated=on_commands_validated,
+ )
+ else:
+ result = self.service.process(
+ **common_arguments,
+ _on_commands_validated=on_commands_validated,
+ )
+ utterance = str(result.get("utterance") or utterance)
+ self.vehicle_display_sync.update_task(
+ status=str(result.get("status", "succeeded")),
+ utterance=utterance,
+ source=source,
+ result=result,
+ )
+ emit({"type": "result", "payload": result})
+ except HardwareAccessRequiredError as exc:
+ self.vehicle_display_sync.update_task(
+ status="failed",
+ utterance=utterance,
+ source=source,
+ error={"code": "hardware_access_required", "message": str(exc)},
+ )
+ emit({
+ "type": "error",
+ "error": "hardware_access_required",
+ "message": str(exc),
+ "access_state": exc.state,
+ })
+ except HardwareBusyError as exc:
+ self.vehicle_display_sync.update_task(
+ status="failed",
+ utterance=utterance,
+ source=source,
+ error={"code": "hardware_busy", "message": str(exc)},
+ )
+ emit({
+ "type": "error",
+ "error": "hardware_busy",
+ "message": str(exc),
+ "retry_after_seconds": exc.retry_after_seconds,
+ })
+ except HardwareSafetyError as exc:
+ self.vehicle_display_sync.update_task(
+ status="failed",
+ utterance=utterance,
+ source=source,
+ error={"code": "hardware_safety_gate", "message": str(exc)},
+ )
+ emit({
+ "type": "error",
+ "error": "hardware_safety_gate",
+ "message": str(exc),
+ })
+ except (ValueError, json.JSONDecodeError) as exc:
+ emit({
+ "type": "error",
+ "error": "invalid_request",
+ "message": str(exc),
+ })
+ except OpenAIIntentError as exc:
+ self.log_error("OpenAI intent request failed: %s", exc)
+ emit({
+ "type": "error",
+ "error": "openai_request_failed",
+ "message": "GPT 意图解析失败，请检查 API Key、模型权限或网络状态。",
+ })
+ except Exception as exc: # noqa: BLE001 - stream a terminal event
+ self.log_error("streaming request failed: %s", exc)
+ emit({
+ "type": "error",
+ "error": "internal_error",
+ "message": "请求处理失败，请查看服务端日志。",
+ })
+
def do_DELETE(self) -> None: # noqa: N802 - stdlib handler API
parsed_url = urlparse(self.path)
path = unquote(parsed_url.path)
diff --git a/tests/e2e/test_seat_web_api.py b/tests/e2e/test_seat_web_api.py
index b073e08..e08bb3c 100755
--- a/tests/e2e/test_seat_web_api.py
+++ b/tests/e2e/test_seat_web_api.py
@@ -332,18 +332,18 @@ 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-101", timeout=3) as response:
+ with self.opener.open(f"{self.base_url}/app.js?v=dual-seat-103", 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
) as response:
gesture_script = response.read().decode("utf-8")
with self.opener.open(
- f"{self.base_url}/console-v2.js?v=sync-layout-41", timeout=3
+ f"{self.base_url}/console-v2.js?v=sync-layout-43", timeout=3
) as response:
console_script = response.read().decode("utf-8")
with self.opener.open(
- f"{self.base_url}/console-v2.css?v=sync-layout-40", timeout=3
+ f"{self.base_url}/console-v2.css?v=sync-layout-41", timeout=3
) as response:
console_styles = response.read().decode("utf-8")
with self.opener.open(
@@ -378,12 +378,12 @@ 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-101"', html)
+ self.assertIn('src="/app.js?v=dual-seat-103"', html)
self.assertIn('href="/styles.css?v=dual-seat-16"', html)
self.assertIn('src="/gesture-control.js?v=android-pad-9"', html)
self.assertNotIn('class="vehicle-system-hud"', html)
- self.assertIn('src="/console-v2.js?v=sync-layout-41"', html)
- self.assertIn('href="/console-v2.css?v=sync-layout-40"', html)
+ self.assertIn('src="/console-v2.js?v=sync-layout-43"', html)
+ self.assertIn('href="/console-v2.css?v=sync-layout-41"', html)
self.assertIn('', html)
self.assertIn('id="v2BootScreen"', html)
self.assertIn('正在加载车机控制与实时状态', html)
@@ -612,7 +612,7 @@ operations:
self.assertIn('client_type: "web_console"', app_script)
self.assertIn('canvas.addEventListener("webglcontextlost"', app_script)
self.assertIn('"cognitive-agent:module-change"', app_script)
- self.assertIn('const CACHE_VERSION = "cognitive-agent-pad-v202"', service_worker)
+ self.assertIn('const CACHE_VERSION = "cognitive-agent-pad-v204"', service_worker)
self.assertIn('"/assets/pets/self-extension-cat.png?v=3"', service_worker)
self.assertIn(".then(() => self.skipWaiting())", service_worker)
self.assertNotIn("const cached = await caches.match(request)", service_worker)
@@ -852,7 +852,7 @@ operations:
) as response:
extension_script = response.read().decode("utf-8")
with self.opener.open(
- f"{self.base_url}/app.js?v=dual-seat-101", timeout=3
+ f"{self.base_url}/app.js?v=dual-seat-103", timeout=3
) as response:
app_script = response.read().decode("utf-8")
@@ -1314,6 +1314,134 @@ operations:
self.assertEqual(result["provider_used"], "local_rules_fallback")
self.assertEqual(result["execution_report"]["spec"]["mode"], "dry_run")
+ def test_two_stage_http_api_returns_local_commands_before_cloud_supplement(
+ self,
+ ) -> None:
+ original_api_key = self.service.openai.api_key
+ self.service.openai.api_key = "two-stage-http-test-key"
+ preview_request = urllib.request.Request(
+ f"{self.base_url}/api/requests/preview",
+ method="POST",
+ headers={"Content-Type": "application/json"},
+ data=json.dumps(
+ {
+ "utterance": "空调温度设置为24度，播放音乐",
+ "provider": "openai",
+ "conversation_id": "conv-e2e-two-stage-http",
+ "execution_mode": "dry_run",
+ "reasoning_effort": "medium",
+ "client_type": "web_console",
+ }
+ ).encode("utf-8"),
+ )
+ model_output = {
+ "detected_domains": ["ambient_light"],
+ "requests": [
+ {
+ "domain": "ambient_light",
+ "capability": "ambient_light.set_power",
+ "target_scope": "vehicle",
+ "operation": "set",
+ "duration_ms": None,
+ "parameters": [{"name": "state", "value": "on"}],
+ "reason": "补充播放场景氛围灯。",
+ }
+ ],
+ "ambiguities": [],
+ "decision_summary": ["仅补充本地未解析片段。"],
+ "confidence": 0.9,
+ }
+ try:
+ with patch.object(
+ self.service.openai,
+ "interpret",
+ side_effect=AssertionError("preview must not invoke GPT"),
+ ):
+ with self.opener.open(preview_request, timeout=3) as response:
+ preview = json.loads(response.read().decode("utf-8"))
+ self.assertEqual(response.status, 201)
+ local_operation = next(
+ operation
+ for config in preview["function_configs"].values()
+ for operation in config["spec"]["operations"]
+ )
+ self.assertEqual(local_operation["parse_source"], "local_rule")
+ self.assertTrue(preview["supplement_required"])
+
+ supplement_request = urllib.request.Request(
+ f"{self.base_url}/api/requests/{preview['request_id']}/supplement",
+ method="POST",
+ headers={"Content-Type": "application/json"},
+ data=json.dumps(
+ {
+ "provider": "openai",
+ "reasoning_effort": "medium",
+ "client_type": "web_console",
+ }
+ ).encode("utf-8"),
+ )
+ with patch.object(
+ self.service.openai,
+ "interpret",
+ return_value=model_output,
+ ) as interpret:
+ with self.opener.open(supplement_request, timeout=3) as response:
+ result = json.loads(response.read().decode("utf-8"))
+ self.assertEqual(response.status, 201)
+ self.assertEqual(interpret.call_args.args[0], "播放音乐")
+ finally:
+ self.service.openai.api_key = original_api_key
+
+ self.assertEqual(result["phase"], "two_stage_complete")
+ sources = {
+ operation["capability"]: operation["parse_source"]
+ for config in result["function_configs"].values()
+ for operation in config["spec"]["operations"]
+ }
+ self.assertEqual(sources["climate.set_temperature"], "local_rule")
+ self.assertEqual(sources["ambient_light.set_power"], "cloud_supplement")
+
+ def test_preview_streams_validated_commands_before_terminal_result(self) -> None:
+ request = urllib.request.Request(
+ f"{self.base_url}/api/requests/preview",
+ method="POST",
+ headers={
+ "Content-Type": "application/json",
+ "Accept": "application/x-ndjson",
+ },
+ data=json.dumps(
+ {
+ "utterance": "空调24度、风量2档",
+ "provider": "local",
+ "conversation_id": "conv-e2e-command-stream",
+ "execution_mode": "dry_run",
+ "client_type": "web_console",
+ },
+ ensure_ascii=False,
+ ).encode("utf-8"),
+ )
+
+ with self.opener.open(request, timeout=3) as response:
+ events = [
+ json.loads(line)
+ for line in response.read().decode("utf-8").splitlines()
+ if line.strip()
+ ]
+
+ self.assertEqual(
+ response.headers.get_content_type(),
+ "application/x-ndjson",
+ )
+ self.assertEqual(
+ [event["type"] for event in events],
+ ["validated_commands", "result"],
+ )
+ validated = events[0]["payload"]
+ self.assertEqual(validated["status"], "processing")
+ self.assertIsNone(validated["execution_report"])
+ self.assertTrue(validated["command_plans"])
+ self.assertEqual(events[1]["payload"]["status"], "succeeded")
+
def test_frontend_suite_covers_every_physical_seat_capability(self) -> None:
suite_path = (
REPOSITORY_ROOT
diff --git a/tests/unit/test_can1_semantic_whitelist.py b/tests/unit/test_can1_semantic_whitelist.py
new file mode 100644
index 0000000..97d8052
--- /dev/null
+++ b/tests/unit/test_can1_semantic_whitelist.py
@@ -0,0 +1,351 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from cognitive_agent.application.native_can_intent import (
+ parse_exact_native_can_command,
+)
+from cognitive_agent.application.native_can_runtime import (
+ NativeCanCompileError,
+ NativeCanRuntime,
+)
+from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
+
+
+HARDWARE_DISABLED_ENV = {
+ "COGNITIVE_AGENT_CH1_HIL_ENABLED": "false",
+ "COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED": "false",
+ "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED": "false",
+}
+
+
+class Can1SemanticWhitelistTests(unittest.TestCase):
+ def setUp(self) -> None:
+ self.temporary_directory = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temporary_directory.cleanup)
+ temporary_root = Path(self.temporary_directory.name)
+ memory_path = temporary_root / "driver-seat-memory.json"
+ memory_path.write_text(
+ json.dumps(
+ {
+ "version": 1,
+ "positions": {
+ str(slot): {
+ "payload_15": "15 00 00 00 00 00 00 00",
+ "payload_16": "16 00 00 00 00 00 00 00",
+ }
+ for slot in (1, 2, 3)
+ },
+ }
+ ),
+ encoding="utf-8",
+ )
+ self.environment = patch.dict(
+ os.environ,
+ {
+ **HARDWARE_DISABLED_ENV,
+ "COGNITIVE_AGENT_DRIVER_SEAT_MEMORY_PATH": str(memory_path),
+ },
+ clear=False,
+ )
+ self.environment.start()
+ self.addCleanup(self.environment.stop)
+ self.runtime = NativeCanRuntime(REPOSITORY_ROOT)
+ self.service = SeatDemoService(REPOSITORY_ROOT)
+ self.service.artifact_root = temporary_root / "artifacts"
+
+ @staticmethod
+ def _valid_exact_commands() -> list[str]:
+ commands = [f"door rear {state}" for state in ("lock", "unlock")]
+ commands.extend(
+ f"window {target} {motion}"
+ for target in (
+ "all",
+ "front",
+ "rear",
+ "front-left",
+ "front-right",
+ "rear-left",
+ "rear-right",
+ )
+ for motion in ("open", "close", "stop")
+ )
+ commands.extend(
+ f"window {target} position {position}"
+ for target in (
+ "front-left",
+ "front-right",
+ "rear-left",
+ "rear-right",
+ )
+ for position in (0, 100)
+ )
+ commands.extend(("ac on", "ac off"))
+ commands.extend(f"ac fan {level}" for level in range(1, 8))
+ commands.extend(
+ f"ac mode {mode}"
+ for mode in (
+ "face",
+ "face-feet",
+ "feet",
+ "feet-defrost",
+ "defrost",
+ "face-feet-defrost",
+ "face-defrost",
+ )
+ )
+ commands.extend(
+ f"ac temp {target} {temperature}"
+ for target in ("driver", "passenger", "rear")
+ for temperature in (18, 32)
+ )
+ commands.extend(("ac control manual", "ac control auto"))
+ commands.extend(
+ f"outlet {target} percent {horizontal} {vertical}"
+ for target in (
+ "driver-left",
+ "driver-right",
+ "passenger-left",
+ "passenger-right",
+ )
+ for horizontal, vertical in ((0, 0), (100, 100))
+ )
+ commands.extend(
+ (
+ "outlet driver-left position -25 -40",
+ "outlet driver-right position 30 30",
+ "outlet passenger-left position -10 -40",
+ "outlet passenger-right position 5 30",
+ )
+ )
+ commands.extend(
+ f"seat driver {feature} {level}"
+ for feature in ("vent", "heat")
+ for level in ("off", "low", "high")
+ )
+ commands.append("seat driver massage off")
+ commands.extend(
+ f"seat driver massage strength {strength}"
+ for strength in ("weak", "medium", "strong")
+ )
+ commands.extend(
+ f"seat driver massage mode {mode}"
+ for mode in ("wave", "pulse", "stretch", "slack", "relax")
+ )
+ commands.extend(
+ (
+ "seat driver horizontal forward 100",
+ "seat driver horizontal backward 2000",
+ "seat driver horizontal stop",
+ "seat driver height up 100",
+ "seat driver height down 2000",
+ "seat driver height stop",
+ "seat driver backrest forward 100",
+ "seat driver backrest backward 2000",
+ "seat driver backrest stop",
+ "seat driver basin up 100",
+ "seat driver basin down 2000",
+ "seat driver basin stop",
+ )
+ )
+ commands.extend(f"seat driver position {slot}" for slot in (1, 2, 3))
+ commands.extend(
+ f"seat driver memory recall {slot}" for slot in (1, 2, 3)
+ )
+ commands.extend(f"light indoor {state}" for state in ("on", "off"))
+ commands.extend(
+ f"light reading all {state}" for state in ("on", "off")
+ )
+ commands.extend(f"light ambient {state}" for state in ("on", "off"))
+ commands.extend(f"light ambient color {value}" for value in (1, 127))
+ commands.extend(
+ f"light ambient brightness {value}" for value in (1, 5)
+ )
+ commands.extend(
+ f"shade {motion}" for motion in ("open", "close", "half", "stop")
+ )
+ commands.extend(
+ f"shade position {position}" for position in (0, 1, 50, 99, 100)
+ )
+ return commands
+
+ def _compile_exact(self, command: str) -> dict:
+ actions = parse_exact_native_can_command(command)
+ self.assertEqual(len(actions), 1, command)
+ action = actions[0]
+ config = self.runtime.build_function_config(
+ action["domain"],
+ actions,
+ "req-can-one-whitelist",
+ "local_rules",
+ )
+ return self.runtime.compile(config)
+
+ @staticmethod
+ def _debug_commands(result: dict) -> list[str]:
+ return [
+ command["signals"]["debug_command"]
+ for plan in result.get("command_plans", {}).values()
+ for command in plan["spec"]["commands"]
+ ]
+
+ def test_every_documented_exact_command_parses_and_compiles(self) -> None:
+ commands = self._valid_exact_commands()
+ self.assertEqual(len(commands), 119)
+
+ for command in commands:
+ with self.subTest(command=command):
+ plan = self._compile_exact(command)
+ self.assertEqual(
+ [
+ item["signals"]["debug_command"]
+ for item in plan["spec"]["commands"]
+ ],
+ [command],
+ )
+
+ def test_out_of_whitelist_exact_values_are_rejected(self) -> None:
+ commands = (
+ "door rear open",
+ "window vehicle open",
+ "window all position 50",
+ "window front-left position -1",
+ "window front-left position 101",
+ "ac fan 0",
+ "ac fan 8",
+ "ac mode auto",
+ "ac temp driver 17",
+ "ac temp driver 33",
+ "ac temp driver 24.5",
+ "ac control off",
+ "outlet driver-left percent -1 50",
+ "outlet driver-left percent 101 50",
+ "seat driver vent medium",
+ "seat driver horizontal forward 99",
+ "seat driver horizontal forward 2001",
+ "seat driver horizontal stop 500",
+ "seat driver position 0",
+ "seat driver position 4",
+ "light ambient color 0",
+ "light ambient color 128",
+ "light ambient brightness 0",
+ "light ambient brightness 6",
+ "shade position -1",
+ "shade position 101",
+ )
+ for command in commands:
+ with self.subTest(command=command):
+ self.assertEqual(parse_exact_native_can_command(command), [])
+
+ def test_calibrated_outlet_values_must_fit_the_selected_outlet(self) -> None:
+ action = parse_exact_native_can_command(
+ "outlet driver-left position -26 -40"
+ )[0]
+ config = self.runtime.build_function_config(
+ action["domain"],
+ [action],
+ "req-can-one-calibration",
+ "local_rules",
+ )
+
+ with self.assertRaisesRegex(
+ NativeCanCompileError, "horizontal_value exceeds the calibrated range"
+ ):
+ self.runtime.compile(config)
+
+ def test_representative_chinese_commands_cover_every_whitelist_family(self) -> None:
+ cases = (
+ ("后排车门上锁", ["door rear lock"]),
+ ("后排门锁解锁", ["door rear unlock"]),
+ ("打开全部车窗", ["window all open"]),
+ ("关闭前排车窗", ["window front close"]),
+ ("停止后排车窗", ["window rear stop"]),
+ ("把副驾车窗开到35%", ["window front-right position 35"]),
+ ("打开空调", ["ac on"]),
+ ("关闭空调", ["ac off"]),
+ ("风量调到7档", ["ac fan 7"]),
+ ("空调出风模式设为吹脸", ["ac mode face"]),
+ ("空调出风模式设为吹面和吹脚", ["ac mode face-feet"]),
+ ("空调出风模式设为吹脚", ["ac mode feet"]),
+ ("空调出风模式设为吹脚除霜", ["ac mode feet-defrost"]),
+ ("空调出风模式设为除霜", ["ac mode defrost"]),
+ ("空调出风模式设为吹脸和吹脚除霜", ["ac mode face-feet-defrost"]),
+ ("空调出风模式设为吹脸除霜", ["ac mode face-defrost"]),
+ ("主驾空调温度设为18度", ["ac temp driver 18"]),
+ ("副驾空调温度设为32度", ["ac temp passenger 32"]),
+ ("后排空调温度设为24度", ["ac temp rear 24"]),
+ ("空调设置为自动控制", ["ac control auto"]),
+ ("空调设置为手动控制", ["ac control manual"]),
+ ("主驾左风口水平25%上下75%", ["outlet driver-left percent 25 75"]),
+ ("副驾右风口左右25%垂直75%", ["outlet passenger-right percent 25 75"]),
+ ("主驾左风口标定坐标左右-25上下-40", ["outlet driver-left position -25 -40"]),
+ ("打开主驾座椅通风", ["seat driver vent low"]),
+ ("关闭主驾座椅加热", ["seat driver heat off"]),
+ ("主驾座椅加热高档", ["seat driver heat high"]),
+ ("关闭主驾座椅按摩", ["seat driver massage off"]),
+ ("主驾座椅按摩强度调为中等", ["seat driver massage strength medium"]),
+ ("主驾座椅按摩模式设为放松", ["seat driver massage mode relax"]),
+ ("主驾座椅向前移动500毫秒", ["seat driver horizontal forward 500"]),
+ ("主驾座椅高度向下调500毫秒", ["seat driver height down 500"]),
+ ("主驾座椅靠背后仰800毫秒", ["seat driver backrest backward 800"]),
+ ("把主驾座椅坐盆抬高300毫秒", ["seat driver basin up 300"]),
+ ("调用主驾座椅记忆位置2", ["seat driver position 2"]),
+ ("打开室内灯", ["light indoor on"]),
+ ("关闭全部阅读灯", ["light reading all off"]),
+ ("打开氛围灯", ["light ambient on"]),
+ ("关闭氛围灯", ["light ambient off"]),
+ ("把氛围灯颜色编号调到70", ["light ambient color 70"]),
+ ("氛围灯亮度调整为3档", ["light ambient brightness 3"]),
+ ("打开遮阳帘", ["shade open"]),
+ ("关闭遮阳帘", ["shade close"]),
+ ("遮阳帘半开", ["shade half"]),
+ ("停止遮阳帘", ["shade stop"]),
+ ("遮阳帘开到20%", ["shade position 20"]),
+ ("遮阳帘打开30%", ["shade position 30"]),
+ )
+ for index, (utterance, expected) in enumerate(cases):
+ with self.subTest(utterance=utterance):
+ result = self.service.process(
+ utterance,
+ provider="local",
+ conversation_id=f"conv-can-one-natural-{index}",
+ execution_mode="dry_run",
+ )
+ self.assertEqual(result["status"], "succeeded")
+ self.assertEqual(self._debug_commands(result), expected)
+
+ rear_context = self.service.conversations.snapshot(
+ "conv-can-one-natural-18"
+ )
+ self.assertEqual(
+ rear_context["domain_devices"]["climate"]["zones"]["rear_row"][
+ "temperature_c"
+ ],
+ 24,
+ )
+
+ def test_fractional_values_are_not_silently_truncated(self) -> None:
+ utterances = (
+ "空调温度设为24.5度",
+ "风量设为2.5档",
+ "氛围灯颜色编号调到70.5",
+ "氛围灯亮度调到3.5档",
+ )
+ for index, utterance in enumerate(utterances):
+ with self.subTest(utterance=utterance):
+ result = self.service.process(
+ utterance,
+ provider="local",
+ conversation_id=f"conv-can-one-fraction-{index}",
+ execution_mode="dry_run",
+ )
+ self.assertEqual(self._debug_commands(result), [])
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/unit/test_ch1_default_targets.py b/tests/unit/test_ch1_default_targets.py
index 3e4a49e..22b2a2d 100644
--- a/tests/unit/test_ch1_default_targets.py
+++ b/tests/unit/test_ch1_default_targets.py
@@ -154,6 +154,44 @@ class Ch1DefaultTargetTests(unittest.TestCase):
"front_right",
)
+ def test_driver_and_passenger_window_aliases_map_to_the_correct_side(self) -> None:
+ cases = (
+ ("打开主驾车窗", "front_left", "window.set_motion", {"motion": "open"}),
+ ("打开主驾驶的车窗", "front_left", "window.set_motion", {"motion": "open"}),
+ ("打开副驾车窗", "front_right", "window.set_motion", {"motion": "open"}),
+ ("打开副驾驶的车窗", "front_right", "window.set_motion", {"motion": "open"}),
+ ("把副驾车窗开到35%", "front_right", "window.set_position", {"position_pct": 35}),
+ )
+ for index, (utterance, target, capability, parameters) in enumerate(cases):
+ with self.subTest(utterance=utterance):
+ result = self.service.process(
+ utterance,
+ provider="local",
+ conversation_id=f"conv-window-occupant-alias-{index}",
+ execution_mode="dry_run",
+ )
+
+ self.assertEqual(result["status"], "succeeded")
+ self.assertEqual(set(result["function_configs"]), {"native_window"})
+ operation = result["function_configs"]["native_window"]["spec"][
+ "operations"
+ ][0]
+ self.assertEqual(operation["target"]["zone"], target)
+ self.assertEqual(operation["capability"], capability)
+ self.assertEqual(operation["parameters"], parameters)
+
+ def test_window_target_does_not_leak_from_a_later_seat_subject(self) -> None:
+ parsed = merge_native_can_intent({}, "打开副驾车窗和主驾座椅通风")
+ windows = [
+ action
+ for action in parsed["native_can_actions"]
+ if action["capability"] == "window.set_motion"
+ ]
+
+ self.assertEqual(len(windows), 1)
+ self.assertEqual(windows[0]["target"], "front_right")
+ self.assertEqual(windows[0]["parameters"], {"motion": "open"})
+
def test_group_or_non_driver_seat_scope_is_not_treated_as_unscoped(self) -> None:
for index, utterance in enumerate(
("打开前排座椅加热", "打开后排座椅加热", "打开所有座椅加热")
diff --git a/tests/unit/test_driver_seat_native_can.py b/tests/unit/test_driver_seat_native_can.py
index 4ec28a7..b783f40 100644
--- a/tests/unit/test_driver_seat_native_can.py
+++ b/tests/unit/test_driver_seat_native_can.py
@@ -211,9 +211,17 @@ class DriverSeatNativeCanTests(unittest.TestCase):
{"axis": "height", "direction": "up", "duration_ms": 500},
),
"主驾座椅调整到记忆位置3": (
- "seat.recall_driver_memory_native",
+ "seat.set_driver_position_native",
{"slot": 3},
),
+ "调用主驾座椅记忆位置2": (
+ "seat.set_driver_position_native",
+ {"slot": 2},
+ ),
+ "恢复主驾座椅到二号记忆位置": (
+ "seat.set_driver_position_native",
+ {"slot": 2},
+ ),
}
for utterance, expected in cases.items():
with self.subTest(utterance=utterance):
diff --git a/tests/unit/test_mqtt_can_transport.py b/tests/unit/test_mqtt_can_transport.py
index 5a96b68..3774d33 100644
--- a/tests/unit/test_mqtt_can_transport.py
+++ b/tests/unit/test_mqtt_can_transport.py
@@ -27,9 +27,13 @@ REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
class _PublishResult:
rc = 0
+ mid = 17
+
+ def is_published(self) -> bool:
+ return True
def wait_for_publish(self, timeout: float) -> bool:
- return timeout > 0
+ raise AssertionError("control publish must never block for acknowledgement")
class _FakeMqttClient:
@@ -571,11 +575,14 @@ class MqttCanTransportTests(unittest.TestCase):
self.assertEqual(payload["data"], "10 00 00 00 00 00 00 00")
self.assertEqual(payload["repeat"], 1)
self.assertEqual(payload["intervalMs"], 0)
+ self.assertFalse(payload["expectResponse"])
self.assertEqual(payload["responseTimeoutMs"], 100)
publication = bus.published_messages[0]
self.assertEqual(publication["topic"], topic)
self.assertEqual(publication["qos"], qos)
self.assertEqual(publication["retain"], retain)
+ self.assertEqual(publication["delivery_status"], "published")
+ self.assertEqual(publication["message_id"], 17)
self.assertEqual(publication["payload"], payload)
self.assertTrue(
{
@@ -606,6 +613,40 @@ class MqttCanTransportTests(unittest.TestCase):
self.assertEqual(len(self.client.published), 2)
self.assertEqual(len(bus.published_messages), 2)
+ def test_publish_returns_queued_without_waiting_for_broker_ack(self) -> None:
+ class QueuedPublishResult:
+ rc = 0
+ mid = 29
+
+ @staticmethod
+ def is_published() -> bool:
+ return False
+
+ @staticmethod
+ def wait_for_publish(timeout: float) -> bool:
+ raise AssertionError(
+ f"wait_for_publish({timeout}) must not be called"
+ )
+
+ with patch.object(
+ self.client,
+ "publish",
+ return_value=QueuedPublishResult(),
+ ):
+ publication = self.transport.publish(
+ CanFrame(
+ arbitration_id=0x431,
+ data=bytes.fromhex("10 00 00 00 00 00 00 00"),
+ channel=1,
+ is_fd=True,
+ bitrate_switch=True,
+ ),
+ command="shade position 20",
+ )
+
+ self.assertEqual(publication["delivery_status"], "queued")
+ self.assertEqual(publication["message_id"], 29)
+
def test_can1_whitelist_publishes_rear_lock_candidate_with_debug_command(self) -> None:
bus = self.transport.bus(channel=1)
bus.set_debug_command("door rear lock")
@@ -717,6 +758,96 @@ class MqttCanTransportTests(unittest.TestCase):
)
)
+ def test_wireless_fixed_controls_and_shade_preset_never_wait_for_state(
+ self,
+ ) -> None:
+ service = SeatDemoService(REPOSITORY_ROOT)
+ service._vehicle_mqtt_transport = self.transport
+ service.vehicle_mqtt_settings = SimpleNamespace(feedback_timeout_ms=1_500)
+ runtime = NativeCanRuntime(REPOSITORY_ROOT)
+ actions_by_domain = {
+ "climate": [
+ {
+ "domain": "climate",
+ "capability": "climate.set_temperature",
+ "target": "driver",
+ "operation": "set",
+ "parameters": {"temperature_c": 24},
+ },
+ {
+ "domain": "climate",
+ "capability": "climate.set_fan_level",
+ "target": "vehicle",
+ "operation": "set",
+ "parameters": {"level": 2},
+ },
+ ],
+ "roof": [
+ {
+ "domain": "roof",
+ "capability": "roof.set_shade_motion",
+ "target": "vehicle",
+ "operation": "set",
+ "parameters": {"motion": "open"},
+ }
+ ],
+ "ambient_light": [
+ {
+ "domain": "ambient_light",
+ "capability": "ambient_light.set_power",
+ "target": "vehicle",
+ "operation": "set",
+ "parameters": {"state": "on"},
+ }
+ ],
+ }
+ configs = {
+ f"native_{domain}": runtime.build_function_config(
+ domain,
+ actions,
+ f"req-wireless-no-wait-{domain}",
+ "local_rules",
+ )
+ for domain, actions in actions_by_domain.items()
+ }
+ plans = {key: runtime.compile(config) for key, config in configs.items()}
+
+ with (
+ patch.object(
+ self.transport,
+ "_receive_after",
+ side_effect=AssertionError(
+ "wireless fixed commands must not wait on can/state"
+ ),
+ ) as receive_after,
+ patch("cognitive_agent.application.native_can_runtime.time.sleep"),
+ ):
+ reports = service._execute_orchestrated_reports(
+ configs,
+ plans,
+ set(configs),
+ {},
+ hardware_request=True,
+ simulated_hardware_request=False,
+ simulation_fault="normal",
+ control_transport="wireless_mqtt",
+ )
+
+ receive_after.assert_not_called()
+ self.assertEqual(len(self.client.published), 10)
+ self.assertTrue(
+ all(
+ report["spec"]["status"] == "succeeded"
+ for report in reports.values()
+ )
+ )
+ self.assertTrue(
+ all(
+ json.loads(payload)["expectResponse"] is False
+ for _topic, payload, _qos, _retain in self.client.published
+ )
+ )
+
def test_model_added_nonwhitelisted_action_does_not_block_whitelisted_publish(
self,
) -> None:
diff --git a/tests/unit/test_requested_frontend_enhancements.py b/tests/unit/test_requested_frontend_enhancements.py
index 0973fa3..c9e217c 100644
--- a/tests/unit/test_requested_frontend_enhancements.py
+++ b/tests/unit/test_requested_frontend_enhancements.py
@@ -261,6 +261,7 @@ def test_parsed_can_command_panel_combines_can0_seat_and_can1_vehicle_commands()
"capability": "seat.set_backrest_recline",
"target": {"zone": "front_right"},
"parameters": {"position_pct": 36, "angle_deg": 120},
+ "parse_source": "local_rule",
},
]
}
@@ -272,6 +273,7 @@ def test_parsed_can_command_panel_combines_can0_seat_and_can1_vehicle_commands()
"capability": "climate.set_temperature",
"target": {"zone": "driver"},
"parameters": {"temperature_c": 24},
+ "parse_source": "cloud_supplement",
}
]
}
@@ -300,11 +302,13 @@ def test_parsed_can_command_panel_combines_can0_seat_and_can1_vehicle_commands()
"category": "副驾座椅靠背调节",
"command": "seat passenger backrest position 36",
"frame": "0x6A0 · C7 64 02 68 01 00 00",
+ "parseSource": "local_rule",
},
{
"channel": "CAN1",
"category": "空调温度",
"command": "ac temp driver 24",
+ "parseSource": "cloud_supplement",
},
]
diff --git a/tests/unit/test_two_stage_can_parsing.py b/tests/unit/test_two_stage_can_parsing.py
new file mode 100644
index 0000000..3838f5e
--- /dev/null
+++ b/tests/unit/test_two_stage_can_parsing.py
@@ -0,0 +1,192 @@
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import patch
+
+from cognitive_agent.application.seat_service import SeatDemoService
+
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def _service(tmp_path: Path) -> SeatDemoService:
+ service = SeatDemoService(ROOT)
+ service.artifact_root = tmp_path / "artifacts"
+ service.openai.api_key = "offline-two-stage-test-key"
+ return service
+
+
+def _operations(result: dict) -> list[dict]:
+ return [
+ operation
+ for config in result.get("function_configs", {}).values()
+ for operation in config.get("spec", {}).get("operations", [])
+ ]
+
+
+def test_preview_validates_local_controls_without_calling_cloud(tmp_path: Path) -> None:
+ service = _service(tmp_path)
+ validated_events: list[dict] = []
+
+ with patch.object(
+ service.openai,
+ "interpret",
+ side_effect=AssertionError("preview must not call hosted parsing"),
+ ) as interpret:
+ result = service.preview_native_can(
+ "空调温度设置为24度，风量设置为2档，播放音乐",
+ provider="openai",
+ conversation_id="conv-two-stage-local-preview",
+ execution_mode="dry_run",
+ on_commands_validated=validated_events.append,
+ )
+
+ interpret.assert_not_called()
+ assert result["phase"] == "deterministic_preview"
+ assert result["status"] == "succeeded"
+ assert result["supplement_required"] is True
+ assert result["unresolved_clauses"] == ["播放音乐"]
+ assert [item["capability"] for item in _operations(result)] == [
+ "climate.set_temperature",
+ "climate.set_fan_level",
+ ]
+ assert {item["parse_source"] for item in _operations(result)} == {
+ "local_rule"
+ }
+ assert len(validated_events) == 1
+ assert validated_events[0]["phase"] == "deterministic_commands_validated"
+ assert validated_events[0]["status"] == "processing"
+ assert validated_events[0]["execution_report"] is None
+ assert _operations(validated_events[0]) == _operations(result)
+ assert validated_events[0]["command_plans"]
+
+
+def test_fully_local_cabin_request_streams_all_commands_without_supplement(
+ tmp_path: Path,
+) -> None:
+ service = _service(tmp_path)
+ validated_events: list[dict] = []
+
+ with patch.object(
+ service.openai,
+ "interpret",
+ side_effect=AssertionError("fully local request must not call GPT"),
+ ):
+ result = service.preview_native_can(
+ "空调24度、风量2档，遮阳帘打开30%，打开氛围灯",
+ provider="openai",
+ conversation_id="conv-two-stage-all-local",
+ execution_mode="dry_run",
+ on_commands_validated=validated_events.append,
+ )
+
+ operations = _operations(result)
+ assert result["supplement_required"] is False
+ assert [item["capability"] for item in operations] == [
+ "climate.set_temperature",
+ "climate.set_fan_level",
+ "roof.set_shade_position",
+ "ambient_light.set_power",
+ ]
+ assert operations[2]["parameters"]["position_pct"] == 30
+ assert {item["parse_source"] for item in operations} == {"local_rule"}
+ assert len(validated_events) == 1
+ assert _operations(validated_events[0]) == operations
+
+
+def test_supplement_receives_only_unresolved_text_and_deduplicates_local_control(
+ tmp_path: Path,
+) -> None:
+ service = _service(tmp_path)
+ preview = service.preview_native_can(
+ "观影时空调温度设置为24度并播放音乐",
+ provider="openai",
+ conversation_id="conv-two-stage-deduplicate",
+ execution_mode="dry_run",
+ )
+ assert [item["capability"] for item in _operations(preview)] == [
+ "climate.set_temperature"
+ ]
+
+ model_output = {
+ "detected_domains": ["climate", "ambient_light"],
+ "requests": [
+ {
+ "domain": "climate",
+ "capability": "climate.set_temperature",
+ "target_scope": "driver",
+ "operation": "set",
+ "duration_ms": None,
+ "parameters": [{"name": "temperature_c", "value": 24}],
+ "reason": "模型重复给出了本地温度动作。",
+ },
+ {
+ "domain": "ambient_light",
+ "capability": "ambient_light.set_power",
+ "target_scope": "vehicle",
+ "operation": "set",
+ "duration_ms": None,
+ "parameters": [{"name": "state", "value": "on"}],
+ "reason": "模型补充观影氛围灯。",
+ },
+ ],
+ "ambiguities": [],
+ "decision_summary": ["仅补充本地规则未覆盖的观影含义。"],
+ "confidence": 0.9,
+ }
+ with patch.object(
+ service.openai,
+ "interpret",
+ return_value=model_output,
+ ) as interpret:
+ result = service.supplement_native_can(
+ preview["request_id"],
+ provider="openai",
+ )
+
+ assert interpret.call_args.args[0] == "观影时空调温度设置为24度并播放音乐"
+ operations = _operations(result)
+ assert [item["capability"] for item in operations].count(
+ "climate.set_temperature"
+ ) == 1
+ assert [item["capability"] for item in operations].count(
+ "ambient_light.set_power"
+ ) == 1
+ assert {
+ item["capability"]: item["parse_source"] for item in operations
+ } == {
+ "climate.set_temperature": "local_rule",
+ "ambient_light.set_power": "cloud_supplement",
+ }
+ cloud_results = (
+ result["phase_execution_reports"]["cloud_supplement"] or {}
+ ).get("spec", {}).get("results", [])
+ assert len(cloud_results) == 1
+
+
+def test_frontend_two_stage_contract_and_source_badges() -> None:
+ app = (ROOT / "apps" / "web-console" / "app.js").read_text(
+ encoding="utf-8"
+ )
+ console = (ROOT / "apps" / "web-console" / "console-v2.js").read_text(
+ encoding="utf-8"
+ )
+ styles = (ROOT / "apps" / "web-console" / "console-v2.css").read_text(
+ encoding="utf-8"
+ )
+ server = (ROOT / "src" / "cognitive_agent" / "server.py").read_text(
+ encoding="utf-8"
+ )
+
+ assert '"/api/requests/preview"' in app
+ assert "`/api/requests/${preview.request_id}/supplement`" in app
+ assert "renderResult(preview);" in app
+ assert 'Accept: "application/x-ndjson"' in app
+ assert '"cognitive-agent:validated-can-commands"' in app
+ assert "本地规则命令已执行；云端补充失败" in app
+ assert 'class="v2-can-command-source"' in console
+ assert 'item.parseSource === "local_rule" ? "本地规则" : "云端模型"' in console
+ assert '"cognitive-agent:validated-can-commands"' in console
+ assert '.v2-can-command-source[data-parse-source="local_rule"]' in styles
+ assert 'path not in {"/api/requests", "/api/requests/preview"}' in server
+ assert '"type": "validated_commands"' in server