"""Read-only, repository-backed templates for common cabin scenes. Scene templates are deliberately separate from saved mode profiles. Matching a template produces a new, unsaved draft; it never makes the template executable as a persisted mode and never lets a later edit overwrite repository YAML. """ from __future__ import annotations import hashlib import json import math import re from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path from typing import Any import yaml from cognitive_agent.application.assistant_plan_guard import ( MAX_PLAN_PARAMETERS, MAX_PLAN_REQUESTS, SCENE_COMPOSABLE_CAPABILITIES, ) from cognitive_agent.application.mode_profile import ModeOperation _TEMPLATE_ID = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") _SURROUNDING_PUNCTUATION = " \t\r\n，。,.!！?？;；:：'\"‘’“”《》<>【】[]（）()" _DOCUMENT_FIELDS = frozenset( {"id", "name", "aliases", "description", "icon", "operations"} ) def normalize_scene_template_name(value: object) -> str: """Normalize an exact user-facing name without enabling fuzzy matching.""" return "".join(str(value or "").strip(_SURROUNDING_PUNCTUATION).casefold().split()) def _clean_text(value: object, *, limit: int, field: str) -> str: text = " ".join(str(value or "").replace("\x00", " ").split()) if not text: raise ValueError(f"scene template {field} must not be empty") if len(text) > limit: raise ValueError(f"scene template {field} must be at most {limit} characters") return text def _json_scalar(value: object) -> bool: return ( isinstance(value, (str, bool, int)) or (isinstance(value, float) and math.isfinite(value)) ) @dataclass(frozen=True) class SceneTemplate: """One immutable common-scene seed made only from absolute set actions.""" id: str name: str aliases: tuple[str, ...] description: str icon: str operations: tuple[ModeOperation, ...] source: str revision: str @classmethod def from_yaml(cls, path: Path) -> "SceneTemplate": document = yaml.safe_load(path.read_text(encoding="utf-8")) if not isinstance(document, Mapping): raise ValueError(f"{path} must contain a YAML object") unknown_fields = set(document) - _DOCUMENT_FIELDS if unknown_fields: raise ValueError( f"{path} contains unsupported fields: {', '.join(sorted(unknown_fields))}" ) template_id = str(document.get("id") or "").strip() if not _TEMPLATE_ID.fullmatch(template_id): raise ValueError(f"{path} contains an invalid scene template id") name = _clean_text(document.get("name"), limit=64, field="name") raw_aliases = document.get("aliases", []) if not isinstance(raw_aliases, list): raise ValueError(f"{path} aliases must be an array") if any(not isinstance(raw_alias, str) for raw_alias in raw_aliases): raise ValueError(f"{path} aliases must contain only strings") alias_values = [name, *raw_aliases] if name.endswith("模式") and len(name) > 2: alias_values.append(name.removesuffix("模式")) aliases: list[str] = [] alias_keys: set[str] = set() for raw_alias in alias_values: alias = _clean_text(raw_alias, limit=64, field="alias") key = normalize_scene_template_name(alias) if not key or key in alias_keys: continue alias_keys.add(key) aliases.append(alias) raw_operations = document.get("operations") if not isinstance(raw_operations, list) or not raw_operations: raise ValueError(f"{path} must contain at least one operation") if len(raw_operations) > MAX_PLAN_REQUESTS: raise ValueError( f"{path} contains more than {MAX_PLAN_REQUESTS} scene operations" ) operations: list[ModeOperation] = [] operation_keys: set[tuple[str, str, str]] = set() for index, raw_operation in enumerate(raw_operations): if not isinstance(raw_operation, dict): raise ValueError(f"{path} operation {index} must be an object") operation = ModeOperation.from_dict(raw_operation) if operation.operation != "set": raise ValueError(f"{path} operation {index} must use absolute set") if not operation.domain or not operation.capability or not operation.target_scope: raise ValueError(f"{path} operation {index} is incomplete") if operation.capability not in SCENE_COMPOSABLE_CAPABILITIES: raise ValueError( f"{path} operation {index} uses a non-composable capability" ) if operation.domain != operation.capability.partition(".")[0]: raise ValueError( f"{path} operation {index} has a capability/domain mismatch" ) if not operation.parameters or len(operation.parameters) > MAX_PLAN_PARAMETERS: raise ValueError(f"{path} operation {index} has invalid parameters") if any( not isinstance(parameter_name, str) or not parameter_name.strip() or not _json_scalar(parameter_value) for parameter_name, parameter_value in operation.parameters.items() ): raise ValueError(f"{path} operation {index} has non-scalar parameters") key = (operation.domain, operation.capability, operation.target_scope) if key in operation_keys: raise ValueError(f"{path} repeats capability and target {key!r}") operation_keys.add(key) operations.append(operation) semantic_document = { "id": template_id, "name": name, "aliases": sorted(aliases, key=normalize_scene_template_name), "description": str(document.get("description") or ""), "icon": str(document.get("icon") or "sparkles"), "operations": sorted( (operation.to_yaml() for operation in operations), key=lambda item: ( str(item.get("domain") or ""), str(item.get("capability") or ""), str(item.get("target_scope") or ""), ), ), } canonical = json.dumps( semantic_document, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ) revision = hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16] return cls( id=template_id, name=name, aliases=tuple(aliases), description=str(document.get("description") or "")[:500], icon=str(document.get("icon") or "sparkles")[:64], operations=tuple(operations), source=str(path), revision=revision, ) def plan(self) -> dict[str, Any]: requests = [] for operation in self.operations: request = operation.to_request(self.name) request["duration_ms"] = None requests.append(request) return { "detected_domains": list( dict.fromkeys(operation.domain for operation in self.operations) ), "requests": requests, "ambiguities": [], "decision_summary": [ f"已命中内置常用模式模板“{self.name}”，等待工程安全链复核。" ], "confidence": 1.0, "metadata": { "source": "scene_template", "template_id": self.id, "template_revision": self.revision, }, } def public_dict(self, *, include_plan: bool = False) -> dict[str, Any]: result: dict[str, Any] = { "id": self.id, "name": self.name, "aliases": list(self.aliases), "description": self.description, "icon": self.icon, "domains": list(dict.fromkeys(item.domain for item in self.operations)), "operation_count": len(self.operations), "template_revision": self.revision, "origin": "builtin_scene_template", "read_only": True, } if include_plan: result["plan"] = self.plan() return result class SceneTemplateRegistry: """Load and exactly resolve the current project's common scene templates.""" def __init__(self, template_root: Path): self.template_root = Path(template_root) self._templates: tuple[SceneTemplate, ...] = () self._by_id: dict[str, SceneTemplate] = {} self._by_name: dict[str, SceneTemplate] = {} self.reload() def reload(self) -> None: templates = tuple( SceneTemplate.from_yaml(path) for path in sorted(self.template_root.rglob("*.yaml")) if path.is_file() ) if self.template_root.exists() else () by_id: dict[str, SceneTemplate] = {} by_name: dict[str, SceneTemplate] = {} for template in templates: if template.id in by_id: raise ValueError(f"duplicate scene template id: {template.id}") by_id[template.id] = template for value in (template.id, template.name, *template.aliases): key = normalize_scene_template_name(value) current = by_name.get(key) if current is not None and current.id != template.id: raise ValueError( f"ambiguous scene template name {value!r}: " f"{current.id}, {template.id}" ) by_name[key] = template self._templates = templates self._by_id = by_id self._by_name = by_name def templates(self) -> list[SceneTemplate]: return list(self._templates) def public_templates(self, *, include_plan: bool = False) -> list[dict[str, Any]]: return [template.public_dict(include_plan=include_plan) for template in self._templates] def resolve_exact_name(self, value: object) -> SceneTemplate | None: key = normalize_scene_template_name(value) return self._by_name.get(key) if key else None def get_template(self, template_id: str) -> SceneTemplate: try: return self._by_id[str(template_id).strip()] except KeyError as exc: raise KeyError(str(template_id).strip()) from exc __all__ = [ "SceneTemplate", "SceneTemplateRegistry", "normalize_scene_template_name", ]