2 Commits

Author SHA1 Message Date
Miayui 1d87a7ffc8 merge: sync feature/N7Web into feature/N7Web-Tech
# Conflicts:
#	apps/web-console/service-worker.js
#	tests/e2e/test_seat_web_api.py
#	tests/unit/test_assistant_frontend_contract.py
#	tests/unit/test_vue_frontend_contract.py
2026-09-16 16:19:13 +08:00
Miayui 75ce391312 feat(n7web): refine spatial UI and NFC-gated extension routing 2026-09-16 16:14:10 +08:00
10 changed files with 658 additions and 87 deletions
+6 -1
View File
@@ -568,7 +568,11 @@
const verificationStatus = String(telemetry.verification_status || "").toLowerCase(); const verificationStatus = String(telemetry.verification_status || "").toLowerCase();
const failureMessage = String(execution?.message || "").trim(); const failureMessage = String(execution?.message || "").trim();
const aborted = abortedStates.has(executionStatus); const aborted = abortedStates.has(executionStatus);
const explicitlyBlocked = executionStatus === "blocked"
|| verificationStatus === "blocked"
|| Boolean(telemetry.authorization_reason);
const executionFailed = executionStatus === "failed" const executionFailed = executionStatus === "failed"
|| explicitlyBlocked
|| verificationStatus === "failed_or_blocked"; || verificationStatus === "failed_or_blocked";
const missingWindowFeedback = /0x388/i.test(failureMessage) && ( const missingWindowFeedback = /0x388/i.test(failureMessage) && (
/no fresh valid/i.test(failureMessage) /no fresh valid/i.test(failureMessage)
@@ -579,7 +583,8 @@
|| /未收到.*(?:有效|新鲜).*遮阳帘.*位置/.test(failureMessage) || /未收到.*(?:有效|新鲜).*遮阳帘.*位置/.test(failureMessage)
); );
const safetyBlocked = executionFailed && !submitted && ( const safetyBlocked = executionFailed && !submitted && (
missingWindowFeedback explicitlyBlocked
|| missingWindowFeedback
|| missingShadeFeedback || missingShadeFeedback
|| /refusing (?:blind )?motion/i.test(failureMessage) || /refusing (?:blind )?motion/i.test(failureMessage)
|| /拒绝(?:启动机构|盲动|执行)/.test(failureMessage) || /拒绝(?:启动机构|盲动|执行)/.test(failureMessage)
+7 -2
View File
@@ -1224,7 +1224,9 @@
} }
showGscuResult(result); showGscuResult(result);
showToast( showToast(
action === "learn" ? `IR ${slot} 学习完成,槽位回读已确认。` : `${action} 指令处理完成。`, action === "learn"
? `IR ${slot} 学习完成并已加入自然语言解析;实际下发仍需 NFC 授权。`
: `${action} 指令处理完成。`,
"success", "success",
); );
} catch (error) { } catch (error) {
@@ -1275,7 +1277,10 @@
$("[data-gscu-action-names]", card).value, $("[data-gscu-action-names]", card).value,
"tcp", "tcp",
).then((value) => { ).then((value) => {
showToast(`IR ${value.slot} 的控制对象与执行动作已保存。`, "success"); showToast(
`IR ${value.slot} 已保存并立即加入自然语言解析;实际下发仍需 NFC 授权。`,
"success",
);
}).catch((error) => { }).catch((error) => {
showToast(error.message, "error"); showToast(error.message, "error");
}).finally(() => { save.disabled = false; }); }).finally(() => { save.disabled = false; });
+1 -1
View File
@@ -1,4 +1,4 @@
const CACHE_VERSION = "cognitive-agent-pad-v316-sidebar-accordion"; const CACHE_VERSION = "cognitive-agent-pad-v318-tech-nfc-routing";
const SHELL_CACHE = `${CACHE_VERSION}-shell`; const SHELL_CACHE = `${CACHE_VERSION}-shell`;
const RUNTIME_CACHE = `${CACHE_VERSION}-runtime`; const RUNTIME_CACHE = `${CACHE_VERSION}-runtime`;
const APP_SHELL = [ const APP_SHELL = [
+178 -57
View File
@@ -380,6 +380,86 @@ class InfraredSlotCatalog:
capabilities.append(sanitized) capabilities.append(sanitized)
return capabilities return capabilities
def list_routing_capabilities(self) -> list[dict[str, Any]]:
"""Return every named IR slot, annotated with execution authorization.
The slot catalog is the source of truth for natural-language discovery,
while ``infrared_capabilities`` remains the source of truth for physical
execution authorization. Keeping those concerns separate lets a newly
learned device become parseable immediately without allowing an
unbound NFC device to emit infrared commands.
"""
slots = self.list_slots()
authorized = self.list_capabilities()
authorized_by_identity: dict[tuple[str, int], dict[str, Any]] = {}
for item in authorized:
try:
identity = (
str(item.get("device_id") or "").strip(),
validate_slot(item.get("slot")),
)
except GatewayError:
continue
if identity[0]:
authorized_by_identity.setdefault(identity, item)
routing: list[dict[str, Any]] = []
for slot_record in slots:
device_id = str(slot_record.get("device_id") or "").strip()
object_names = _safe_peripheral_identity_names(
slot_record.get("object_names")
)
action_names = [
str(value or "").strip()
for value in slot_record.get("action_names", [])
if str(value or "").strip()
][:24]
if (
not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,63}", device_id)
or not object_names
or not action_names
):
continue
slot = validate_slot(slot_record.get("slot"))
persisted = authorized_by_identity.get((device_id, slot))
item = dict(persisted or {})
card_uid = _normalized_card_uid(item.get("card_uid"))
enabled = item.get("enabled", True) is not False
execution_authorized = bool(card_uid and enabled)
item.update(
{
"id": str(item.get("id") or f"ir-slot:{device_id}:{slot}"),
"status": "learned",
"enabled": enabled if card_uid else False,
"parseable": True,
"execution_authorized": execution_authorized,
"authorization_state": (
"authorized"
if execution_authorized
else "disabled"
if card_uid
else "nfc_binding_required"
),
"device_id": device_id,
"device_name": object_names[0],
"device_aliases": object_names,
"object_names": object_names,
"device_type": str(item.get("device_type") or "other")[:48],
"card_uid": card_uid or None,
"action": action_names[0],
"aliases": action_names,
"action_names": action_names,
"risk": str(item.get("risk") or "low"),
"slot": slot,
"transport": str(item.get("transport") or "gscu_tcp")[:32],
"routing_source": "infrared_slot_catalog",
}
)
routing.append(item)
return routing
def sync_capabilities( def sync_capabilities(
self, self,
catalog: object, catalog: object,
@@ -1244,6 +1324,28 @@ def resolve_active_peripheral_command(
catalog, catalog,
nfc_snapshot, nfc_snapshot,
require_enabled=True, require_enabled=True,
require_binding=True,
)
def resolve_parseable_peripheral_command(
utterance: str,
catalog: list[dict[str, Any]],
nfc_snapshot: dict[str, Any],
) -> dict[str, Any] | None:
"""Resolve a named learned IR action without granting execution rights.
This is intentionally a parsing-only operation. Callers must inspect the
returned capability's persisted ``card_uid`` and ``enabled`` state before
allowing any physical infrared emit.
"""
return _resolve_bound_peripheral_command(
utterance,
catalog,
nfc_snapshot,
require_enabled=False,
require_binding=False,
) )
@@ -1253,6 +1355,7 @@ def _resolve_bound_peripheral_command(
nfc_snapshot: dict[str, Any], nfc_snapshot: dict[str, Any],
*, *,
require_enabled: bool, require_enabled: bool,
require_binding: bool = True,
) -> dict[str, Any] | None: ) -> dict[str, Any] | None:
"""Resolve a persistently card-bound command, including disabled entries. """Resolve a persistently card-bound command, including disabled entries.
@@ -1279,7 +1382,7 @@ def _resolve_bound_peripheral_command(
): ):
continue continue
bound_uid = _normalized_card_uid(raw_item.get("card_uid")) bound_uid = _normalized_card_uid(raw_item.get("card_uid"))
if not bound_uid: if require_binding and not bound_uid:
continue continue
device_score = _device_match_score(compact_utterance, raw_item) device_score = _device_match_score(compact_utterance, raw_item)
if not device_score: if not device_score:
@@ -1310,6 +1413,17 @@ def _resolve_bound_peripheral_command(
resolved = dict(matches[0][2]) resolved = dict(matches[0][2])
resolved["slot"] = validate_slot(resolved.get("slot")) resolved["slot"] = validate_slot(resolved.get("slot"))
resolved["card_uid"] = _normalized_card_uid(resolved.get("card_uid")) resolved["card_uid"] = _normalized_card_uid(resolved.get("card_uid"))
resolved["parseable"] = True
resolved["execution_authorized"] = bool(
resolved["card_uid"] and resolved.get("enabled", True) is not False
)
resolved["authorization_state"] = (
"authorized"
if resolved["execution_authorized"]
else "disabled"
if resolved["card_uid"]
else "nfc_binding_required"
)
return resolved return resolved
@@ -1326,10 +1440,12 @@ def partition_active_peripheral_intent(
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Claim only explicit learned-peripheral clauses and preserve CAN text. """Claim only explicit learned-peripheral clauses and preserve CAN text.
The function is deliberately conservative: a clause must contain both a The function is deliberately conservative: a clause must contain both an
persistently NFC-bound device identity and one learned action. When no peripheral explicit learned device identity and one learned action. A matching but
clause is claimed, ``residual_utterance`` is byte-for-byte the original unbound device is returned as an authorization denial, never as residual
trimmed input so the existing CAN parser observes exactly its old input. vehicle text. When no peripheral clause is claimed, ``residual_utterance``
is byte-for-byte the original trimmed input so the existing CAN parser
observes exactly its old input.
""" """
original = str(utterance or "").strip() original = str(utterance or "").strip()
@@ -1342,12 +1458,16 @@ def partition_active_peripheral_intent(
"denied_peripheral_actions": [], "denied_peripheral_actions": [],
} }
bound_catalog = [ routing_catalog = [
item item
for item in catalog[:MAX_CATALOG_ITEMS] for item in catalog[:MAX_CATALOG_ITEMS]
if isinstance(item, dict) if isinstance(item, dict)
and item.get("status") == "learned" and item.get("status") == "learned"
and bool(_normalized_card_uid(item.get("card_uid"))) ]
bound_catalog = [
item
for item in routing_catalog
if bool(_normalized_card_uid(item.get("card_uid")))
] ]
active_catalog = [ active_catalog = [
item item
@@ -1355,8 +1475,8 @@ def partition_active_peripheral_intent(
if item.get("enabled", True) is not False if item.get("enabled", True) is not False
] ]
compact_original = _compact_match_text(original) compact_original = _compact_match_text(original)
if not bound_catalog or not any( if not routing_catalog or not any(
_device_match_score(compact_original, item) for item in bound_catalog _device_match_score(compact_original, item) for item in routing_catalog
): ):
return { return {
"peripheral_actions": [], "peripheral_actions": [],
@@ -1380,36 +1500,30 @@ def partition_active_peripheral_intent(
contextual_device_name = "" contextual_device_name = ""
for index, clause in enumerate(clauses): for index, clause in enumerate(clauses):
compact_clause = _compact_match_text(clause) compact_clause = _compact_match_text(clause)
explicit_bound_candidates = [ explicit_routing_candidates = [
item item
for item in bound_catalog for item in routing_catalog
if _device_match_score(compact_clause, item) if _device_match_score(compact_clause, item)
] ]
explicit_candidates = [ if explicit_routing_candidates and _has_unshadowed_vehicle_device_phrase(
item
for item in explicit_bound_candidates
if item.get("enabled", True) is not False
]
if explicit_bound_candidates and _has_unshadowed_vehicle_device_phrase(
compact_clause, compact_clause,
explicit_bound_candidates, explicit_routing_candidates,
): ):
residual.append(clause) residual.append(clause)
contextual_device_id = None contextual_device_id = None
contextual_device_name = "" contextual_device_name = ""
continue continue
clause_candidates = explicit_candidates clause_candidates = explicit_routing_candidates
bound_clause_candidates = explicit_bound_candidates
resolution_text = clause resolution_text = clause
if explicit_bound_candidates: if explicit_routing_candidates:
candidate_device_ids = { candidate_device_ids = {
str(item.get("device_id") or "") str(item.get("device_id") or "")
for item in explicit_bound_candidates for item in explicit_routing_candidates
} }
if len(candidate_device_ids) == 1: if len(candidate_device_ids) == 1:
contextual_device_id = next(iter(candidate_device_ids)) contextual_device_id = next(iter(candidate_device_ids))
identity_names: list[str] = [] identity_names: list[str] = []
for item in explicit_bound_candidates: for item in explicit_routing_candidates:
for field in ("object_names", "device_aliases"): for field in ("object_names", "device_aliases"):
values = item.get(field, []) values = item.get(field, [])
if isinstance(values, str): if isinstance(values, str):
@@ -1454,56 +1568,51 @@ def partition_active_peripheral_intent(
): ):
clause_candidates = [ clause_candidates = [
item item
for item in active_catalog for item in routing_catalog
if str(item.get("device_id") or "") == contextual_device_id
]
bound_clause_candidates = [
item
for item in bound_catalog
if str(item.get("device_id") or "") == contextual_device_id if str(item.get("device_id") or "") == contextual_device_id
] ]
resolution_text = f"{contextual_device_name},{clause}" resolution_text = f"{contextual_device_name},{clause}"
if not clause_candidates and not bound_clause_candidates: if not clause_candidates:
residual.append(clause) residual.append(clause)
continue continue
active_resolution_catalog = ( routing_resolution_catalog = (
clause_candidates if not explicit_candidates else active_catalog clause_candidates if not explicit_routing_candidates else routing_catalog
) )
resolved = resolve_active_peripheral_command( resolved = resolve_parseable_peripheral_command(
resolution_text, active_resolution_catalog, nfc_snapshot resolution_text, routing_resolution_catalog, nfc_snapshot
) )
if resolved is None: if resolved is None:
bound_resolution_catalog = ( # Explicit third-party device text must never fall through into
bound_clause_candidates # the vehicle CAN model as if it were a cabin actuator request.
if not explicit_bound_candidates unresolved.append(clause)
else bound_catalog authorized_device_ids = {
str(item.get("device_id") or "") for item in clause_candidates
}
unresolved_candidates.update(
(str(item.get("id")), item)
for item in active_catalog
if item.get("id")
and str(item.get("device_id") or "") in authorized_device_ids
) )
bound_match = _resolve_bound_peripheral_command( continue
resolution_text, if not _normalized_card_uid(resolved.get("card_uid")):
bound_resolution_catalog,
nfc_snapshot,
require_enabled=False,
)
if (
bound_match is not None
and bound_match.get("enabled", True) is False
):
denied.append( denied.append(
{ {
"index": index, "index": index,
"clause": clause, "clause": clause,
"capability": bound_match, "capability": resolved,
"reason": "nfc_command_disabled", "reason": "nfc_binding_required",
} }
) )
continue continue
# Explicit third-party device text must never fall through into if resolved.get("enabled", True) is False:
# the vehicle CAN model as if it were a cabin actuator request. denied.append(
unresolved.append(clause) {
unresolved_candidates.update( "index": index,
(str(item.get("id")), item) "clause": clause,
for item in clause_candidates "capability": resolved,
if item.get("id") "reason": "nfc_command_disabled",
}
) )
continue continue
actions.append( actions.append(
@@ -2275,6 +2384,11 @@ class SelfExtensionService:
return self.infrared_catalog.list_capabilities() return self.infrared_catalog.list_capabilities()
def list_peripheral_routing_capabilities(self) -> list[dict[str, Any]]:
"""Return parseable IR actions with explicit authorization metadata."""
return self.infrared_catalog.list_routing_capabilities()
def sync_peripheral_catalog(self, catalog: object) -> dict[str, Any]: def sync_peripheral_catalog(self, catalog: object) -> dict[str, Any]:
"""Persist the complete learned capability catalog from the NFC UI.""" """Persist the complete learned capability catalog from the NFC UI."""
@@ -2441,6 +2555,13 @@ class SelfExtensionService:
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Execute a persistently NFC-bound match through the main request API.""" """Execute a persistently NFC-bound match through the main request API."""
if (
capability.get("enabled", True) is False
or not _normalized_card_uid(capability.get("card_uid"))
):
raise ExtensionIntegrationError(
"红外能力尚未完成持久 NFC 授权,已拒绝物理下发。"
)
if not _peripheral_execution_identity_authorized(utterance, capability): if not _peripheral_execution_identity_authorized(utterance, capability):
raise ExtensionIntegrationError( raise ExtensionIntegrationError(
"外设名称未在原始指令中被明确提及,已拒绝红外执行。" "外设名称未在原始指令中被明确提及,已拒绝红外执行。"
+207 -17
View File
@@ -742,7 +742,7 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
).strip() ).strip()
peripheral_routing = self._speech_peripheral_routing(raw_text) peripheral_routing = self._speech_peripheral_routing(raw_text)
peripheral_transcript = bool( peripheral_transcript = bool(
peripheral_routing.get("execution_eligible") and raw_text peripheral_routing.get("routing_eligible") and raw_text
) )
expected_text = raw_text if peripheral_transcript else clean_text expected_text = raw_text if peripheral_transcript else clean_text
if utterance != expected_text: if utterance != expected_text:
@@ -942,6 +942,18 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
HTTPStatus.INTERNAL_SERVER_ERROR, HTTPStatus.INTERNAL_SERVER_ERROR,
) )
def _peripheral_routing_catalog(self) -> list[dict[str, object]]:
"""Read the parse catalog while remaining compatible with test doubles."""
provider = getattr(
self.extension,
"list_peripheral_routing_capabilities",
None,
)
if callable(provider):
return list(provider())
return list(self.extension.list_peripheral_capabilities())
def _execute_active_peripheral_request( def _execute_active_peripheral_request(
self, self,
utterance: str, utterance: str,
@@ -1053,14 +1065,16 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
Vehicle speech keeps using its cached structured semantic payload. A Vehicle speech keeps using its cached structured semantic payload. A
learned peripheral cannot be represented by the static vehicle learned peripheral cannot be represented by the static vehicle
capability catalog, so it may instead reuse the exact server-owned ASR capability catalog, so it may instead reuse the exact server-owned ASR
transcript only after the durable NFC binding and learned action resolve. transcript once the learned device and action resolve. Execution
authorization is checked separately so an unbound device can be shown
as parsed-and-blocked without ever reaching the infrared transport.
""" """
text = str(utterance or "").strip() text = str(utterance or "").strip()
if not text: if not text:
return {"execution_eligible": False, "action_count": 0} return {"execution_eligible": False, "action_count": 0}
try: try:
catalog = self.extension.list_peripheral_capabilities() catalog = self._peripheral_routing_catalog()
nfc_snapshot = self.nfc.snapshot() nfc_snapshot = self.nfc.snapshot()
partition = partition_active_peripheral_intent( partition = partition_active_peripheral_intent(
text, text,
@@ -1089,9 +1103,12 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
action_count = len(entries) or (1 if direct is not None else 0) action_count = len(entries) or (1 if direct is not None else 0)
if plan_catalog: if plan_catalog:
action_count = max(action_count, 1) action_count = max(action_count, 1)
routing_eligible = bool(action_count or unresolved or denied)
return { return {
"execution_eligible": bool(action_count and not unresolved and not denied), "execution_eligible": bool(action_count and not unresolved and not denied),
"routing_eligible": routing_eligible,
"action_count": action_count, "action_count": action_count,
"denied_action_count": len(denied),
"has_ordered_plan": bool(plan_catalog), "has_ordered_plan": bool(plan_catalog),
} }
@@ -1395,7 +1412,7 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
result: dict[str, object], result: dict[str, object],
entries: list[dict[str, object]], entries: list[dict[str, object]],
) -> dict[str, object]: ) -> dict[str, object]:
"""Expose disabled bound commands without discarding allowed work.""" """Expose parsed-but-unauthorized IR commands without dispatching them."""
if not entries: if not entries:
return result return result
@@ -1419,13 +1436,156 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
result["denied_peripheral_actions"] = denied_actions result["denied_peripheral_actions"] = denied_actions
result["authorization_filtered"] = True result["authorization_filtered"] = True
blocked_operations: list[dict[str, object]] = []
blocked_commands: list[dict[str, object]] = []
blocked_results: list[dict[str, object]] = []
for position, item in enumerate(denied_actions, start=1):
reason = str(item.get("reason") or "nfc_command_disabled")
device_name = str(
item.get("device_name") or item.get("device_id") or "未命名外设"
)
action_name = str(item.get("action") or "未命名动作")
operation_ref = (
f"{item.get('capability_id') or 'infrared'}-authorization-{position}"
)
message = (
f"已识别“{device_name} · {action_name}”,但设备尚未完成 NFC "
"绑定授权,未下发。"
if reason == "nfc_binding_required"
else f"已识别“{device_name} · {action_name}”,但该 NFC 能力已被禁用,未下发。"
)
blocked_operations.append(
{
"id": operation_ref,
"capability": "infrared.emit",
"target": {
"device": item.get("device_id"),
"name": item.get("device_name"),
},
"parameters": {
"action": item.get("action"),
"slot": item.get("slot"),
},
"parse_source": "local_rule",
"authorization": {
"authorized": False,
"reason": reason,
},
}
)
blocked_commands.append(
{
"operation_ref": operation_ref,
"bus": "GSCU_TCP",
"message": "infrared.emit",
"signals": {
"slot": item.get("slot"),
"authorization_status": "blocked",
"authorization_reason": reason,
},
}
)
blocked_results.append(
{
"ref": operation_ref,
"domain": "self_extension",
"status": "blocked",
"message": message,
"telemetry": {
"transport": "gscu_tcp",
"slot": item.get("slot"),
"delivery_status": "not_submitted",
"transport_submitted": False,
"gateway_acknowledged": False,
"physical_effect_confirmed": False,
"verification_status": "failed_or_blocked",
"authorization_reason": reason,
},
}
)
function_configs = deepcopy(result.get("function_configs") or {})
command_plans = deepcopy(result.get("command_plans") or {})
config_key = "self_extension_blocked"
suffix = 2
while config_key in function_configs or config_key in command_plans:
config_key = f"self_extension_blocked_{suffix}"
suffix += 1
blocked_config = {
"apiVersion": "cognitive-agent.io/v1alpha1",
"kind": "FunctionConfig",
"metadata": {
"name": config_key,
"labels": {
"execution_scope": "nfc_authorization_required",
},
},
"spec": {
"domain": "self_extension",
"operations": blocked_operations,
},
}
blocked_plan = {
"apiVersion": "cognitive-agent.io/v1alpha1",
"kind": "CommandPlan",
"metadata": {
"name": f"{config_key}-gscu",
"labels": {"authorization_status": "blocked"},
},
"spec": {
"domain": "self_extension",
"platform": "gscu1b",
"commands": blocked_commands,
},
}
function_configs[config_key] = blocked_config
command_plans[config_key] = blocked_plan
result["function_configs"] = function_configs
result["command_plans"] = command_plans
if not isinstance(result.get("function_config"), dict):
result["function_config"] = blocked_config
if not isinstance(result.get("command_plan"), dict):
result["command_plan"] = blocked_plan
result["operation_results"] = [
*deepcopy(list(result.get("operation_results") or [])),
*blocked_results,
]
result["trace_events"] = [
*deepcopy(list(result.get("trace_events") or [])),
{
"stage": "nfc_peripheral_authorization",
"status": "blocked",
"summary": "红外指令已由本地规则解析,但未通过 NFC 执行授权。",
"duration_ms": 0,
},
]
summaries = list(result.get("decision_summary") or []) summaries = list(result.get("decision_summary") or [])
binding_required = [
item
for item in denied_actions
if item.get("reason") == "nfc_binding_required"
]
disabled = [
item
for item in denied_actions
if item.get("reason") != "nfc_binding_required"
]
if binding_required:
labels = "、".join( labels = "、".join(
str(item.get("clause") or item.get("action") or "未命名动作") str(item.get("clause") or item.get("action") or "未命名动作")
for item in denied_actions for item in binding_required
) )
summaries.append( summaries.append(
f"持久绑定中有 {len(denied_actions)} 条红外能力未启用,已过滤:{labels}。" f"已解析 {len(binding_required)} 条红外控制指令,但设备尚未完成 NFC "
f"绑定授权,未下发:{labels}。"
)
if disabled:
labels = "、".join(
str(item.get("clause") or item.get("action") or "未命名动作")
for item in disabled
)
summaries.append(
f"持久绑定中有 {len(disabled)} 条红外能力未启用,已过滤:{labels}。"
) )
result["decision_summary"] = summaries result["decision_summary"] = summaries
@@ -1433,11 +1593,12 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
if prior_status in {"succeeded", "partially_succeeded", "no_action"}: if prior_status in {"succeeded", "partially_succeeded", "no_action"}:
result["status"] = "partially_succeeded" result["status"] = "partially_succeeded"
result["success"] = False result["success"] = False
result["outcome"] = "failed_or_blocked"
execution_report = result.get("execution_report") execution_report = result.get("execution_report")
if isinstance(execution_report, dict): if isinstance(execution_report, dict):
spec = execution_report.get("spec") spec = execution_report.get("spec")
if isinstance(spec, dict): if isinstance(spec, dict):
spec["status"] = "partially_succeeded" spec["status"] = str(result.get("status") or "blocked")
return result return result
def _execute_request_with_peripherals( def _execute_request_with_peripherals(
@@ -1474,7 +1635,7 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
**mode_arguments, **mode_arguments,
_on_commands_validated=callback, _on_commands_validated=callback,
) )
catalog = self.extension.list_peripheral_capabilities() catalog = self._peripheral_routing_catalog()
nfc_snapshot = self.nfc.snapshot() nfc_snapshot = self.nfc.snapshot()
partition = partition_active_peripheral_intent( partition = partition_active_peripheral_intent(
utterance, utterance,
@@ -1488,7 +1649,7 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
parsed_peripheral_clauses = list( parsed_peripheral_clauses = list(
dict.fromkeys( dict.fromkeys(
str(entry.get("clause") or "").strip() str(entry.get("clause") or "").strip()
for entry in entries for entry in [*entries, *denied]
if isinstance(entry, dict) and str(entry.get("clause") or "").strip() if isinstance(entry, dict) and str(entry.get("clause") or "").strip()
) )
) )
@@ -1839,14 +2000,43 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
"supplement_required": True, "supplement_required": True,
} }
else: else:
labels = "、".join( local_elapsed_ms = round(
str(item.get("clause") or "未命名动作") (time.perf_counter() - local_stage_started) * 1000,
for item in denied 3,
if isinstance(item, dict)
)
raise ExtensionIntegrationError(
f"以下持久绑定红外能力未启用,无法执行:{labels}。"
) )
request_id = f"req-local-blocked-{os.urandom(8).hex()}"
result = {
"request_id": request_id,
"preview_request_id": request_id,
"utterance": utterance,
"conversation_id": conversation_id,
"status": "blocked",
"success": False,
"outcome": "failed_or_blocked",
"phase": "authorization_blocked",
"orchestration_route": "local_peripheral_authorization",
"provider_requested": provider,
"provider_used": "infrared_local_rules",
"execution_mode": execution_mode,
"control_transport": "gscu_tcp",
"decision_summary": [],
"function_config": None,
"function_configs": {},
"command_plan": None,
"command_plans": {},
"trace_events": [],
"validation": [],
"execution_report": None,
"operation_results": [],
"parsed_control_signatures": [],
"preview_elapsed_ms": local_elapsed_ms,
"local_control_budget_ms": LOCAL_CONTROL_BUDGET_MS,
"local_control_elapsed_ms": local_elapsed_ms,
"local_control_within_budget": (
local_elapsed_ms < LOCAL_CONTROL_BUDGET_MS
),
"supplement_required": False,
}
result["natural_language_routing"] = deepcopy(natural_language_routing) result["natural_language_routing"] = deepcopy(natural_language_routing)
for key in ("parsed_clauses", "resolved_clauses", "unresolved_clauses"): for key in ("parsed_clauses", "resolved_clauses", "unresolved_clauses"):
@@ -3538,7 +3728,7 @@ class SeatDemoHandler(BaseHTTPRequestHandler):
result = {**result, "input_mode": input_mode} result = {**result, "input_mode": input_mode}
raw_text = str(result.get("raw_text") or "").strip() raw_text = str(result.get("raw_text") or "").strip()
peripheral_routing = self._speech_peripheral_routing(raw_text) peripheral_routing = self._speech_peripheral_routing(raw_text)
if peripheral_routing.get("execution_eligible"): if peripheral_routing.get("routing_eligible"):
result = { result = {
**result, **result,
"routing_text": raw_text, "routing_text": raw_text,
+91 -1
View File
@@ -1363,7 +1363,7 @@ operations:
self.assertIn('canvas.addEventListener("webglcontextlost"', app_script) self.assertIn('canvas.addEventListener("webglcontextlost"', app_script)
self.assertIn('"cognitive-agent:module-change"', app_script) self.assertIn('"cognitive-agent:module-change"', app_script)
self.assertIn( self.assertIn(
'const CACHE_VERSION = "cognitive-agent-pad-v316-sidebar-accordion"', 'const CACHE_VERSION = "cognitive-agent-pad-v318-tech-nfc-routing"',
service_worker, service_worker,
) )
self.assertIn('/console-v2.css?v=delivery-state-69', service_worker) self.assertIn('/console-v2.css?v=delivery-state-69', service_worker)
@@ -2116,6 +2116,96 @@ operations:
finally: finally:
SeatDemoHandler.extension = original_extension SeatDemoHandler.extension = original_extension
def test_new_tcp_slot_is_immediately_parsed_but_blocked_until_nfc_binding(
self,
) -> None:
original_extension = SeatDemoHandler.extension
original_nfc = SeatDemoHandler.nfc
with tempfile.TemporaryDirectory() as directory:
extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(directory, "unbound-cactus.sqlite3"),
)
SeatDemoHandler.extension = extension
SeatDemoHandler.nfc = SimpleNamespace(
snapshot=lambda: {
"revision": 1,
"reader": {"online": True},
"card": None,
}
)
save_request = urllib.request.Request(
f"{self.base_url}/api/extension/gscu/slots/11",
method="PUT",
headers={"Content-Type": "application/json"},
data=json.dumps(
{
"device_id": "toy-cacti-002",
"object_names": ["仙人掌"],
"action_names": ["打开", "开启"],
"source": "tcp",
},
ensure_ascii=False,
).encode("utf-8"),
)
control_request = urllib.request.Request(
f"{self.base_url}/api/requests",
method="POST",
headers={
"Content-Type": "application/json",
"Accept": "application/x-ndjson",
},
data=json.dumps(
{
"utterance": "打开仙人掌",
"provider": "local",
"conversation_id": "conv-unbound-cactus-e2e",
"execution_mode": "hardware",
"client_type": "web_console",
},
ensure_ascii=False,
).encode("utf-8"),
)
try:
with self.opener.open(save_request, timeout=3) as response:
self.assertEqual(response.status, 200)
with patch.object(
extension,
"execute_gateway",
side_effect=AssertionError("unbound IR command must not be emitted"),
):
with self.opener.open(control_request, timeout=3) as response:
events = [
json.loads(line)
for line in response.read().decode("utf-8").splitlines()
if line.strip()
]
finally:
SeatDemoHandler.extension = original_extension
SeatDemoHandler.nfc = original_nfc
self.assertEqual(events[-1]["type"], "result")
result = events[-1]["payload"]
self.assertEqual(result["status"], "blocked")
self.assertLess(result["local_control_elapsed_ms"], 1000)
self.assertEqual(result["parsed_clauses"], ["打开仙人掌"])
self.assertEqual(result["unresolved_clauses"], [])
self.assertEqual(
result["denied_peripheral_actions"][0]["reason"],
"nfc_binding_required",
)
operation = next(
operation
for config in result["function_configs"].values()
for operation in config["spec"]["operations"]
)
self.assertEqual(operation["target"]["device"], "toy-cacti-002")
self.assertEqual(operation["parameters"], {"action": "打开", "slot": 11})
self.assertEqual(result["operation_results"][0]["status"], "blocked")
self.assertFalse(
result["operation_results"][0]["telemetry"]["transport_submitted"]
)
def test_main_request_routes_durable_nfc_peripheral_after_card_removal(self) -> None: def test_main_request_routes_durable_nfc_peripheral_after_card_removal(self) -> None:
original_extension = SeatDemoHandler.extension original_extension = SeatDemoHandler.extension
original_nfc = SeatDemoHandler.nfc original_nfc = SeatDemoHandler.nfc
+11 -1
View File
@@ -27,7 +27,7 @@ def test_console_contains_ai_assistant_surface():
def test_codex_assistant_assets_are_versioned_together_in_the_service_worker(): def test_codex_assistant_assets_are_versioned_together_in_the_service_worker():
service_worker = (ROOT / "apps/web-console/service-worker.js").read_text(encoding="utf-8") service_worker = (ROOT / "apps/web-console/service-worker.js").read_text(encoding="utf-8")
assert ( assert (
'const CACHE_VERSION = "cognitive-agent-pad-v316-sidebar-accordion"' 'const CACHE_VERSION = "cognitive-agent-pad-v318-tech-nfc-routing"'
in service_worker in service_worker
) )
assert '"/assistant-chat.css?v=codex-assistant-29"' in service_worker assert '"/assistant-chat.css?v=codex-assistant-29"' in service_worker
@@ -120,6 +120,16 @@ def test_validated_controls_are_mirrored_into_a_compact_assistant_card():
assert ".assistant-dispatched-controls[open]" in css assert ".assistant-dispatched-controls[open]" in css
def test_explicit_nfc_authorization_block_is_rendered_as_safety_blocked():
console = (ROOT / "apps/web-console/console-v2.js").read_text(encoding="utf-8")
assert 'executionStatus === "blocked"' in console
assert "Boolean(telemetry.authorization_reason)" in console
assert 'deliveryState = aborted' in console
assert '? "blocked"' in console
assert '? "安全阻止"' in console
def test_latest_dispatched_control_group_is_the_only_group_expanded(): def test_latest_dispatched_control_group_is_the_only_group_expanded():
assistant = (ROOT / "apps/web-console/assistant-chat.js").read_text( assistant = (ROOT / "apps/web-console/assistant-chat.js").read_text(
encoding="utf-8" encoding="utf-8"
+78
View File
@@ -696,6 +696,84 @@ class SelfExtensionTests(unittest.TestCase):
self.assertEqual(changed["device_id"], "toy-light-v2") self.assertEqual(changed["device_id"], "toy-light-v2")
self.assertEqual(service.list_peripheral_capabilities(), []) self.assertEqual(service.list_peripheral_capabilities(), [])
def test_new_structured_slot_is_parseable_before_nfc_authorization(self) -> None:
with TemporaryDirectory() as directory:
service = SelfExtensionService(
SimulationGateway(),
catalog_db_path=Path(directory, "infrared-slots.sqlite3"),
)
service.save_infrared_slot(
11,
{
"device_id": "toy-cacti-002",
"object_names": ["仙人掌"],
"action_names": ["打开", "开启"],
"source": "tcp",
},
)
self.assertEqual(service.list_peripheral_capabilities(), [])
routing = service.list_peripheral_routing_capabilities()
self.assertEqual(len(routing), 1)
self.assertEqual(routing[0]["id"], "ir-slot:toy-cacti-002:11")
self.assertEqual(routing[0]["authorization_state"], "nfc_binding_required")
self.assertTrue(routing[0]["parseable"])
self.assertFalse(routing[0]["execution_authorized"])
partition = partition_active_peripheral_intent(
"开启仙人掌",
routing,
{"card": None},
)
self.assertEqual(partition["peripheral_actions"], [])
self.assertEqual(partition["residual_utterance"], "")
self.assertEqual(partition["unresolved_peripheral_clauses"], [])
self.assertEqual(
[
(
item["capability"]["id"],
item["capability"]["slot"],
item["reason"],
)
for item in partition["denied_peripheral_actions"]
],
[("ir-slot:toy-cacti-002:11", 11, "nfc_binding_required")],
)
self.assertIsNone(
resolve_active_peripheral_command(
"开启仙人掌",
routing,
{"card": None},
)
)
service.sync_peripheral_catalog(
[
{
**routing[0],
"enabled": True,
"card_uid": "DBA5B203",
}
]
)
authorized_routing = service.list_peripheral_routing_capabilities()
self.assertEqual(
authorized_routing[0]["authorization_state"], "authorized"
)
authorized = partition_active_peripheral_intent(
"开启仙人掌",
authorized_routing,
{"card": None},
)
self.assertEqual(
[item["capability"]["slot"] for item in authorized["peripheral_actions"]],
[11],
)
self.assertEqual(authorized["denied_peripheral_actions"], [])
service.delete_infrared_slot(11)
self.assertEqual(service.list_peripheral_routing_capabilities(), [])
def test_structured_slot_rejects_invalid_or_missing_device_id(self) -> None: def test_structured_slot_rejects_invalid_or_missing_device_id(self) -> None:
with TemporaryDirectory() as directory: with TemporaryDirectory() as directory:
service = SelfExtensionService( service = SelfExtensionService(
+72
View File
@@ -5,6 +5,10 @@ from types import SimpleNamespace
from unittest.mock import patch from unittest.mock import patch
from cognitive_agent.application.seat_service import SeatDemoService from cognitive_agent.application.seat_service import SeatDemoService
from cognitive_agent.application.self_extension import (
SelfExtensionService,
SimulationGateway,
)
from cognitive_agent.server import SeatDemoHandler from cognitive_agent.server import SeatDemoHandler
@@ -591,6 +595,74 @@ def test_pure_unresolved_peripheral_returns_cloud_ready_local_preview(
).is_file() ).is_file()
def test_new_unbound_infrared_device_is_parsed_but_never_dispatched(
tmp_path: Path,
) -> None:
handler = SeatDemoHandler.__new__(SeatDemoHandler)
handler.service = _service(tmp_path)
handler.extension = SelfExtensionService(
SimulationGateway(),
catalog_db_path=tmp_path / "infrared-slots.sqlite3",
)
handler.extension.save_infrared_slot(
11,
{
"device_id": "toy-cacti-002",
"object_names": ["仙人掌"],
"action_names": ["打开", "开启"],
"source": "tcp",
},
)
handler.nfc = SimpleNamespace(snapshot=lambda: {"card": None})
validated_events: list[dict[str, object]] = []
with patch.object(handler.extension, "execute_gateway") as execute_gateway:
result = handler._execute_request_with_peripherals(
"/api/requests",
{
"utterance": "打开仙人掌",
"provider": "local",
"conversation_id": "conv-unbound-cactus",
"execution_mode": "hardware",
},
on_commands_validated=validated_events.append,
)
execute_gateway.assert_not_called()
assert result is not None
assert result["status"] == "blocked"
assert result["phase"] == "authorization_blocked"
assert result["success"] is False
assert result["local_control_elapsed_ms"] < 1000
assert result["natural_language_routing"] == {
"original_utterance": "打开仙人掌",
"parsed_clauses": ["打开仙人掌"],
"resolved_clauses": ["打开仙人掌"],
"unresolved_clauses": [],
}
assert result["denied_peripheral_actions"][0]["reason"] == (
"nfc_binding_required"
)
operation = next(
operation
for config in result["function_configs"].values()
for operation in config["spec"]["operations"]
)
assert operation["capability"] == "infrared.emit"
assert operation["target"]["device"] == "toy-cacti-002"
assert operation["parameters"] == {"action": "打开", "slot": 11}
assert operation["authorization"] == {
"authorized": False,
"reason": "nfc_binding_required",
}
blocked = result["operation_results"][0]
assert blocked["status"] == "blocked"
assert blocked["telemetry"]["delivery_status"] == "not_submitted"
assert blocked["telemetry"]["transport_submitted"] is False
assert "尚未完成 NFC 绑定授权" in blocked["message"]
assert validated_events == [result]
def test_mixed_request_finishes_all_local_phases_before_cloud_ir() -> None: def test_mixed_request_finishes_all_local_phases_before_cloud_ir() -> None:
handler = SeatDemoHandler.__new__(SeatDemoHandler) handler = SeatDemoHandler.__new__(SeatDemoHandler)
order: list[str] = [] order: list[str] = []
+1 -1
View File
@@ -31,7 +31,7 @@ def test_vue_vite_application_is_the_primary_frontend_entry() -> None:
assert package["scripts"]["build"] == "vite build" assert package["scripts"]["build"] == "vite build"
assert 'import vue from "@vitejs/plugin-vue"' in vite assert 'import vue from "@vitejs/plugin-vue"' in vite
assert 'outDir: outputRoot' in vite assert 'outDir: outputRoot' in vite
assert 'const CACHE_VERSION = "cognitive-agent-pad-v316-sidebar-accordion"' in service_worker assert 'const CACHE_VERSION = "cognitive-agent-pad-v318-tech-nfc-routing"' in service_worker
assert "unpkg.com" not in html assert "unpkg.com" not in html
assert "cdn.jsdelivr.net" not in html assert "cdn.jsdelivr.net" not in html