Voice asks which device you mean when two fit the name

custom_components/jev/conversation.py

@@ -26,7 +26,8 @@
 import logging
 import re
 from collections.abc import Mapping
-from dataclasses import asdict, dataclass
+from dataclasses import asdict, dataclass, field, replace
+from datetime import datetime
 from typing import Any, Literal
 
 from homeassistant.components import conversation
@@ -38,10 +39,20 @@
 from homeassistant.helpers import entity_registry as er
 from homeassistant.helpers import intent as ha_intent
 from homeassistant.helpers import template, translation
+from homeassistant.helpers.chat_session import CONVERSATION_TIMEOUT
 from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
 from homeassistant.util import dt as dt_util
 from homeassistant.util import language as language_util
-from jevclient import ChoiceAnswer, JevAuthError, JevError, JevResponse, NoulAnswer
+from jevclient import (
+    Choice,
+    ChoiceAnswer,
+    JevAuthError,
+    JevError,
+    JevResponse,
+    Noul,
+    NoulAnswer,
+    Question,
+)
 
 from .const import (
     CONF_ALLOW_WHOLE_HOME,
@@ -53,9 +64,9 @@
 )
 from .coordinator import JevRuntimeData
 from .entity import build_device_info
-from .interpret import build_questions, interpret
+from .interpret import NONE, Interpretation, build_questions, interpret, spoken_name
 from .payload import payload_bytes
-from .snapshot import async_snapshot
+from .snapshot import HomeSnapshot, async_heard_in, async_snapshot
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -79,11 +90,25 @@
     ),
     "auth_failed": "TypeSafe rejected the API key. Check it in the Jev settings.",
     "unavailable": "TypeSafe did not answer. Try again in a moment.",
+    "which_device": "Do you mean {first} or {second}?",
 }
 
 PARALLEL_UPDATES = 0
 
 
+@dataclass(slots=True)
+class _Pending:
+    """A command held while the agent asks which device it meant."""
+
+    text: str
+    response: JevResponse
+    snapshot: HomeSnapshot
+    candidates: tuple[str, str]
+    expires: datetime = field(
+        default_factory=lambda: dt_util.utcnow() + CONVERSATION_TIMEOUT
+    )
+
+
 async def async_setup_entry(
     hass: HomeAssistant,
     entry: ConfigEntry,
@@ -104,6 +129,8 @@ def __init__(self, entry: ConfigEntry) -> None:
         self._attr_unique_id = f"{entry.entry_id}_conversation"
         runtime: JevRuntimeData = entry.runtime_data
         self._attr_device_info = build_device_info(entry.entry_id, runtime)
+        # Conversation id to the command waiting on its reply.
+        self._pending: dict[str, _Pending] = {}
 
     @property
     def supported_languages(self) -> list[str] | Literal["*"]:
@@ -159,12 +186,56 @@ async def _async_handle_message(
         runtime: JevRuntimeData = self._entry.runtime_data
 
         runtime.usage.roll_over(dt_util.now().date())
+        if (pending := self._take_pending(chat_log.conversation_id)) is not None:
+            resolved = await self._resolve(user_input, chat_log, pending)
+            if resolved is not None:
+                return resolved
+
         snapshot = async_snapshot(self.hass, MAX_CONVERSATION_ENTITIES)
         if not snapshot.entities:
             return await self._fall_back(user_input, "no entities are exposed to Assist")
 
         questions = build_questions(user_input.text, snapshot, MAX_CONVERSATION_ENTITIES)
         state = snapshot.as_state() | {"command": user_input.text}
+        response = await self._ask(user_input, state, questions)
+        if isinstance(response, conversation.ConversationResult):
+            return response
+
+        decision = interpret(
+            response,
+            user_input.text,
+            snapshot,
+            self._min_confidence,
+            heard_in=async_heard_in(
+                self.hass, user_input.satellite_id, user_input.device_id
+            ),
+        )
+        self._trace(
+            chat_log,
+            response,
+            {
+                "text": user_input.text,
+                "exposed_entities": len(snapshot.entities),
+                **asdict(decision),
+            },
+        )
+
+        if decision.candidates is not None:
+            return await self._ask_which(
+                user_input,
+                chat_log,
+                _Pending(user_input.text, response, snapshot, decision.candidates),
+            )
+        return await self._act(user_input, decision, user_input.text)
+
+    async def _ask(
+        self,
+        user_input: conversation.ConversationInput,
+        state: dict[str, Any],
+        questions: dict[str, Question],
+    ) -> JevResponse | conversation.ConversationResult:
+        """One call to Jev, inside the budget. A result means it did not answer."""
+        runtime: JevRuntimeData = self._entry.runtime_data
 
         # The budget covers voice as well as sensors, because a satellite that
         # mishears a wake word all night is exactly the runaway it exists to stop.
@@ -198,14 +269,20 @@ async def _async_handle_message(
         runtime.usage.record(response.usage.input_tokens, request_bytes)
         runtime.model_version = response.model or runtime.model_version
         runtime.usage.notify()
+        return response
 
-        decision = interpret(response, user_input.text, snapshot, self._min_confidence)
+    def _trace(
+        self,
+        chat_log: conversation.ChatLog,
+        response: JevResponse,
+        record: dict[str, Any],
+    ) -> None:
+        """Keep what one call decided, for diagnostics and the Assist dialog."""
+        runtime: JevRuntimeData = self._entry.runtime_data
         trace = {
-            "text": user_input.text,
             "latency_ms": response.latency_ms,
             "input_tokens": response.usage.input_tokens,
-            "exposed_entities": len(snapshot.entities),
-            **asdict(decision),
+            **record,
         }
         runtime.conversation_traces.appendleft(trace)
         # The pipeline records a chat log delta as an intent-progress event: the
@@ -221,6 +298,125 @@ async def _async_handle_message(
                 {"role": "assistant", "thinking_content": _reasoning(trace, response)},
             )
 
+    # --- asking which device ---
+
+    def _take_pending(self, conversation_id: str) -> _Pending | None:
+        """The command waiting on this conversation's reply, if it is still live.
+
+        A pending command lasts as long as Home Assistant keeps the chat session,
+        so a reply that arrives in a new session is never read as an answer.
+        """
+        now = dt_util.utcnow()
+        for key in [k for k, v in self._pending.items() if v.expires < now]:
+            del self._pending[key]
+        return self._pending.pop(conversation_id, None)
+
+    async def _ask_which(
+        self,
+        user_input: conversation.ConversationInput,
+        chat_log: conversation.ChatLog,
+        pending: _Pending,
+    ) -> conversation.ConversationResult:
+        """Ask which of two devices the command meant, and keep the command."""
+        first, second = (pending.snapshot.by_id(e) for e in pending.candidates)
+        assert first is not None and second is not None
+        names = spoken_name(first, second)
+        assert names is not None
+        language = user_input.language or self.hass.config.language
+        text = (await self._lines(language))["which_device"].format(
+            first=names[0], second=names[1]
+        )
+        self._pending[chat_log.conversation_id] = pending
+        # In the chat log, the next turn in this conversation carries the question
+        # it answers, and an LLM fallback agent reads the same history.
+        chat_log.async_add_assistant_content_without_tools(
+            conversation.AssistantContent(agent_id=self.entity_id, content=text)
+        )
+        response = ha_intent.IntentResponse(language=user_input.language)
+        response.async_set_speech(text)
+        # A satellite opens the microphone again for the answer.
+        return conversation.ConversationResult(
+            response=response,
+            conversation_id=chat_log.conversation_id,
+            continue_conversation=True,
+        )
+
+    async def _resolve(
+        self,
+        user_input: conversation.ConversationInput,
+        chat_log: conversation.ChatLog,
+        pending: _Pending,
+    ) -> conversation.ConversationResult | None:
+        """Run the kept command on the device the reply picked.
+
+        None when the reply picked neither, so the reply is handled as a new
+        command: "no, the kitchen light" and "never mind, lock up" both are one.
+        """
+        snapshot = pending.snapshot
+        options: dict[str, Any] = {
+            entity_id: described.as_option()
+            for entity_id in pending.candidates
+            if (described := snapshot.by_id(entity_id)) is not None
+        }
+        options[NONE] = "Neither of these, or a different request"
+        questions: dict[str, Question] = {
+            "which": Choice("Which device does the reply pick?", options),
+            # "Never mind, turn off the lamp in the bedroom" names one of the two, so
+            # the choice alone picks it and the kept command, turn on, runs on it.
+            # Measured on a development instance, twelve replies to "turn on the
+            # lamp": the six that only pick a device scored 0.08 to 0.26 here, and
+            # the six that ask for something else, that one included, 0.91 to 0.97.
+            "new_request": Noul(
+                "Does the reply ask for something of its own, rather than only "
+                "saying which device the command meant?",
+                true="The reply is a new instruction, or changes what should happen",
+                false="The reply only picks a device, however it is phrased",
+            ),
+        }
+        state = {"command": pending.text, "reply": user_input.text}
+        response = await self._ask(user_input, state, questions)
+        if isinstance(response, conversation.ConversationResult):
+            return response
+
+        answer = response.answers.get("which")
+        new_request = response.answers.get("new_request")
+        picked = (
+            answer.choice
+            if isinstance(answer, ChoiceAnswer)
+            and answer.choice in pending.candidates
+            and answer.confidence >= self._min_confidence
+            and not (isinstance(new_request, NoulAnswer) and new_request.noul >= 0.5)
+            else None
+        )
+        self._trace(
+            chat_log,
+            response,
+            {"text": user_input.text, "answers_command": pending.text, "picked": picked},
+        )
+        if picked is None:
+            return None
+
+        # The first call's answers stand, with the entity question settled. The
+        # command is read from its own sentence again, so a brightness it named
+        # still comes from the text.
+        settled = ChoiceAnswer(choice=picked, probabilities={picked: 1.0}, confidence=1.0)
+        first = replace(
+            pending.response, answers=pending.response.answers | {"entity": settled}
+        )
+        decision = interpret(
+            first, pending.text, snapshot, self._min_confidence, ask_back=False
+        )
+        return await self._act(user_input, decision, pending.text)
+
+    # --- acting ---
+
+    async def _act(
+        self,
+        user_input: conversation.ConversationInput,
+        decision: Interpretation,
+        text: str,
+    ) -> conversation.ConversationResult:
+        """Carry out one decision, or say why not."""
         if decision.already_satisfied is not None:
             name, settled = decision.already_satisfied
             return await self._speak(user_input, f"already_{settled}", name=name)
@@ -247,7 +443,7 @@ async def _async_handle_message(
                 DOMAIN,
                 decision.intent_type,
                 decision.slots,
-                user_input.text,
+                text,
                 user_input.context,
                 language=user_input.language,
                 assistant=conversation.DOMAIN,
@@ -364,11 +560,17 @@ async def _speak(
 
 def _reasoning(trace: Mapping[str, Any], response: JevResponse) -> str:
     """The trace as lines a person reads in the Assist dialog."""
-    lines = [
-        f"Jev: {trace['action'] or 'no action'}, {trace['reason']}, "
-        f"confidence {trace['confidence']:.2f}",
-        f"Slots: {json.dumps(trace['slots'], ensure_ascii=False)}",
-    ]
+    if "answers_command" in trace:
+        lines = [
+            f'Jev: a reply to "{trace["answers_command"]}", '
+            f"picked {trace['picked'] or 'neither'}"
+        ]
+    else:
+        lines = [
+            f"Jev: {trace['action'] or 'no action'}, {trace['reason']}, "
+            f"confidence {trace['confidence']:.2f}",
+            f"Slots: {json.dumps(trace['slots'], ensure_ascii=False)}",
+        ]
     for key, answer in response.answers.items():
         if isinstance(answer, ChoiceAnswer):
             ranked = sorted((answer.probabilities or {}).items(), key=lambda kv: -kv[1])[
@@ -380,10 +582,11 @@ def _reasoning(trace: Mapping[str, Any], response: JevResponse) -> str:
             lines.append(f"{key}: {answer.noul:.2f}")
         else:
             lines.append(f"{key}: {json.dumps(asdict(answer), ensure_ascii=False)}")
-    lines.append(
-        f"{response.model}, {trace['input_tokens']} input tokens, "
-        f"{trace['latency_ms']:.0f} ms, {trace['exposed_entities']} entities"
-    )
+    footer = f"{response.model}, {trace['input_tokens']} input tokens, "
+    footer += f"{trace['latency_ms']:.0f} ms"
+    if "exposed_entities" in trace:
+        footer += f", {trace['exposed_entities']} entities"
+    lines.append(footer)
     return "\n".join(lines)