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

feat(n7web): refine spatial UI and NFC-gated extension routing

Browse Source
This commit is contained in:
Miayui
2026-09-16 16:14:10 +08:00
parent 0e4328f548
commit 75ce391312
16 changed files with 925 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cognitive_agent/application/self_extension.py
+180 -59
View File
@@ -380,6 +380,86 @@ class InfraredSlotCatalog:
capabilities.append(sanitized)
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(
self,
catalog: object,
@@ -1244,6 +1324,28 @@ def resolve_active_peripheral_command(
catalog,
nfc_snapshot,
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],
*,
require_enabled: bool,
require_binding: bool = True,
) -> dict[str, Any] | None:
"""Resolve a persistently card-bound command, including disabled entries.
@@ -1279,7 +1382,7 @@ def _resolve_bound_peripheral_command(
):
continue
bound_uid = _normalized_card_uid(raw_item.get("card_uid"))
if not bound_uid:
if require_binding and not bound_uid:
continue
device_score = _device_match_score(compact_utterance, raw_item)
if not device_score:
@@ -1310,6 +1413,17 @@ def _resolve_bound_peripheral_command(
resolved = dict(matches[0][2])
resolved["slot"] = validate_slot(resolved.get("slot"))
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
@@ -1326,10 +1440,12 @@ def partition_active_peripheral_intent(
) -> dict[str, Any]:
"""Claim only explicit learned-peripheral clauses and preserve CAN text.
The function is deliberately conservative: a clause must contain both a
persistently NFC-bound device identity and one learned action. 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.
The function is deliberately conservative: a clause must contain both an
explicit learned device identity and one learned action. A matching but
unbound device is returned as an authorization denial, never as residual
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()
@@ -1342,12 +1458,16 @@ def partition_active_peripheral_intent(
"denied_peripheral_actions": [],
}
bound_catalog = [
routing_catalog = [
item
for item in catalog[:MAX_CATALOG_ITEMS]
if isinstance(item, dict)
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 = [
item
@@ -1355,8 +1475,8 @@ def partition_active_peripheral_intent(
if item.get("enabled", True) is not False
]
compact_original = _compact_match_text(original)
if not bound_catalog or not any(
_device_match_score(compact_original, item) for item in bound_catalog
if not routing_catalog or not any(
_device_match_score(compact_original, item) for item in routing_catalog
):
return {
"peripheral_actions": [],
@@ -1380,36 +1500,30 @@ def partition_active_peripheral_intent(
contextual_device_name = ""
for index, clause in enumerate(clauses):
compact_clause = _compact_match_text(clause)
explicit_bound_candidates = [
explicit_routing_candidates = [
item
for item in bound_catalog
for item in routing_catalog
if _device_match_score(compact_clause, item)
]
explicit_candidates = [
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(
if explicit_routing_candidates and _has_unshadowed_vehicle_device_phrase(
compact_clause,
explicit_bound_candidates,
explicit_routing_candidates,
):
residual.append(clause)
contextual_device_id = None
contextual_device_name = ""
continue
clause_candidates = explicit_candidates
bound_clause_candidates = explicit_bound_candidates
clause_candidates = explicit_routing_candidates
resolution_text = clause
if explicit_bound_candidates:
if explicit_routing_candidates:
candidate_device_ids = {
str(item.get("device_id") or "")
for item in explicit_bound_candidates
for item in explicit_routing_candidates
}
if len(candidate_device_ids) == 1:
contextual_device_id = next(iter(candidate_device_ids))
identity_names: list[str] = []
for item in explicit_bound_candidates:
for item in explicit_routing_candidates:
for field in ("object_names", "device_aliases"):
values = item.get(field, [])
if isinstance(values, str):
@@ -1454,56 +1568,51 @@ def partition_active_peripheral_intent(
):
clause_candidates = [
item
for item in active_catalog
if str(item.get("device_id") or "") == contextual_device_id
]
bound_clause_candidates = [
item
for item in bound_catalog
for item in routing_catalog
if str(item.get("device_id") or "") == contextual_device_id
]
resolution_text = f"{contextual_device_name}，{clause}"
if not clause_candidates and not bound_clause_candidates:
if not clause_candidates:
residual.append(clause)
continue
active_resolution_catalog = (
clause_candidates if not explicit_candidates else active_catalog
routing_resolution_catalog = (
clause_candidates if not explicit_routing_candidates else routing_catalog
)
resolved = resolve_active_peripheral_command(
resolution_text, active_resolution_catalog, nfc_snapshot
resolved = resolve_parseable_peripheral_command(
resolution_text, routing_resolution_catalog, nfc_snapshot
)
if resolved is None:
bound_resolution_catalog = (
bound_clause_candidates
if not explicit_bound_candidates
else bound_catalog
)
bound_match = _resolve_bound_peripheral_command(
resolution_text,
bound_resolution_catalog,
nfc_snapshot,
require_enabled=False,
)
if (
bound_match is not None
and bound_match.get("enabled", True) is False
):
denied.append(
{
"index": index,
"clause": clause,
"capability": bound_match,
"reason": "nfc_command_disabled",
}
)
continue
# Explicit third-party device text must never fall through into
# the vehicle CAN model as if it were a cabin actuator request.
unresolved.append(clause)
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 clause_candidates
for item in active_catalog
if item.get("id")
and str(item.get("device_id") or "") in authorized_device_ids
)
continue
if not _normalized_card_uid(resolved.get("card_uid")):
denied.append(
{
"index": index,
"clause": clause,
"capability": resolved,
"reason": "nfc_binding_required",
}
)
continue
if resolved.get("enabled", True) is False:
denied.append(
{
"index": index,
"clause": clause,
"capability": resolved,
"reason": "nfc_command_disabled",
}
)
continue
actions.append(
@@ -2275,6 +2384,11 @@ class SelfExtensionService:
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]:
"""Persist the complete learned capability catalog from the NFC UI."""
@@ -2441,6 +2555,13 @@ class SelfExtensionService:
) -> dict[str, Any]:
"""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):
raise ExtensionIntegrationError(
"外设名称未在原始指令中被明确提及，已拒绝红外执行。"