Voice: Jabra fallback when the car is away + owner's trailing @eclass

carwatch/agent.py

@@ -221,7 +221,14 @@ def _mentions_me(msg: dict, handle: str, owner: str = "") -> bool:
     # #32. A message is addressed to the car when the handle LEADS it, or
     # when it is a reply to one of the car's own posts. A mention buried in
     # a sentence is somebody talking about the car; stay quiet.
-    if not _addressed_to(msg, handle):
+    # The owner typing the literal @handle anywhere is talking TO the car:
+    # "can you speak using the jabra @eclass" (petrus, 26 Sep) went
+    # unanswered because the handle trailed. #32's third-person false
+    # triggers came from agents, which the owner gate below still stops.
+    owner_tagged = (handle.lower() in body.lower()
+                    and bool((owner or "").strip())
+                    and _owner_ok(sender, owner))
+    if not owner_tagged and not _addressed_to(msg, handle):
         return False
     # ONLY the owner addresses the car through the room. Fellow agents
     # DISCUSSING the car ("eclass", "E Class" in ordinary sentences) kept

carwatch/listen.py

@@ -433,8 +433,15 @@ def _speak(text: str) -> bool:
                 subprocess.run(["bluetoothctl", "connect", car],
                                capture_output=True, timeout=10)
                 time.sleep(2)
-            target = f"bluealsa:DEV={car},PROFILE=a2dp"
-            bt = True
+            # Only a LIVE link counts. The saved MAC outlives the car: on the
+            # kitchen table (26 Sep, Jabra) every answer went to the absent
+            # car's A2DP and the speakerphone on the desk stayed silent.
+            if _bt_pcm_mac("a2dpsrc/sink") == car:
+                target = f"bluealsa:DEV={car},PROFILE=a2dp"
+                bt = True
+            else:
+                print(f"speak: car {car} not connected, trying USB/headset",
+                      flush=True)
         if not target:
             target = _usb_audio_device("playback")
         if not target:

carwatch/selfstate.py

@@ -15,6 +15,7 @@
 
 import glob
 import os
+import re
 import shutil
 import subprocess
 
@@ -170,6 +171,42 @@ def manual_status() -> str | None:
     return None
 
 
+def _usb_audio_names(aplay_l: str) -> list[str]:
+    """Card names of USB audio devices in `aplay -l` output, e.g.
+    'card 0: MS [Jabra Speak2 40 MS], device 0: USB Audio [USB Audio]'."""
+    names = []
+    for line in (aplay_l or "").splitlines():
+        m = re.match(r"card \d+: \S+ \[([^\]]+)\], device \d+: USB Audio", line)
+        if m and m.group(1) not in names:
+            names.append(m.group(1))
+    return names
+
+
+def voice() -> str | None:
+    """What the car speaks and listens through right now. Without this the
+    car told petrus it had no Jabra while one sat plugged into it (26 Sep)."""
+    usb = _usb_audio_names(_run(["aplay", "-l"]) or "")
+    car = None
+    try:
+        mac = open(os.path.expanduser("~/.carwatch/car-bt-mac")).read().strip()
+        if mac:
+            info = _run(["bluetoothctl", "info", mac]) or ""
+            car = "Connected: yes" in info
+    except Exception:
+        pass
+    parts = []
+    if car:
+        parts.append("you speak through the car's own speakers over Bluetooth")
+    if usb:
+        dev = " and ".join(usb)
+        parts.append(f"a USB speakerphone is plugged into your onboard computer ({dev}); "
+                     "you hear through its microphone"
+                     + ("" if car else " and speak your answers aloud through its speaker"))
+    if car is False:
+        parts.append("the car's Bluetooth audio is not connected right now")
+    return "; ".join(parts) or None
+
+
 def live_facts() -> dict[str, str]:
     """Sensor readings the car may legitimately assert about itself."""
     facts: dict[str, str] = {}
@@ -193,6 +230,7 @@ def live_facts() -> dict[str, str]:
         ("brain", serving_model()),
         ("network", network()),
         ("your manual", manual_status()),
+        ("your voice", voice()),
     ):
         if val:
             facts[key] = val

scripts/car-speak.sh

@@ -40,6 +40,18 @@ speak() {
         # Bond can exist while A2DP is down (measured: Paired yes, Connected no).
         bluetoothctl connect "$mac" >/dev/null 2>&1 || true
         sleep 2
+        # The saved MAC outlives the car: away from it (26 Sep, Jabra on the
+        # kitchen table) speak to a USB speakerphone instead of into silence.
+        if ! bluetoothctl info "$mac" 2>/dev/null | grep -q "Connected: yes"; then
+            local usb
+            usb=$(aplay -l 2>/dev/null | awk '/^card [0-9]+:.*USB Audio/{sub(":","",$2); print $2; exit}')
+            if [ -n "$usb" ]; then
+                timeout 60 aplay -q -D "plughw:$usb,0" "$wav" 2>/dev/null || \
+                echo "USB playback failed - is the voice listener holding the speakerphone?"
+                rm -f "$wav"
+                return
+            fi
+        fi
         # aplay = send direction; bluealsa-aplay was the receive tool and
         # blocked forever (27 Aug). Timeout guards the wedge.
         timeout 30 aplay --buffer-time=1000000 -D "bluealsa:DEV=$mac,PROFILE=a2dp" "$wav" 2>/dev/null || \

tests/test_speak_route.py

@@ -0,0 +1,111 @@
+"""Voice answers were silent on the Jabra (petrus, 26 Sep 2026, vadelma).
+
+~/.carwatch/car-bt-mac held the car's MAC while the car was away, so
+_speak aimed every answer at bluealsa A2DP for a head unit that was not
+connected and never tried the USB speakerphone. The room gate also dropped
+"can you speak using the jabra @eclass" because the handle trailed. These
+tests pin both: the car is used only when its A2DP link is live, and the
+owner's literal @handle anywhere addresses the car while agents' mentions
+still do not. subprocess and piper are mocked: nothing touches real audio.
+"""
+import os
+import sys
+import unittest
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+CAR = "88:5A:85:66:67:C2"
+
+
+class TestSpeakRoute(unittest.TestCase):
+    def setUp(self):
+        from carwatch import listen
+        self.L = listen
+        wav = os.path.join(os.path.dirname(__file__), "_speak_route.wav")
+        open(wav, "wb").close()
+        self.wav = wav
+        self.addCleanup(lambda: os.path.exists(wav) and os.unlink(wav))
+
+    def _route(self, a2dp_mac):
+        played = []
+
+        def run(cmd, **kw):
+            if cmd[0] == "aplay":
+                played.append(cmd[cmd.index("-D") + 1])
+            return mock.Mock(returncode=0)
+
+        from carwatch import voiceroom
+        with mock.patch.object(voiceroom, "tts_wav", return_value=self.wav), \
+                mock.patch.object(self.L, "_car_a2dp_mac", return_value=CAR), \
+                mock.patch.object(self.L, "_bt_pcm_mac",
+                                  side_effect=lambda s: a2dp_mac if s == "a2dpsrc/sink" else None), \
+                mock.patch.object(self.L, "_usb_audio_device", return_value="plughw:0,0"), \
+                mock.patch.object(self.L, "_echo_tail_sec", return_value=0.0), \
+                mock.patch.object(self.L.subprocess, "run", side_effect=run), \
+                mock.patch.object(self.L.time, "sleep"):
+            self.assertTrue(self.L._speak("hello"))
+        return played
+
+    def test_absent_car_falls_back_to_usb(self):
+        self.assertEqual(self._route(None), ["plughw:0,0"])
+
+    def test_connected_car_still_wins(self):
+        self.assertEqual(self._route(CAR), [f"bluealsa:DEV={CAR},PROFILE=a2dp"])
+
+
+class TestOwnerTrailingHandle(unittest.TestCase):
+    def setUp(self):
+        from carwatch import agent
+        self.agent = agent
+
+    def test_owner_trailing_handle_is_addressed(self):
+        msg = {"from": "petrus", "body": "can you speak using the jabra @eclass"}
+        self.assertTrue(self.agent._mentions_me(msg, "@eclass", owner="petrus"))
+
+    def test_agent_trailing_handle_stays_quiet(self):
+        msg = {"from": "@claudeMB", "body": "the line @eclass just posted"}
+        self.assertFalse(self.agent._mentions_me(msg, "@eclass", owner="petrus"))
+
+    def test_owner_spoken_name_mid_sentence_stays_quiet(self):
+        msg = {"from": "petrus", "body": "I parked the eclass outside"}
+        self.assertFalse(self.agent._mentions_me(msg, "@eclass", owner="petrus"))
+
+
+class TestVoiceFact(unittest.TestCase):
+    """The car said "I don't have a Jabra speaker" with one plugged in."""
+
+    APLAY = ("**** List of PLAYBACK Hardware Devices ****\n"
+             "card 0: MS [Jabra Speak2 40 MS], device 0: USB Audio [USB Audio]\n"
+             "  Subdevices: 1/1\n"
+             "card 1: vc4hdmi0 [vc4-hdmi-0], device 0: MAI PCM i2s-hifi-0 [MAI PCM i2s-hifi-0]\n")
+
+    def setUp(self):
+        from carwatch import selfstate
+        self.S = selfstate
+
+    def test_usb_names_skip_hdmi(self):
+        self.assertEqual(self.S._usb_audio_names(self.APLAY), ["Jabra Speak2 40 MS"])
+
+    def _voice(self, bt_info):
+        def run(cmd, timeout=5):
+            return self.APLAY if cmd[0] == "aplay" else bt_info
+        with mock.patch.object(self.S, "_run", side_effect=run), \
+                mock.patch("builtins.open", mock.mock_open(read_data=CAR)):
+            return self.S.voice()
+
+    def test_car_away_names_the_jabra_as_the_speaker(self):
+        v = self._voice("Device x\n\tConnected: no\n")
+        self.assertIn("Jabra Speak2 40 MS", v)
+        self.assertIn("speak your answers aloud through its speaker", v)
+        self.assertIn("Bluetooth audio is not connected", v)
+
+    def test_car_connected_speaks_through_the_car(self):
+        v = self._voice("Device x\n\tConnected: yes\n")
+        self.assertIn("car's own speakers", v)
+        self.assertNotIn("aloud through its speaker", v)
+
+    def test_nothing_plugged_in_says_nothing(self):
+        with mock.patch.object(self.S, "_run", return_value=""), \
+                mock.patch("builtins.open", side_effect=OSError):
+            self.assertIsNone(self.S.voice())