Merge branch 'main' into dependabot/github_actions/actions/setup-python-7

Cancel was the only timer event to drop `timer_data` and emit a bare
`idle`. Since `idle` also ends an ordinary conversation, a peripheral
could not tell a cancel from "finished speaking", nor which timer had
gone.

Emit `timer_updated` with the timer's identity and `seconds_left` zeroed,
then go idle as before. Shipped peripherals already read `seconds_left`
from `timer_updated` and none treats zero as ringing, so the countdown
clears with no peripheral changes. The zeroed update has to land before
`idle`, or a peripheral that re-checks on `idle` stays in the timer
state.

Fixes #367

.github/workflows/lint.yml

@@ -16,7 +16,7 @@ jobs:
     steps:
     - uses: actions/checkout@v7
     - name: Set up Python 3.13
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@v7
       with:
         python-version: "3.13"
     - name: Install dependencies
@@ -31,7 +31,7 @@ jobs:
     steps:
     - uses: actions/checkout@v7
     - name: Set up Python 3.13
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@v7
       with:
         python-version: "3.13"
     - name: Install dependencies
@@ -46,7 +46,7 @@ jobs:
     steps:
     - uses: actions/checkout@v7
     - name: Set up Python 3.13
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@v7
       with:
         python-version: "3.13"
     - name: Install dependencies
@@ -61,7 +61,7 @@ jobs:
     steps:
     - uses: actions/checkout@v7
     - name: Set up Python 3.13
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@v7
       with:
         python-version: "3.13"
     - name: Install dependencies
@@ -77,7 +77,7 @@ jobs:
     steps:
     - uses: actions/checkout@v7
     - name: Set up Python 3.13
-      uses: actions/setup-python@v5
+      uses: actions/setup-python@v7
       with:
         python-version: "3.13"
     - name: Install dependencies

.github/workflows/tests.yml

@@ -19,7 +19,7 @@ jobs:
         uses: actions/checkout@v7
 
       - name: Set up Python 3.12
-        uses: actions/setup-python@v5
+        uses: actions/setup-python@v7
         with:
           python-version: "3.12"
 

CHANGELOG.md

@@ -1,3 +1,29 @@
+<a id="v1.1.15"></a>
+# [v1.1.15](https://github.com/OHF-Voice/linux-voice-assistant/releases/tag/v1.1.15) - 2026-08-02
+
+<!-- Release notes generated using configuration in .github/release.yml at main -->
+
+## What's Changed
+### Hotfix
+* Fix variable handling in docker entrypoint script
+
+### Exciting New Features 🎉
+* Add --music-output-device to route music separately from voice by [@matthewdva](https://github.com/matthewdva) in [#350](https://github.com/OHF-Voice/linux-voice-assistant/pull/350)
+
+### Other Changes
+* Update Jabra Speak DOCS.md  by [@mrbubble62](https://github.com/mrbubble62) in [#387](https://github.com/OHF-Voice/linux-voice-assistant/pull/387)
+* Bump docker/login-action from 4.5.1 to 4.6.0 by [@dependabot](https://github.com/dependabot)[bot] in [#386](https://github.com/OHF-Voice/linux-voice-assistant/pull/386)
+* Documentation Improvements by [@omaramin-2000](https://github.com/omaramin-2000) in [#385](https://github.com/OHF-Voice/linux-voice-assistant/pull/385)
+
+## New Contributors
+* [@mrbubble62](https://github.com/mrbubble62) made their first contribution in [#387](https://github.com/OHF-Voice/linux-voice-assistant/pull/387)
+* [@matthewdva](https://github.com/matthewdva) made their first contribution in [#350](https://github.com/OHF-Voice/linux-voice-assistant/pull/350)
+
+**Full Changelog**: https://github.com/OHF-Voice/linux-voice-assistant/compare/v1.1.14...v1.1.15
+
+[Changes][v1.1.15]
+
+
 <a id="v1.1.14"></a>
 # [v1.1.14](https://github.com/OHF-Voice/linux-voice-assistant/releases/tag/v1.1.14) - 2026-07-28
 
@@ -364,6 +390,7 @@ Initial release
 [Changes][v1.0.0]
 
 
+[v1.1.15]: https://github.com/OHF-Voice/linux-voice-assistant/compare/v1.1.14...v1.1.15
 [v1.1.14]: https://github.com/OHF-Voice/linux-voice-assistant/compare/v1.1.13...v1.1.14
 [v1.1.13]: https://github.com/OHF-Voice/linux-voice-assistant/compare/v1.1.12...v1.1.13
 [v1.1.12]: https://github.com/OHF-Voice/linux-voice-assistant/compare/v1.1.11...v1.1.12

README.md

@@ -53,7 +53,7 @@ Alternatively if on a lower budget then suggest could use other microphone-array
 For HA OS, we provide a finished [Assist Satellite](https://github.com/OHF-Voice/apps/tree/main/assist_satellite) app (formerly add-on), which uses the Linux Voice Assistant runtime to turn your HA host into a voice satellite. 
 
 > [!NOTE]
-> For now you first have to add the [OHF-Voice apps](https://github.com/OHF-Voice/apps) repo manually to the App Store repositroy inside Home Assistant before you can install it.
+> For now you first have to add the [OHF-Voice apps](https://github.com/OHF-Voice/apps) repo manually to the App Store repository inside Home Assistant before you can install it.
 
 Later you will be able to install it directly from the official add-on repository (but it is not yet published publicly there):
 

docs/enabling_aec.md

@@ -16,14 +16,13 @@ load-module module-echo-cancel source_name=aec_mic aec_method=webrtc
 
 ### PipeWire
 ```sh
-pw-cli load-module libpipewire-module-echo-cancel '{ aec.method=webrtc source.props={ node.name=aec_mic } }'
+pw-cli load-module libpipewire-module-echo-cancel '{ source.props={ node.name=aec_mic } }'
 ```
 
 To make permanent, add the following to the `context.modules` section of `/etc/pipewire/pipewire.conf`:
 ```
 { name = libpipewire-module-echo-cancel
     args = {
-      aec.method = webrtc
       source.props = { node.name = aec_mic }
     }
 }
@@ -34,7 +33,6 @@ Or if you are using a standard PipeWire installation without a custom `pipewire.
 context.modules = [
   { name = libpipewire-module-echo-cancel
     args = {
-      aec.method = webrtc
       source.props = { node.name = aec_mic }
     }
   }
@@ -57,14 +55,13 @@ pactl load-module module-echo-cancel source_name=aec_mic aec_method=webrtc sourc
 
 ### PipeWire
 ```sh
-pw-cli load-module libpipewire-module-echo-cancel '{ aec.method=webrtc source.props={ node.name=aec_mic node.description="Echo Cancelled Mic" } sink.props={ node.name=aec_sink } capture.props={ node.name=<your-mic-node-name> } playback.props={ node.name=<your-speaker-node-name> } }'
+pw-cli load-module libpipewire-module-echo-cancel '{ source.props={ node.name=aec_mic node.description="Echo Cancelled Mic" } sink.props={ node.name=aec_sink } capture.props={ node.name=<your-mic-node-name> } playback.props={ node.name=<your-speaker-node-name> } }'
 ```
 
 To make permanent, add to `context.modules` in `/etc/pipewire/pipewire.conf` or a file in `pipewire.conf.d/`:
 ```
 { name = libpipewire-module-echo-cancel
     args = {
-      aec.method = webrtc
       source.props = {
         node.name = aec_mic
         node.description = "Echo Cancelled Mic"

docs/peripheral_api.md

@@ -65,6 +65,7 @@ Immediately after connecting, LVA sends a snapshot so your client can synchronis
     "muted": false,
     "volume": 0.8,
     "ha_connected": true,
+    "button_controls_disabled": false,
     "last_stt_text": "set a timer for five minutes",
     "last_tts_text": "Sure, I've set a timer for five minutes."
   }
@@ -124,13 +125,14 @@ These are all the events LVA emits. Your peripheral script receives them and rea
 | `disconnected` | — | The TCP connection to Home Assistant was lost. Show a "no connection" animation and keep it until you see `zeroconf` with `status: connected`. Note: if LVA itself is not running, your client will see a WebSocket connection failure instead — treat that the same way. |
 | `muted` | `{"muted": bool}` | The microphone mute state changed. `true` = muted (show a muted indicator on your LEDs, e.g. red at mic positions), `false` = unmuted. Emitted on every transition in both directions, so peripherals can track mute state without inferring it from `idle`. |
 | `zeroconf` | `{"status": "getting_started" \| "connected"}` | Reports LVA's connection lifecycle. `getting_started` is emitted at startup before HA connects; `connected` is emitted once the HA TCP handshake completes. Use `connected` to clear a "no connection" animation. |
+| `button_controls_disabled` | `{"disabled": bool}` | Reflects the "Disable button controls" switch on the Home Assistant device page. `true` = on-board buttons should be ignored by the peripheral; `false` = normal button behaviour. Off by default. Also included directly in the `snapshot` payload for newly-connecting clients. |
 
 ### Timer events
 
 | Event | Data | Description |
 |-------|------|-------------|
 | `timer_ticking` | `{"id": str, "name": str, "total_seconds": int, "seconds_left": int}` | A timer has been set and is counting down. Show a countdown animation proportional to `seconds_left / total_seconds`. |
-| `timer_updated` | `{"id": str, "name": str, "total_seconds": int, "seconds_left": int}` | A running timer was adjusted. Update your countdown display. |
+| `timer_updated` | `{"id": str, "name": str, "total_seconds": int, "seconds_left": int}` | A running timer was adjusted. Update your countdown display. Also sent with `seconds_left` of `0`, followed by `idle`, when a timer is cancelled: treat zero as the timer being gone and clear it. A timer expiring normally arrives as `timer_ringing`. |
 | `timer_ringing` | `{"id": str, "name": str, "total_seconds": int, "seconds_left": int}` | The timer has expired and the alarm sound is playing. Show a repeating alert animation. Send `stop_timer_ringing` to dismiss. |
 
 ### Media and volume events
@@ -165,6 +167,7 @@ For HA to see your entity, your peripheral must register before HA enumerates th
 |---------|------|-------------|
 | `register_light` | `{"name": str, "object_id": str, "effects": [str], "supports_rgb": bool, "supports_brightness": bool}` | Register a Light entity for an LED strip, ring, or single LED. HA exposes it as `light.<satellite>_<object_id>` with on/off, brightness, RGB, and a selectable effect from the declared list. Subsequent HA changes are delivered as `light_command` events. Send once after connecting; repeat registrations for the same `object_id` are idempotent (no-op). Example: `{"command": "register_light", "data": {"name": "LEDs", "object_id": "leds", "effects": ["Voice Assistant"], "supports_rgb": true, "supports_brightness": true}}` |
 | `register_button` | `{"name": Button Press, "button_press_event": str}` | Register a Button entity for a physical button on your peripheral. When the user presses the button in HA, LVA emits a `button_press` event to all connected peripherals. Send once after connecting; repeat registrations for the same `object_id` are idempotent (no-op). Example: `{"command": "register_button"}` |
+| `register_button_lock` | `{"name": Button Lock, "button_lock_event": str}` | Register a Button Lock entity for a "Disable button controls" switch on the HA device page. When the user toggles it, LVA emits a `button_lock_changed` event to all connected peripherals. Send once after connecting; repeat registrations for the same `object_id` are idempotent (no-op). Example: `{"command": "register_button_lock"}` |
 
 ### Voice pipeline
 
@@ -322,7 +325,7 @@ asyncio.run(main())
 The repository ships with ready-to-use peripheral controllers for popular hardware boards. Each example is a standalone Docker container that connects to LVA over this API.
 
 - [Peripheral Web Console](https://github.com/OHF-Voice/linux-voice-assistant/blob/main/examples/Peripheral%20web%20console/DOCS.md) — browser-based real-time dashboard showing all events and a command palette; useful for debugging your setup or testing automations without any hardware
-- [Satellite1 HAT Board](https://github.com/OHF-Voice/linux-voice-assistant/blob/main/examples/Satellite1%20HAT%20Board/DOCS.md) — 12-LED SK6812 ring + 4 buttons; animations mirror the Home Assistant Voice PE firmware exactly
+- [Satellite1 HAT Board](https://github.com/OHF-Voice/linux-voice-assistant/blob/main/examples/Satellite1%20HAT%20Board/DOCS.md) — 24-LED WS2812 ring + 4 buttons; animations mirror the Home Assistant Voice PE firmware exactly
 - [ReSpeaker 2-Mic Pi HAT](https://github.com/OHF-Voice/linux-voice-assistant/blob/main/examples/ReSpeaker%202mic%20HAT/DOCS.md) — 3 APA102 LEDs + 1 onboard button; compact single-button controller
 - [ReSpeaker 4-Mic Array HAT](https://github.com/OHF-Voice/linux-voice-assistant/blob/main/examples/ReSpeaker%204mic%20HAT/DOCS.md) — 12 APA102 LEDs + external GPIO buttons; four-microphone circular array
 - [ReSpeaker Mic Array v2.0 (USB)](https://github.com/OHF-Voice/linux-voice-assistant/blob/main/examples/ReSpeaker%20Mic%20Array%20v2.0%20(USB)/DOCS.md) — 12 APA102 LEDs driven over USB HID; plug-and-play, no GPIO required

examples/Jabra Speak 410/DOCS.md

@@ -67,6 +67,28 @@ Togges microphone mute. Sends `mute_mic` when unmuted, `unmute_mic` when muted.
 
 > **Mute button caveat:** The Jabra Speak 410 does not expose the mute button state over HID when not in an active call. The controller detects mute by reading the raw audio stream from PipeWire — all-zero audio indicates hardware mute has engaged. This requires PipeWire to be exposed to the container (see installation).
 
+### Disabling button controls from Home Assistant
+
+On connect, the controller registers a **Disable button controls** switch
+with LVA, which appears on the device page as
+`switch.<satellite>_disable_button_controls` (off by default).
+
+When turned **on** in Home Assistant, the call/hang-up button, volume
+buttons, and mute button all stop sending commands to LVA (and, when
+`VOLUME_CONTROL=pipewire` is set, volume buttons also stop adjusting the
+PipeWire sink directly). When turned **off**, normal button behavior
+resumes immediately.
+
+> **Note:** Unlike the other example peripherals, the mute button on this
+> device is also gated by the lock, since it is a physical button command
+> like any other rather than a dedicated hardware mute switch. If you'd
+> prefer mute to remain functional while locked, that's a small change in
+> `main.py`'s `Telephony.mute` / `Volume.mute` handling.
+
+The lock state is persisted by LVA across restarts. On (re)connect, the
+controller reads the current value from the `snapshot` event's
+`button_controls_locked` field.
+
 ---
 
 ## Installation

examples/Jabra Speak 410/main.py

@@ -200,6 +200,9 @@ class JabraSpeak:
                 if not vol_ctrl:
                     print("ignoring volume command, env not set")
                     continue
+                if buttons_locked:
+                    print("ignoring volume command, buttons locked")
+                    continue                    
                 if event & Volume.vol_up:
                     print("jabra to lva: volume up detected")
                     match vol_ctrl:
@@ -250,6 +253,9 @@ async def write_to_jabra(state: LEDs | LEDState):
 
 
 async def write_to_lva(command: LVACommand, data: dict = None):
+    if buttons_locked:
+        print(f"to lva: buttons locked, ignoring {command}")
+        return    
     if lva_sock:
         global last_lva_write
         last_lva_write = command
@@ -299,6 +305,8 @@ current_state: None | LVAEvent = None
 
 muted: bool = False
 
+buttons_locked: bool = False
+
 
 async def run_cmd(cmd: list[str]):
     print("run_cmd ", cmd)
@@ -342,12 +350,19 @@ async def wsloop():
             async with websockets.connect(LVA_WS_URL) as websocket:
                 print(f"Connected to LVA at {LVA_WS_URL}")
                 lva_sock = websocket
+                await websocket.send(json.dumps({"command": "register_button_lock"}))
                 while True:
                     data = await websocket.recv()
                     print(f"from lva: {data}")
                     json_data = json.loads(data)
                     if json_data["event"] == "snapshot":
                         await set_mute(json_data["data"]["muted"], write_lva=False)
+                        global buttons_locked
+                        buttons_locked = json_data["data"].get("button_controls_locked", False)
+                    elif json_data["event"] == "button_lock_changed":
+                        buttons_locked = json_data["data"].get("locked", False)
+                        print(f"Button controls {'locked' if buttons_locked else 'unlocked'}")
+                        continue                        
                     global current_state
                     try:
                         current_state = LVAEvent(json_data["event"])

examples/Peripheral web console/DOCS.md

@@ -126,6 +126,26 @@ Send commands to LVA by clicking buttons:
 - **Single press**, **Double press**, **Triple press**, **Long press** - Simulate physical button inputs
 - Useful for testing button-based interactions if LVA is connected to hardware buttons
 
+### Button Lock Testing
+
+The console registers a **Disable button controls** switch with LVA
+(`register_button_lock`) so it can be used to test and demonstrate the
+button-lock feature that hardware peripherals implement.
+
+- The **Buttons Locked** field in the status area reflects the current
+  switch state, sourced from the `snapshot` event and updated live on
+  `button_lock_changed` events.
+- When the switch is turned **on** from the Home Assistant device page,
+  every button in the **Commands** section is disabled (greyed out) and
+  clicking it is a no-op, mirroring how a real peripheral should ignore
+  its physical buttons while locked.
+- Turning the switch **off** re-enables the command buttons immediately.
+
+This is useful for verifying the lock behavior end-to-end without needing
+physical hardware: toggle the switch in Home Assistant and confirm the
+console's command grid disables/enables accordingly, and that no
+commands are sent to LVA while locked.
+
 ## Example Workflow
 
 1. **Start LVA server** with WebSocket peripheral API enabled

examples/Peripheral web console/peripheral_web_console.html

@@ -117,6 +117,10 @@
         <span class="meta-label">HA Connected</span>
         <span class="meta-value" id="meta-ha">—</span>
       </div>
+      <div class="meta-item">
+        <span class="meta-label">Buttons Locked</span>
+        <span class="meta-value" id="meta-locked">—</span>
+      </div>      
     </div>
 
     <!-- Volume bar -->
@@ -311,6 +315,9 @@
       // Register the Button entity with LVA so HA can receive button events.
       ws.send(JSON.stringify({command: 'register_button'}));
       log('sent', '↑ register_button');
+
+      ws.send(JSON.stringify({command: 'register_button_lock'}));
+      log('sent', '↑ register_button_lock');
     };
 
     ws.onclose = () => {
@@ -340,6 +347,8 @@
         metaMuted.textContent = data.muted !== undefined ? (data.muted ? 'yes' : 'no') : '—';
         metaHa.textContent    = data.ha_connected !== undefined ? (data.ha_connected ? 'yes' : 'no') : '—';
         volBar.style.width    = (data.volume !== undefined ? data.volume * 100 : 0).toFixed(0) + '%';
+        buttonsLocked = !!data.button_controls_locked;
+        applyButtonLockUI();
 
         // Restore last conversation context
         if (data.last_stt_text) {
@@ -361,7 +370,12 @@
         volBar.style.width  = pct + '%';
       }
       if (ev === 'volume_muted' && data) {
-        metaMuted.textContent = data.muted ? 'yes' : 'no';
+        const isMuted = !!data.muted;
+        metaMuted.textContent = isMuted ? 'yes' : 'no';
+        if (isMuted) {
+          metaVol.textContent = '0%';
+          volBar.style.width = '0%';
+        }
       }
       if (ev === 'zeroconf' && data) {
         metaHa.textContent = data.status === 'connected' ? 'yes' : 'connecting…';
@@ -393,6 +407,10 @@
           timerBox.classList.remove('hidden');
         }
       }
+      if (ev === 'button_lock_changed' && data) {
+        buttonsLocked = !!data.locked;
+        applyButtonLockUI();
+      }
 
       setState(ev);
       log('recv', ev, dataStr);
@@ -411,6 +429,10 @@
       log('err-item', '✗ not connected — connect first');
       return;
     }
+    if (buttonsLocked) {
+      log('err-item', '✗ button controls locked — ignoring ' + command);
+      return;
+    }    
     const msg = { command };
     if (data) msg.data = data;
     ws.send(JSON.stringify(msg));
@@ -430,4 +452,15 @@
     send('set_volume', { volume: clamped });
     volumeInput.value = '';  // Clear after sending
   }
+
+  let buttonsLocked = false;
+
+  function applyButtonLockUI() {
+    document.getElementById('meta-locked').textContent = buttonsLocked ? 'yes' : 'no';
+    document.querySelectorAll('.cmd-grid .cmd-btn').forEach((el) => {
+      el.disabled = buttonsLocked;
+      el.style.opacity = buttonsLocked ? '0.4' : '';
+      el.style.cursor = buttonsLocked ? 'not-allowed' : '';
+    });
+  }  
 </script>

examples/ReSpeaker 2mic HAT/DOCS.md

@@ -75,6 +75,7 @@ LED 0 sits above MIC_L and LED 2 sits above MIC_R. When the microphone is muted,
 | **Timer ringing** | **Blue flash (repeating)** | **All 3** |
 | Timer ticking | Dim cyan, brightness ∝ time left | All 3 |
 | Media playing | Dim green steady | All 3 |
+| **Volume muted** | **Solid red** | **LED 1 only (center led)** |
 
 "User color" comes from the Home Assistant Light entity described below (default: HAVPE-style blue). HA brightness scales every animation in this table. Semantic colors (Thinking yellow, Speaking green, Muted red, Timer cyan/blue) are hardcoded so they remain recognisable across user customisation.
 
@@ -132,6 +133,22 @@ Multi-press commands are useful for triggering custom Home Assistant automations
 
 These commands are exposed as button press events to Home Assistant (via the registered button sensor), allowing you to create custom automations via `button_press_event` triggers.
 
+### Disabling button controls from Home Assistant
+
+On connect, the controller registers a **Disable button controls** switch
+with LVA, which appears on the device page as
+`switch.<satellite>_disable_button_controls` (off by default).
+
+When turned **on** in Home Assistant, the onboard button's single-press
+context action and all multi-press gestures (double, triple, long press)
+stop sending commands to LVA. When turned **off**, normal behavior
+resumes immediately.
+
+The lock state is persisted by LVA across restarts. On (re)connect, the
+controller reads the current value from the `snapshot` event's
+`button_controls_locked` field, so it never has a stale/unknown lock
+state after a reconnect.
+
 ---
 
 ## Installation

examples/ReSpeaker 2mic HAT/respeaker_2mic_hat.py

@@ -20,7 +20,7 @@ LED behaviours
   pipeline_error   : red flash on all 3 LEDs
   timer_ringing    : blue flash on all 3 LEDs (repeating)
   timer_ticking    : all 3 dim cyan, brightness proportional to time left
-  media_playing    : dim green steady on all 3 LEDs
+  volume_muted     : LED 1 solid red, LED 0 and LED 2 off
   not_ready/no_ha  : dim red pulse on all 3 LEDs
 
 On connect the script registers an HA Light entity with LVA via the
@@ -130,7 +130,7 @@ RECONNECT_DELAY_S = 3.0
 # is used to route incoming light_command events back to this script.
 LIGHT_OBJECT_ID = "leds"
 LIGHT_NAME      = "LEDs"
-LIGHT_ICON      = "mdi:circle-outline"
+LIGHT_ICON      = "mdi:led-strip-varient"
 
 
 # ===========================================================================
@@ -157,6 +157,7 @@ class AssistState(str, Enum):
     TIMER_TICKING = "timer_ticking"
     TIMER_RINGING = "timer_ringing"
     MEDIA_PLAYING = "media_player_playing"
+    VOLUME_MUTED  = "volume_muted"
 
 
 # Effect name. Must match the LEDLightEntity effects list the peripheral
@@ -175,8 +176,9 @@ class SharedState:
         self._lock = threading.Lock()
         self.assist_state: AssistState = AssistState.NOT_READY
         self.ha_connected: bool = False
-        self.muted: bool = False
+        self.muted: bool = False          # mic mute
         self.volume: float = 1.0
+        self.volume_muted: bool = False   # media volume zero
         self.timer_total_seconds: int = 0
         self.timer_seconds_left: int = 0
         # Light entity state, driven by HA via light_command events.
@@ -192,6 +194,7 @@ class SharedState:
         # Monotonic deadline used by _timer_tick to fade brightness
         # smoothly between sparse timer_updated events.
         self.timer_ends_at: float = 0.0
+        self.buttons_locked: bool = False
 
     def update(self, **kwargs) -> None:
         with self._lock:
@@ -213,6 +216,7 @@ class SharedState:
                 "ha_connected":        self.ha_connected,
                 "muted":               self.muted,
                 "volume":              self.volume,
+                "volume_muted":        self.volume_muted,
                 "timer_total_seconds": self.timer_total_seconds,
                 "timer_seconds_left":  self.timer_seconds_left,
                 "light_is_on":         self.light_is_on,
@@ -221,6 +225,7 @@ class SharedState:
                 "light_green":         self.light_green,
                 "light_blue":          self.light_blue,
                 "timer_ends_at":       self.timer_ends_at,
+                "buttons_locked":       self.buttons_locked,
             }
 
 
@@ -379,6 +384,9 @@ class LEDAnimator:
             # Left and right LEDs red (mic positions), centre off
             self._task = asyncio.create_task(self._muted())
 
+        elif state == AssistState.VOLUME_MUTED:
+            self._task = asyncio.create_task(self._volume_muted())
+      
         elif state == AssistState.ERROR:
             self._task = asyncio.create_task(
                 self._flash_all(RED, flashes=3, on_ms=150, off_ms=100, then_off=True)
@@ -456,6 +464,13 @@ class LEDAnimator:
         self._leds.set(2, RED)
         self._leds.show(self._brightness(0.6))
 
+    async def _volume_muted(self) -> None:
+        """Centre LED red; left and right off."""
+        self._leds.set(0, OFF)
+        self._leds.set(1, RED)
+        self._leds.set(2, OFF)
+        self._leds.show(self._brightness(0.6))
+  
     async def _flash_all(
         self,
         color: ColorSource,
@@ -588,6 +603,9 @@ class ButtonHandler:
             self._on_press()
 
     def _send(self, command: str) -> None:
+        if self._state.snapshot["buttons_locked"]:
+            _LOGGER.debug("Button controls locked — ignoring %s", command)
+            return
         _LOGGER.info("Button → %s", command)
         asyncio.run_coroutine_threadsafe(
             self._queue.put(command), self._loop
@@ -652,6 +670,9 @@ class ButtonMultipressHandler:
         self._single_click_action = action
 
     def _send(self, command: str) -> None:
+        if self._state.snapshot["buttons_locked"]:
+            _LOGGER.debug("Button controls locked — ignoring %s", command)
+            return
         _LOGGER.info("Button → %s", command)
         asyncio.run_coroutine_threadsafe(
             self._queue.put(command), self._loop
@@ -793,6 +814,7 @@ class LVAClient:
             # creates a Button Press event entity in Home Assistant.
             # Idempotent: safe to send on every reconnect.
             await ws.send(json.dumps({"command": "register_button"}))
+            await ws.send(json.dumps({"command": "register_button_lock"}))
             recv_task = asyncio.create_task(self._recv_loop(ws))
             send_task = asyncio.create_task(self._send_loop(ws))
             done, pending = await asyncio.wait(
@@ -835,6 +857,7 @@ class LVAClient:
                 muted=muted,
                 volume=data.get("volume", 1.0),
                 ha_connected=data.get("ha_connected", False),
+                buttons_locked=data.get("button_controls_locked", False),
             )
             if muted:
                 self._animator.set_state(AssistState.MUTED)
@@ -844,6 +867,12 @@ class LVAClient:
                 self._animator.set_state(AssistState.IDLE)
             return
 
+        elif event == "button_lock_changed":
+            locked = data.get("locked", False)
+            self._state.update(buttons_locked=locked)
+            _LOGGER.info("Button controls %s", "locked" if locked else "unlocked")
+            return
+
         if event == "wake_word_detected":
             self._state.update(assist_state=AssistState.WAKE_WORD)
 
@@ -922,11 +951,11 @@ class LVAClient:
             self._state.update(volume=data.get("volume", 1.0))
 
         elif event == "volume_muted":
-            muted = data.get("muted", False)
-            self._state.update(muted=muted)
-            if muted:
-                self._state.update(assist_state=AssistState.MUTED)
-            elif self._state.assist_state == AssistState.MUTED:
+            volume_muted = data.get("muted", True)
+            self._state.update(volume_muted=volume_muted)
+            if volume_muted:
+                self._state.update(assist_state=AssistState.VOLUME_MUTED)
+            elif self._state.assist_state == AssistState.VOLUME_MUTED:
                 self._state.update(assist_state=AssistState.IDLE)
 
         elif event == "zeroconf":

examples/ReSpeaker 4mic HAT/DOCS.md

@@ -44,7 +44,7 @@ The controller runs as a separate Docker container on the same Raspberry Pi. It
 All animations mirror the Home Assistant Voice PE ESPHome firmware exactly.
 
 | LVA state                    | Animation                   | Description                                                                                       |
-|---|---|---|
+| ---------------------------- | --------------------------- | ------------------------------------------------------------------------------------------------- |
 | No HA connection / not ready | Red twinkle                 | Random red sparkle across all LEDs                                                                |
 | Idle                         | Off                         | All LEDs off                                                                                      |
 | Wake word detected           | Slow clockwise spin         | Two trailing arcs at opposing positions                                                           |
@@ -55,6 +55,8 @@ All animations mirror the Home Assistant Voice PE ESPHome firmware exactly.
 | Error                        | Red pulse                   | All LEDs red, pulsing                                                                             |
 | Timer ticking                | Countdown arc               | Arc proportional to `seconds_left / total_seconds`                                                |
 | Timer ringing                | Pulse + optional red        | Full ring pulsing; red at corners if muted                                                        |
+| Volume changed               | Volume Display arc          | Temporary arc (2.5s) showing the current volume level, starting from LED 0 and sweeping clockwise |
+| Volume muted                 | Solid ring + red indicator  | Red at the bottom-mid LED                                                                         |
 
 "User color" comes from the Home Assistant Light entity described below (default: HAVPE-style blue). HA brightness scales every animation in this table. Semantic colors (Waiting/Listening cyan spin, Thinking pulse, Replying anticlockwise, Muted solid + red, Timer pulse/arc, Error red pulse) are hardcoded so they remain recognisable across user customisation.
 

examples/ReSpeaker 4mic HAT/respeaker_4mic_hat.py

@@ -93,6 +93,22 @@ SPI_DEVICE     = 1
 # 10 ≈ 32 % (gentle default), 31 = maximum.
 LED_BRIGHTNESS = 10
 
+# Volume Display effect
+# Mirrors the ESPHome Voice PE "Volume Display" addressable_lambda. Origin
+# is LED 6 (bottom-mid) — the same reference LED already used for the
+# volume_muted indicator (VOLUME_MUTED_LED_INDEX) — so the arc grows from
+# the same spot that turns solid red when volume is silenced.
+VOLUME_DISPLAY_SECONDS = 2.5
+
+# LED that lights up solid red while the media player (speaker) output is
+# muted at volume 0 — the `volume_muted` peripheral event. Ring index 6 is
+# the bottom-middle position (180°, directly opposite index 0 at the top),
+# following the same 0=top / 3=right / 6=bottom / 9=left convention used for
+# the mic corner indicators below. Sits between the two bottom mic corners
+# (LEDs 4 and 7) so it reads as a distinct "speaker muted" dot rather than
+# overlapping a mic indicator.
+VOLUME_MUTED_LED_INDEX = 6
+
 # Default ring colour (R, G, B) — matches HA Voice PE default
 DEFAULT_R, DEFAULT_G, DEFAULT_B = 24, 187, 242
 
@@ -124,6 +140,7 @@ class AssistState(str, Enum):
     TIMER_TICKING = "timer_ticking"
     TIMER_RINGING = "timer_ringing"
     MEDIA_PLAYING = "media_player_playing"
+    VOLUME_MUTED  = "volume_muted"
 
 
 # ---------------------------------------------------------------------------
@@ -135,8 +152,13 @@ class SharedState:
         self._lock = threading.Lock()
         self.assist_state: AssistState = AssistState.NOT_READY
         self.ha_connected: bool = False
-        self.muted: bool = False
+        self.muted: bool = False          # mic mute
         self.volume: float = 1.0
+        # Monotonic deadline until which the Volume Display arc takes over
+        # the ring, set on each volume_changed event. 0 (or in the past)
+        # means the arc is not showing.
+        self.volume_display_until: float = 0.0      
+        self.volume_muted: bool = False   # media player (speaker) mute
         self.timer_total_seconds: int = 0
         self.timer_seconds_left: int = 0
         # Light entity state, driven by HA via light_command events.
@@ -162,6 +184,8 @@ class SharedState:
                 "ha_connected":        self.ha_connected,
                 "muted":               self.muted,
                 "volume":              self.volume,
+                "volume_display_until": self.volume_display_until,
+                "volume_muted":        self.volume_muted,
                 "timer_total_seconds": self.timer_total_seconds,
                 "timer_seconds_left":  self.timer_seconds_left,
                 "light_is_on":         self.light_is_on,
@@ -317,6 +341,19 @@ class LEDRing:
         self._set(7,  _scale(RED, factor))
         self._set(10, _scale(RED, factor))
 
+    def _apply_volume_muted_indicator(self) -> None:
+        """
+        Overlay a solid red dot on the bottom-middle LED (index
+        ``VOLUME_MUTED_LED_INDEX``) when the media player output is muted.
+
+        Applied after the current pipeline animation has already rendered
+        its frame, so the indicator shows up consistently no matter which
+        animation is active — mirrors how ``_apply_mic_indicators`` marks
+        the microphone corners, but for the distinct speaker-mute state.
+        """
+        self._set(VOLUME_MUTED_LED_INDEX, RED)
+        self._write()
+
     # ------------------------------------------------------------------
     # Animation implementations
     # ------------------------------------------------------------------
@@ -453,6 +490,32 @@ class LEDRing:
         self._write()
         return 0.1
 
+    def _anim_volume_display(self, color: RGB, volume: float) -> float:
+        """
+        Arc showing the current volume level, originating at LED 6
+        (bottom-mid — same LED as VOLUME_MUTED_LED_INDEX) and sweeping
+        clockwise around the ring.
+
+        Mirrors the ESPHome Voice PE "Volume Display" addressable_lambda —
+        proportional arc with a partial-brightness leading LED, plus a red
+        indicator on the origin LED when volume is silenced.
+        """
+        silenced_color = RED
+        volume_ratio = LED_COUNT * max(0.0, min(1.0, volume))
+
+        for i in range(LED_COUNT):
+            if i <= volume_ratio:
+                brightness = min(volume_ratio - i, 1.0)
+                self._set((VOLUME_MUTED_LED_INDEX + i) % LED_COUNT, _scale(color, brightness))
+            else:
+                self._set((VOLUME_MUTED_LED_INDEX + i) % LED_COUNT, BLACK)
+
+        if volume <= 0.0:
+            self._set(VOLUME_MUTED_LED_INDEX, silenced_color)
+
+        self._write()
+        return 0.05  # matches the 50ms update_interval of the ESPHome effect
+  
     # ------------------------------------------------------------------
     # Animation loop
     # ------------------------------------------------------------------
@@ -465,10 +528,16 @@ class LEDRing:
             snap    = self._state.snapshot
             color   = self._color()
             muted   = snap["muted"]
+            volume_muted = snap["volume_muted"]
             t_total = snap["timer_total_seconds"]
             t_left  = snap["timer_seconds_left"]
 
-            if anim == self.ANIM_OFF:
+            # Volume Display takes over the ring temporarily, regardless of
+            # the current pipeline state, then falls back to the normal
+            # animation once its deadline passes.
+            if snap["volume_display_until"] > time.monotonic():
+                sleep = self._anim_volume_display(color, snap["volume"])
+            elif anim == self.ANIM_OFF:
                 sleep = self._anim_off()
             elif anim == self.ANIM_IDLE:
                 sleep = self._anim_idle()
@@ -495,6 +564,9 @@ class LEDRing:
             else:
                 sleep = self._anim_off()
 
+            if volume_muted:
+                self._apply_volume_muted_indicator()
+
             time.sleep(sleep)
 
 
@@ -687,10 +759,23 @@ class LVAClient:
             self._state.update(assist_state=AssistState.MEDIA_PLAYING)
 
         elif event == "volume_changed":
-            self._state.update(volume=data.get("volume", 1.0))
+            self._state.update(
+                volume=data.get("volume", 1.0),
+                volume_display_until=time.monotonic() + VOLUME_DISPLAY_SECONDS,
+            )
 
         elif event == "volume_muted":
-            self._state.update(muted=data.get("muted", False))
+            # Media player (speaker) mute state, carried as {"muted": bool}
+            # per the peripheral API — distinct from the microphone `muted`
+            # event above. Drives the bottom-middle LED indicator rather
+            # than replacing the whole ring animation, since the assist
+            # pipeline can still be doing something else at the same time.
+            volume_muted = data.get("muted", True)
+            self._state.update(volume_muted=volume_muted)
+            if volume_muted:
+                self._state.update(assist_state=AssistState.VOLUME_MUTED)
+            elif self._state.assist_state == AssistState.VOLUME_MUTED:
+                self._state.update(assist_state=AssistState.IDLE)
 
         elif event == "zeroconf":
             status = data.get("status", "")

examples/ReSpeaker Mic Array v2.0 (USB)/DOCS.md

@@ -36,6 +36,8 @@ All animations mirror the Home Assistant Voice PE ESPHome firmware exactly.
 | Error | Red pulse | All LEDs red, pulsing |
 | Timer ticking | Countdown arc | Arc proportional to `seconds_left / total_seconds` |
 | Timer ringing | Pulse + optional red | Full ring pulsing; red at mic positions if muted |
+| Volume changed | Volume Display arc | Temporary arc (2.5s) showing the current volume level, starting from LED 9 (bottom, between the micro-USB and AUX jack) and sweeping clockwise |
+| Volume muted  | Red mic indicator | LED 9 (bottom at the AUX jack corner) solid red to indicate the media player volume is zero; other LEDs off |
 
 "User color" comes from the Home Assistant Light entity described below (default: HAVPE-style blue). HA brightness scales every animation in this table. Semantic colors (Waiting/Listening cyan spin, Thinking pulse, Replying anticlockwise, Muted solid + red, Timer pulse/arc, Error red pulse) are hardcoded so they remain recognisable across user customisation.
 

examples/ReSpeaker Mic Array v2.0 (USB)/respeaker_usb_mic_array.py

@@ -86,6 +86,21 @@ USB_PRODUCT_ID    = 0x0018
 LED_COUNT         = 12
 LED_BRIGHTNESS    = 10    # 0–31 (APA102 5-bit global brightness register)
 
+# Volume Display effect
+# Mirrors the ESPHome Voice PE "Volume Display" addressable_lambda. Origin
+# is LED 9 (bottom, DOA 270°) — the same reference LED already used for
+# the volume_muted indicator (VOLUME_MUTED_LED_INDEX), sitting between the
+# micro-USB connector and the 3.5 mm AUX jack — so the arc grows from the
+# same spot that turns solid red when volume is silenced.
+VOLUME_DISPLAY_SECONDS = 2.5
+
+# LED that lights up solid red while the media player (speaker) output is
+# muted at volume 0 — the `volume_muted` peripheral event. LED 9 is the
+# board's bottom-middle position (DOA 270°), sitting right between the
+# micro-USB connector and the 3.5 mm AUX jack on the board edge — the same
+# "bottom" position already used by the mic indicator mapping below.
+VOLUME_MUTED_LED_INDEX = 9
+
 RECONNECT_DELAY_S = 3.0
 
 
@@ -113,6 +128,7 @@ class AssistState(str, Enum):
     TIMER_TICKING = "timer_ticking"
     TIMER_RINGING = "timer_ringing"
     MEDIA_PLAYING = "media_player_playing"
+    VOLUME_MUTED = "volume_muted"
 
 
 # ---------------------------------------------------------------------------
@@ -124,8 +140,13 @@ class SharedState:
         self._lock = threading.Lock()
         self.assist_state: AssistState = AssistState.NOT_READY
         self.ha_connected: bool = False
-        self.muted: bool = False
+        self.muted: bool = False         # mic mute
         self.volume: float = 1.0
+        # Monotonic deadline until which the Volume Display arc takes over
+        # the ring, set on each volume_changed event. 0 (or in the past)
+        # means the arc is not showing.
+        self.volume_display_until: float = 0.0      
+        self.volume_muted: bool = False  # media player volume zero
         self.timer_total_seconds: int = 0
         self.timer_seconds_left: int = 0
         # Light entity state, driven by HA via light_command events.
@@ -151,6 +172,8 @@ class SharedState:
                 "ha_connected":        self.ha_connected,
                 "muted":               self.muted,
                 "volume":              self.volume,
+                "volume_display_until": self.volume_display_until,
+                "volume_muted":        self.volume_muted,
                 "timer_total_seconds": self.timer_total_seconds,
                 "timer_seconds_left":  self.timer_seconds_left,
                 "light_is_on":         self.light_is_on,
@@ -407,6 +430,48 @@ class LEDRing:
         self._leds.set(6, _scale(RED, factor))
         self._leds.set(9, _scale(RED, factor))
 
+    def _anim_volume_display(self, color: RGB, volume: float) -> float:
+        """
+        Arc showing the current volume level, originating at LED 9
+        (bottom, between the micro-USB and AUX jack — same LED as
+        VOLUME_MUTED_LED_INDEX) and sweeping clockwise around the ring.
+
+        Mirrors the ESPHome Voice PE "Volume Display" addressable_lambda —
+        proportional arc with a partial-brightness leading LED, plus a red
+        indicator on the origin LED when volume is silenced.
+        """
+        silenced_color = RED
+        volume_ratio = LED_COUNT * max(0.0, min(1.0, volume))
+
+        for i in range(LED_COUNT):
+            if i <= volume_ratio:
+                brightness = min(volume_ratio - i, 1.0)
+                self._leds.set((VOLUME_MUTED_LED_INDEX + i) % LED_COUNT, _scale(color, brightness))
+            else:
+                self._leds.set((VOLUME_MUTED_LED_INDEX + i) % LED_COUNT, BLACK)
+
+        if volume <= 0.0:
+            self._leds.set(VOLUME_MUTED_LED_INDEX, silenced_color)
+
+        self._leds.show()
+        return 0.05  # matches the 50ms update_interval of the ESPHome effect
+  
+    def _apply_volume_muted_indicator(self) -> None:
+        """
+        Overlay a solid red dot on the bottom-middle LED (index
+        ``VOLUME_MUTED_LED_INDEX``, DOA 270°) when the media player output
+        is muted at volume 0.
+
+        Sits right between the board's micro-USB connector and 3.5 mm AUX
+        jack. Applied after the current pipeline animation has already
+        rendered its frame, so the indicator shows up consistently no
+        matter which animation is active — mirrors how
+        ``_apply_mic_indicators`` marks the microphone cardinal points, but
+        for the distinct speaker-mute state.
+        """
+        self._leds.set(VOLUME_MUTED_LED_INDEX, RED)
+        self._leds.show()
+
     # ------------------------------------------------------------------
     # Animations — each returns sleep time in seconds
     # ------------------------------------------------------------------
@@ -555,10 +620,16 @@ class LEDRing:
             snap    = self._state.snapshot
             color   = self._color()
             muted   = snap["muted"]
+            volume_muted = snap["volume_muted"]
             t_total = snap["timer_total_seconds"]
             t_left  = snap["timer_seconds_left"]
 
-            if anim == self.ANIM_OFF:
+            # Volume Display takes over the ring temporarily, regardless of
+            # the current pipeline state, then falls back to the normal
+            # animation once its deadline passes.
+            if snap["volume_display_until"] > time.monotonic():
+                sleep = self._anim_volume_display(color, snap["volume"])
+            elif anim == self.ANIM_OFF:
                 sleep = self._anim_off()
             elif anim == self.ANIM_IDLE:
                 sleep = self._anim_idle()
@@ -585,6 +656,9 @@ class LEDRing:
             else:
                 sleep = self._anim_off()
 
+            if volume_muted:
+                self._apply_volume_muted_indicator()
+
             time.sleep(sleep)
 
 
@@ -780,10 +854,18 @@ class LVAClient:
             self._state.update(assist_state=AssistState.MEDIA_PLAYING)
 
         elif event == "volume_changed":
-            self._state.update(volume=data.get("volume", 1.0))
+            self._state.update(
+                volume=data.get("volume", 1.0),
+                volume_display_until=time.monotonic() + VOLUME_DISPLAY_SECONDS,
+            )
 
         elif event == "volume_muted":
-            self._state.update(muted=data.get("muted", False))
+            Volume_muted = data.get("muted", True)
+            self._state.update(volume_muted=Volume_muted)
+            if Volume_muted:
+                self._state.update(assist_state=AssistState.VOLUME_MUTED)
+            elif self._state.assist_state == AssistState.VOLUME_MUTED:
+                self._state.update(assist_state=AssistState.IDLE)
 
         elif event == "zeroconf":
             status = data.get("status", "")

examples/Satellite1 HAT Board/DOCS.md

@@ -4,7 +4,7 @@
 
 Peripheral controller for the [Satellite1 HAT Board](https://futureproofhomes.net/products/satellite1-top-microphone-board) by FutureProofHomes, running alongside the Linux Voice Assistant (LVA) container.
 
-The controller runs as a separate Docker container on the same Raspberry Pi. It connects to LVA's peripheral WebSocket API, drives the 12-LED SK6812 ring with animations that mirror the [Home Assistant Voice PE](https://www.home-assistant.io/voice-pe/) LED behaviour, and maps the four hardware buttons to LVA commands with multipress support on the context action button.
+The controller runs as a separate Docker container on the same Raspberry Pi. It connects to LVA's peripheral WebSocket API, drives the 24-LED WS2812 ring with animations that mirror the [Home Assistant Voice PE](https://www.home-assistant.io/voice-pe/) LED behaviour, and maps the four hardware buttons to LVA commands with multipress support on the context action button.
 
 ---
 
@@ -13,7 +13,7 @@ The controller runs as a separate Docker container on the same Raspberry Pi. It
 | Component | Details |
 |---|---|
 | Microphone array | XMOS XVF3800 (USB, far-field, onboard DSP) |
-| LED ring | 12 × SK6812 RGBW NeoPixels |
+| LED ring | 24 × WS2812 RGBW NeoPixels |
 | Buttons | 4 × tactile (top, bottom, left, right) |
 | Audio output | I2S line-out (optional) |
 | Interface | Raspberry Pi 40-pin HAT connector |
@@ -68,6 +68,7 @@ In Home Assistant, you can:
    - Thinking → Pulse in your color
    - Speaking → Reverse spin in your color
    - Muted → Solid ring in your color + red at mic positions
+   - Volume muted → Solid ring in your color + red at AUX jack corner LEDs
    - Timer → Countdown arc in your color
 
 3. **Set brightness** — scales all animations by the chosen brightness (0–100%). Useful for adapting to ambient lighting.
@@ -103,10 +104,14 @@ target:
 ## Button behaviour
 
 ### Right button — Volume Up
-Sends `volume_up` to LVA. Each press increases volume by one step.
+Sends `volume_up` to LVA. A quick press increases volume by one step.
+
+**Hold to adjust gradually:** keep the button held past 1000 ms and the controller keeps sending `volume_up` every ~150 ms — volume rises smoothly for as long as the button stays down, instead of one step per press. Release the button to stop.
 
 ### Left button — Volume Down
-Sends `volume_down` to LVA. Each press decreases volume by one step.
+Sends `volume_down` to LVA. A quick press decreases volume by one step.
+
+**Hold to adjust gradually:** the same behaviour applies in reverse — holding the button past 1000 ms repeats `volume_down` every ~150 ms until released.
 
 ### Top button — Mute / Unmute
 Toggles microphone mute. Sends `mute_mic` when unmuted, `unmute_mic` when muted. The LED ring switches to the muted indicator pattern immediately.
@@ -126,6 +131,8 @@ Context-aware command based on current assistant state, mirroring the Home Assis
 | Music / media playing | `stop_media_player` |
 | Any other (idle) | `start_listening` |
 
+**LED feedback while pressed:** whenever the button is physically held down — even for a very short tap — the LED ring lights up at the currently chosen color (or the default cyan if no color has been picked yet), then turns off again the instant the button is released. This only kicks in when the HA Light entity is currently **off**; if the light is already on, the idle animation is already showing the color, so nothing changes.
+
 #### Multi-press gestures
 
 When the action button is released quickly (not held), it detects multiple presses:
@@ -154,6 +161,10 @@ Hold down the action button and use the volume buttons to rotate through the HSV
 | **Hold + Volume Up** | Rotate hue +10° (clockwise around color wheel) |
 | **Hold + Volume Down** | Rotate hue -10° (counter-clockwise around color wheel) |
 
+**Hold the volume button to keep rotating:** a quick press of a volume button while holding the action button steps the hue by 10° once, same as above. If you keep the volume button held past 1000 ms, the controller keeps stepping the hue every ~150 ms — the color keeps rotating around the wheel for as long as you hold it, and stops the moment you release the volume button (the action button can stay held throughout).
+
+**Real-time preview:** every hue step — whether from a single tap or the continuous hold-rotation — is applied to shared state immediately and shows up on the LED ring right away, without waiting on Home Assistant to echo the change back. This means the ring always reflects exactly the color you're about to select while you're still picking it.
+
 The color wheel provides **36 distinct color stops** (0°–360° in 10° increments):
 - **0°** – Red
 - **60°** – Yellow
@@ -162,14 +173,30 @@ The color wheel provides **36 distinct color stops** (0°–360° in 10° increm
 - **240°** – Blue
 - **300°** – Magenta
 
-Each press changes the hue by 10°, allowing smooth navigation around the full color spectrum. The LED ring updates immediately to show the new color during idle/standby (when the light is on).
+Each press (or hold-repeat step) changes the hue by 10°, allowing smooth navigation around the full color spectrum. The LED ring updates immediately to show the new color during idle/standby (when the light is on).
 
 **Example workflow:**
 1. Press and hold the action button
-2. Press volume up repeatedly to cycle through warm colors (red → orange → yellow)
-3. Release action button to confirm and exit color mode
+2. Press and hold volume up to rotate continuously through warm colors (red → orange → yellow), or tap it repeatedly for single 10° steps
+3. Release the volume button once you like the color, then release the action button to confirm and exit color mode
 4. The new color persists in Home Assistant
 
+#### Disabling button controls from Home Assistant
+
+On connect, the controller registers a **Disable button controls** switch
+with LVA, which appears on the device page as
+`switch.<satellite>_disable_button_controls` (off by default).
+
+When turned **on** in Home Assistant, the action button (context command), volume buttons, mute button, and the hold + volume color-wheel gesture are all ignored — pressing them has no effect on LVA or the LED ring color.
+
+When turned **off**, button behavior resumes immediately.
+
+The switch state is persisted by LVA across restarts, so a lock set from
+Home Assistant survives a reboot of the satellite. On reconnect, the
+controller reads the current state from the `snapshot` event's
+`button_controls_locked` field and applies it immediately, without
+waiting for a change.
+
 ---
 
 ## LED ring animations
@@ -188,6 +215,9 @@ All animations mirror the Home Assistant Voice PE ESPHome firmware exactly. The
 | Error | Red pulse | All LEDs red, pulsing |
 | Timer ticking | Countdown arc | Arc length proportional to `seconds_left / total_seconds`, tinted by your color |
 | Timer ringing | Pulse + optional red | Full ring pulsing in your color; red at 3 & 9 if muted |
+| Volume changed | Volume Display arc | Temporary arc (2.5s) showing the current volume level, starting from the AUX jack corner LED (D1, index 0) and sweeping clockwise across all 24 LEDs |
+| Volume muted | Solid ring + red indicator | red at the AUX jack corner LEDs positions 11, 0, 1 to indicate the media player volume is zero |
+| Action button held (light off) | Solid ring in chosen/default color | Overrides whatever else is showing for as long as the action button is physically held down, only while the HA Light entity is off — gives instant tactile feedback and previews color-wheel changes live |
 
 **Note:** All animations except twinkle, off, and error animations use the color from the HA Light entity. This means you can customize the entire LED experience from Home Assistant by adjusting the light's RGB color and brightness.
 
@@ -202,7 +232,7 @@ All animations mirror the Home Assistant Voice PE ESPHome firmware exactly. The
 Edit `/boot/firmware/config.txt` (or `/boot/config.txt` on older Raspberry Pi OS):
 
 ```ini
-# Enable PWM on GPIO 12 for the SK6812 LED ring
+# Enable PWM on GPIO 12 for the WS2812 LED ring
 dtoverlay=pwm,pin=12,func=4
 
 # Disable onboard audio — it shares PWM0 with GPIO 12
@@ -293,13 +323,18 @@ BTN_ACTION      = 23
 
 BTN_DEBOUNCE_MS = 150   # Button debounce in milliseconds
 
+# Volume button hold-to-repeat (also drives continuous color wheel rotation
+# while the action button is held — see "Hold + Volume buttons" above)
+VOLUME_HOLD_THRESHOLD_MS = 1000  # Hold time before gradual repeat kicks in
+VOLUME_HOLD_REPEAT_S     = 0.15  # Seconds between repeated steps while held
+
 # Button multipress timing
 MULTIPRESS_TIMEOUT_MS = 500    # Time window between presses (ms)
 LONG_PRESS_MS         = 1000   # Duration to detect long press (ms)
 
 
 # LED ring
-LED_COUNT      = 12
+LED_COUNT      = 24
 LED_BRIGHTNESS = 168    # 0–255, default is 66 % (168)
 
 # HA Light entity registration
@@ -414,7 +449,13 @@ automation:
 2. Confirm onboard audio is disabled (`dtparam=audio=on` is commented out).
 3. Check the container is running as root (`user: "0:0"` in compose file) — `rpi-ws281x` requires root for DMA access via `/dev/mem`.
 4. Run with `--debug` and look for `LED ring initialised` in the logs. If `rpi_ws281x not found` appears, the Python package failed to install — rebuild the image.
-5. **If you see only pipeline animations (listening, thinking, etc.) but no idle glow:** The light entity defaults to **off** in Home Assistant to match Voice PE behavior. Turn on the `light.<satellite>_leds` entity in HA to show the idle LED color.
+5. **If you see only pipeline animations (listening, thinking, etc.) but no idle glow:** The light entity defaults to **off** in Home Assistant to match Voice PE behavior. Turn on the `light.<satellite>_leds` entity in HA to show the idle LED color, or simply hold the action button — it lights the ring at the current color while held whenever the light is off.
+
+### Volume doesn't change smoothly while holding the button
+
+1. Confirm you're holding the button past 1000 ms (`VOLUME_HOLD_THRESHOLD_MS`) — a shorter press only sends a single step, by design.
+2. Run with `--debug` — each repeated step logs `Button → volume_up` / `Button → volume_down` roughly every 150 ms while held.
+3. If the repeat feels too fast/slow, adjust `VOLUME_HOLD_REPEAT_S` at the top of `Satellite1_HAT_Board.py`.
 
 ### Buttons do not respond
 

examples/Satellite1 HAT Board/Satellite1_HAT_Board.py

@@ -7,7 +7,7 @@ hardware buttons to LVA peripheral API commands.
 
 Hardware layout (Satellite 1 HAT on Raspberry Pi)
 --------------------------------------------------
-  LED ring   : 12 × SK6812 RGBW NeoPixels  →  GPIO 12 (PWM0)
+  LED ring   : 24 x WS2812 RGBW NeoPixels  →  GPIO 12 (PWM0)
   Right btn  : Volume Up                    →  GPIO 17
   Left btn   : Volume Down                  →  GPIO 27
   Top btn    : Mute / Unmute mic            →  GPIO 22
@@ -47,7 +47,7 @@ import sys
 import threading
 import time
 from enum import Enum
-from typing import List, Optional, Tuple
+from typing import Dict, List, Optional, Tuple
 
 # ---------------------------------------------------------------------------
 # Hardware dependencies (gracefully stubbed when not on real Pi hardware so
@@ -90,14 +90,43 @@ BTN_ACTION     = 23   # Bottom button
 
 BTN_DEBOUNCE_MS = 30  # Milliseconds
 
+# Continuous hold-repeat behaviour for the volume buttons (and, while the
+# action button is held at the same time, the color wheel rotation). Once a
+# volume button has been held past VOLUME_HOLD_THRESHOLD_MS, the associated
+# step (volume up/down, or hue up/down when in color change mode) repeats
+# every VOLUME_HOLD_REPEAT_S seconds until the button is released.
+VOLUME_HOLD_THRESHOLD_MS = 1000  # Milliseconds before continuous repeat starts
+VOLUME_HOLD_REPEAT_S = 0.15  # Seconds between repeated steps while held
+
 # LED ring
-LED_COUNT      = 12
+LED_COUNT      = 24
 LED_FREQ_HZ    = 800_000
 LED_DMA        = 10
 LED_BRIGHTNESS = 168   # 66 % of 255 – matches ESPHome default
 LED_INVERT     = False
 LED_CHANNEL    = 0
 
+# Four mic positions at the cardinal points (top, right, bottom, left),
+# spaced LED_COUNT/4 apart. On the 24-LED ring: LEDs 0, 6, 12, 18.
+MIC_LED_POSITIONS = tuple(i * (LED_COUNT // 4) for i in range(4))
+
+# LEDs physically sitting in the AUX (3.5 mm) audio jack corner of the board,
+# identified by their silkscreen designators on the Satellite1 HAT (D35, D1,
+# D2). These three ring positions light up solid red whenever the media
+# player volume is at zero (the `volume_muted` peripheral event), independent
+# of whatever pipeline animation is currently running. Scaled to the 24-LED
+# ring to cover the same angular span as the original 12-LED mapping —
+# verify against your board revision and adjust if it differs.
+AUX_JACK_LED_INDICES = (22, 23, 0, 1, 2)
+
+# Volume Display effect
+# Mirrors the ESPHome Voice PE "Volume Display" addressable_lambda, but
+# rotated so the arc originates at the AUX jack corner LED (D1, index 0)
+# instead of Voice PE's origin — matching this board's physical AUX jack
+# placement. Shown as a temporary overlay for this many seconds after
+# volume_changed.
+VOLUME_DISPLAY_SECONDS = 2.5
+
 # Default ring color  (ESPHome default: 9.4 % R, 73.3 % G, 94.9 % B)
 DEFAULT_R, DEFAULT_G, DEFAULT_B = 24, 187, 242
 
@@ -223,6 +252,7 @@ class AssistState(str, Enum):
     TIMER_TICKING = "timer_ticking"
     TIMER_RINGING = "timer_ringing"
     MEDIA_PLAYING = "media_player_playing"
+    VOLUME_MUTED  = "volume_muted"
 
 
 # ---------------------------------------------------------------------------
@@ -234,8 +264,13 @@ class SharedState:
         self._lock = threading.Lock()
         self.assist_state: AssistState = AssistState.NOT_READY
         self.ha_connected: bool = False
-        self.muted: bool = False
+        self.muted: bool = False  # mic mute
         self.volume: float = 1.0
+        # Monotonic deadline until which the Volume Display arc takes over
+        # the ring, set on each volume_changed event. 0 (or in the past)
+        # means the arc is not showing.
+        self.volume_display_until: float = 0.0        
+        self.volume_muted: bool = False  # media player volume zero
         self.timer_total_seconds: int = 0
         self.timer_seconds_left: int = 0
         # Light entity state, driven by HA via light_command events.
@@ -251,6 +286,12 @@ class SharedState:
         self.hsv_hue: int = DEFAULT_HUE
         self.hsv_saturation: float = DEFAULT_SATURATION
         self.hsv_value: float = DEFAULT_VALUE
+        # True for as long as the action (bottom) button is physically held
+        # down, regardless of press duration. Drives the "light up while
+        # pressed" tactile feedback in LEDRing (only applied when the HA
+        # Light entity is off — see LEDRing._anim_action_held).
+        self.action_button_down: bool = False
+        self.buttons_locked: bool = False
 
     def update(self, **kwargs) -> None:
         with self._lock:
@@ -265,6 +306,8 @@ class SharedState:
                 "ha_connected":         self.ha_connected,
                 "muted":                self.muted,
                 "volume":               self.volume,
+                "volume_display_until": self.volume_display_until,
+                "volume_muted":         self.volume_muted,
                 "timer_total_seconds":  self.timer_total_seconds,
                 "timer_seconds_left":   self.timer_seconds_left,
                 "light_is_on":          self.light_is_on,
@@ -275,6 +318,8 @@ class SharedState:
                 "hsv_hue":              self.hsv_hue,
                 "hsv_saturation":       self.hsv_saturation,
                 "hsv_value":            self.hsv_value,
+                "action_button_down":   self.action_button_down,
+                "buttons_locked":       self.buttons_locked,
             }
 
 
@@ -425,6 +470,25 @@ class LEDRing:
             self._all_off()
         return 0.1
 
+    def _anim_action_held(self, color: RGB, brightness: float) -> float:
+        """
+        Force the ring fully on at the current (or default, if none has been
+        chosen yet) light color while the action button is physically held
+        down and the HA Light entity is off.
+
+        Only takes over when the Light entity is off — if it's already on,
+        the idle animation is already showing the color and the ring is
+        left alone. Gives instant visual feedback for every press,
+        including a very short tap (turns off the moment the button is
+        released), and refreshes quickly so that rotating the color wheel
+        (holding action + a volume button) shows up on the ring in real
+        time.
+        """
+        for i in range(LED_COUNT):
+            self._set(i, _scale(color, brightness))
+        self._write()
+        return 0.05
+
     def _anim_off(self) -> float:
         self._all_off()
         return 0.1
@@ -487,18 +551,12 @@ class LEDRing:
     def _anim_muted(self, color: RGB, muted: bool) -> float:
         """
         Solid ring with red indicators at all 4 mic positions when muted.
-        Satellite 1 HAT has mics at 12, 3, 6 and 9 o'clock → LEDs 0, 3, 6, 9.
         """
         for i in range(LED_COUNT):
             self._set(i, color)
 
         if muted:
-            # 4 mic positions: top (0), right (3), bottom (6), left (9)
-            # Blank the immediate neighbours so the red indicators stand out
-            self._set(11, BLACK); self._set(0, RED);  self._set(1, BLACK)
-            self._set(2,  BLACK); self._set(3, RED);  self._set(4, BLACK)
-            self._set(5,  BLACK); self._set(6, RED);  self._set(7, BLACK)
-            self._set(8,  BLACK); self._set(9, RED);  self._set(10, BLACK)
+            self._apply_mic_indicators()
 
         self._write()
         return 0.016
@@ -555,16 +613,69 @@ class LEDRing:
                 self._set(i, BLACK)
 
         if muted:
-            # 4 mic positions: LEDs 0, 3, 6, 9
-            self._set(11, BLACK); self._set(0, RED);  self._set(1, BLACK)
-            self._set(2,  BLACK); self._set(3, RED);  self._set(4, BLACK)
-            self._set(5,  BLACK); self._set(6, RED);  self._set(7, BLACK)
-            self._set(8,  BLACK); self._set(9, RED);  self._set(10, BLACK)
+            self._apply_mic_indicators()
 
         self._index = (LED_COUNT + self._index - 1) % LED_COUNT
         self._write()
         return 0.1
 
+    def _anim_volume_display(self, color: RGB, volume: float) -> float:
+        """
+        Arc showing the current volume level, originating at the AUX jack
+        corner LED (D1, index 0) and sweeping clockwise around the ring.
+
+        Mirrors the ESPHome Voice PE "Volume Display" addressable_lambda
+        exactly, except the arc's origin is rotated from LED 6 to LED 0 to
+        match this board's AUX jack corner LED placement.
+        """
+        silenced_color = RED
+        volume_ratio = LED_COUNT * max(0.0, min(1.0, volume))
+
+        for i in range(LED_COUNT):
+            if i <= volume_ratio:
+                brightness = min(volume_ratio - i, 1.0)
+                self._set(i, _scale(color, brightness))
+            else:
+                self._set(i, BLACK)
+
+        if volume <= 0.0:
+            self._set(0, silenced_color)
+
+        self._write()
+        return 0.05  # matches the 50ms update_interval of the ESPHome effect
+
+    def _apply_mic_indicators(self) -> None:
+        """
+        Mark all 4 mic positions (cardinal points) red on the already
+        rendered frame, blanking immediate neighbours so each indicator
+        reads as a distinct dot. Positions are LED_COUNT/4 apart —
+        LEDs 0, 6, 12, 18 on the 24-LED ring.
+        """
+        for pos in MIC_LED_POSITIONS:
+            self._set((pos - 1) % LED_COUNT, BLACK)
+            self._set(pos, RED)
+            self._set((pos + 1) % LED_COUNT, BLACK)
+
+    def _apply_mic_indicators_pulsed(self, factor: float) -> None:
+        """Pulsed version for timer_ring — keeps indicators in sync with ring brightness."""
+        for pos in MIC_LED_POSITIONS:
+            self._set(pos, _scale(RED, factor))
+
+    def _apply_volume_muted_indicator(self) -> None:
+        """
+        Overlay a solid red indicator on the AUX jack corner LEDs
+        (designators D35, D1, D2 — see ``AUX_JACK_LED_INDICES``) when the
+        media player output is muted.
+
+        Applied as a post-render overlay after the current pipeline
+        animation has already written its frame, so the indicator shows up
+        consistently no matter which animation is active, without having to
+        thread the flag through every individual ``_anim_*`` method.
+        """
+        for i in AUX_JACK_LED_INDICES:
+            self._set(i, RED)
+        self._write()
+
     # ------------------------------------------------------------------
     # Animation loop
     # ------------------------------------------------------------------
@@ -577,11 +688,29 @@ class LEDRing:
             snap = self._state.snapshot
             color        = self._color()
             muted        = snap["muted"]
+            volume_muted = snap["volume_muted"]
             t_total      = snap["timer_total_seconds"]
             t_left       = snap["timer_seconds_left"]
             ha_connected = snap["ha_connected"]
+            action_button_down = snap["action_button_down"]
 
-            if anim == self.ANIM_IDLE:
+            # Volume Display takes over the ring temporarily, regardless of
+            # the current pipeline state, then falls back to the normal
+            # animation once its deadline passes.
+            if snap["volume_display_until"] > time.monotonic():
+                sleep = self._anim_volume_display(color, snap["volume"])
+
+            # The action button forces the ring on at the current (or
+            # default) color while physically held down — but only when the
+            # HA Light entity is off; if it's already on, the idle animation
+            # is already showing the color and there's nothing extra to do.
+            # This also carries live updates while the color wheel is being
+            # rotated (holding action + a volume button), since that
+            # rotation writes the new color straight into shared state.
+            elif action_button_down and not snap["light_is_on"]:
+                sleep = self._anim_action_held(color, snap["light_brightness"])
+
+            elif anim == self.ANIM_IDLE:
                 sleep = self._anim_idle()
 
             elif anim == self.ANIM_OFF:
@@ -620,6 +749,9 @@ class LEDRing:
             else:
                 sleep = self._anim_off()
 
+            if volume_muted:
+                self._apply_volume_muted_indicator()
+
             time.sleep(sleep)
 
 
@@ -678,11 +810,24 @@ class ButtonHandler:
     coroutines onto the asyncio event loop thread-safely.
     
     Supports:
-    - Volume +/- for volume control
-    - Mute button for microphone toggle
+    - Volume +/- for volume control. A short press sends a single step;
+      holding the button past VOLUME_HOLD_THRESHOLD_MS repeats that step
+      every VOLUME_HOLD_REPEAT_S seconds until released, so volume moves
+      gradually while held instead of one step per press.
+    - Mute button for microphone toggle.
     - Action button for context-aware commands or color changing:
       * Hold + Volume +: Increase hue by 10°
       * Hold + Volume -: Decrease hue by 10°
+      Holding the volume button past VOLUME_HOLD_THRESHOLD_MS while the
+      action button is held rotates continuously through the color wheel
+      using the same repeat mechanism, until the volume button is released.
+      Color changes are applied to the LED ring immediately so the wheel
+      rotation previews in real time.
+    - The action button also lights up the LED ring at the chosen (or
+      default) color for as long as it's physically held down — even for a
+      quick tap — turning off again on release. This only happens when the
+      HA Light entity is off; if it's already on, the idle animation is
+      already showing the color.
     """
 
     def __init__(
@@ -699,6 +844,15 @@ class ButtonHandler:
         self._action_button_held = False
         self._action_button_obj = None
 
+        # Continuous hold-repeat state for the volume buttons, keyed by
+        # "up" / "down". `_volume_hold_timers` fires once the button has
+        # been held past VOLUME_HOLD_THRESHOLD_MS to kick off the repeat
+        # loop; `_volume_repeat_stop` / `_volume_repeat_threads` drive that
+        # loop until the button is released.
+        self._volume_hold_timers: Dict[str, Optional[threading.Timer]] = {"up": None, "down": None}
+        self._volume_repeat_stop: Dict[str, Optional[threading.Event]] = {"up": None, "down": None}
+        self._volume_repeat_threads: Dict[str, Optional[threading.Thread]] = {"up": None, "down": None}
+
     def setup(self) -> None:
         if not _HAS_GPIO:
             _LOGGER.warning("GPIO unavailable – buttons not configured")
@@ -711,12 +865,14 @@ class ButtonHandler:
         btn_mute = Button(BTN_MUTE,        pull_up=True, bounce_time=debounce)
         btn_act  = Button(BTN_ACTION,      pull_up=True, bounce_time=debounce)
 
-        btn_up.when_pressed   = self._on_volume_up
-        btn_down.when_pressed = self._on_volume_down
+        btn_up.when_pressed    = lambda: self._on_volume_button_pressed("up")
+        btn_up.when_released   = lambda: self._on_volume_button_released("up")
+        btn_down.when_pressed  = lambda: self._on_volume_button_pressed("down")
+        btn_down.when_released = lambda: self._on_volume_button_released("down")
         btn_mute.when_pressed = self._on_mute
-        btn_act.when_pressed  = self._on_action
+        btn_act.when_pressed  = self._on_action_pressed
         btn_act.when_held     = self._on_action_held
-        btn_act.when_released = self._on_action_released
+        btn_act.when_released = self._on_action_button_released
         btn_act.hold_time     = 0.1  # 100ms to detect hold
 
         # Keep references — gpiozero Buttons are released when garbage-collected
@@ -728,6 +884,9 @@ class ButtonHandler:
         )
 
     def cleanup(self) -> None:
+        for which in ("up", "down"):
+            self._cancel_volume_hold_timer(which)
+            self._stop_volume_repeat(which)
         for btn in self._buttons:
             btn.close()
         self._buttons.clear()
@@ -750,6 +909,94 @@ class ButtonHandler:
         else:
             self._send("volume_down")
 
+    def _on_volume_button_pressed(self, which: str) -> None:
+        """
+        Handle a Volume Up/Down press: fire the immediate single step (volume,
+        or hue when the action button is held for color changing), then arm a
+        timer that starts continuous repeat if the button is still held past
+        VOLUME_HOLD_THRESHOLD_MS.
+        """
+        if which == "up":
+            self._on_volume_up()
+        else:
+            self._on_volume_down()
+
+        self._cancel_volume_hold_timer(which)
+        timer = threading.Timer(
+            VOLUME_HOLD_THRESHOLD_MS / 1000.0, self._start_volume_repeat, args=(which,)
+        )
+        timer.daemon = True
+        self._volume_hold_timers[which] = timer
+        timer.start()
+
+    def _on_volume_button_released(self, which: str) -> None:
+        """Stop any pending hold detection and any active continuous repeat."""
+        self._cancel_volume_hold_timer(which)
+        self._stop_volume_repeat(which)
+
+    def _cancel_volume_hold_timer(self, which: str) -> None:
+        timer = self._volume_hold_timers.get(which)
+        if timer is not None:
+            timer.cancel()
+            self._volume_hold_timers[which] = None
+
+    def _start_volume_repeat(self, which: str) -> None:
+        """
+        Begin repeating the volume/hue step every VOLUME_HOLD_REPEAT_S while
+        the button stays held. Runs in its own daemon thread so the GPIO
+        event-detection thread is never blocked; stopped via a threading.Event
+        set from `_on_volume_button_released`.
+        """
+        self._stop_volume_repeat(which)  # Safety: never overlap two loops
+
+        stop_event = threading.Event()
+        self._volume_repeat_stop[which] = stop_event
+
+        def _repeat_loop() -> None:
+            while not stop_event.wait(VOLUME_HOLD_REPEAT_S):
+                if which == "up":
+                    self._on_volume_up()
+                else:
+                    self._on_volume_down()
+
+        thread = threading.Thread(
+            target=_repeat_loop, daemon=True, name=f"volume-hold-repeat-{which}"
+        )
+        self._volume_repeat_threads[which] = thread
+        thread.start()
+
+    def _stop_volume_repeat(self, which: str) -> None:
+        stop_event = self._volume_repeat_stop.get(which)
+        if stop_event is not None:
+            stop_event.set()
+
+        thread = self._volume_repeat_threads.get(which)
+        if thread is not None and thread.is_alive():
+            thread.join(timeout=0.5)
+
+        self._volume_repeat_stop[which] = None
+        self._volume_repeat_threads[which] = None
+
+    def _on_action_pressed(self) -> None:
+        """
+        Action button physically pressed down.
+
+        Fires the usual context-sensitive command immediately (unchanged
+        behaviour), and also marks the button as held down so the LED ring
+        can react instantly — even for a very short tap. The ring itself
+        only lights up in response to this when the HA Light entity is off
+        (see LEDRing._anim_action_held); if it's already on, the idle
+        animation is already showing the color and there's nothing extra
+        to do.
+        """
+        self._state.update(action_button_down=True)
+        self._on_action()
+
+    def _on_action_button_released(self) -> None:
+        """Action button released – exit color-change mode and stop forcing the ring on."""
+        self._on_action_released()
+        self._state.update(action_button_down=False)
+
     def _on_action_held(self) -> None:
         """Action button held – enter color change mode."""
         self._action_button_held = True
@@ -772,13 +1019,26 @@ class ButtonHandler:
         self._send_color_command(new_hue, snap["hsv_saturation"], snap["hsv_value"])
     
     def _send_color_command(self, hue: int, saturation: float, value: float) -> None:
-        """Convert HSV to RGB and send light_command to LVA."""
-        # Update local HSV state
-        self._state.update(hsv_hue=hue, hsv_saturation=saturation, hsv_value=value)
-        
-        # Convert HSV to RGB for light command
+        """
+        Convert HSV to RGB, apply it to shared state immediately for a
+        real-time preview on the LED ring, and notify LVA/Home Assistant.
+        """
         r, g, b = hsv_to_rgb(hue, saturation, value)
 
+        # Update local state right away so the ring reflects the new color
+        # as it's being picked, without waiting on a light_command
+        # round-trip through LVA/Home Assistant. This also turns the light
+        # "on" — picking a color is an explicit, deliberate action.
+        self._state.update(
+            hsv_hue=hue,
+            hsv_saturation=saturation,
+            hsv_value=value,
+            light_is_on=True,
+            light_red=r / 255.0,
+            light_green=g / 255.0,
+            light_blue=b / 255.0,
+        )
+
         # Normalize RGB to 0-1 range
         light_cmd = {
             "command": "light_command",
@@ -832,8 +1092,12 @@ class ButtonMultipressHandler:
         self.button_pin = BTN_ACTION
         self.last_press_time = 0
         self.press_count = 0
+        self._state = state
 
     def _send(self, command: str) -> None:
+        if self._state.snapshot["buttons_locked"]:
+            _LOGGER.debug("Button controls locked — ignoring %s", command)
+            return
         _LOGGER.info("Button → %s", command)
         asyncio.run_coroutine_threadsafe(
             self._queue.put(command), self._loop
@@ -932,6 +1196,7 @@ class LVAClient:
             # creates a Button Press event entity in Home Assistant.
             # Idempotent: safe to send on every reconnect.
             await ws.send(json.dumps({"command": "register_button"}))
+            await ws.send(json.dumps({"command": "register_button_lock"}))
             recv_task = asyncio.create_task(self._recv_loop(ws))
             send_task = asyncio.create_task(self._send_loop(ws))
             done, pending = await asyncio.wait(
@@ -986,7 +1251,11 @@ class LVAClient:
             self._state.update(
                 muted=data.get("muted", False),
                 volume=data.get("volume", 1.0),
+                # Not currently sent by LVA's snapshot payload, but read it
+                # defensively in case a future LVA version adds it.
+                volume_muted=data.get("volume_muted", False),
                 ha_connected=data.get("ha_connected", False),
+                buttons_locked=data.get("button_controls_locked", False),
             )
             snap = self._state.snapshot
             anim = state_to_animation(
@@ -995,6 +1264,12 @@ class LVAClient:
             self._leds.set_animation(anim)
             return
 
+        elif event == "button_lock_changed":
+            locked = data.get("locked", False)
+            self._state.update(buttons_locked=locked)
+            _LOGGER.info("Button controls %s", "locked" if locked else "unlocked")
+            return
+      
         # --- Voice pipeline events -----------------------------------------
         elif event == "wake_word_detected":
             self._state.update(assist_state=AssistState.WAKE_WORD)
@@ -1069,10 +1344,23 @@ class LVAClient:
             self._state.update(assist_state=AssistState.MEDIA_PLAYING)
 
         elif event == "volume_changed":
-            self._state.update(volume=data.get("volume", 1.0))
+            self._state.update(
+                volume=data.get("volume", 1.0),
+                volume_display_until=time.monotonic() + VOLUME_DISPLAY_SECONDS,
+            )
 
         elif event == "volume_muted":
-            self._state.update(muted=data.get("muted", False))
+            # Media player (speaker) mute state, distinct from the
+            # microphone `muted` event above. Drives the red indicator on
+            # the AUX jack corner LEDs (see AUX_JACK_LED_INDICES) rather
+            # than swapping the whole ring animation, since the assist
+            # pipeline can still be doing something else at the same time.
+            volume_muted = data.get("muted", True)
+            self._state.update(volume_muted=volume_muted)
+            if volume_muted:
+                self._state.update(assist_state=AssistState.VOLUME_MUTED)
+            elif self._state.assist_state == AssistState.VOLUME_MUTED:
+                self._state.update(assist_state=AssistState.IDLE)
 
         # --- Zeroconf / connection events ----------------------------------
         elif event == "zeroconf":

linux_voice_assistant/__main__.py

@@ -18,7 +18,7 @@ from getmac import get_mac_address  # type: ignore
 from pymicro_wakeword import MicroWakeWord, MicroWakeWordFeatures
 from pyopen_wakeword import OpenWakeWord, OpenWakeWordFeatures
 
-from .models import Preferences, ServerState, WakeWordType
+from .models import Preferences, ServerState, WakeWordType, initial_stop_word_threshold
 from .mpv_player import MpvMediaPlayer
 from .peripheral_api import LVAEvent, PeripheralAPIServer
 from .satellite import VoiceSatelliteProtocol
@@ -379,6 +379,19 @@ async def main() -> None:
     initial_volume = max(0.0, min(1.0, float(initial_volume)))
     preferences.volume = initial_volume
 
+    # Load stop word sensitivity from preferences on startup, and ensure it's between 0.0 and 1.0
+    initial_threshold = initial_stop_word_threshold(preferences.stop_word_sensitivity)
+    preferences.stop_word_sensitivity = initial_threshold
+
+    # Load button-lock state from preferences on startup. This must happen
+    # before ServerState is constructed (and therefore before the peripheral
+    # WebSocket server can accept any connection), because the very first
+    # snapshot sent to a (re)connecting peripheral must already reflect the
+    # persisted value — peripherals only learn the lock state from that
+    # snapshot or a later button_lock_changed event, and nothing re-sends
+    # a corrected snapshot after the fact.
+    initial_button_controls_locked = bool(preferences.button_controls_locked)
+
     if args.enable_thinking_sound:
         preferences.thinking_sound = 1
 
@@ -441,6 +454,8 @@ async def main() -> None:
         output_only=args.output_only,
         download_dir=args.download_dir,
         volume=initial_volume,
+        stop_word_threshold=initial_threshold,
+        button_controls_locked=initial_button_controls_locked,
         mic_volume=preferences.mic_volume,
         mic_auto_gain=preferences.mic_auto_gain,
         mic_noise_suppression=preferences.mic_noise_suppression,

linux_voice_assistant/entity.py

@@ -182,6 +182,8 @@ class MediaPlayerEntity(ESPHomeEntity):
                         self.music_player.set_volume(0)
                         self.announce_player.set_volume(0)
                         self.muted = True
+                        if hasattr(self.server, "state") and getattr(self.server, "state", None) is not None:
+                            self.server.state.persist_volume(self.volume)
                     yield self._update_state(self.state)
 
                 elif command == MediaPlayerCommand.UNMUTE:
@@ -191,6 +193,8 @@ class MediaPlayerEntity(ESPHomeEntity):
                         self.music_player.set_volume(int(self.volume * 100))
                         self.announce_player.set_volume(int(self.volume * 100))
                         self.muted = False
+                        if hasattr(self.server, "state") and getattr(self.server, "state", None) is not None:
+                            self.server.state.persist_volume(self.volume)
                     yield self._update_state(self.state)
 
             elif msg.has_volume:
@@ -797,6 +801,65 @@ class ButtonEventSensorEntity(ESPHomeEntity):
         )
 
 
+class ButtonLockEntity(ESPHomeEntity):
+    """Switch entity for disabling a peripheral's on-board buttons.
+
+    Registered by a peripheral via register_button_lock, mirroring the
+    opt-in pattern used by register_light. LVA only exposes the switch
+    in Home Assistant and owns its state; it has no opinion on what
+    "locked" means for the hardware. When Home Assistant changes the
+    switch, on_changed fires so the peripheral API server can broadcast
+    a button_lock_changed event back to the peripheral, which decides
+    whether to keep executing button-triggered commands.
+    """
+
+    def __init__(
+        self,
+        server: APIServer,
+        key: int,
+        name: str,
+        object_id: str,
+        get_locked: Callable[[], bool],
+        set_locked: Callable[[bool], None],
+        icon: str = "mdi:button-pointer",
+    ) -> None:
+        ESPHomeEntity.__init__(self, server)
+        self.key = key
+        self.name = name
+        self.object_id = object_id
+        self.icon = icon
+        self._get_locked = get_locked
+        self._set_locked = set_locked
+        self._switch_state = self._get_locked()  # Sync internal state with actual value on init
+
+    def update_get_locked(self, get_locked: Callable[[], bool]) -> None:
+        self._get_locked = get_locked
+
+    def update_set_locked(self, set_locked: Callable[[bool], None]) -> None:
+        self._set_locked = set_locked
+
+    def sync_with_state(self) -> None:
+        self._switch_state = self._get_locked()
+
+    def handle_message(self, msg: message.Message) -> Iterable[message.Message]:
+        if isinstance(msg, SwitchCommandRequest) and (msg.key == self.key):
+            new_state = bool(msg.state)
+            self._switch_state = new_state
+            self._set_locked(new_state)
+            yield SwitchStateResponse(key=self.key, state=self._switch_state)
+        elif isinstance(msg, ListEntitiesRequest):
+            yield ListEntitiesSwitchResponse(
+                object_id=self.object_id,
+                key=self.key,
+                name=self.name,
+                entity_category=EntityCategory.CONFIG,
+                icon=self.icon,
+            )
+        elif isinstance(msg, SubscribeHomeAssistantStatesRequest):
+            self.sync_with_state()
+            yield SwitchStateResponse(key=self.key, state=self._switch_state)
+
+
 # Backward compatibility export aliases
 __all__ = [
     "ESPHomeEntity",
@@ -805,6 +868,7 @@ __all__ = [
     "ThinkingSoundEntity",
     "LEDLightEntity",
     "ButtonEventSensorEntity",
+    "ButtonLockEntity",
     "WakeWord1SensitivityNumberEntity",
     "WakeWord2SensitivityNumberEntity",
     "StopWordSensitivityNumberEntity",

linux_voice_assistant/models.py

@@ -16,6 +16,7 @@ if TYPE_CHECKING:
 
     from .entity import (
         ButtonEventSensorEntity,
+        ButtonLockEntity,
         ESPHomeEntity,
         LEDLightEntity,
         MediaPlayerEntity,
@@ -85,6 +86,7 @@ class Preferences:
     active_wake_words: List[Optional[str]] = field(default_factory=list)
     volume: Optional[float] = None
     thinking_sound: int = 0  # 0 = disabled, 1 = enabled
+    button_controls_locked: int = 0  # 0 = buttons enabled (default), 1 = buttons locked
     wake_word_1_sensitivity: Optional[float] = None
     wake_word_2_sensitivity: Optional[float] = None
     stop_word_sensitivity: Optional[float] = None
@@ -146,6 +148,13 @@ class ServerState:
     # Survives HA reconnects so the entity is re-registered automatically.
     pending_button: bool = False
 
+    # True once a peripheral sends register_button_lock. Gates creation of
+    # ButtonLockEntity so the HA device page only shows the "Disable button
+    # controls" switch when hardware that actually supports it is present.
+    # Survives HA reconnects so the entity is re-registered automatically.
+    pending_button_lock: bool = False
+    button_lock_entity: "Optional[ButtonLockEntity]" = None
+
     # Optional peripheral WebSocket API (LEDs, buttons, HAT boards).
     # Assigned in __main__ before the event loop starts.
     peripheral_api: "Optional[Any]" = None  # PeripheralAPIServer at runtime
@@ -159,6 +168,7 @@ class ServerState:
     wake_words_changed: bool = False
     refractory_seconds: float = 2.0
     thinking_sound_enabled: bool = False
+    button_controls_locked: bool = False
     output_only: bool = False
     muted: bool = False
     connected: bool = False
@@ -217,6 +227,7 @@ class ServerState:
             _LOGGER.debug("Skipping save - volume unchanged")
             return
 
+        previous_muted = self.volume == 0.0
         self.volume = clamped_volume
         self.preferences.volume = clamped_volume
         _LOGGER.info("Saving volume %s to %s", clamped_volume, self.preferences_path)
@@ -230,6 +241,10 @@ class ServerState:
 
             api.emit_event_sync(LVAEvent.VOLUME_CHANGED, {"volume": round(clamped_volume, 3)})
 
+            new_muted = clamped_volume == 0.0
+            if previous_muted != new_muted:
+                api.emit_event_sync(LVAEvent.VOLUME_MUTED, {"muted": new_muted})
+
     def persist_mic_gain(self, gain: float) -> None:
         """Persist the microphone auto gain value."""
         gain_int = int(gain)
@@ -260,3 +275,14 @@ class ServerState:
         self.preferences.mic_volume = volume_int
         _LOGGER.info("Saving mic_volume %s to %s", volume_int, self.preferences_path)
         self.save_preferences()
+
+
+def initial_stop_word_threshold(saved_sensitivity: Optional[float]) -> float:
+    """
+    Resolve the stop word probability cutoff to start from, clamped to 0.0-1.0.
+    :param saved_sensitivity: Value persisted in preferences, or None if it has never been set.
+    """
+    if saved_sensitivity is None:
+        return ServerState.stop_word_threshold
+
+    return max(0.0, min(1.0, float(saved_sensitivity)))

linux_voice_assistant/peripheral_api.py

@@ -54,6 +54,16 @@ Feedback events emitted by LVA
               effect names are those the peripheral declared at
               registration; e.g. "Voice Assistant" runs the pipeline
               animations.
+  button_lock_changed   data: {"locked": bool}
+              Fires when the user toggles the "Disable button controls"
+              switch on the Home Assistant device page for a peripheral
+              that registered it via register_button_lock. The current
+              value is also included as "button_controls_locked" in the
+              snapshot payload for newly-connecting/reconnecting clients.
+              A peripheral should stop executing button-triggered LVA
+              commands (start_listening, stop_pipeline, mute_mic/
+              unmute_mic, volume_up/down, button_*_press, etc.) while
+              locked is true.
 
 Commands accepted from the peripheral container
 ------------------------------------------------
@@ -92,6 +102,17 @@ Commands accepted from the peripheral container
               events to Home Assistant when the corresponding
               button_* commands are sent. Send once after connecting;
               duplicate registrations are ignored.
+  register_button_lock
+              The peripheral declares that it has on-board buttons whose
+              behavior can be toggled from Home Assistant. LVA creates a
+              "Disable button controls" switch entity (visible as
+              switch.<satellite>_disable_button_controls, disabled/off
+              by default) and routes HA changes back to the peripheral
+              as button_lock_changed events. The peripheral is
+              responsible for honoring the lock state — LVA does not
+              gate anything on its end. Send once after connecting;
+              duplicate registrations are ignored and preserve the
+              current lock state.
 """
 
 from __future__ import annotations
@@ -139,6 +160,7 @@ class LVAEvent(str, Enum):
     VOLUME_MUTED = "volume_muted"
     ZEROCONF = "zeroconf"
     LIGHT_COMMAND = "light_command"
+    BUTTON_LOCK_CHANGED = "button_lock_changed"
 
 
 class LVACommand(str, Enum):
@@ -161,6 +183,7 @@ class LVACommand(str, Enum):
     BUTTON_LONG_PRESS = "button_long_press"
     REGISTER_LIGHT = "register_light"
     REGISTER_BUTTON = "register_button"
+    REGISTER_BUTTON_LOCK = "register_button_lock"
 
 
 # ---------------------------------------------------------------------------
@@ -278,7 +301,9 @@ class PeripheralAPIServer:
                 "data": {
                     "muted": state.muted,
                     "volume": round(state.volume, 3),
+                    "volume_muted": state.volume == 0.0,
                     "ha_connected": state.connected,
+                    "button_controls_locked": state.button_controls_locked,
                     "last_stt_text": self._last_stt_text,
                     "last_tts_text": self._last_tts_text,
                 },
@@ -480,6 +505,9 @@ class PeripheralAPIServer:
         elif command == LVACommand.REGISTER_BUTTON:
             self._register_button(satellite)
 
+        elif command == LVACommand.REGISTER_BUTTON_LOCK:
+            self._register_button_lock(satellite)
+
     def _register_light(self, data: Dict[str, Any], satellite: Any) -> None:
         """Register a Light declared by a peripheral.
 
@@ -549,6 +577,35 @@ class PeripheralAPIServer:
 
         self._schedule_ha_reconnect_for_late_entity("button", "button_press_event")
 
+    def _register_button_lock(self, satellite: Any) -> None:
+        """Register the button-lock switch declared by a peripheral.
+
+        Idempotent: repeat registrations from a reconnecting peripheral
+        are a no-op — the existing entity and its current lock state
+        are preserved.
+
+        When the satellite is already running, the entity is materialised
+        immediately so a subsequent switch toggle routes correctly. HA
+        only sees the new entity after the integration is reloaded, but
+        the same startup-wait window that applies to register_light
+        applies here too (see --peripheral-startup-wait).
+        """
+        state = self._state
+        if state is None:
+            return
+
+        if state.pending_button_lock:
+            # Already registered; keep the existing entity and its state.
+            return
+
+        state.pending_button_lock = True
+        _LOGGER.info("Button lock switch registered by peripheral")
+
+        if satellite is not None:
+            satellite.register_pending_button_lock()
+
+        self._schedule_ha_reconnect_for_late_entity("button_lock", "disable_button_controls")
+
     # ------------------------------------------------------------------
     # Helpers
     # ------------------------------------------------------------------
@@ -659,6 +716,7 @@ class PeripheralAPIServer:
             LVAEvent.MEDIA_PLAYER_PLAYING,
             LVAEvent.DISCONNECTED,
             LVAEvent.PIPELINE_ERROR,
+            LVAEvent.VOLUME_MUTED,
         }
         if event in _STATE_EVENTS:
             self._current_state = event

linux_voice_assistant/satellite.py

@@ -52,6 +52,7 @@ from pyopen_wakeword import OpenWakeWord
 from .api_server import APIServer
 from .entity import (
     ButtonEventSensorEntity,
+    ButtonLockEntity,
     LEDLightEntity,
     MediaPlayerEntity,
     MicSettingEntity,
@@ -335,6 +336,9 @@ class VoiceSatelliteProtocol(APIServer):
         # button support before this satellite was constructed (e.g. on an HA
         # reconnect while the peripheral container stayed connected to LVA).
         self.register_pending_button()
+        # Materialise ButtonLockEntity if a peripheral already registered
+        # the button-lock switch before this satellite was constructed.
+        self.register_pending_button_lock()
 
         # ---- Instance variables ----
 
@@ -431,6 +435,40 @@ class VoiceSatelliteProtocol(APIServer):
         self.state.button_event_sensor_entity = entity
         _LOGGER.info("Button event sensor entity materialised")
 
+    def register_pending_button_lock(self) -> None:
+        """Materialise the button-lock switch once a peripheral has registered it.
+
+        Called from __init__ (handles HA reconnects where the peripheral
+        container stayed connected to LVA and pending_button_lock is already
+        True) and from PeripheralAPIServer._register_button_lock() when the
+        command arrives at runtime.
+
+        Safe to call multiple times: idempotent — if the entity already
+        exists it is only reattached to the current satellite server
+        instance, preserving its current lock state.
+        """
+        if not self.state.pending_button_lock:
+            return
+
+        if self.state.button_lock_entity is not None:
+            self.state.button_lock_entity.server = self
+            if self.state.button_lock_entity not in self.state.entities:
+                self.state.entities.append(self.state.button_lock_entity)
+            return
+
+        entity = ButtonLockEntity(
+            server=self,
+            key=len(self.state.entities),
+            name="Disable button controls",
+            object_id="disable_button_controls",
+            get_locked=lambda: self.state.button_controls_locked,
+            set_locked=self._set_button_controls_locked,
+        )
+        self.state.entities.append(entity)
+        self.state.button_lock_entity = entity
+        entity.sync_with_state()
+        _LOGGER.info("Button lock switch entity materialised")
+
     def _on_led_light_changed(self, object_id: str) -> None:
         """Forward an HA Light entity change to peripherals as light_command.
 
@@ -442,6 +480,10 @@ class VoiceSatelliteProtocol(APIServer):
             return
         self._emit(LVAEvent.LIGHT_COMMAND, entity.state_dict())
 
+    def _on_button_lock_changed(self, is_locked: bool) -> None:
+        """Forward a button-lock switch change to peripherals as button_lock_changed."""
+        self._emit(LVAEvent.BUTTON_LOCK_CHANGED, {"locked": is_locked})
+
     # ------------------------------------------------------------------
     # Mute / thinking sound
     # ------------------------------------------------------------------
@@ -457,6 +499,21 @@ class VoiceSatelliteProtocol(APIServer):
             pass
         self.state.save_preferences()
 
+    def _set_button_controls_locked(self, new_state: bool) -> None:
+        self.state.button_controls_locked = bool(new_state)
+        self.state.preferences.button_controls_locked = 1 if self.state.button_controls_locked else 0
+        self.state.save_preferences()
+
+        if self.state.button_controls_locked:
+            _LOGGER.debug("On-board button controls locked")
+        else:
+            _LOGGER.debug("On-board button controls unlocked")
+
+        if self.state.button_lock_entity is not None:
+            self.state.button_lock_entity.sync_with_state()
+
+        self._emit(LVAEvent.BUTTON_LOCK_CHANGED, {"locked": self.state.button_controls_locked})
+
     def _set_sensitivity_1(self, new_value: float) -> None:
         self.state.wake_word_1_threshold = float(new_value)
         self.state.preferences.wake_word_1_sensitivity = float(new_value)
@@ -599,6 +656,12 @@ class VoiceSatelliteProtocol(APIServer):
             self._emit(LVAEvent.TIMER_UPDATED, timer_data)
 
         elif event_type == VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_CANCELLED:
+            # Report the cancel as a zeroed update before going idle. Every
+            # shipped peripheral already reads `seconds_left` from
+            # `timer_updated`, and none of them treats zero as "ringing",
+            # that is `timer_ringing`, so the countdown clears without any
+            # peripheral having to learn a new event.
+            self._emit(LVAEvent.TIMER_UPDATED, {**timer_data, "seconds_left": 0})
             self._emit(LVAEvent.IDLE)
 
         elif event_type == VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_FINISHED:

tests/unit/test_models.py

@@ -1,7 +1,7 @@
 """Unit tests for shared models."""
 
 import json
-from unittest.mock import patch
+from unittest.mock import Mock, patch
 
 import pytest
 
@@ -192,6 +192,17 @@ class TestPersistVolume:
         state.persist_volume(1.0)
         assert state.volume == 1.0
 
+    def test_emits_volume_muted_event_when_volume_zero(self, tmp_path):
+        from linux_voice_assistant.peripheral_api import LVAEvent
+
+        state = make_server_state(tmp_path)
+        state.peripheral_api = Mock()
+
+        state.persist_volume(0.0)
+
+        state.peripheral_api.emit_event_sync.assert_any_call(LVAEvent.VOLUME_CHANGED, {"volume": 0.0})
+        state.peripheral_api.emit_event_sync.assert_any_call(LVAEvent.VOLUME_MUTED, {"muted": True})
+
 
 # ---------------------------------------------------------------------------
 # ServerState.persist_mic_gain()
@@ -271,3 +282,30 @@ class TestPersistMicNoise:
         with patch.object(state, "save_preferences") as mock_save:
             state.persist_mic_noise(4.0)
             mock_save.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# initial_stop_word_threshold()
+# ---------------------------------------------------------------------------
+
+
+class TestInitialStopWordThreshold:
+    def test_saved_sensitivity_is_restored(self):
+        from linux_voice_assistant.models import initial_stop_word_threshold
+
+        assert initial_stop_word_threshold(0.9) == pytest.approx(0.9)
+
+    def test_falls_back_to_server_state_default_when_never_set(self):
+        from linux_voice_assistant.models import ServerState, initial_stop_word_threshold
+
+        assert initial_stop_word_threshold(None) == pytest.approx(ServerState.stop_word_threshold)
+
+    def test_clamped_to_one_when_above(self):
+        from linux_voice_assistant.models import initial_stop_word_threshold
+
+        assert initial_stop_word_threshold(1.5) == pytest.approx(1.0)
+
+    def test_clamped_to_zero_when_below(self):
+        from linux_voice_assistant.models import initial_stop_word_threshold
+
+        assert initial_stop_word_threshold(-0.5) == pytest.approx(0.0)

tests/unit/test_peripheral_api.py

@@ -0,0 +1,915 @@
+"""Unit tests for the peripheral WebSocket API (PeripheralAPIServer)."""
+
+import asyncio
+import json
+from typing import Any, Dict, Optional
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from linux_voice_assistant.peripheral_api import LVACommand, LVAEvent, PeripheralAPIServer
+from tests.unit.conftest import make_state
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def make_server(state=None) -> PeripheralAPIServer:
+    """Return a PeripheralAPIServer, optionally attached to a ServerState."""
+    server = PeripheralAPIServer()
+    if state is not None:
+        server.set_state(state)
+    return server
+
+
+def make_media_entity(key: int = 1, volume: float = 0.5, muted: bool = False):
+    entity = MagicMock()
+    entity.key = key
+    entity.volume = volume
+    entity.muted = muted
+    entity.previous_volume = volume
+    return entity
+
+
+async def dispatch(server: PeripheralAPIServer, command: str, data: Optional[dict] = None) -> None:
+    payload: Dict[str, Any] = {"command": command}
+    if data is not None:
+        payload["data"] = data
+    await server._dispatch_command(json.dumps(payload))  # pylint: disable=protected-access
+
+
+# ---------------------------------------------------------------------------
+# __init__ / set_state
+# ---------------------------------------------------------------------------
+
+
+class TestInit:
+    def test_default_host_and_port(self):
+        server = PeripheralAPIServer()
+        assert server._host == "0.0.0.0"
+        assert server._port == 6055
+
+    def test_custom_host_and_port(self):
+        server = PeripheralAPIServer(host="127.0.0.1", port=9999)
+        assert server._host == "127.0.0.1"
+        assert server._port == 9999
+
+    def test_default_volume_step(self):
+        server = PeripheralAPIServer()
+        assert server._volume_step == PeripheralAPIServer.DEFAULT_VOLUME_STEP
+
+    def test_custom_volume_step(self):
+        server = PeripheralAPIServer(volume_step=0.1)
+        assert server._volume_step == pytest.approx(0.1)
+
+    def test_no_clients_on_init(self):
+        server = PeripheralAPIServer()
+        assert server._clients == set()
+
+    def test_no_state_on_init(self):
+        server = PeripheralAPIServer()
+        assert server._state is None
+
+    def test_no_current_state_on_init(self):
+        server = PeripheralAPIServer()
+        assert server._current_state is None
+
+
+class TestSetState:
+    def test_attaches_state(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server()
+        server.set_state(state)
+        assert server._state is state
+
+
+# ---------------------------------------------------------------------------
+# start() / stop()
+# ---------------------------------------------------------------------------
+
+
+class TestStartStop:
+    @pytest.mark.asyncio
+    async def test_start_creates_server(self):
+        server = make_server()
+        fake_server = MagicMock()
+
+        with patch("websockets.server.serve", AsyncMock(return_value=fake_server)) as mock_serve:
+            await server.start()
+
+        mock_serve.assert_called_once()
+        assert server._server is fake_server
+
+    @pytest.mark.asyncio
+    async def test_start_uses_configured_host_and_port(self):
+        server = make_server()
+        server._host = "127.0.0.1"
+        server._port = 1234
+
+        with patch("websockets.server.serve", AsyncMock(return_value=MagicMock())) as mock_serve:
+            await server.start()
+
+        _, args, kwargs = mock_serve.mock_calls[0]
+        assert "127.0.0.1" in args or kwargs.get("host") == "127.0.0.1"
+
+    @pytest.mark.asyncio
+    async def test_start_sets_loop(self):
+        server = make_server()
+
+        with patch("websockets.server.serve", AsyncMock(return_value=MagicMock())):
+            await server.start()
+
+        assert server._loop is asyncio.get_running_loop()
+
+    @pytest.mark.asyncio
+    async def test_stop_closes_server(self):
+        server = make_server()
+        fake_server = MagicMock()
+        fake_server.wait_closed = AsyncMock()
+        server._server = fake_server
+
+        await server.stop()
+
+        fake_server.close.assert_called_once()
+        fake_server.wait_closed.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_stop_without_server_does_not_raise(self):
+        server = make_server()
+        await server.stop()  # should not raise
+
+    @pytest.mark.asyncio
+    async def test_stop_cancels_pending_reconnect_task(self):
+        server = make_server()
+        pending_task = MagicMock()
+        server._pending_entity_reconnect_task = pending_task
+
+        await server.stop()
+
+        pending_task.cancel.assert_called_once()
+        assert server._pending_entity_reconnect_task is None
+
+
+# ---------------------------------------------------------------------------
+# _send_snapshot()
+# ---------------------------------------------------------------------------
+
+
+class TestSendSnapshot:
+    @pytest.mark.asyncio
+    async def test_no_state_does_nothing(self):
+        server = make_server()
+        websocket = AsyncMock()
+        await server._send_snapshot(websocket)  # pylint: disable=protected-access
+        websocket.send.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_sends_snapshot_payload(self, tmp_path):
+        state = make_state(tmp_path, volume=0.75, muted=False, connected=True)
+        server = make_server(state)
+        websocket = AsyncMock()
+
+        await server._send_snapshot(websocket)  # pylint: disable=protected-access
+
+        sent = json.loads(websocket.send.call_args_list[0].args[0])
+        assert sent["event"] == "snapshot"
+        assert sent["data"]["volume"] == pytest.approx(0.75)
+        assert sent["data"]["muted"] is False
+        assert sent["data"]["ha_connected"] is True
+
+    @pytest.mark.asyncio
+    async def test_snapshot_includes_last_conversation_text(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+        server._last_stt_text = "turn on the lights"  # pylint: disable=protected-access
+        server._last_tts_text = "OK"  # pylint: disable=protected-access
+        websocket = AsyncMock()
+
+        await server._send_snapshot(websocket)  # pylint: disable=protected-access
+
+        sent = json.loads(websocket.send.call_args_list[0].args[0])
+        assert sent["data"]["last_stt_text"] == "turn on the lights"
+        assert sent["data"]["last_tts_text"] == "OK"
+
+    @pytest.mark.asyncio
+    async def test_no_current_state_skips_replay(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+        websocket = AsyncMock()
+
+        await server._send_snapshot(websocket)  # pylint: disable=protected-access
+
+        assert websocket.send.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_replays_current_state(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+        server._current_state = LVAEvent.THINKING  # pylint: disable=protected-access
+        websocket = AsyncMock()
+
+        await server._send_snapshot(websocket)  # pylint: disable=protected-access
+
+        assert websocket.send.await_count == 2
+        replayed = json.loads(websocket.send.call_args_list[1].args[0])
+        assert replayed["event"] == "thinking"
+
+    @pytest.mark.asyncio
+    async def test_replays_current_state_with_data(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+        server._current_state = LVAEvent.TIMER_TICKING  # pylint: disable=protected-access
+        server._current_state_data = {"id": "1", "seconds_left": 30}  # pylint: disable=protected-access
+        websocket = AsyncMock()
+
+        await server._send_snapshot(websocket)  # pylint: disable=protected-access
+
+        replayed = json.loads(websocket.send.call_args_list[1].args[0])
+        assert replayed["data"]["seconds_left"] == 30
+
+    @pytest.mark.asyncio
+    async def test_skips_replay_of_disconnected_when_connected(self, tmp_path):
+        state = make_state(tmp_path, connected=True)
+        server = make_server(state)
+        server._current_state = LVAEvent.DISCONNECTED  # pylint: disable=protected-access
+        websocket = AsyncMock()
+
+        await server._send_snapshot(websocket)  # pylint: disable=protected-access
+
+        # Only the snapshot itself should have been sent
+        assert websocket.send.await_count == 1
+
+
+# ---------------------------------------------------------------------------
+# _dispatch_command() — basic parsing
+# ---------------------------------------------------------------------------
+
+
+class TestDispatchCommandParsing:
+    @pytest.mark.asyncio
+    async def test_invalid_json_does_not_raise(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+        await server._dispatch_command("not json")  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_missing_command_is_noop(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+        await server._dispatch_command(json.dumps({}))  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_no_state_attached_is_noop(self):
+        server = make_server()
+        await dispatch(server, LVACommand.START_LISTENING.value)  # should not raise
+
+
+# ---------------------------------------------------------------------------
+# _dispatch_command() — voice pipeline
+# ---------------------------------------------------------------------------
+
+
+class TestDispatchStartListening:
+    @pytest.mark.asyncio
+    async def test_calls_satellite_start_listening(self, tmp_path):
+        state = make_state(tmp_path, muted=False)
+        satellite = MagicMock()
+        state.satellite = satellite
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.START_LISTENING.value)
+
+        satellite.start_listening.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_noop_when_muted(self, tmp_path):
+        state = make_state(tmp_path, muted=True)
+        satellite = MagicMock()
+        state.satellite = satellite
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.START_LISTENING.value)
+
+        satellite.start_listening.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_noop_when_no_satellite(self, tmp_path):
+        state = make_state(tmp_path, muted=False)
+        state.satellite = None
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.START_LISTENING.value)  # should not raise
+
+
+class TestDispatchStopPipeline:
+    @pytest.mark.asyncio
+    async def test_calls_satellite_stop(self, tmp_path):
+        state = make_state(tmp_path)
+        satellite = MagicMock()
+        state.satellite = satellite
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.STOP_PIPELINE.value)
+
+        satellite.stop.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_noop_when_no_satellite(self, tmp_path):
+        state = make_state(tmp_path)
+        state.satellite = None
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.STOP_PIPELINE.value)  # should not raise
+
+
+# ---------------------------------------------------------------------------
+# _dispatch_command() — microphone mute
+# ---------------------------------------------------------------------------
+
+
+class TestDispatchMute:
+    @pytest.mark.asyncio
+    async def test_mute_mic_calls_set_muted(self, tmp_path):
+        state = make_state(tmp_path, muted=False)
+        satellite = MagicMock()
+        state.satellite = satellite
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.MUTE_MIC.value)
+
+        satellite._set_muted.assert_called_once_with(True)  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_mute_mic_noop_when_already_muted(self, tmp_path):
+        state = make_state(tmp_path, muted=True)
+        satellite = MagicMock()
+        state.satellite = satellite
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.MUTE_MIC.value)
+
+        satellite._set_muted.assert_not_called()  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_unmute_mic_calls_set_muted(self, tmp_path):
+        state = make_state(tmp_path, muted=True)
+        satellite = MagicMock()
+        state.satellite = satellite
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.UNMUTE_MIC.value)
+
+        satellite._set_muted.assert_called_once_with(False)  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_unmute_mic_noop_when_already_unmuted(self, tmp_path):
+        state = make_state(tmp_path, muted=False)
+        satellite = MagicMock()
+        state.satellite = satellite
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.UNMUTE_MIC.value)
+
+        satellite._set_muted.assert_not_called()  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_mute_mic_pushes_switch_state_to_ha(self, tmp_path):
+        state = make_state(tmp_path, muted=False)
+        state.mute_switch_entity = MagicMock(key=1)
+        satellite = MagicMock()
+        state.satellite = satellite
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.MUTE_MIC.value)
+
+        satellite.send_messages.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# _dispatch_command() — volume
+# ---------------------------------------------------------------------------
+
+
+class TestDispatchVolume:
+    @pytest.mark.asyncio
+    async def test_volume_up_increases_volume(self, tmp_path):
+        state = make_state(tmp_path, volume=0.5)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.VOLUME_UP.value)
+
+        assert state.volume == pytest.approx(0.5 + PeripheralAPIServer.DEFAULT_VOLUME_STEP)
+
+    @pytest.mark.asyncio
+    async def test_volume_down_decreases_volume(self, tmp_path):
+        state = make_state(tmp_path, volume=0.5)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.VOLUME_DOWN.value)
+
+        assert state.volume == pytest.approx(0.5 - PeripheralAPIServer.DEFAULT_VOLUME_STEP)
+
+    @pytest.mark.asyncio
+    async def test_volume_up_clamped_at_one(self, tmp_path):
+        state = make_state(tmp_path, volume=1.0)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.VOLUME_UP.value)
+
+        assert state.volume == pytest.approx(1.0)
+
+    @pytest.mark.asyncio
+    async def test_volume_down_clamped_at_zero(self, tmp_path):
+        state = make_state(tmp_path, volume=0.0)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.VOLUME_DOWN.value)
+
+        assert state.volume == pytest.approx(0.0)
+
+    @pytest.mark.asyncio
+    async def test_volume_up_updates_music_and_tts_players(self, tmp_path):
+        state = make_state(tmp_path, volume=0.5)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.VOLUME_UP.value)
+
+        state.music_player.set_volume.assert_called_once()
+        state.tts_player.set_volume.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_volume_up_pushes_media_player_state(self, tmp_path):
+        state = make_state(tmp_path, volume=0.5)
+        state.media_player_entity = make_media_entity()
+        satellite = MagicMock()
+        state.satellite = satellite
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.VOLUME_UP.value)
+
+        satellite.send_messages.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_set_volume_sets_exact_value(self, tmp_path):
+        state = make_state(tmp_path, volume=0.2)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.SET_VOLUME.value, {"volume": 0.9})
+
+        assert state.volume == pytest.approx(0.9)
+
+    @pytest.mark.asyncio
+    async def test_set_volume_clamps_above_one(self, tmp_path):
+        state = make_state(tmp_path, volume=0.2)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.SET_VOLUME.value, {"volume": 5.0})
+
+        assert state.volume == pytest.approx(1.0)
+
+    @pytest.mark.asyncio
+    async def test_set_volume_clamps_below_zero(self, tmp_path):
+        state = make_state(tmp_path, volume=0.2)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.SET_VOLUME.value, {"volume": -5.0})
+
+        assert state.volume == pytest.approx(0.0)
+
+    @pytest.mark.asyncio
+    async def test_set_volume_ignores_non_numeric(self, tmp_path):
+        state = make_state(tmp_path, volume=0.2)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.SET_VOLUME.value, {"volume": "loud"})
+
+        assert state.volume == pytest.approx(0.2)
+
+    @pytest.mark.asyncio
+    async def test_set_volume_accepts_int(self, tmp_path):
+        state = make_state(tmp_path, volume=0.2)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.SET_VOLUME.value, {"volume": 1})
+
+        assert state.volume == pytest.approx(1.0)
+
+
+# ---------------------------------------------------------------------------
+# _dispatch_command() — timer
+# ---------------------------------------------------------------------------
+
+
+class TestDispatchStopTimerRinging:
+    @pytest.mark.asyncio
+    async def test_stops_ringing_timer(self, tmp_path):
+        state = make_state(tmp_path)
+        satellite = MagicMock()
+        satellite._timer_finished = True  # pylint: disable=protected-access
+        state.satellite = satellite
+        state.active_wake_words.add(state.stop_word.id)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.STOP_TIMER_RINGING.value)
+
+        assert satellite._timer_finished is False  # pylint: disable=protected-access
+        assert state.stop_word.id not in state.active_wake_words
+        state.tts_player.stop.assert_called_once()
+        satellite.unduck.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_noop_when_not_ringing(self, tmp_path):
+        state = make_state(tmp_path)
+        satellite = MagicMock()
+        satellite._timer_finished = False  # pylint: disable=protected-access
+        state.satellite = satellite
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.STOP_TIMER_RINGING.value)
+
+        state.tts_player.stop.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_noop_when_no_satellite(self, tmp_path):
+        state = make_state(tmp_path)
+        state.satellite = None
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.STOP_TIMER_RINGING.value)  # should not raise
+
+
+# ---------------------------------------------------------------------------
+# _dispatch_command() — media player
+# ---------------------------------------------------------------------------
+
+
+class TestDispatchMediaPlayer:
+    @pytest.mark.asyncio
+    async def test_stop_media_player_stops_music(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.STOP_MEDIA_PLAYER.value)
+
+        state.music_player.stop.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_pause_media_player_pauses_music(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.PAUSE_MEDIA_PLAYER.value)
+
+        state.music_player.pause.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_resume_media_player_resumes_music(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.RESUME_MEDIA_PLAYER.value)
+
+        state.music_player.resume.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_stop_media_player_updates_entity_state(self, tmp_path):
+        from aioesphomeapi.model import MediaPlayerState  # pylint: disable=import-outside-toplevel
+
+        state = make_state(tmp_path)
+        state.media_player_entity = make_media_entity()
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.STOP_MEDIA_PLAYER.value)
+
+        assert state.media_player_entity.state == MediaPlayerState.IDLE
+
+
+# ---------------------------------------------------------------------------
+# _dispatch_command() — button events
+# ---------------------------------------------------------------------------
+
+
+class TestDispatchButtonPress:
+    @pytest.mark.asyncio
+    async def test_single_press_updates_entity(self, tmp_path):
+        state = make_state(tmp_path)
+        state.button_event_sensor_entity = MagicMock()
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.BUTTON_SINGLE_PRESS.value)
+
+        state.button_event_sensor_entity.update_state.assert_called_once_with("single_press")
+
+    @pytest.mark.asyncio
+    async def test_single_press_does_not_play_sound(self, tmp_path):
+        state = make_state(tmp_path)
+        state.button_event_sensor_entity = MagicMock()
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.BUTTON_SINGLE_PRESS.value)
+
+        state.tts_player.play.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_double_press_plays_sound_and_updates_entity(self, tmp_path):
+        state = make_state(tmp_path)
+        state.button_event_sensor_entity = MagicMock()
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.BUTTON_DOUBLE_PRESS.value)
+
+        state.tts_player.play.assert_called_once_with(state.button_double_press_sound)
+        state.button_event_sensor_entity.update_state.assert_called_once_with("double_press")
+
+    @pytest.mark.asyncio
+    async def test_triple_press_plays_sound_and_updates_entity(self, tmp_path):
+        state = make_state(tmp_path)
+        state.button_event_sensor_entity = MagicMock()
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.BUTTON_TRIPLE_PRESS.value)
+
+        state.tts_player.play.assert_called_once_with(state.button_triple_press_sound)
+        state.button_event_sensor_entity.update_state.assert_called_once_with("triple_press")
+
+    @pytest.mark.asyncio
+    async def test_long_press_plays_sound_and_updates_entity(self, tmp_path):
+        state = make_state(tmp_path)
+        state.button_event_sensor_entity = MagicMock()
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.BUTTON_LONG_PRESS.value)
+
+        state.tts_player.play.assert_called_once_with(state.button_long_press_sound)
+        state.button_event_sensor_entity.update_state.assert_called_once_with("long_press")
+
+    @pytest.mark.asyncio
+    async def test_button_press_noop_when_entity_not_registered(self, tmp_path):
+        state = make_state(tmp_path)
+        state.button_event_sensor_entity = None
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.BUTTON_SINGLE_PRESS.value)  # should not raise
+
+
+# ---------------------------------------------------------------------------
+# register_light() / register_button()
+# ---------------------------------------------------------------------------
+
+
+class TestRegisterLight:
+    def test_missing_object_id_is_ignored(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+
+        server._register_light({}, None)  # pylint: disable=protected-access
+
+        assert state.pending_lights == []
+
+    def test_adds_light_registration(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+
+        server._register_light(  # pylint: disable=protected-access
+            {"name": "LEDs", "object_id": "leds", "effects": ["Voice Assistant"]},
+            None,
+        )
+
+        assert len(state.pending_lights) == 1
+        assert state.pending_lights[0].object_id == "leds"
+
+    def test_duplicate_object_id_is_ignored(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+        data = {"name": "LEDs", "object_id": "leds"}
+
+        server._register_light(data, None)  # pylint: disable=protected-access
+        server._register_light(data, None)  # pylint: disable=protected-access
+
+        assert len(state.pending_lights) == 1
+
+    def test_calls_register_pending_lights_on_satellite(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+        satellite = MagicMock()
+
+        server._register_light({"object_id": "leds"}, satellite)  # pylint: disable=protected-access
+
+        satellite.register_pending_lights.assert_called_once()
+
+    def test_no_state_is_noop(self):
+        server = make_server()
+        server._register_light({"object_id": "leds"}, None)  # pylint: disable=protected-access  # should not raise
+
+
+class TestRegisterButton:
+    def test_sets_pending_button(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+
+        server._register_button(None)  # pylint: disable=protected-access
+
+        assert state.pending_button is True
+
+    def test_idempotent_when_already_registered(self, tmp_path):
+        state = make_state(tmp_path)
+        state.pending_button = True
+        server = make_server(state)
+        satellite = MagicMock()
+
+        server._register_button(satellite)  # pylint: disable=protected-access
+
+        satellite.register_pending_button.assert_not_called()
+
+    def test_calls_register_pending_button_on_satellite(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+        satellite = MagicMock()
+
+        server._register_button(satellite)  # pylint: disable=protected-access
+
+        satellite.register_pending_button.assert_called_once()
+
+    def test_no_state_is_noop(self):
+        server = make_server()
+        server._register_button(None)  # pylint: disable=protected-access  # should not raise
+
+
+# ---------------------------------------------------------------------------
+# _dispatch_command() — register_light / register_button routing
+# ---------------------------------------------------------------------------
+
+
+class TestDispatchRegistration:
+    @pytest.mark.asyncio
+    async def test_register_light_command_routes_correctly(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.REGISTER_LIGHT.value, {"object_id": "leds"})
+
+        assert len(state.pending_lights) == 1
+
+    @pytest.mark.asyncio
+    async def test_register_button_command_routes_correctly(self, tmp_path):
+        state = make_state(tmp_path)
+        server = make_server(state)
+
+        await dispatch(server, LVACommand.REGISTER_BUTTON.value)
+
+        assert state.pending_button is True
+
+
+# ---------------------------------------------------------------------------
+# emit_event()
+# ---------------------------------------------------------------------------
+
+
+class TestEmitEvent:
+    @pytest.mark.asyncio
+    async def test_broadcasts_to_all_clients(self):
+        server = make_server()
+        ws1, ws2 = AsyncMock(), AsyncMock()
+        server._clients = {ws1, ws2}  # pylint: disable=protected-access
+
+        await server.emit_event(LVAEvent.IDLE)
+
+        ws1.send.assert_awaited_once()
+        ws2.send.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_no_clients_does_not_raise(self):
+        server = make_server()
+        await server.emit_event(LVAEvent.IDLE)  # should not raise
+
+    @pytest.mark.asyncio
+    async def test_payload_includes_data(self):
+        server = make_server()
+        ws = AsyncMock()
+        server._clients = {ws}  # pylint: disable=protected-access
+
+        await server.emit_event(LVAEvent.STT_TEXT, {"text": "hello"})
+
+        sent = json.loads(ws.send.call_args.args[0])
+        assert sent["data"]["text"] == "hello"
+
+    @pytest.mark.asyncio
+    async def test_dead_client_is_removed(self):
+        server = make_server()
+        dead_ws = AsyncMock()
+        dead_ws.send.side_effect = ConnectionError
+        alive_ws = AsyncMock()
+        server._clients = {dead_ws, alive_ws}  # pylint: disable=protected-access
+
+        await server.emit_event(LVAEvent.IDLE)
+
+        assert dead_ws not in server._clients  # pylint: disable=protected-access
+        assert alive_ws in server._clients  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_caches_last_stt_text(self):
+        server = make_server()
+        await server.emit_event(LVAEvent.STT_TEXT, {"text": "turn on the lights"})
+        assert server._last_stt_text == "turn on the lights"  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_caches_last_tts_text(self):
+        server = make_server()
+        await server.emit_event(LVAEvent.TTS_TEXT, {"text": "OK"})
+        assert server._last_tts_text == "OK"  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_listening_clears_cached_text(self):
+        server = make_server()
+        server._last_stt_text = "old"  # pylint: disable=protected-access
+        server._last_tts_text = "old"  # pylint: disable=protected-access
+
+        await server.emit_event(LVAEvent.LISTENING)
+
+        assert server._last_stt_text is None  # pylint: disable=protected-access
+        assert server._last_tts_text is None  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_tracks_current_state_for_state_events(self):
+        server = make_server()
+        await server.emit_event(LVAEvent.TTS_SPEAKING)
+        assert server._current_state == LVAEvent.TTS_SPEAKING  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_does_not_track_transient_events(self):
+        server = make_server()
+        await server.emit_event(LVAEvent.STT_TEXT, {"text": "hi"})
+        assert server._current_state is None  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_timer_updated_refreshes_ticking_data(self):
+        server = make_server()
+        await server.emit_event(LVAEvent.TIMER_TICKING, {"seconds_left": 30})
+        await server.emit_event(LVAEvent.TIMER_UPDATED, {"seconds_left": 25})
+
+        assert server._current_state == LVAEvent.TIMER_TICKING  # pylint: disable=protected-access
+        assert server._current_state_data["seconds_left"] == 25  # pylint: disable=protected-access
+
+    @pytest.mark.asyncio
+    async def test_timer_updated_ignored_when_not_ticking(self):
+        server = make_server()
+        await server.emit_event(LVAEvent.IDLE)
+        await server.emit_event(LVAEvent.TIMER_UPDATED, {"seconds_left": 25})
+
+        assert server._current_state == LVAEvent.IDLE  # pylint: disable=protected-access
+
+
+# ---------------------------------------------------------------------------
+# emit_event_sync()
+# ---------------------------------------------------------------------------
+
+
+class TestEmitEventSync:
+    def test_noop_when_no_loop(self):
+        server = make_server()
+        server.emit_event_sync(LVAEvent.IDLE)  # should not raise
+
+    def test_noop_when_loop_closed(self):
+        server = make_server()
+        loop = MagicMock()
+        loop.is_closed.return_value = True
+        server._loop = loop  # pylint: disable=protected-access
+
+        with patch("asyncio.run_coroutine_threadsafe") as mock_run:
+            server.emit_event_sync(LVAEvent.IDLE)
+
+        mock_run.assert_not_called()
+
+    def test_schedules_coroutine_on_loop(self):
+        server = make_server()
+        loop = MagicMock()
+        loop.is_closed.return_value = False
+        server._loop = loop  # pylint: disable=protected-access
+
+        with patch("asyncio.run_coroutine_threadsafe") as mock_run:
+            server.emit_event_sync(LVAEvent.IDLE, {"foo": "bar"})
+
+        mock_run.assert_called_once()
+        assert mock_run.call_args.args[1] is loop
+
+
+# ---------------------------------------------------------------------------
+# _create_media_player_response()
+# ---------------------------------------------------------------------------
+
+
+class TestCreateMediaPlayerResponse:
+    def test_returns_response_with_entity_fields(self, tmp_path):
+        from aioesphomeapi.model import MediaPlayerState  # pylint: disable=import-outside-toplevel
+
+        state = make_state(tmp_path)
+        state.media_player_entity = make_media_entity(key=3, volume=0.4, muted=True)
+        server = make_server(state)
+
+        response = server._create_media_player_response(MediaPlayerState.PLAYING)  # pylint: disable=protected-access
+
+        assert response.key == 3
+        assert response.volume == pytest.approx(0.4)
+        assert response.muted is True
+        assert response.state == MediaPlayerState.PLAYING

tests/unit/test_satellite.py

@@ -367,3 +367,69 @@ class TestConnectionLost:
         sat = make_satellite(tmp_path)
         sat.connection_lost(None)
         sat.state.tts_player.stop.assert_called()
+
+
+# ---------------------------------------------------------------------------
+# Timer events
+# ---------------------------------------------------------------------------
+
+
+def _timer_msg(seconds_left=60, total_seconds=300, timer_id="t1", name="pasta"):
+    msg = MagicMock()
+    msg.timer_id = timer_id
+    msg.name = name
+    msg.total_seconds = total_seconds
+    msg.seconds_left = seconds_left
+    return msg
+
+
+class TestHandleTimerEvent:
+    def _sat(self, tmp_path):
+        sat = make_satellite(tmp_path)
+        sat.state.peripheral_api = MagicMock()
+        return sat
+
+    def _emitted(self, sat):
+        return [c.args for c in sat.state.peripheral_api.emit_event_sync.call_args_list]
+
+    def test_started_emits_ticking_with_data(self, tmp_path):
+        from aioesphomeapi.model import VoiceAssistantTimerEventType
+
+        from linux_voice_assistant.peripheral_api import LVAEvent
+
+        sat = self._sat(tmp_path)
+        sat.handle_timer_event(VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_STARTED, _timer_msg())
+        event, data = self._emitted(sat)[0]
+        assert event == LVAEvent.TIMER_TICKING
+        assert data["seconds_left"] == 60
+
+    def test_cancelled_zeroes_the_countdown(self, tmp_path):
+        from aioesphomeapi.model import VoiceAssistantTimerEventType
+
+        from linux_voice_assistant.peripheral_api import LVAEvent
+
+        sat = self._sat(tmp_path)
+        sat.handle_timer_event(
+            VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_CANCELLED,
+            _timer_msg(seconds_left=42),
+        )
+        emitted = self._emitted(sat)
+        assert [e for e, *_ in emitted] == [LVAEvent.TIMER_UPDATED, LVAEvent.IDLE]
+
+        _, data = emitted[0]
+        assert data["seconds_left"] == 0
+        # The identity is preserved, so a peripheral tracking several timers
+        # knows which one went away.
+        assert data["id"] == "t1"
+        assert data["total_seconds"] == 300
+
+    # Peripherals that re-check a running timer on `idle` must see the zeroed
+    # update first, or they stay in the countdown state.
+    def test_cancelled_still_ends_with_idle(self, tmp_path):
+        from aioesphomeapi.model import VoiceAssistantTimerEventType
+
+        from linux_voice_assistant.peripheral_api import LVAEvent
+
+        sat = self._sat(tmp_path)
+        sat.handle_timer_event(VoiceAssistantTimerEventType.VOICE_ASSISTANT_TIMER_CANCELLED, _timer_msg())
+        assert self._emitted(sat)[-1][0] == LVAEvent.IDLE