add startup self-check for clashing custom intents

README.md

@@ -230,6 +230,7 @@ The integration can be set up entirely in the UI (**Settings** -> **Devices & se
 | `builtin_allowlist` | `[]` | A list of builtin intents to include for matching, even if `include_builtins` is off. Highly recommended over `include_builtins`. Allows you to e.g. enable `HassTurnOn` separately. |
 | `slot_extraction` | `true` | Extract slot values from the user's speech and substitute them into the canonical sentence. Disable to make the integration only correct slot-less phrases. (Why would you do this though, this is the best part!)|
 | `fallback_agent` | `conversation.home_assistant` | Agent consulted if no match for the canonical sentence is found. Default is Hassil itself, i.e. "no fallback". Set to an LLM agent if you want one. |
+| `startup_self_check` | `true` | After the candidate pool is built, feed every custom intent's own canonical form back through the matcher. Any intent whose perfect input does not route to itself is reported as a HomeAssistant repair issue. Useful for spotting two intents that shadow each other. Disable if the repair issue is noisy. |
 
 #### YAML configuration
 
@@ -243,6 +244,7 @@ closest_intent:
   builtin_allowlist: []
   slot_extraction: true
   fallback_agent: conversation.home_assistant
+  startup_self_check: true
 ```
 
  

custom_components/closest_intent/__init__.py

@@ -22,11 +22,13 @@
     CONF_INCLUDE_BUILTINS,
     CONF_SLOT_EXTRACTION,
     CONF_SLOT_THRESHOLD,
+    CONF_STARTUP_SELF_CHECK,
     CONF_THRESHOLD,
     DEFAULT_EXPANSION_CAP,
     DEFAULT_FALLBACK_AGENT,
     DEFAULT_INCLUDE_BUILTINS,
     DEFAULT_SLOT_EXTRACTION,
+    DEFAULT_STARTUP_SELF_CHECK,
     DEFAULT_THRESHOLD,
     DOMAIN,
     KEY_AGENT_INSTANCES,
@@ -60,6 +62,9 @@
                 vol.Optional(CONF_INCLUDE_BUILTINS, default=DEFAULT_INCLUDE_BUILTINS): cv.boolean,
                 vol.Optional(CONF_BUILTIN_ALLOWLIST, default=None): vol.Any(None, [cv.string]),
                 vol.Optional(CONF_SLOT_EXTRACTION, default=DEFAULT_SLOT_EXTRACTION): cv.boolean,
+                vol.Optional(
+                    CONF_STARTUP_SELF_CHECK, default=DEFAULT_STARTUP_SELF_CHECK
+                ): cv.boolean,
                 # Conversation entity to forward the canonical sentence to after a fuzzy match.
                 # Default is HA's bundled agent.
                 vol.Optional(CONF_FALLBACK_AGENT, default=DEFAULT_FALLBACK_AGENT): cv.string,

custom_components/closest_intent/config_flow.py

@@ -21,11 +21,13 @@
     CONF_INCLUDE_BUILTINS,
     CONF_SLOT_EXTRACTION,
     CONF_SLOT_THRESHOLD,
+    CONF_STARTUP_SELF_CHECK,
     CONF_THRESHOLD,
     DEFAULT_EXPANSION_CAP,
     DEFAULT_FALLBACK_AGENT,
     DEFAULT_INCLUDE_BUILTINS,
     DEFAULT_SLOT_EXTRACTION,
+    DEFAULT_STARTUP_SELF_CHECK,
     DEFAULT_THRESHOLD,
     DOMAIN,
     KEY_CONVERSATION_INTENTS,
@@ -121,6 +123,14 @@ def _build_schema(
                 default=defaults.get(CONF_SLOT_EXTRACTION, DEFAULT_SLOT_EXTRACTION),
                 description="Extract slot values from user speech",
             ): selector.BooleanSelector(),
+            vol.Required(
+                CONF_STARTUP_SELF_CHECK,
+                default=defaults.get(CONF_STARTUP_SELF_CHECK, DEFAULT_STARTUP_SELF_CHECK),
+                description=(
+                    "On startup, verify that each custom intent's own patterns still "
+                    "select that intent. Raises a repair issue if clashes are found."
+                ),
+            ): selector.BooleanSelector(),
             vol.Required(
                 CONF_INCLUDE_BUILTINS,
                 default=defaults.get(CONF_INCLUDE_BUILTINS, DEFAULT_INCLUDE_BUILTINS),

custom_components/closest_intent/const.py

@@ -27,11 +27,13 @@ def _read_version() -> str:
 CONF_BUILTIN_ALLOWLIST = "builtin_allowlist"
 CONF_SLOT_EXTRACTION = "slot_extraction"
 CONF_FALLBACK_AGENT = "fallback_agent"
+CONF_STARTUP_SELF_CHECK = "startup_self_check"
 
 DEFAULT_THRESHOLD = 70
 DEFAULT_EXPANSION_CAP = 16
 DEFAULT_INCLUDE_BUILTINS = False
 DEFAULT_SLOT_EXTRACTION = True
+DEFAULT_STARTUP_SELF_CHECK = True
 # Fallback conversation agent, used only when hassil errors or returns no
 # intent match. The canonical sentence itself always goes to hassil first.
 # Be careful not to create a loop...

custom_components/closest_intent/conversation.py

@@ -26,11 +26,13 @@
         CONF_INCLUDE_BUILTINS,
         CONF_SLOT_EXTRACTION,
         CONF_SLOT_THRESHOLD,
+        CONF_STARTUP_SELF_CHECK,
         CONF_THRESHOLD,
         DEFAULT_EXPANSION_CAP,
         DEFAULT_FALLBACK_AGENT,
         DEFAULT_INCLUDE_BUILTINS,
         DEFAULT_SLOT_EXTRACTION,
+        DEFAULT_STARTUP_SELF_CHECK,
         DEFAULT_THRESHOLD,
         DOMAIN,
         KEY_AGENT_INSTANCES,
@@ -59,11 +61,13 @@
         CONF_INCLUDE_BUILTINS,
         CONF_SLOT_EXTRACTION,
         CONF_SLOT_THRESHOLD,
+        CONF_STARTUP_SELF_CHECK,
         CONF_THRESHOLD,
         DEFAULT_EXPANSION_CAP,
         DEFAULT_FALLBACK_AGENT,
         DEFAULT_INCLUDE_BUILTINS,
         DEFAULT_SLOT_EXTRACTION,
+        DEFAULT_STARTUP_SELF_CHECK,
         DEFAULT_THRESHOLD,
         DOMAIN,
         KEY_AGENT_INSTANCES,
@@ -112,6 +116,7 @@ def opt(key, default):
         builtin_allowlist=opt(CONF_BUILTIN_ALLOWLIST, None),
         slot_extraction=opt(CONF_SLOT_EXTRACTION, DEFAULT_SLOT_EXTRACTION),
         fallback_agent_id=opt(CONF_FALLBACK_AGENT, DEFAULT_FALLBACK_AGENT),
+        startup_self_check=opt(CONF_STARTUP_SELF_CHECK, DEFAULT_STARTUP_SELF_CHECK),
         entry_id=entry.entry_id,
     )
     hass.data.setdefault(DOMAIN, {}).setdefault(KEY_AGENT_INSTANCES, {})[entry.entry_id] = agent
@@ -142,6 +147,7 @@ def opt(key, default):
         builtin_allowlist=opt(CONF_BUILTIN_ALLOWLIST, None),
         slot_extraction=opt(CONF_SLOT_EXTRACTION, DEFAULT_SLOT_EXTRACTION),
         fallback_agent_id=opt(CONF_FALLBACK_AGENT, DEFAULT_FALLBACK_AGENT),
+        startup_self_check=opt(CONF_STARTUP_SELF_CHECK, DEFAULT_STARTUP_SELF_CHECK),
     )
 
 
@@ -162,6 +168,7 @@ def __init__(
         builtin_allowlist: list[str] | None,
         slot_extraction: bool,
         fallback_agent_id: str,
+        startup_self_check: bool = DEFAULT_STARTUP_SELF_CHECK,
         entry_id: str,
     ) -> None:
         self.hass = hass
@@ -173,6 +180,7 @@ def __init__(
         self._builtin_allowlist = set(builtin_allowlist) if builtin_allowlist else None
         self._slot_extraction = slot_extraction
         self._fallback_agent_id = fallback_agent_id
+        self._startup_self_check = startup_self_check
         self._entry_id = entry_id
 
         # Per-language pools: built lazily on first request for that
@@ -184,6 +192,7 @@ def __init__(
         self._pools: dict[str, tuple[Resolver, list[Candidate], list[Candidate]]] = {}
         self._pool_locks: dict[str, asyncio.Lock] = {}
         self._builtin_intents_cache: dict[str, dict[str, list[str]]] = {}
+        self._self_check_issue_ids: dict[str, str] = {}
         self._rebuild_handle = None  # async_call_later cancel handle
         self._unsub_listeners: list = []
 
@@ -238,6 +247,7 @@ def apply_options(
         builtin_allowlist: list[str] | None,
         slot_extraction: bool,
         fallback_agent_id: str,
+        startup_self_check: bool = DEFAULT_STARTUP_SELF_CHECK,
     ) -> None:
         self._threshold = threshold
         self._slot_threshold = slot_threshold
@@ -247,6 +257,7 @@ def apply_options(
         self._builtin_allowlist = set(builtin_allowlist) if builtin_allowlist else None
         self._slot_extraction = slot_extraction
         self._fallback_agent_id = fallback_agent_id
+        self._startup_self_check = startup_self_check
         # Anything affecting candidate composition invalidates the pools.
         self._pools.clear()
         self._builtin_intents_cache.clear()
@@ -310,7 +321,17 @@ async def _async_get_pool(
             )
             self._pools[language] = pool
             self._builtin_intents_cache[language] = builtin_intents
-            return pool
+
+        if self._startup_self_check:
+            try:
+                clashes = await self.hass.async_add_executor_job(
+                    self._self_check, language, pool[0], pool[1], pool[2]
+                )
+            except Exception:  # pragma: no cover
+                _LOGGER.exception("[%s] self-check raised", language)
+                clashes = []
+            self._publish_self_check_issue(language, clashes)
+        return pool
 
     def _find_default_agent(self):
         get_agent = getattr(conversation, "async_get_agent", None)
@@ -711,6 +732,96 @@ def _best_extractable_sibling(
                 return (c, captured, s)
         return None
 
+    def _self_check(
+        self,
+        language: str,
+        resolver: Resolver,
+        user_candidates: list[Candidate],
+        builtin_candidates: list[Candidate],
+    ) -> list[dict]:
+        """
+        For each user candidate, feed its own canonical form back through the matcher.
+        Anything that does not round-trip to the same intent is reported as a clash.
+        """
+        if not user_candidates:
+            return []
+        combined = user_candidates + builtin_candidates
+        clashes: list[dict] = []
+        for c in user_candidates:
+            perfect = _materialise_candidate_input(c)
+            if not perfect:
+                continue
+            try:
+                detail = self._match(perfect, resolver, combined)
+            except Exception:  # pragma: no cover
+                _LOGGER.exception("self_check: match raised for %r", perfect)
+                continue
+            if detail is None:
+                clashes.append(
+                    {
+                        "expected_intent": c.intent,
+                        "pattern": _pretty_pattern(c),
+                        "input": perfect,
+                        "got_intent": None,
+                        "got_pattern": None,
+                        "score": None,
+                    }
+                )
+                continue
+            winner, _, score_value, _ = detail
+            if winner.intent != c.intent:
+                clashes.append(
+                    {
+                        "expected_intent": c.intent,
+                        "pattern": _pretty_pattern(c),
+                        "input": perfect,
+                        "got_intent": winner.intent,
+                        "got_pattern": _pretty_pattern(winner),
+                        "score": score_value,
+                    }
+                )
+        if clashes:
+            _LOGGER.warning(
+                "[%s] self-check found %d intent clash(es); see repairs UI for details",
+                language,
+                len(clashes),
+            )
+        else:
+            _LOGGER.debug("[%s] self-check passed (%d intents)", language, len(user_candidates))
+        return clashes
+
+    def _publish_self_check_issue(self, language: str, clashes: list[dict]) -> None:
+        try:
+            from homeassistant.helpers import issue_registry as ir  # type: ignore
+        except ImportError:  # pragma: no cover - test stub path
+            return
+        prior_id = self._self_check_issue_ids.pop(language, None)
+        if prior_id is not None:
+            try:
+                ir.async_delete_issue(self.hass, DOMAIN, prior_id)
+            except Exception:  # pragma: no cover
+                _LOGGER.exception("[%s] failed to delete prior self-check issue", language)
+        if not clashes:
+            return
+        issue_id = f"self_check_{self._entry_id}_{language}_{len(clashes)}"
+        try:
+            ir.async_create_issue(
+                self.hass,
+                DOMAIN,
+                issue_id,
+                is_fixable=False,
+                severity=ir.IssueSeverity.WARNING,
+                translation_key="self_check_clashes",
+                translation_placeholders={
+                    "language": language,
+                    "count": str(len(clashes)),
+                    "details": _format_clashes(clashes),
+                },
+            )
+            self._self_check_issue_ids[language] = issue_id
+        except Exception:  # pragma: no cover
+            _LOGGER.exception("[%s] failed to publish self-check issue", language)
+
     async def parse_sentence(
         self,
         language: str,
@@ -967,6 +1078,7 @@ def _keep(intent_name: str) -> bool:
             "slot_extraction": self._slot_extraction,
             "fallback_agent_id": self._fallback_agent_id,
             "denylist": sorted(self._denylist) if self._denylist else None,
+            "startup_self_check": self._startup_self_check,
             "languages": {},
         }
         if intent_filter is not None:
@@ -1017,6 +1129,79 @@ def _no_match(
     )
 
 
+_SELFCHECK_SLOT_SENTINEL_PREFIX = "zqzqxslotx"
+_SELFCHECK_SLOT_SENTINEL_SUFFIX = "xzqzq"
+
+
+def _materialise_candidate_input(c: Candidate) -> str:
+    """Return the candidate text with each ``SLOT_WILDCARD`` replaced by a unique sentinel."""
+    if SLOT_WILDCARD not in c.text:
+        return re.sub(r"\s+", " ", c.text).strip()
+    parts = c.text.split(SLOT_WILDCARD)
+    pieces = [parts[0]]
+    for i, part in enumerate(parts[1:]):
+        sentinel = f" {_SELFCHECK_SLOT_SENTINEL_PREFIX}{i}{_SELFCHECK_SLOT_SENTINEL_SUFFIX} "
+        pieces.append(sentinel)
+        pieces.append(part)
+    return re.sub(r"\s+", " ", "".join(pieces)).strip()
+
+
+def _pretty_pattern(c: Candidate) -> str:
+    """Render a candidate's pattern with ``{slot_name}`` placeholders restored."""
+    src = c.display_text or c.text
+    if SLOT_WILDCARD not in src:
+        return re.sub(r"\s+", " ", src).strip()
+    parts = src.split(SLOT_WILDCARD)
+    n_slots = len(parts) - 1
+    names = list(c.slot_names) + ["slot"] * max(0, n_slots - len(c.slot_names))
+    pieces = [parts[0]]
+    for slot_name, part in zip(names[:n_slots], parts[1:], strict=False):
+        pieces.append("{" + slot_name + "}")
+        pieces.append(part)
+    return re.sub(r"\s+", " ", "".join(pieces)).strip()
+
+
+def _format_clashes(clashes: list[dict]) -> str:
+    """Markdown summary grouped by source intent."""
+    grouped: dict[str, list[dict]] = {}
+    for c in clashes:
+        grouped.setdefault(c["expected_intent"], []).append(c)
+
+    sections: list[str] = []
+    for source_intent in sorted(grouped):
+        entries = grouped[source_intent]
+        seen: set[tuple] = set()
+        unique: list[dict] = []
+        for c in entries:
+            key = (c["pattern"], c.get("got_intent"), c.get("got_pattern"))
+            if key in seen:
+                continue
+            seen.add(key)
+            unique.append(c)
+
+        # Sort: real shadowers alphabetically, "matched nothing" at the end.
+        unique.sort(
+            key=lambda c: (
+                c.get("got_intent") is None,
+                (c.get("got_intent") or "").lower(),
+                c["pattern"].lower(),
+            )
+        )
+
+        lines = [f"### `{source_intent}`", ""]
+        for c in unique:
+            if c.get("got_intent") is None:
+                lines.append(f"- `{c['pattern']}` matched nothing above threshold")
+            else:
+                lines.append(
+                    f"- {c['pattern']}` is matched as `{c['got_pattern']}`"
+                    f"`from `{c['got_intent']}` (score {c['score']})"
+                )
+        sections.append("\n".join(lines))
+
+    return "\n\n".join(sections)
+
+
 def _parse_raw_list_values(raw_def) -> list[str]:
     """
     Pull plain string values from the raw YAML form of a Hassil slot list.

custom_components/closest_intent/strings.json

@@ -0,0 +1,8 @@
+{
+  "issues": {
+    "self_check_clashes": {
+      "title": "Closest Intent: {count} clashing custom intent(s)",
+      "description": "Closest Intent's startup self-check feeds each of your custom intents' canonical sentence through the fuzzy matcher. For **{count}** intent(s) in language `{language}`, the matcher routed these perfect inputs to a *different* intent rather than the one it came from.\n\nThis usually means two custom intents have fuzzy-overlapping patterns and one is consistently shadowing the other on perfect input.\n\n**Clashes, grouped by the intent that should have been matched:**\n\n{details}\n\n**Before opening an issue about these**, please try to determine if these matches are legitimate fuzzy-matching behavior errors, or configuration errors. Use the `parse_sentence` and `dump_candidates` debug tools.\n\nIf you believe the errors are genuinely bugs in Closest Intent, please open an issue with as much detail as you can provide! In the meantime, here's some stop-gap measure you can take:\n\n- Narrow the source pattern: add more anchoring words so it stops matching the shadowing input.\n- Add the shadowing intent's name to `denylist` if you do not actually use it.\n- Set `startup_self_check: false` if the clashes are acceptable for your usage."
+    }
+  }
+}