test: restore regression coverage #120071 dropped as dead or redundant

Review on #120071 (yoniebans) found tests deleted as 'dead' that were only
mis-gated, and regression tests with no remaining equivalent:

- Discord voice receive (real NaCl/RTP: wrong-key drop, DAVE decrypt
  failure, malformed padding, non-allowlisted speaker) moves out of the
  never-selected tests/integration/ to tests/plugins/platforms/, where no
  conftest stubs discord.py; 34 pass with the messaging extra.
- Five *_windows_live.py files had a bare skipif(win32), so
  list_os_marked_tests.py never selected them; now windows_only.
- iron-proxy token-swap E2E (opt-in gate) is back.
- Gateway regressions: cached agent iteration cap (#48127), clarify JSON
  never renders as progress (#52374), handoff watcher DB work off the event
  loop, delivery ledger single connection, checkpoint prune and memory trim
  housekeeping ticks.
- Desktop: command-screenshot IPC rejection, inactive WSL bridge never
  spawns wsl.exe, bootstrap runner git binary.

apps/desktop/electron/bootstrap-runner-git-binary.test.ts

@@ -0,0 +1,21 @@
+import assert from 'node:assert/strict'
+import type * as ChildProcess from 'node:child_process'
+
+import { test, vi } from 'vitest'
+
+const execFileSyncMock = vi.fn((..._args: unknown[]) => 'abcdef1234567890\n')
+
+vi.mock('node:child_process', async importOriginal => ({
+  ...(await importOriginal<typeof ChildProcess>()),
+  execFileSync: execFileSyncMock
+}))
+
+test('resolveCheckoutHead runs the probed git binary, not the first PATH hit (#114718)', async () => {
+  // The bootstrap marker pin used a bare `git`, which on the reported Mac is
+  // an Intel-only /usr/local/bin/git that fails at spawn, silently dropping
+  // the real pin. main.ts now passes the binary it already probed.
+  const { resolveCheckoutHead } = await import('./bootstrap-runner')
+
+  assert.equal(resolveCheckoutHead('/repo', { gitBinary: '/usr/bin/git' }), 'abcdef1234567890')
+  assert.equal(execFileSyncMock.mock.calls[0]?.[0], '/usr/bin/git')
+})

apps/desktop/electron/command-screenshot.test.ts

@@ -0,0 +1,143 @@
+import { EventEmitter } from 'node:events'
+import { mkdtemp, readFile, rm } from 'node:fs/promises'
+import os from 'node:os'
+import path from 'node:path'
+
+import { afterEach, describe, expect, it, vi } from 'vitest'
+
+const native = vi.hoisted(() => ({ start: vi.fn(), stop: vi.fn() }))
+
+const electron = vi.hoisted(() => ({
+  handlers: new Map(),
+  windows: [] as any[],
+  focused: null as any,
+  screenPermission: 'granted',
+  directory: ''
+}))
+
+vi.mock('./command-screenshot-monitor', () => ({
+  CommandScreenshotMonitor: class {
+    start = native.start
+    stop = native.stop
+  }
+}))
+vi.mock('electron', async () => {
+  const { EventEmitter } = await import('node:events')
+  const ipcMain = new EventEmitter() as any
+  ipcMain.handle = (channel: string, handler: unknown) => electron.handlers.set(channel, handler)
+  ipcMain.removeHandler = (channel: string) => electron.handlers.delete(channel)
+
+  return {
+    app: Object.assign(new EventEmitter(), { getPath: () => electron.directory, getAppPath: () => '/app' }),
+    ipcMain,
+    BrowserWindow: {
+      fromWebContents: (wc: unknown) => electron.windows.find(win => win.webContents === wc),
+      getFocusedWindow: () => electron.focused,
+      getAllWindows: () => electron.windows
+    },
+    desktopCapturer: {
+      getSources: vi.fn(async () => [
+        { id: 'window:42:0', thumbnail: { isEmpty: () => false, toPNG: () => new Uint8Array([1]) } }
+      ])
+    },
+    systemPreferences: { getMediaAccessStatus: () => electron.screenPermission },
+    shell: { openExternal: vi.fn() }
+  }
+})
+
+import { app, desktopCapturer, ipcMain } from 'electron'
+
+import { installCommandScreenshot } from './command-screenshot'
+
+const cleanups: (() => void)[] = []
+afterEach(async () => {
+  cleanups.splice(0).forEach(fn => fn())
+  await rm(electron.directory, { recursive: true, force: true })
+  electron.windows = []
+  electron.focused = null
+  electron.screenPermission = 'granted'
+  vi.clearAllMocks()
+})
+
+function window(id: number, url = 'http://127.0.0.1:5174/') {
+  const frame = { url }
+
+  const wc = Object.assign(new EventEmitter(), {
+    id,
+    mainFrame: frame,
+    getURL: () => url,
+    isDestroyed: () => false,
+    send: vi.fn()
+  })
+
+  const win = { webContents: wc, isDestroyed: () => false }
+  electron.windows.push(win)
+
+  return { win, event: { sender: wc, senderFrame: frame }, wc }
+}
+
+async function setup() {
+  electron.directory = await mkdtemp(path.join(os.tmpdir(), 'hermes-screenshot-'))
+  cleanups.push(installCommandScreenshot({ rendererUrl: 'http://127.0.0.1:5174/' }))
+}
+
+const call = (channel: string, event: unknown, ...args: unknown[]) =>
+  electron.handlers.get(`hermes:screenshot:${channel}`)(event, ...args)
+
+describe.skipIf(process.platform !== 'darwin')('Command screenshot native bridge', () => {
+  it('persists opt-in, routes to the last focused subscribed window while backgrounded, and revokes on disable', async () => {
+    await setup()
+    const first = window(1)
+    const second = window(2)
+    expect(await call('settings:get', first.event)).toEqual({ enabled: false, state: 'disabled' })
+    expect(native.start).not.toHaveBeenCalled()
+    ipcMain.emit('subscribe', first.event) // unrelated IPC grants nothing
+    ipcMain.emit('hermes:screenshot:subscribe', first.event, true)
+    ipcMain.emit('hermes:screenshot:subscribe', second.event, true)
+    app.emit('browser-window-focus', {}, second.win)
+    electron.focused = null
+    await call('settings:set', first.event, true)
+    expect(JSON.parse(await readFile(path.join(electron.directory, 'screenshot.json'), 'utf8'))).toEqual({
+      enabled: true
+    })
+    const [capture, status] = native.start.mock.calls.at(-1)!
+    status({ type: 'ready' })
+    capture({ type: 'capture', windowId: 42, width: 600, height: 400 })
+    const requests = second.wc.send.mock.calls.filter(([channel]) => channel === 'hermes:screenshot:request')
+    expect(requests).toHaveLength(1)
+    expect(first.wc.send.mock.calls.some(([channel]) => channel === 'hermes:screenshot:request')).toBe(false)
+    expect(await call('capture', first.event, requests[0]![1])).toEqual({ ok: false, reason: 'expired' })
+    await call('settings:set', second.event, false)
+    expect(await call('capture', second.event, requests[0]![1])).toEqual({ ok: false, reason: 'expired' })
+    expect(native.stop).toHaveBeenCalled()
+
+    for (let i = 0; i < 3; i += 1) {
+      ipcMain.emit('hermes:screenshot:subscribe', second.event, false)
+      ipcMain.emit('hermes:screenshot:subscribe', second.event, true)
+    }
+
+    expect(second.wc.listenerCount('destroyed')).toBe(1)
+    ipcMain.emit('hermes:screenshot:subscribe', second.event, false)
+    expect(second.wc.listenerCount('destroyed')).toBe(0)
+  })
+
+  it('rejects guest/foreign frames and reports denied screen access rather than ready', async () => {
+    await setup()
+    const trusted = window(3)
+    const foreign = window(4, 'https://example.org/')
+    await expect(call('settings:set', foreign.event, true)).rejects.toThrow()
+    await expect(
+      call('settings:set', { ...trusted.event, senderFrame: { url: 'https://example.org/' } }, true)
+    ).rejects.toThrow()
+    expect(native.start).not.toHaveBeenCalled()
+    electron.screenPermission = 'denied'
+    await call('settings:set', trusted.event, true)
+    native.start.mock.calls.at(-1)![1]({ type: 'ready' })
+    expect(desktopCapturer.getSources).toHaveBeenCalledWith({
+      types: ['window'],
+      thumbnailSize: { width: 1, height: 1 },
+      fetchWindowIcons: false
+    })
+    expect(await call('settings:get', trusted.event)).toEqual({ enabled: true, state: 'screen-permission' })
+  })
+})

apps/desktop/electron/wsl-path-bridge-gate.test.ts

@@ -0,0 +1,75 @@
+/**
+ * Windows-platform regression for the WSL path-bridge gate (#66433).
+ *
+ * The behavioural tests in wsl-path-bridge.test.ts prove the no-op contract
+ * (paths pass through unchanged when the bridge is inactive). This file goes
+ * one rung further: with `process.platform` stubbed to `win32` and
+ * `child_process.execFileSync` mocked, it proves the actual `wsl.exe` spawn is
+ * suppressed — not just that the return value looks right.
+ *
+ * Each test re-imports the module fresh (vi.resetModules) so IS_WINDOWS is
+ * re-evaluated against the stubbed platform.
+ */
+import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'
+
+const execFileSyncMock = vi.fn(() => 'Ubuntu\n')
+
+vi.mock('node:child_process', () => ({ execFileSync: execFileSyncMock }))
+
+describe('WSL bridge gate on Windows (#66433)', () => {
+  const realPlatform = process.platform
+
+  beforeEach(() => {
+    Object.defineProperty(process, 'platform', { value: 'win32', configurable: true })
+    vi.resetModules()
+    execFileSyncMock.mockClear()
+  })
+
+  afterEach(() => {
+    Object.defineProperty(process, 'platform', { value: realPlatform, configurable: true })
+  })
+
+  test('wsl.exe IS probed for a POSIX path when the bridge is active (control)', async () => {
+    const { resolveLocalReadPath } = await import('./wsl-path-bridge')
+    resolveLocalReadPath('/home/ubuntu/project')
+    expect(execFileSyncMock).toHaveBeenCalled()
+    // Sanity: it really was wsl.exe, not some other binary.
+    expect(execFileSyncMock).toHaveBeenNthCalledWith(
+      1,
+      'wsl.exe',
+      expect.arrayContaining(['-l', '-q']),
+      expect.anything()
+    )
+  })
+
+  test('wsl.exe is NEVER probed when the bridge is inactive — even for POSIX paths', async () => {
+    const { resolveLocalReadPath, setWslBridgeActive } = await import('./wsl-path-bridge')
+    setWslBridgeActive(false)
+    // A POSIX path that WOULD trigger bridging (and the wsl.exe probe) when
+    // active — but with the bridge off, resolveDefaultWslDistro is never
+    // reached because resolveLocalReadPath returns before it.
+    const result = resolveLocalReadPath('/home/ubuntu/project')
+    expect(execFileSyncMock).not.toHaveBeenCalled()
+    expect(result).toBe('/home/ubuntu/project')
+  })
+
+  test('the picker default-path also skips the wsl.exe probe when inactive', async () => {
+    const { resolvePickerDefaultPath, setWslBridgeActive } = await import('./wsl-path-bridge')
+    setWslBridgeActive(false)
+    const result = resolvePickerDefaultPath('/home/ubuntu')
+    expect(execFileSyncMock).not.toHaveBeenCalled()
+    expect(result).toBe('/home/ubuntu')
+  })
+
+  test('re-enabling the bridge restores wsl.exe probing', async () => {
+    const { resolveLocalReadPath, setWslBridgeActive } = await import('./wsl-path-bridge')
+    setWslBridgeActive(false)
+    resolveLocalReadPath('/home/ubuntu/project')
+    expect(execFileSyncMock).not.toHaveBeenCalled()
+
+    setWslBridgeActive(true)
+    execFileSyncMock.mockClear()
+    resolveLocalReadPath('/home/ubuntu/project')
+    expect(execFileSyncMock).toHaveBeenCalled()
+  })
+})

tests/agent/test_iron_proxy_e2e.py

@@ -0,0 +1,368 @@
+"""End-to-end smoke test for the iron-proxy egress integration.
+
+Spins up the REAL iron-proxy binary (auto-installed if not present), routes
+a curl request through it against a local fake upstream, and verifies that
+the Authorization header was swapped from a proxy token to a real secret.
+
+Gated on the network. Skipped by default in CI unless the user explicitly
+opts in with --run-e2e or HERMES_RUN_E2E=1.  This is intentional — the test
+downloads ~16MB and requires both `openssl` and `curl` to be present.
+"""
+
+from __future__ import annotations
+
+import os
+import socket
+import subprocess
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+from typing import Optional
+
+import pytest
+
+from agent.proxy_sources import iron_proxy as ip
+
+
+pytestmark = pytest.mark.skipif(
+    os.environ.get("HERMES_RUN_E2E", "0") != "1",
+    reason="E2E proxy test — set HERMES_RUN_E2E=1 to run (requires network + curl + openssl)",
+)
+
+
+@pytest.fixture
+def hermes_home(tmp_path, monkeypatch):
+    home = tmp_path / "hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    return home
+
+
+def _free_port() -> int:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+class _CaptureHandler(BaseHTTPRequestHandler):
+    """Records the Authorization header of every incoming request."""
+
+    captured_auth: Optional[str] = None  # class-level so tests can read it
+
+    def do_GET(self):
+        type(self).captured_auth = self.headers.get("Authorization")
+        body = b'{"ok": true}'
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *args, **kwargs):
+        return  # silence access log
+
+
+def test_iron_proxy_swaps_authorization_header_end_to_end(hermes_home, monkeypatch):
+    """Real binary, real CA, real curl. Verify the proxy swaps a proxy-token
+    Authorization header for the real bearer value before forwarding."""
+
+    if not __import__("shutil").which("curl"):
+        pytest.skip("curl not available")
+    if not __import__("shutil").which("openssl"):
+        pytest.skip("openssl not available")
+
+    # ----- fake upstream ----------------------------------------------------
+    upstream_port = _free_port()
+    server = HTTPServer(("127.0.0.1", upstream_port), _CaptureHandler)
+    server_thread = threading.Thread(target=server.serve_forever, daemon=True)
+    server_thread.start()
+
+    try:
+        # ----- iron-proxy install + CA + config ---------------------------
+        binary = ip.install_iron_proxy()
+        assert binary.exists()
+        ca_crt, ca_key = ip.ensure_ca_cert()
+        assert ca_crt.exists()
+
+        real_secret = "sk-real-upstream-value-deadbeef"
+        monkeypatch.setenv("TEST_UPSTREAM_KEY", real_secret)
+        proxy_token = ip.mint_proxy_token("test")
+
+        mapping = ip.TokenMapping(
+            proxy_token=proxy_token,
+            real_env_name="TEST_UPSTREAM_KEY",
+            upstream_hosts=("127.0.0.1",),
+        )
+
+        tunnel_port = _free_port()
+        cfg = ip.build_proxy_config(
+            mappings=[mapping],
+            ca_cert=ca_crt,
+            ca_key=ca_key,
+            tunnel_port=tunnel_port,
+            allowed_hosts=["127.0.0.1"],
+            # Test target is on loopback — clear the default IMDS+loopback
+            # deny list so iron-proxy will dial 127.0.0.1.
+            upstream_deny_cidrs=[],
+            # Hermetic: pin the bind to loopback.  Without this, Linux
+            # hosts with docker0 present would bind the bridge gateway
+            # (the production default) and the loopback curl below would
+            # never reach the proxy.
+            http_listen=[f"127.0.0.1:{tunnel_port}"],
+        )
+        ip.write_proxy_config(cfg)
+        ip.write_mappings([mapping])
+
+        # ----- start the proxy --------------------------------------------
+        try:
+            status = ip.start_proxy()
+        except RuntimeError as exc:
+            pytest.skip(f"iron-proxy could not start in this environment: {exc}")
+        assert status.pid is not None
+
+        # Wait up to 10s for the listener to come up.
+        for _ in range(50):
+            if ip._port_listening("127.0.0.1", tunnel_port):
+                break
+            time.sleep(0.2)
+        else:
+            pytest.fail("iron-proxy never started listening on the tunnel port")
+
+        # ----- request through the proxy ----------------------------------
+        # The fake upstream listens on plain HTTP (not HTTPS).  Plain-HTTP
+        # absolute-form forwards are served by the http_listen listener on
+        # tunnel_port + 1 (tunnel_port itself is the CONNECT/MITM listener
+        # that HTTPS_PROXY traffic hits).  The secrets transform fires on
+        # the plain forward too, swapping the Authorization header.
+        result = subprocess.run(
+            [
+                "curl",
+                "--silent",
+                "--max-time", "10",
+                "-x", f"http://127.0.0.1:{tunnel_port + 1}",
+                "-H", f"Authorization: Bearer {proxy_token}",
+                f"http://127.0.0.1:{upstream_port}/",
+            ],
+            capture_output=True,
+            text=True,
+        )
+        assert result.returncode == 0, f"curl failed: {result.stderr}"
+        # Some iron-proxy versions return 200 with no body; only the swap matters.
+        captured = _CaptureHandler.captured_auth
+        assert captured is not None, "upstream never received the request"
+        assert real_secret in captured, (
+            f"Authorization header was not swapped — upstream saw: {captured!r}"
+        )
+        assert proxy_token not in captured, (
+            f"Proxy token leaked through to upstream: {captured!r}"
+        )
+
+    finally:
+        # ----- cleanup ------------------------------------------------------
+        try:
+            ip.stop_proxy()
+        except Exception:
+            pass
+        server.shutdown()
+        server.server_close()
+
+
+class _CaptureXApiKeyHandler(BaseHTTPRequestHandler):
+    """Records the x-api-key header of every incoming request."""
+
+    captured_key: Optional[str] = None
+
+    def do_GET(self):
+        type(self).captured_key = self.headers.get("x-api-key")
+        body = b'{"ok": true}'
+        self.send_response(200)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *args, **kwargs):
+        return
+
+
+def test_iron_proxy_swaps_x_api_key_header_end_to_end(hermes_home, monkeypatch):
+    """Header-auth providers: the secrets transform must swap the proxy
+    token out of a NON-Authorization header (x-api-key — the Anthropic
+    native scheme) on the pinned binary."""
+
+    if not __import__("shutil").which("curl"):
+        pytest.skip("curl not available")
+    if not __import__("shutil").which("openssl"):
+        pytest.skip("openssl not available")
+
+    upstream_port = _free_port()
+    server = HTTPServer(("127.0.0.1", upstream_port), _CaptureXApiKeyHandler)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+
+    try:
+        binary = ip.install_iron_proxy()
+        assert binary.exists()
+        ca_crt, ca_key = ip.ensure_ca_cert()
+
+        real_secret = "sk-ant-real-value-cafebabe"
+        monkeypatch.setenv("TEST_XAPI_KEY", real_secret)
+        proxy_token = ip.mint_proxy_token("anthropic")
+
+        mapping = ip.TokenMapping(
+            proxy_token=proxy_token,
+            real_env_name="TEST_XAPI_KEY",
+            upstream_hosts=("127.0.0.1",),
+            match_headers=("x-api-key", "Authorization"),
+        )
+
+        tunnel_port = _free_port()
+        cfg = ip.build_proxy_config(
+            mappings=[mapping],
+            ca_cert=ca_crt,
+            ca_key=ca_key,
+            tunnel_port=tunnel_port,
+            allowed_hosts=["127.0.0.1"],
+            upstream_deny_cidrs=[],
+            http_listen=[f"127.0.0.1:{tunnel_port}"],
+        )
+        ip.write_proxy_config(cfg)
+        ip.write_mappings([mapping])
+
+        try:
+            status = ip.start_proxy()
+        except RuntimeError as exc:
+            pytest.skip(f"iron-proxy could not start in this environment: {exc}")
+        assert status.pid is not None
+
+        for _ in range(50):
+            if ip._port_listening("127.0.0.1", tunnel_port):
+                break
+            time.sleep(0.2)
+        else:
+            pytest.fail("iron-proxy never started listening on the tunnel port")
+
+        result = subprocess.run(
+            [
+                "curl",
+                "--silent",
+                "--max-time", "10",
+                "-x", f"http://127.0.0.1:{tunnel_port + 1}",
+                "-H", f"x-api-key: {proxy_token}",
+                f"http://127.0.0.1:{upstream_port}/",
+            ],
+            capture_output=True,
+            text=True,
+        )
+        assert result.returncode == 0, f"curl failed: {result.stderr}"
+        captured = _CaptureXApiKeyHandler.captured_key
+        assert captured is not None, "upstream never received the request"
+        assert real_secret in captured, (
+            f"x-api-key header was not swapped — upstream saw: {captured!r}"
+        )
+        assert proxy_token not in captured, (
+            f"Proxy token leaked through to upstream: {captured!r}"
+        )
+
+    finally:
+        try:
+            ip.stop_proxy()
+        except Exception:
+            pass
+        server.shutdown()
+        server.server_close()
+
+
+def test_iron_proxy_management_reload_end_to_end(hermes_home, monkeypatch):
+    """Real binary: the management listener comes up, an authenticated
+    POST /v1/reload succeeds after a config edit, and the edited ruleset
+    takes effect WITHOUT a restart (same pid)."""
+
+    if not __import__("shutil").which("curl"):
+        pytest.skip("curl not available")
+    if not __import__("shutil").which("openssl"):
+        pytest.skip("openssl not available")
+
+    upstream_port = _free_port()
+    server = HTTPServer(("127.0.0.1", upstream_port), _CaptureHandler)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+
+    try:
+        binary = ip.install_iron_proxy()
+        assert binary.exists()
+        ca_crt, ca_key = ip.ensure_ca_cert()
+
+        real_secret = "sk-real-reload-value-0badf00d"
+        monkeypatch.setenv("TEST_RELOAD_KEY", real_secret)
+        token_v1 = ip.mint_proxy_token("v1")
+
+        def _write_cfg(mapping):
+            cfg = ip.build_proxy_config(
+                mappings=[mapping],
+                ca_cert=ca_crt,
+                ca_key=ca_key,
+                tunnel_port=tunnel_port,
+                allowed_hosts=["127.0.0.1"],
+                upstream_deny_cidrs=[],
+                http_listen=[f"127.0.0.1:{tunnel_port}"],
+            )
+            ip.write_proxy_config(cfg)
+            ip.write_mappings([mapping])
+
+        tunnel_port = _free_port()
+        _write_cfg(ip.TokenMapping(
+            proxy_token=token_v1,
+            real_env_name="TEST_RELOAD_KEY",
+            upstream_hosts=("127.0.0.1",),
+        ))
+
+        try:
+            status = ip.start_proxy()
+        except RuntimeError as exc:
+            pytest.skip(f"iron-proxy could not start in this environment: {exc}")
+        pid_before = status.pid
+        assert pid_before is not None
+
+        for _ in range(50):
+            if ip._port_listening("127.0.0.1", tunnel_port):
+                break
+            time.sleep(0.2)
+        else:
+            pytest.fail("iron-proxy never started listening")
+
+        # Rotate the sandbox-visible token in the config, then hot-reload.
+        token_v2 = ip.mint_proxy_token("v2")
+        _write_cfg(ip.TokenMapping(
+            proxy_token=token_v2,
+            real_env_name="TEST_RELOAD_KEY",
+            upstream_hosts=("127.0.0.1",),
+        ))
+        assert ip.reload_proxy() is True
+
+        # Same daemon (no restart) ...
+        assert ip.get_status().pid == pid_before
+
+        # ... but the NEW token now swaps.
+        _CaptureHandler.captured_auth = None
+        result = subprocess.run(
+            [
+                "curl", "--silent", "--max-time", "10",
+                "-x", f"http://127.0.0.1:{tunnel_port + 1}",
+                "-H", f"Authorization: Bearer {token_v2}",
+                f"http://127.0.0.1:{upstream_port}/",
+            ],
+            capture_output=True, text=True,
+        )
+        assert result.returncode == 0, f"curl failed: {result.stderr}"
+        captured = _CaptureHandler.captured_auth
+        assert captured is not None and real_secret in captured, (
+            f"post-reload token was not swapped — upstream saw: {captured!r}"
+        )
+
+    finally:
+        try:
+            ip.stop_proxy()
+        except Exception:
+            pass
+        server.shutdown()
+        server.server_close()

tests/gateway/test_cached_agent_max_iterations.py

@@ -0,0 +1,77 @@
+"""Regression tests for PR #48127: cached agent max_iterations refresh.
+
+When a long-lived gateway reuses an agent from its cache, the agent must run
+the *current* configured iteration budget — not the budget it was constructed
+with on the first turn of that session. Two pieces make that true:
+
+1. ``GatewayRunner._init_cached_agent_for_turn`` must NOT reset
+   ``max_iterations`` itself (the gateway refreshes it explicitly right after,
+   from current config). If this helper ever started clobbering it, the
+   gateway's refresh would be silently undone.
+2. The per-turn budget object is rebuilt from ``agent.max_iterations`` at the
+   start of every turn (``agent/turn_context.py`` -> ``IterationBudget``), so
+   refreshing ``max_iterations`` on the cached agent is sufficient to change
+   the operative cap the agent loop checks.
+
+These tests exercise the real code paths rather than asserting a plain
+assignment, so they fail if either contract regresses.
+"""
+
+import time
+from types import SimpleNamespace
+
+from agent.iteration_budget import IterationBudget
+from agent.session_activity import ActivityProvenance
+
+
+def _make_cached_agent(max_iterations: int) -> SimpleNamespace:
+    """A minimal stand-in cached agent with the attributes the helpers touch."""
+    # The turn loop checks both api_call_count >= max_iterations AND
+    # iteration_budget.remaining <= 0 (turn_finalizer.py), so the budget must
+    # also reflect the new cap. Seed it with the stale value to prove the
+    # refresh propagates.
+    return SimpleNamespace(
+        _last_activity_ts=time.time() - 1000,
+        _last_activity_desc="previous turn",
+        _last_activity_provenance=ActivityProvenance.AGENT_COMPRESSION,
+        _api_call_count=42,
+        _last_flushed_db_idx=5,
+        max_iterations=max_iterations,
+        iteration_budget=IterationBudget(max_iterations),
+    )
+
+
+def test_init_cached_agent_for_turn_does_not_touch_max_iterations():
+    """The per-turn reset helper must leave max_iterations untouched.
+
+    The gateway refreshes max_iterations explicitly right after calling this
+    helper; if the helper ever reset it, that refresh would be undone.
+    """
+    from gateway.run import GatewayRunner
+
+    agent = _make_cached_agent(90)
+    GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=0)
+
+    # Per-turn state was reset...
+    assert agent._api_call_count == 0
+    assert agent._last_activity_desc == "starting new turn (cached)"
+    assert agent._last_activity_provenance is ActivityProvenance.UNKNOWN
+    assert agent._last_flushed_db_idx == 0
+    # ...but the iteration budget was NOT changed by the helper itself.
+    assert agent.max_iterations == 90
+
+
+def test_init_cached_agent_preserves_max_iterations_on_interrupt_depth():
+    """Interrupt-recursive turns must also leave max_iterations alone."""
+    from gateway.run import GatewayRunner
+
+    agent = _make_cached_agent(200)
+    GatewayRunner._init_cached_agent_for_turn(agent, interrupt_depth=1)
+
+    # Activity timestamps preserved for the inactivity watchdog (#15654)...
+    assert agent._last_activity_desc == "previous turn"
+    assert agent._last_activity_provenance is ActivityProvenance.AGENT_COMPRESSION
+    # ...and max_iterations untouched.
+    assert agent.max_iterations == 200
+
+

tests/gateway/test_checkpoint_prune_housekeeping.py

@@ -0,0 +1,31 @@
+"""Checkpoint store pruning rides the gateway housekeeping tick, not the constructor.
+
+Its ``git gc`` repacks the whole store (tens of seconds on a GB store); run at construction it
+delayed the control socket, adapters and the code_sha stamp, so the first restart of the day (the
+``hermes update`` one) looked hung and failed fleet verification.
+"""
+
+import gateway.run as gateway_run
+
+
+class _OneTickStopEvent:
+    def __init__(self):
+        self.waited = False
+
+    def is_set(self):
+        return self.waited
+
+    def wait(self, timeout=None):
+        self.waited = True
+        return True
+
+
+def test_gateway_housekeeping_runs_the_checkpoint_prune(monkeypatch):
+    import tools.checkpoint_manager as cm
+
+    calls = []
+    monkeypatch.setattr(cm, "auto_prune_from_config", lambda: calls.append(True) or {"skipped": False})
+
+    gateway_run._start_gateway_housekeeping(_OneTickStopEvent(), interval=0)
+
+    assert calls == [True]

tests/gateway/test_clarify_progress_leak.py

@@ -0,0 +1,156 @@
+"""Regression tests for #52374 — raw clarify tool-call JSON must never leak
+into the chat as a tool-progress bubble.
+
+The adapter's ``send_clarify`` is the user-facing rendering of a clarify
+prompt (interactive buttons, or the numbered-text fallback).  The gateway's
+tool-progress callback used to also render a progress bubble for the
+``clarify`` tool.started event — in verbose mode that bubble contains the raw
+tool-call args JSON (``{"question": ..., "choices": [...]}``), and because the
+progress queue drains on a background task the JSON landed right underneath
+the rendered interactive prompt on Slack.
+"""
+
+import importlib
+import sys
+import time
+import types
+
+import pytest
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import BasePlatformAdapter, SendResult
+from gateway.session import SessionSource
+
+
+class ProgressCaptureAdapter(BasePlatformAdapter):
+    """Records every send so the test can assert nothing leaked."""
+
+    def __init__(self, platform=Platform.SLACK):
+        super().__init__(PlatformConfig(enabled=True, token="***"), platform)
+        self.sent = []
+        self.edits = []
+
+    async def connect(self, *, is_reconnect: bool = False) -> bool:
+        return True
+
+    async def disconnect(self) -> None:
+        return None
+
+    async def send(self, chat_id, content, reply_to=None, metadata=None) -> SendResult:
+        self.sent.append({"chat_id": chat_id, "content": content})
+        return SendResult(success=True, message_id="m-1")
+
+    async def edit_message(self, chat_id, message_id, content) -> SendResult:
+        self.edits.append({"chat_id": chat_id, "message_id": message_id, "content": content})
+        return SendResult(success=True, message_id=message_id)
+
+    async def send_typing(self, chat_id, metadata=None) -> None:
+        return None
+
+    async def stop_typing(self, chat_id) -> None:
+        return None
+
+    async def get_chat_info(self, chat_id: str):
+        return {"id": chat_id}
+
+
+class ClarifyThenToolAgent:
+    """Emits a clarify tool.started (with raw args) then a normal tool."""
+
+    def __init__(self, **kwargs):
+        self.tool_progress_callback = kwargs.get("tool_progress_callback")
+        self.tools = []
+
+    def run_conversation(self, message, conversation_history=None, task_id=None):
+        cb = self.tool_progress_callback
+        if cb is not None:
+            cb(
+                "tool.started",
+                "clarify",
+                "Which environment?",
+                {"question": "Which environment?", "choices": ["staging", "production"]},
+            )
+            time.sleep(0.35)
+            cb("tool.started", "terminal", "pwd", {})
+            time.sleep(0.35)
+        return {"final_response": "done", "messages": [], "api_calls": 1}
+
+
+def _make_runner(adapter):
+    gateway_run = importlib.import_module("gateway.run")
+    GatewayRunner = gateway_run.GatewayRunner
+    runner = object.__new__(GatewayRunner)
+    runner.adapters = {adapter.platform: adapter}
+    runner._voice_mode = {}
+    runner._prefill_messages = []
+    runner._ephemeral_system_prompt = ""
+    runner._reasoning_config = None
+    runner._provider_routing = {}
+    runner._fallback_model = None
+    runner._session_db = None
+    runner._running_agents = {}
+    runner._session_run_generation = {}
+    runner.hooks = types.SimpleNamespace(loaded_hooks=False)
+    runner.config = types.SimpleNamespace(
+        thread_sessions_per_user=False,
+        group_sessions_per_user=False,
+        stt_enabled=False,
+    )
+    return runner
+
+
+def _install_fakes(monkeypatch, mode):
+    monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", mode)
+
+    fake_dotenv = types.ModuleType("dotenv")
+    fake_dotenv.load_dotenv = lambda *a, **k: None
+    monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv)
+
+    fake_run_agent = types.ModuleType("run_agent")
+    fake_run_agent.AIAgent = ClarifyThenToolAgent
+    monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
+    import tools.terminal_tool  # noqa: F401 — register terminal emoji
+
+    gateway_run = importlib.import_module("gateway.run")
+    monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"})
+    return gateway_run
+
+
+@pytest.mark.parametrize("mode", ["verbose", "all"])
+@pytest.mark.asyncio
+async def test_clarify_tool_never_renders_progress_bubble(monkeypatch, tmp_path, mode):
+    """No progress bubble for clarify — in any mode, especially verbose.
+
+    Verbose mode used to dump the raw args JSON
+    (``{"question": ..., "choices": [...]}``) into the chat right under the
+    interactive prompt (#52374).
+    """
+    adapter = ProgressCaptureAdapter()
+    runner = _make_runner(adapter)
+    gateway_run = _install_fakes(monkeypatch, mode)
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+
+    source = SessionSource(platform=Platform.SLACK, chat_id="C1", chat_type="dm")
+
+    result = await runner._run_agent(
+        message="hello",
+        context_prompt="",
+        history=[],
+        source=source,
+        session_id="sess-clarify-leak",
+        session_key="agent:main:slack:dm:C1",
+    )
+
+    assert result["final_response"] == "done"
+    all_content = "\n".join(
+        [m["content"] for m in adapter.sent] + [e["content"] for e in adapter.edits]
+    )
+    # Raw clarify args JSON must not leak anywhere.
+    assert '"question"' not in all_content
+    assert '"choices"' not in all_content
+    assert "Which environment?" not in all_content
+    # No clarify progress line at all (verb "Asking" / tool name).
+    assert "clarify" not in all_content
+    assert "Asking" not in all_content
+    # The unrelated terminal tool still renders progress normally.
+    assert "pwd" in all_content

tests/gateway/test_control_socket_windows_live.py

@@ -0,0 +1,151 @@
+"""LIVE Windows E2E for the gateway control socket named-pipe transport.
+
+Runs ONLY on a real Windows host (the on-demand ``windows-venv-e2e.yml``
+lane). Spawns a REAL child process that binds the REAL named pipe via the
+proactor event loop with the DEFAULT verb handlers, then drives the real
+sync client and the real fleet consumers against it — no mocks anywhere.
+
+Proves, on windows-latest:
+  1. `GatewayControlServer` binds ``\\\\.\\pipe\\hermes-gateway-<hash>`` via
+     ``loop.start_serving_pipe`` and answers ``identify``/``status``.
+  2. The sync client's pipe transport (open/write/read/busy-retry) works
+     against a live server and returns the child's true pid + code identity.
+  3. ``collect_fleet_versions()`` prefers the socket (``source: socket``).
+  4. After the server process is force-killed, the client returns None
+     (FileNotFoundError on the pipe — no stale-file hazard on Windows) and
+     consumers fall back to the state-file/scan layer.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.windows_only
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+
+_CHILD_CODE = r"""
+import asyncio, os, sys
+sys.path.insert(0, sys.argv[1])
+os.environ["HERMES_HOME"] = sys.argv[2]
+from gateway.control_socket import GatewayControlServer
+
+async def main():
+    server = GatewayControlServer()
+    ok = await server.start()
+    # Print our REAL pid: on Windows uv venvs, python.exe is a trampoline
+    # that spawns the actual interpreter as a child, so Popen.pid is the
+    # shim, not the server process. (That spawner-view-vs-reality gap is
+    # the exact bug class the control socket exists to eliminate.)
+    print(f"SERVER_STARTED {os.getpid()}" if ok else "SERVER_FAILED", flush=True)
+    if not ok:
+        return
+    await asyncio.sleep(120)
+
+asyncio.run(main())
+"""
+
+
+@pytest.fixture()
+def live_server(tmp_path: Path):
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    proc = subprocess.Popen(
+        [sys.executable, "-c", _CHILD_CODE, str(PROJECT_ROOT), str(home)],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        cwd=str(PROJECT_ROOT),
+    )
+    line = proc.stdout.readline().strip()
+    if not line.startswith("SERVER_STARTED"):
+        err = proc.stderr.read() if proc.poll() is not None else ""
+        proc.kill()
+        pytest.fail(f"pipe server child failed to start: {line!r} {err}")
+    server_pid = int(line.split()[1])
+    yield proc, home, server_pid
+    _kill_tree(proc)
+
+
+def _kill_tree(proc: subprocess.Popen) -> None:
+    """Kill the spawned child AND its descendants (uv trampoline shims)."""
+    if proc.poll() is None:
+        subprocess.run(
+            ["taskkill", "/PID", str(proc.pid), "/T", "/F"],
+            capture_output=True,
+        )
+        proc.wait()
+
+
+def test_named_pipe_identify_status_and_fleet_consumer(live_server, monkeypatch):
+    proc, home, server_pid = live_server
+    from gateway.control_socket import identify_gateway, query_gateway_control
+
+    ident = identify_gateway(home, timeout=5.0)
+    assert ident is not None, "identify returned None against a live pipe server"
+    # Compare against the server's SELF-reported pid, not Popen.pid — uv's
+    # Windows trampoline makes the spawner's view wrong (see _CHILD_CODE).
+    assert ident["pid"] == server_pid
+    assert ident["protocol"] == 1
+    assert ident["kind"] == "hermes-gateway"
+    assert ident["supervisor"] in {"systemd", "launchd", "desktop", "external", "manual"}
+
+    status = query_gateway_control(home, "status", timeout=5.0)
+    assert status is not None
+    assert status["answering_pid"] == server_pid
+
+    # Real fleet consumer prefers the pipe
+    import hermes_cli.update_receipt as ur
+
+    monkeypatch.setattr(
+        "hermes_cli.build_info.get_code_identity",
+        lambda refresh=False: {"sha": ident.get("code_sha") or "X", "version": "t"},
+    )
+    monkeypatch.setattr("hermes_cli.profiles._get_default_hermes_home", lambda: home)
+    monkeypatch.setattr(
+        "hermes_cli.profiles._get_profiles_root", lambda: home / "no-profiles"
+    )
+    fleet = ur.collect_fleet_versions()
+    assert len(fleet) == 1, fleet
+    assert fleet[0]["source"] == "socket"
+    assert fleet[0]["pid"] == server_pid
+
+
+def test_pipe_gone_after_kill_falls_back(live_server, monkeypatch):
+    proc, home, server_pid = live_server
+    from gateway.control_socket import identify_gateway
+
+    assert identify_gateway(home, timeout=5.0) is not None
+    _kill_tree(proc)
+    time.sleep(0.5)
+
+    assert identify_gateway(home, timeout=2.0) is None
+
+    # Consumer falls back to the state file (live pid = this test process)
+    import hermes_cli.update_receipt as ur
+
+    (home / "gateway_state.json").write_text(
+        json.dumps(
+            {"pid": os.getpid(), "code_sha": "OLD", "kind": "hermes-gateway"}
+        ),
+        encoding="utf-8",
+    )
+    monkeypatch.setattr(
+        "hermes_cli.build_info.get_code_identity",
+        lambda refresh=False: {"sha": "NEW", "version": "t"},
+    )
+    monkeypatch.setattr("hermes_cli.profiles._get_default_hermes_home", lambda: home)
+    monkeypatch.setattr(
+        "hermes_cli.profiles._get_profiles_root", lambda: home / "no-profiles"
+    )
+    fleet = ur.collect_fleet_versions()
+    assert len(fleet) == 1, fleet
+    assert "source" not in fleet[0]
+    assert fleet[0]["state"] == "stale"

tests/gateway/test_delivery_ledger_single_connection.py

@@ -0,0 +1,31 @@
+"""Recording a reply opens exactly one SQLite connection.
+
+``record_obligation`` runs on every outbound final response. Its retention prune must run on the
+recording connection, inside the recording transaction — the way the cron ledgers prune — not on a
+second connection opened after the first one closed.
+"""
+from __future__ import annotations
+
+import sqlite3
+
+from gateway import delivery_ledger as dl
+
+
+def test_recording_a_reply_does_not_open_a_second_connection(tmp_path, monkeypatch):
+    monkeypatch.setattr(dl, "_db_path", lambda: tmp_path / "state.db")
+    real_connect = sqlite3.connect
+    opened: list[str] = []
+
+    def counting_connect(*args, **kwargs):
+        opened.append(str(args[0]) if args else str(kwargs.get("database")))
+        return real_connect(*args, **kwargs)
+
+    monkeypatch.setattr(sqlite3, "connect", counting_connect)
+    dl.record_obligation(obligation_id="ob-warm", session_key="s", platform="p", chat_id="c",
+                         thread_id=None, content="warm-up (schema)")
+    opened.clear()
+
+    dl.record_obligation(obligation_id="ob-1", session_key="s", platform="p", chat_id="c",
+                         thread_id=None, content="hello")
+
+    assert len(opened) == 1, opened

tests/gateway/test_handoff_watcher_async_db.py

@@ -0,0 +1,134 @@
+"""Regression test for #40695 (salvage of keystone PR #40782).
+
+The Discord gateway heartbeat was stalling because the handoff watcher
+(``GatewayRunner._handoff_watcher``) polled the synchronous, blocking
+SQLite-backed ``SessionDB`` directly on the asyncio event loop every 2s
+('Shard ID None heartbeat blocked for more than N seconds').
+
+The fix routes every blocking ``SessionDB`` call in the watcher through the
+``AsyncSessionDB`` facade, which offloads each call via ``asyncio.to_thread`` so
+the SQLite I/O runs on a worker thread and never blocks the event loop / Discord
+heartbeat.
+
+These tests assert that behaviour contract. They are mutation-survivable:
+reverting any ``await self._session_db.<call>(...)`` back to a direct synchronous
+call on the loop makes the relevant assertion fail.
+"""
+
+import asyncio
+import types
+
+import pytest
+
+import gateway.run as run
+
+
+class _RecordingSessionDB:
+    """SessionDB stand-in that records the thread each method runs on.
+
+    If the watcher calls these methods directly on the event loop (the bug),
+    they run on the loop thread. If they are wrapped in ``asyncio.to_thread``
+    (the fix), they run on a *different* worker thread.
+    """
+
+    def __init__(self, loop_thread_ident):
+        self._loop_thread_ident = loop_thread_ident
+        self.threads = {}
+        self.calls = []
+
+    def _record(self, name):
+        import threading
+
+        self.threads.setdefault(name, []).append(threading.get_ident())
+        self.calls.append(name)
+
+    def ran_off_loop(self, name):
+        """True iff every call to ``name`` ran on a non-loop thread."""
+        idents = self.threads.get(name, [])
+        return bool(idents) and all(i != self._loop_thread_ident for i in idents)
+
+    def list_pending_handoffs(self):
+        self._record("list_pending_handoffs")
+        return [{"id": "sess-1"}]
+
+    def claim_handoff(self, session_id):
+        self._record("claim_handoff")
+        return True
+
+    def complete_handoff(self, session_id):
+        self._record("complete_handoff")
+
+    def fail_handoff(self, session_id, error):
+        self._record("fail_handoff")
+
+
+def _make_fake_runner(session_db, *, fail_process=False):
+    """Build a minimal object that exposes exactly what the loop body touches.
+
+    The watcher now talks to the SessionDB through the AsyncSessionDB facade,
+    so wrap the recording stand-in the same way the gateway does.
+    """
+    from hermes_state import AsyncSessionDB
+
+    fake = types.SimpleNamespace()
+    fake._session_db = AsyncSessionDB(session_db)
+    # _running yields True for the first loop check, then False so the loop
+    # exits after a single tick.
+    states = iter([True, False])
+
+    class _Running:
+        def __bool__(_self):
+            try:
+                return next(states)
+            except StopIteration:
+                return False
+
+    fake._running = _Running()
+
+    async def _process_handoff(row):
+        if fail_process:
+            raise RuntimeError("boom")
+
+    fake._process_handoff = _process_handoff
+    return fake
+
+
+async def _run_one_tick(fake, monkeypatch):
+    """Run the watcher for a single tick with sleeps neutralised."""
+
+    async def _no_sleep(_seconds):
+        return None
+
+    monkeypatch.setattr(run.asyncio, "sleep", _no_sleep)
+    # Bind the real (patched) method onto our minimal stand-in.
+    coro = run.GatewayRunner._handoff_watcher(fake, interval=0.0)
+    await asyncio.wait_for(coro, timeout=5)
+
+
+@pytest.mark.asyncio
+async def test_watcher_wraps_calls_via_asyncio_to_thread(monkeypatch):
+    """Explicitly assert the offload goes through asyncio.to_thread.
+
+    Patches the AsyncSessionDB facade's ``asyncio.to_thread`` (it lives in
+    hermes_state) and records which SessionDB callables were handed to it.
+    Mutation-survivable: dropping any await removes its callable from the set.
+    """
+    import hermes_state
+
+    db = _RecordingSessionDB(loop_thread_ident=-1)
+    fake = _make_fake_runner(db, fail_process=False)
+
+    wrapped = []
+    real_to_thread = hermes_state.asyncio.to_thread
+
+    async def _spy_to_thread(func, *args, **kwargs):
+        wrapped.append(getattr(func, "__name__", repr(func)))
+        return await real_to_thread(func, *args, **kwargs)
+
+    monkeypatch.setattr(hermes_state.asyncio, "to_thread", _spy_to_thread)
+
+    await _run_one_tick(fake, monkeypatch)
+
+    assert "list_pending_handoffs" in wrapped
+    assert "claim_handoff" in wrapped
+    assert "complete_handoff" in wrapped

tests/gateway/test_memory_trim_housekeeping.py

@@ -0,0 +1,32 @@
+"""Memory-trim coverage for the long-lived messaging gateway housekeeper."""
+
+import gateway.run as gateway_run
+
+
+class _OneTickStopEvent:
+    """Run one housekeeping tick without a sleep or background thread."""
+
+    def __init__(self):
+        self.waited = False
+
+    def is_set(self):
+        return self.waited
+
+    def wait(self, timeout=None):
+        self.waited = True
+        return True
+
+
+def test_gateway_housekeeping_calls_periodic_memory_trim(monkeypatch):
+    import hermes_cli.mem_trim as mem_trim
+
+    calls = []
+    monkeypatch.setattr(
+        mem_trim,
+        "trim_memory",
+        lambda **kwargs: calls.append(kwargs) or True,
+    )
+
+    gateway_run._start_gateway_housekeeping(_OneTickStopEvent(), interval=0)
+
+    assert calls == [{"reason": "messaging gateway housekeeping"}]

tests/hermes_cli/test_desktop_lifecycle_windows_live.py

@@ -0,0 +1,142 @@
+"""LIVE Windows E2E for the Desktop-lifecycle cold-start skip (#76129/#76745).
+
+Runs ONLY on a real Windows host (the on-demand ``windows-venv-e2e.yml``
+lane). Exercises the REAL ownership predicate against REAL processes:
+
+ 1. A real child process self-registers in the REAL spawn ledger as a
+    ``serve`` purpose with THIS process as its live spawner — ownership
+    must hold, and ``_pause_windows_gateways_for_update`` must return None
+    (no cold-start plan) even with an autostart artifact present.
+ 2. Kill the child (dead serve) — ownership drops, the pause plan carries
+    ``cold_start_if_installed`` again.
+ 3. Venv-holder fallback rung: a real venv-python process with true
+    ``serve`` argv is detected by the scan and, having a live parent,
+    confers ownership; the token classifier rejects a ``kanban
+    --preserve-cache`` lookalike (the #90778 class the salvage fixed).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+
+pytestmark = pytest.mark.windows_only
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+
+
+@pytest.fixture()
+def sleeper():
+    procs: list[subprocess.Popen] = []
+
+    def _spawn(*tail: str) -> subprocess.Popen:
+        p = subprocess.Popen(
+            [sys.executable, "-c", "import time; time.sleep(120)", *tail],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        )
+        procs.append(p)
+        time.sleep(0.5)
+        assert p.poll() is None
+        return p
+
+    yield _spawn
+    for p in procs:
+        if p.poll() is None:
+            p.kill()
+            p.wait()
+
+
+def _write_ledger(entries: list[dict]) -> None:
+    from hermes_cli import process_identity as pid_mod
+
+    path = pid_mod._ledger_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(entries), encoding="utf-8")
+
+
+def _entry(proc: subprocess.Popen, purpose: str = "serve") -> dict:
+    import psutil
+
+    from hermes_cli import process_identity as pid_mod
+
+    return {
+        "install": pid_mod.install_id(None),
+        "pid": proc.pid,
+        "create_time": psutil.Process(proc.pid).create_time(),
+        "purpose": purpose,
+        "spawner_pid": os.getpid(),
+        "spawner_create": psutil.Process(os.getpid()).create_time(),
+    }
+
+
+def test_live_supervised_serve_suppresses_cold_start(sleeper, monkeypatch, tmp_path):
+    from hermes_cli import gateway as hermes_gateway
+    from hermes_cli import gateway_windows
+    from hermes_cli import update_cmd
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    (tmp_path / ".hermes").mkdir()
+
+    serve = sleeper()
+    _write_ledger([_entry(serve)])
+
+    assert update_cmd._desktop_owns_gateway_lifecycle() is True
+
+    # Autostart artifact present + no gateway running: WITHOUT ownership the
+    # pause phase would plan a cold start; WITH it, no plan.
+    monkeypatch.setattr(hermes_gateway, "find_gateway_pids", lambda **_k: [])
+    monkeypatch.setattr(gateway_windows, "is_installed", lambda: True)
+    assert update_cmd._pause_windows_gateways_for_update() is None
+
+    # Dead serve → ownership drops → plan returns.
+    serve.kill()
+    serve.wait()
+    time.sleep(0.5)
+    assert update_cmd._desktop_owns_gateway_lifecycle() is False
+    token = update_cmd._pause_windows_gateways_for_update()
+    assert token is not None and token.get("cold_start_if_installed") is True
+
+
+def test_holder_scan_fallback_respects_token_classifier(sleeper, monkeypatch, tmp_path):
+    from hermes_cli import update_cmd
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
+    (tmp_path / ".hermes").mkdir()
+    _write_ledger([])  # force the fallback rung
+
+    # Real process whose argv carries genuine serve shape, visible to psutil.
+    serve_like = sleeper("-m", "hermes_cli.main", "serve")
+    # Lookalike from the #90778 class — must NOT confer ownership.
+    kanban_like = sleeper("-m", "hermes_cli.main", "kanban", "--preserve-cache")
+
+    def fake_holders():
+        import psutil
+
+        out = []
+        for p in (serve_like, kanban_like):
+            proc = psutil.Process(p.pid)
+            out.append((p.pid, proc.name(), " ".join(proc.cmdline())))
+        return out
+
+    monkeypatch.setattr(
+        "hermes_cli.main._detect_venv_python_processes", fake_holders
+    )
+
+    # serve-shaped holder with a live parent (us) → owns
+    assert update_cmd._desktop_owns_gateway_lifecycle() is True
+
+    # Only the kanban lookalike left → classifier rejects → does not own
+    serve_like.kill()
+    serve_like.wait()
+    monkeypatch.setattr(
+        "hermes_cli.main._detect_venv_python_processes",
+        lambda: [fake_holders()[1]],
+    )
+    assert update_cmd._desktop_owns_gateway_lifecycle() is False

tests/hermes_cli/test_fleet_config_migration_windows_live.py

@@ -0,0 +1,51 @@
+"""LIVE Windows E2E for the fleet-wide config migration (wine2e lane).
+
+Real profile homes on the Windows filesystem, real migration pipeline,
+fresh-process semantics via HERMES_HOME env — mirrors the Linux live E2E.
+"""
+import sys
+from pathlib import Path
+
+import pytest
+import yaml
+
+WORKTREE = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(WORKTREE))
+
+pytestmark = pytest.mark.windows_only
+
+
+def test_fleet_config_migration_live_windows(tmp_path, monkeypatch):
+    active = tmp_path / "hermes-home"
+    profiles = active / "profiles"
+    for name, ver in [("research", 12), ("work", 25)]:
+        home = profiles / name
+        home.mkdir(parents=True)
+        (home / "config.yaml").write_text(
+            yaml.safe_dump({"_config_version": ver, "model": {"provider": "nous"}}),
+            encoding="utf-8",
+        )
+    active.mkdir(exist_ok=True)
+    (active / "config.yaml").write_text(
+        yaml.safe_dump({"_config_version": 12}), encoding="utf-8"
+    )
+    monkeypatch.setenv("HERMES_HOME", str(active))
+
+    import hermes_cli.update_cmd as update_cmd
+    from hermes_cli.config import DEFAULT_CONFIG
+
+    latest = int(DEFAULT_CONFIG["_config_version"])
+    migrated = update_cmd._migrate_sibling_profile_configs()
+
+    by_name = {m[0]: m for m in migrated}
+    assert set(by_name) == {"research", "work"}, migrated
+    assert by_name["research"][1] == 12 and by_name["research"][2] == latest
+
+    for name in ("research", "work"):
+        on_disk = yaml.safe_load((profiles / name / "config.yaml").read_text())
+        assert on_disk["_config_version"] == latest
+        assert on_disk["model"]["provider"] == "nous"
+
+    # active untouched; idempotent second run
+    assert yaml.safe_load((active / "config.yaml").read_text())["_config_version"] == 12
+    assert update_cmd._migrate_sibling_profile_configs() == []

tests/hermes_cli/test_plan_reconciliation_windows_live.py

@@ -0,0 +1,95 @@
+"""LIVE Windows E2E for plan-reconciliation (#92902) on windows-latest.
+
+Real processes with real Hermes-shaped argv, real inventory collection
+(PID-file discovery + supervisor detection on REAL Windows), real
+reconciliation. No mocks on the components under test.
+
+ 1. Spawn a real process registered as a profile gateway (PID file + state
+    file in a temp HERMES_HOME) — real collect_runtime_inventory() must find
+    it, classify supervisor=manual, mechanism id 'manual'.
+ 2. Reconcile with bookkeeping that MISSES it -> unaccounted + escalation.
+ 3. Reconcile with it in killed_pids -> 'stopped', no escalation.
+ 4. Machine-id contract holds on Windows (no display strings leak in).
+"""
+import io, contextlib, json, os, subprocess, sys, tempfile, time
+from pathlib import Path
+
+WORKTREE = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(WORKTREE))
+
+import pytest
+
+pytestmark = pytest.mark.windows_only
+
+
+def test_plan_reconciliation_live_windows(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+
+    # Real live process standing in for a manual gateway
+    child = subprocess.Popen(
+        [sys.executable, "-c", "import time; time.sleep(120)"],
+        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+    )
+    try:
+        time.sleep(0.5)
+        assert child.poll() is None
+
+        import psutil
+
+        create_time = psutil.Process(child.pid).create_time()
+        (home / "gateway_state.json").write_text(json.dumps({
+            "pid": child.pid,
+            "create_time": create_time,
+            "gateway_state": "running",
+            "kind": "hermes-gateway",
+            "code_sha": "f" * 40,
+            "code_version": "0.20.5",
+        }), encoding="utf-8")
+
+        import hermes_cli.profiles as profiles_mod
+        monkeypatch.setattr(profiles_mod, "_get_default_hermes_home", lambda: home)
+        monkeypatch.setattr(profiles_mod, "_get_profiles_root", lambda: tmp_path / "none")
+
+        from hermes_cli.update_inventory import (
+            collect_runtime_inventory,
+            match_runtime_outcomes,
+            report_unaccounted_runtimes,
+        )
+
+        plan = collect_runtime_inventory()
+        rows = [r for r in plan.runtimes if r.pid == child.pid]
+        assert rows, f"real inventory missed the live process: {plan.runtimes}"
+        rt = rows[0]
+        assert rt.supervisor == "manual"
+        assert rt.restart_via == "manual", "machine id, not display string"
+        assert rt.code_sha == "f" * 40
+
+        # 2. bookkeeping misses it -> unaccounted + escalation
+        buf = io.StringIO()
+        with contextlib.redirect_stdout(buf):
+            outcomes = match_runtime_outcomes(
+                plan, restarted_services=[], relaunched_profiles=[],
+                externally_supervised_profiles=[], killed_pids=set(),
+                failed_units=[],
+            )
+            escalated = report_unaccounted_runtimes(outcomes)
+        mine = [o for o in outcomes if o["pid"] == child.pid]
+        assert mine and mine[0]["outcome"] == "unaccounted"
+        assert escalated is True
+        assert "never touched" in buf.getvalue()
+
+        # 3. accounted as stopped -> clean
+        outcomes2 = match_runtime_outcomes(
+            plan, restarted_services=[], relaunched_profiles=[],
+            externally_supervised_profiles=[], killed_pids={child.pid},
+            failed_units=[],
+        )
+        mine2 = [o for o in outcomes2 if o["pid"] == child.pid]
+        assert mine2 and mine2[0]["outcome"] == "stopped"
+        assert report_unaccounted_runtimes(outcomes2) is False
+    finally:
+        if child.poll() is None:
+            child.kill()
+        child.wait()

tests/hermes_cli/test_shim_fail_closed_windows_live.py

@@ -0,0 +1,151 @@
+"""LIVE Windows E2E for the fail-closed shim quarantine (#87331).
+
+Runs ONLY on a real Windows host (the on-demand ``windows-venv-e2e.yml``
+lane). Reproduces the REAL lock shape from the field report: a process
+holding ``hermes.exe`` open WITHOUT FILE_SHARE_DELETE, exactly like a
+running launcher — then proves the strict quarantine refuses before any
+installer runs, and that the non-contended path still installs.
+
+No mocks: real files, a real child process holding a real Windows handle,
+the real rename attempt hitting the real sharing violation.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import pytest
+from hermes_cli import main_install_repair
+
+pytestmark = pytest.mark.windows_only
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+
+# Child that opens a file with GENERIC_READ and NO FILE_SHARE_DELETE —
+# the exact sharing mode a running .exe image / desktop backend exhibits.
+_HOLDER_CODE = r"""
+import ctypes, sys, time
+GENERIC_READ = 0x80000000
+FILE_SHARE_READ = 0x1  # note: NO FILE_SHARE_DELETE
+OPEN_EXISTING = 3
+h = ctypes.windll.kernel32.CreateFileW(
+    sys.argv[1], GENERIC_READ, FILE_SHARE_READ, None, OPEN_EXISTING, 0, None
+)
+if h == -1 or h == 0xFFFFFFFF:
+    print("OPEN_FAILED", flush=True)
+    sys.exit(1)
+print("HOLDING", flush=True)
+time.sleep(120)
+"""
+
+
+@pytest.fixture()
+def held_shim(tmp_path: Path):
+    scripts = tmp_path / "venv" / "Scripts"
+    scripts.mkdir(parents=True)
+    shim = scripts / "hermes.exe"
+    shim.write_bytes(b"MZ fake shim")
+    (scripts / "hermes-gateway.exe").write_bytes(b"MZ fake shim")
+    holder = subprocess.Popen(
+        [sys.executable, "-c", _HOLDER_CODE, str(shim)],
+        stdout=subprocess.PIPE,
+        text=True,
+    )
+    line = holder.stdout.readline().strip()
+    if line != "HOLDING":
+        holder.kill()
+        pytest.fail(f"lock-holder child failed: {line!r}")
+    yield scripts, shim
+    holder.kill()
+    holder.wait()
+
+
+def test_locked_shim_really_cannot_be_renamed(held_shim):
+    """Premise check: the no-FILE_SHARE_DELETE handle blocks rename."""
+    _scripts, shim = held_shim
+    with pytest.raises(OSError):
+        os.rename(shim, shim.with_name("hermes.exe.old.premise"))
+
+
+def test_strict_quarantine_refuses_against_real_lock(held_shim, monkeypatch):
+    import hermes_cli.main as cli_main
+    import hermes_cli.main_install_repair as hermes_cli_main_install_repair
+
+    scripts, _shim = held_shim
+    install_ran: list = []
+    monkeypatch.setattr(
+        cli_main,
+        "_run_install_with_heartbeat",
+        lambda cmd, env=None: install_ran.append(cmd),
+    )
+    monkeypatch.setattr(
+        hermes_cli_main_install_repair,
+        "_run_install_with_heartbeat",
+        lambda cmd, env=None: install_ran.append(cmd),
+    )
+
+    with pytest.raises(main_install_repair.ShimQuarantineError) as exc_info:
+        main_install_repair._run_quarantined_install(
+            ["would-be", "uv", "pip", "install"],
+            scripts_dir=scripts,
+            strict_quarantine=True,
+        )
+
+    assert install_ran == [], "installer ran against a contended venv"
+    assert "hermes.exe" in exc_info.value.failed_shims
+    # The unlocked sibling's rename was rolled back — venv untouched.
+    assert (scripts / "hermes-gateway.exe").exists()
+    assert not list(scripts.glob("*.old.*"))
+
+
+def test_recovery_installer_refuses_against_real_lock(held_shim, monkeypatch):
+    import hermes_cli._install_repair as ir
+
+    scripts, _shim = held_shim
+    monkeypatch.setattr(ir, "_venv_scripts_dir", lambda root: scripts)
+    run_calls: list = []
+    monkeypatch.setattr(ir.subprocess, "run", lambda *a, **k: run_calls.append(a))
+
+    with pytest.raises(ir.ShimQuarantineError):
+        ir._run_install_cmd(["fake"], env=None, root=scripts.parent.parent)
+    assert run_calls == []
+
+
+def test_release_then_strict_quarantine_succeeds(tmp_path, monkeypatch):
+    """After the holder exits, the same strict path proceeds normally."""
+    import hermes_cli.main as cli_main
+    import hermes_cli.main_install_repair as hermes_cli_main_install_repair
+
+    scripts = tmp_path / "venv" / "Scripts"
+    scripts.mkdir(parents=True)
+    (scripts / "hermes.exe").write_bytes(b"MZ fake shim")
+
+    holder = subprocess.Popen(
+        [sys.executable, "-c", _HOLDER_CODE, str(scripts / "hermes.exe")],
+        stdout=subprocess.PIPE,
+        text=True,
+    )
+    assert holder.stdout.readline().strip() == "HOLDING"
+    holder.kill()
+    holder.wait()
+    time.sleep(0.3)  # handle teardown
+
+    install_ran: list = []
+    monkeypatch.setattr(
+        cli_main,
+        "_run_install_with_heartbeat",
+        lambda cmd, env=None: install_ran.append(cmd),
+    )
+    monkeypatch.setattr(
+        hermes_cli_main_install_repair,
+        "_run_install_with_heartbeat",
+        lambda cmd, env=None: install_ran.append(cmd),
+    )
+    main_install_repair._run_quarantined_install(
+        ["fake"], scripts_dir=scripts, strict_quarantine=True
+    )
+    assert install_ran == [["fake"]]

tests/plugins/platforms/test_discord_voice_receive.py

@@ -0,0 +1,761 @@
+"""Integration tests for Discord voice channel audio flow.
+
+Uses real NaCl encryption and Opus codec (no mocks for crypto/codec).
+Does NOT require a Discord connection — tests the VoiceReceiver
+packet processing pipeline end-to-end.
+
+Requires: PyNaCl>=1.5.0, discord.py[voice] (opus codec)
+"""
+
+import struct
+import time
+import pytest
+
+
+
+# Skip entire module if voice deps are missing
+pytest.importorskip("nacl.secret", reason="PyNaCl required for voice integration tests")
+discord = pytest.importorskip("discord", reason="discord.py required for voice integration tests")
+
+import nacl.secret
+
+try:
+    if not discord.opus.is_loaded():
+        import ctypes.util
+        opus_path = ctypes.util.find_library("opus")
+        if not opus_path:
+            for p in ("/opt/homebrew/lib/libopus.dylib", "/usr/local/lib/libopus.dylib"):
+                import os
+                if os.path.isfile(p):
+                    opus_path = p
+                    break
+        if opus_path:
+            discord.opus.load_opus(opus_path)
+    OPUS_AVAILABLE = discord.opus.is_loaded()
+except Exception:
+    OPUS_AVAILABLE = False
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+from plugins.platforms.discord.adapter import VoiceReceiver
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _make_secret_key():
+    """Generate a random 32-byte key."""
+    import os
+    return os.urandom(32)
+
+
+def _build_encrypted_rtp_packet(secret_key, opus_payload, ssrc=100, seq=1, timestamp=960):
+    """Build a real NaCl-encrypted RTP packet matching Discord's format.
+
+    Format: RTP header (12 bytes) + encrypted(opus) + 4-byte nonce
+    Encryption: aead_xchacha20_poly1305 with RTP header as AAD.
+    """
+    # RTP header: version=2, payload_type=0x78, no extension, no CSRC
+    header = struct.pack(">BBHII", 0x80, 0x78, seq, timestamp, ssrc)
+
+    # Encrypt with NaCl AEAD
+    box = nacl.secret.Aead(secret_key)
+    nonce_counter = struct.pack(">I", seq)  # 4-byte counter as nonce seed
+    # Full 24-byte nonce: counter in first 4 bytes, rest zeros
+    full_nonce = nonce_counter + b'\x00' * 20
+
+    enc_msg = box.encrypt(opus_payload, header, full_nonce)
+    ciphertext = enc_msg.ciphertext  # without nonce prefix
+
+    # Discord format: header + ciphertext + 4-byte nonce
+    return header + ciphertext + nonce_counter
+
+
+def _build_padded_rtp_packet(
+    secret_key, opus_payload, pad_len, ssrc=100, seq=1, timestamp=960,
+    declared_pad_len=None, ext_words=0,
+):
+    """Build a NaCl-encrypted RTP packet with the P bit set and padding appended.
+
+    Per RFC 3550 §5.1, the last padding byte declares how many trailing bytes
+    (including itself) to discard. ``pad_len`` is the actual padding appended;
+    ``declared_pad_len`` lets a test forge a mismatched declared length to
+    exercise the validation path. ``ext_words`` > 0 also sets the X bit and
+    prepends a synthetic extension block (4-byte preamble in cleartext header,
+    ext_words*4 bytes of encrypted extension data prepended to the payload).
+    """
+    if pad_len < 1:
+        raise ValueError("pad_len must be >= 1 (last byte includes itself)")
+    declared = pad_len if declared_pad_len is None else declared_pad_len
+    if declared < 0 or declared > 255:
+        raise ValueError("declared_pad_len must fit in one byte")
+
+    has_extension = ext_words > 0
+    first_byte = 0xA0 | (0x10 if has_extension else 0)  # V=2, P=1, [X=?], CC=0
+    fixed_header = struct.pack(">BBHII", first_byte, 0x78, seq, timestamp, ssrc)
+    if has_extension:
+        # 4-byte extension preamble: 2 bytes "defined by profile" + 2 bytes length-in-words
+        ext_preamble = struct.pack(">HH", 0xBEDE, ext_words)
+        header = fixed_header + ext_preamble
+        ext_data = b"\xab" * (ext_words * 4)
+    else:
+        header = fixed_header
+        ext_data = b""
+
+    padding = b"\x00" * (pad_len - 1) + bytes([declared])
+    plaintext = ext_data + opus_payload + padding
+
+    box = nacl.secret.Aead(secret_key)
+    nonce_counter = struct.pack(">I", seq)
+    full_nonce = nonce_counter + b"\x00" * 20
+
+    enc_msg = box.encrypt(plaintext, header, full_nonce)
+    ciphertext = enc_msg.ciphertext
+
+    return header + ciphertext + nonce_counter
+
+
+def _make_voice_receiver(secret_key, dave_session=None, bot_ssrc=9999,
+                         allowed_user_ids=None, members=None):
+    """Create a VoiceReceiver with real secret key."""
+    vc = MagicMock()
+    vc._connection.secret_key = list(secret_key)
+    vc._connection.dave_session = dave_session
+    vc._connection.ssrc = bot_ssrc
+    vc._connection.add_socket_listener = MagicMock()
+    vc._connection.remove_socket_listener = MagicMock()
+    vc._connection.hook = None
+    vc.user = SimpleNamespace(id=bot_ssrc)
+    vc.channel = MagicMock()
+    vc.channel.members = members or []
+    receiver = VoiceReceiver(vc, allowed_user_ids=allowed_user_ids)
+    receiver.start()
+    return receiver
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+class TestRealNaClDecrypt:
+    """End-to-end: real NaCl encrypt → _on_packet decrypt → buffer."""
+
+    def test_valid_encrypted_packet_buffered(self):
+        """Real NaCl encrypted packet → decrypted → buffered."""
+        key = _make_secret_key()
+        opus_silence = b'\xf8\xff\xfe'
+        receiver = _make_voice_receiver(key)
+
+        packet = _build_encrypted_rtp_packet(key, opus_silence, ssrc=100)
+        receiver._on_packet(packet)
+
+        assert 100 in receiver._buffers
+        assert len(receiver._buffers[100]) > 0
+
+    def test_wrong_key_packet_dropped(self):
+        """Packet encrypted with wrong key → NaCl fails → not buffered."""
+        real_key = _make_secret_key()
+        wrong_key = _make_secret_key()
+        opus_silence = b'\xf8\xff\xfe'
+        receiver = _make_voice_receiver(real_key)
+
+        packet = _build_encrypted_rtp_packet(wrong_key, opus_silence, ssrc=100)
+        receiver._on_packet(packet)
+
+        assert len(receiver._buffers.get(100, b"")) == 0
+
+    def test_bot_ssrc_ignored(self):
+        """Packet from bot's own SSRC → ignored."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key, bot_ssrc=9999)
+
+        packet = _build_encrypted_rtp_packet(key, b'\xf8\xff\xfe', ssrc=9999)
+        receiver._on_packet(packet)
+
+        assert len(receiver._buffers) == 0
+
+    def test_multiple_packets_accumulate(self):
+        """Multiple valid packets → buffer grows."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+
+        for seq in range(1, 6):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver._on_packet(packet)
+
+        assert 100 in receiver._buffers
+        buf_size = len(receiver._buffers[100])
+        assert buf_size > 0, "Multiple packets should accumulate in buffer"
+
+    def test_different_ssrcs_separate_buffers(self):
+        """Packets from different SSRCs → separate buffers."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+
+        for ssrc in [100, 200, 300]:
+            packet = _build_encrypted_rtp_packet(key, b'\xf8\xff\xfe', ssrc=ssrc)
+            receiver._on_packet(packet)
+
+        assert len(receiver._buffers) == 3
+        for ssrc in [100, 200, 300]:
+            assert ssrc in receiver._buffers
+
+
+class TestRealNaClWithDAVE:
+    """NaCl decrypt + DAVE passthrough scenarios with real crypto."""
+
+    def test_dave_unknown_ssrc_passthrough(self):
+        """DAVE enabled but SSRC unknown → skip DAVE, buffer audio."""
+        key = _make_secret_key()
+        dave = MagicMock()  # DAVE session present but SSRC not mapped
+        receiver = _make_voice_receiver(key, dave_session=dave)
+
+        packet = _build_encrypted_rtp_packet(key, b'\xf8\xff\xfe', ssrc=100)
+        receiver._on_packet(packet)
+
+        # DAVE decrypt not called (SSRC unknown)
+        dave.decrypt.assert_not_called()
+        # Audio still buffered via passthrough
+        assert 100 in receiver._buffers
+        assert len(receiver._buffers[100]) > 0
+
+    def test_dave_unencrypted_error_passthrough(self):
+        """DAVE raises 'Unencrypted' → use NaCl-decrypted data as-is."""
+        key = _make_secret_key()
+        dave = MagicMock()
+        dave.decrypt.side_effect = Exception(
+            "DecryptionFailed(UnencryptedWhenPassthroughDisabled)"
+        )
+        receiver = _make_voice_receiver(key, dave_session=dave)
+        receiver.map_ssrc(100, 42)
+
+        packet = _build_encrypted_rtp_packet(key, b'\xf8\xff\xfe', ssrc=100)
+        receiver._on_packet(packet)
+
+        # DAVE was called but failed → passthrough
+        dave.decrypt.assert_called_once()
+        assert 100 in receiver._buffers
+        assert len(receiver._buffers[100]) > 0
+
+    def test_dave_real_error_drops(self):
+        """DAVE raises non-Unencrypted error → packet dropped."""
+        key = _make_secret_key()
+        dave = MagicMock()
+        dave.decrypt.side_effect = Exception("KeyRotationFailed")
+        receiver = _make_voice_receiver(key, dave_session=dave)
+        receiver.map_ssrc(100, 42)
+
+        packet = _build_encrypted_rtp_packet(key, b'\xf8\xff\xfe', ssrc=100)
+        receiver._on_packet(packet)
+
+        assert len(receiver._buffers.get(100, b"")) == 0
+
+
+class TestRTPPaddingStrip:
+    """RFC 3550 §5.1 — strip RTP padding before DAVE/Opus decode."""
+
+    def test_padded_packet_stripped_and_buffered(self):
+        """P bit set → trailing padding stripped → opus payload decoded."""
+        key = _make_secret_key()
+        opus_silence = b"\xf8\xff\xfe"
+        receiver = _make_voice_receiver(key)
+
+        # 5 bytes of padding (4 zeros + count byte = 5)
+        packet = _build_padded_rtp_packet(key, opus_silence, pad_len=5, ssrc=100)
+        receiver._on_packet(packet)
+
+        assert 100 in receiver._buffers
+        assert len(receiver._buffers[100]) > 0
+
+    def test_padded_packet_matches_unpadded_output(self):
+        """Same opus payload with/without padding → same decoded PCM."""
+        key = _make_secret_key()
+        opus_silence = b"\xf8\xff\xfe"
+
+        recv_plain = _make_voice_receiver(key)
+        recv_plain._on_packet(
+            _build_encrypted_rtp_packet(key, opus_silence, ssrc=100)
+        )
+
+        recv_padded = _make_voice_receiver(key)
+        recv_padded._on_packet(
+            _build_padded_rtp_packet(key, opus_silence, pad_len=7, ssrc=100)
+        )
+
+        assert bytes(recv_plain._buffers[100]) == bytes(recv_padded._buffers[100])
+
+    def test_padding_with_dave_passthrough(self):
+        """Padding stripped before DAVE → passthrough buffers cleanly."""
+        key = _make_secret_key()
+        opus_silence = b"\xf8\xff\xfe"
+        dave = MagicMock()  # SSRC unmapped → DAVE skipped, passthrough used
+        receiver = _make_voice_receiver(key, dave_session=dave)
+
+        packet = _build_padded_rtp_packet(key, opus_silence, pad_len=4, ssrc=100)
+        receiver._on_packet(packet)
+
+        dave.decrypt.assert_not_called()
+        assert 100 in receiver._buffers
+        assert len(receiver._buffers[100]) > 0
+
+    def test_invalid_padding_length_zero_dropped(self):
+        """Declared pad_len=0 is invalid (RFC requires count includes itself)."""
+        key = _make_secret_key()
+        opus_silence = b"\xf8\xff\xfe"
+        receiver = _make_voice_receiver(key)
+
+        packet = _build_padded_rtp_packet(
+            key, opus_silence, pad_len=4, declared_pad_len=0, ssrc=100
+        )
+        receiver._on_packet(packet)
+
+        assert len(receiver._buffers.get(100, b"")) == 0
+
+    def test_invalid_padding_length_overflow_dropped(self):
+        """Declared pad_len > payload size → packet dropped."""
+        key = _make_secret_key()
+        opus_silence = b"\xf8\xff\xfe"
+        receiver = _make_voice_receiver(key)
+
+        packet = _build_padded_rtp_packet(
+            key, opus_silence, pad_len=4, declared_pad_len=255, ssrc=100
+        )
+        receiver._on_packet(packet)
+
+        assert len(receiver._buffers.get(100, b"")) == 0
+
+    def test_padding_consuming_entire_payload_dropped(self):
+        """Padding consumes entire payload → no opus data → dropped."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+
+        # Empty opus payload, 6 bytes of padding (count byte declares 6)
+        packet = _build_padded_rtp_packet(key, b"", pad_len=6, ssrc=100)
+        receiver._on_packet(packet)
+
+        assert len(receiver._buffers.get(100, b"")) == 0
+
+    def test_padding_with_extension_stripped_correctly(self):
+        """X+P bits both set → strip extension from start, padding from end."""
+        key = _make_secret_key()
+        opus_silence = b"\xf8\xff\xfe"
+
+        # Same opus payload sent two ways: plain, and with both ext+padding
+        recv_plain = _make_voice_receiver(key)
+        recv_plain._on_packet(
+            _build_encrypted_rtp_packet(key, opus_silence, ssrc=100)
+        )
+
+        recv_ext_pad = _make_voice_receiver(key)
+        recv_ext_pad._on_packet(
+            _build_padded_rtp_packet(
+                key, opus_silence, pad_len=5, ext_words=2, ssrc=100
+            )
+        )
+
+        # Both must yield identical decoded PCM — ext data and padding both
+        # stripped before opus decode.
+        assert bytes(recv_plain._buffers[100]) == bytes(recv_ext_pad._buffers[100])
+
+
+class TestFullVoiceFlow:
+    """End-to-end: encrypt → receive → buffer → silence detect → complete."""
+
+    def test_single_utterance_flow(self):
+        """Encrypt packets → buffer → silence → check_silence returns utterance."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+        receiver.map_ssrc(100, 42)
+
+        # Send enough packets to exceed MIN_SPEECH_DURATION (0.5s)
+        # At 48kHz stereo 16-bit, each Opus silence frame decodes to ~3840 bytes
+        # Need 96000 bytes = ~25 frames
+        for seq in range(1, 30):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver._on_packet(packet)
+
+        # Simulate silence by setting last_packet_time in the past
+        receiver._last_packet_time[100] = time.monotonic() - 3.0
+
+        completed = receiver.check_silence()
+        assert len(completed) == 1
+        user_id, pcm_data = completed[0]
+        assert user_id == 42
+        assert len(pcm_data) > 0
+
+    def test_utterance_with_ssrc_automap(self):
+        """No SPEAKING event → auto-map sole allowed user → utterance processed."""
+        key = _make_secret_key()
+        members = [
+            SimpleNamespace(id=9999, name="Bot"),
+            SimpleNamespace(id=42, name="Alice"),
+        ]
+        receiver = _make_voice_receiver(
+            key, allowed_user_ids={"42"}, members=members
+        )
+        # No map_ssrc call — simulating missing SPEAKING event
+
+        for seq in range(1, 30):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver._on_packet(packet)
+
+        receiver._last_packet_time[100] = time.monotonic() - 3.0
+
+        completed = receiver.check_silence()
+        assert len(completed) == 1
+        assert completed[0][0] == 42  # auto-mapped to sole allowed user
+
+    def test_pause_blocks_during_playback(self):
+        """Pause receiver → packets ignored → resume → packets accepted."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+
+        # Pause (echo prevention during TTS playback)
+        receiver.pause()
+        packet = _build_encrypted_rtp_packet(key, b'\xf8\xff\xfe', ssrc=100)
+        receiver._on_packet(packet)
+        assert len(receiver._buffers.get(100, b"")) == 0
+
+        # Resume
+        receiver.resume()
+        receiver._on_packet(packet)
+        assert 100 in receiver._buffers
+        assert len(receiver._buffers[100]) > 0
+
+    def test_corrupted_packet_ignored(self):
+        """Corrupted/truncated packet → silently ignored."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+
+        # Too short
+        receiver._on_packet(b"\x00" * 5)
+        assert len(receiver._buffers) == 0
+
+        # Wrong RTP version
+        bad_header = struct.pack(">BBHII", 0x00, 0x78, 1, 960, 100)
+        receiver._on_packet(bad_header + b"\x00" * 20)
+        assert len(receiver._buffers) == 0
+
+        # Wrong payload type
+        bad_pt = struct.pack(">BBHII", 0x80, 0x00, 1, 960, 100)
+        receiver._on_packet(bad_pt + b"\x00" * 20)
+        assert len(receiver._buffers) == 0
+
+    def test_stop_cleans_everything(self):
+        """stop() clears all state cleanly."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+        receiver.map_ssrc(100, 42)
+
+        for seq in range(1, 10):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver._on_packet(packet)
+
+        assert len(receiver._buffers[100]) > 0
+
+        receiver.stop()
+        assert receiver._running is False
+        assert len(receiver._buffers) == 0
+        assert len(receiver._ssrc_to_user) == 0
+        assert len(receiver._decoders) == 0
+
+
+class TestSPEAKINGHook:
+    """SPEAKING event hook correctly maps SSRC to user_id."""
+
+    def test_speaking_hook_installed(self):
+        """start() installs speaking hook on connection."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+        conn = receiver._vc._connection
+        # hook should be set (wrapped)
+        assert conn.hook is not None
+
+    def test_map_ssrc_via_speaking(self):
+        """SPEAKING op 5 event maps SSRC to user_id."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+        receiver.map_ssrc(500, 12345)
+        assert receiver._ssrc_to_user[500] == 12345
+
+    def test_map_ssrc_overwrites(self):
+        """New SPEAKING event for same SSRC overwrites old mapping."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+        receiver.map_ssrc(500, 111)
+        receiver.map_ssrc(500, 222)
+        assert receiver._ssrc_to_user[500] == 222
+
+    def test_speaking_mapped_audio_processed(self):
+        """After SSRC is mapped, audio from that SSRC gets correct user_id."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+        receiver.map_ssrc(100, 42)
+
+        for seq in range(1, 30):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver._on_packet(packet)
+
+        receiver._last_packet_time[100] = time.monotonic() - 3.0
+        completed = receiver.check_silence()
+        assert len(completed) == 1
+        assert completed[0][0] == 42
+
+
+class TestAuthFiltering:
+    """Only allowed users' audio should be processed."""
+
+    def test_allowed_user_audio_processed(self):
+        """Allowed user's utterance is returned by check_silence."""
+        key = _make_secret_key()
+        members = [
+            SimpleNamespace(id=9999, name="Bot"),
+            SimpleNamespace(id=42, name="Alice"),
+        ]
+        receiver = _make_voice_receiver(
+            key, allowed_user_ids={"42"}, members=members,
+        )
+        receiver.map_ssrc(100, 42)
+
+        for seq in range(1, 30):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver._on_packet(packet)
+
+        receiver._last_packet_time[100] = time.monotonic() - 3.0
+        completed = receiver.check_silence()
+        assert len(completed) == 1
+        assert completed[0][0] == 42
+
+    def test_automap_rejects_unallowed_user(self):
+        """Auto-map refuses to map SSRC to user not in allowed list."""
+        key = _make_secret_key()
+        members = [
+            SimpleNamespace(id=9999, name="Bot"),
+            SimpleNamespace(id=42, name="Alice"),
+        ]
+        receiver = _make_voice_receiver(
+            key, allowed_user_ids={"99"},  # Alice not allowed
+            members=members,
+        )
+        # No map_ssrc — SSRC unknown, auto-map should reject
+
+        for seq in range(1, 30):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver._on_packet(packet)
+
+        receiver._last_packet_time[100] = time.monotonic() - 3.0
+        completed = receiver.check_silence()
+        assert len(completed) == 0
+
+    def test_empty_allowlist_allows_all(self):
+        """Empty allowed_user_ids means no restriction."""
+        key = _make_secret_key()
+        members = [
+            SimpleNamespace(id=9999, name="Bot"),
+            SimpleNamespace(id=42, name="Alice"),
+        ]
+        receiver = _make_voice_receiver(
+            key, allowed_user_ids=None, members=members,
+        )
+
+        for seq in range(1, 30):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver._on_packet(packet)
+
+        receiver._last_packet_time[100] = time.monotonic() - 3.0
+        completed = receiver.check_silence()
+        # Auto-mapped to sole non-bot member
+        assert len(completed) == 1
+        assert completed[0][0] == 42
+
+
+class TestRejoinFlow:
+    """Leave and rejoin: state cleanup and fresh receiver."""
+
+    def test_stop_then_new_receiver_clean_state(self):
+        """After stop(), a new receiver starts with empty state."""
+        key = _make_secret_key()
+        receiver1 = _make_voice_receiver(key)
+        receiver1.map_ssrc(100, 42)
+
+        for seq in range(1, 10):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver1._on_packet(packet)
+
+        assert len(receiver1._buffers[100]) > 0
+        receiver1.stop()
+
+        # New receiver (simulates rejoin)
+        receiver2 = _make_voice_receiver(key)
+        assert len(receiver2._buffers) == 0
+        assert len(receiver2._ssrc_to_user) == 0
+        assert len(receiver2._decoders) == 0
+
+    def test_rejoin_new_ssrc_works(self):
+        """After rejoin, user may get new SSRC — still works."""
+        key = _make_secret_key()
+        receiver1 = _make_voice_receiver(key)
+        receiver1.map_ssrc(100, 42)  # old SSRC
+        receiver1.stop()
+
+        receiver2 = _make_voice_receiver(key)
+        receiver2.map_ssrc(200, 42)  # new SSRC after rejoin
+
+        for seq in range(1, 30):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=200, seq=seq, timestamp=960 * seq
+            )
+            receiver2._on_packet(packet)
+
+        receiver2._last_packet_time[200] = time.monotonic() - 3.0
+        completed = receiver2.check_silence()
+        assert len(completed) == 1
+        assert completed[0][0] == 42
+
+    def test_rejoin_without_speaking_event_automap(self):
+        """Rejoin without SPEAKING event — auto-map sole allowed user."""
+        key = _make_secret_key()
+        members = [
+            SimpleNamespace(id=9999, name="Bot"),
+            SimpleNamespace(id=42, name="Alice"),
+        ]
+
+        # First session
+        receiver1 = _make_voice_receiver(
+            key, allowed_user_ids={"42"}, members=members,
+        )
+        receiver1.stop()
+
+        # Rejoin — new key (Discord may assign new secret_key)
+        new_key = _make_secret_key()
+        receiver2 = _make_voice_receiver(
+            new_key, allowed_user_ids={"42"}, members=members,
+        )
+        # No map_ssrc — simulating missing SPEAKING event
+
+        for seq in range(1, 30):
+            packet = _build_encrypted_rtp_packet(
+                new_key, b'\xf8\xff\xfe', ssrc=300, seq=seq, timestamp=960 * seq
+            )
+            receiver2._on_packet(packet)
+
+        receiver2._last_packet_time[300] = time.monotonic() - 3.0
+        completed = receiver2.check_silence()
+        assert len(completed) == 1
+        assert completed[0][0] == 42
+
+
+class TestMultiGuildIsolation:
+    """Each guild has independent voice state."""
+
+    def test_separate_receivers_independent(self):
+        """Two receivers (different guilds) don't interfere."""
+        key1 = _make_secret_key()
+        key2 = _make_secret_key()
+
+        receiver1 = _make_voice_receiver(key1, bot_ssrc=1111)
+        receiver2 = _make_voice_receiver(key2, bot_ssrc=2222)
+
+        receiver1.map_ssrc(100, 42)
+        receiver2.map_ssrc(200, 99)
+
+        # Send to receiver1
+        for seq in range(1, 10):
+            packet = _build_encrypted_rtp_packet(
+                key1, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver1._on_packet(packet)
+
+        # receiver2 should be empty
+        assert len(receiver2._buffers) == 0
+        assert 100 in receiver1._buffers
+
+    def test_stop_one_doesnt_affect_other(self):
+        """Stopping one receiver doesn't affect another."""
+        key1 = _make_secret_key()
+        key2 = _make_secret_key()
+
+        receiver1 = _make_voice_receiver(key1)
+        receiver2 = _make_voice_receiver(key2)
+
+        receiver1.map_ssrc(100, 42)
+        receiver2.map_ssrc(200, 99)
+
+        for seq in range(1, 10):
+            packet = _build_encrypted_rtp_packet(
+                key2, b'\xf8\xff\xfe', ssrc=200, seq=seq, timestamp=960 * seq
+            )
+            receiver2._on_packet(packet)
+
+        receiver1.stop()
+
+        # receiver2 still has data
+        assert receiver2._running is True
+        assert len(receiver2._buffers[200]) > 0
+
+
+class TestEchoPreventionFlow:
+    """Receiver pause/resume during TTS playback prevents echo."""
+
+    def test_audio_during_pause_ignored(self):
+        """Audio arriving while paused is completely ignored."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+        receiver.map_ssrc(100, 42)
+        receiver.pause()
+
+        for seq in range(1, 30):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver._on_packet(packet)
+
+        assert len(receiver._buffers.get(100, b"")) == 0
+
+    def test_audio_after_resume_processed(self):
+        """Audio arriving after resume is processed normally."""
+        key = _make_secret_key()
+        receiver = _make_voice_receiver(key)
+        receiver.map_ssrc(100, 42)
+
+        # Pause → send packets → resume → send more packets
+        receiver.pause()
+        for seq in range(1, 5):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver._on_packet(packet)
+        assert len(receiver._buffers.get(100, b"")) == 0
+
+        receiver.resume()
+        for seq in range(5, 35):
+            packet = _build_encrypted_rtp_packet(
+                key, b'\xf8\xff\xfe', ssrc=100, seq=seq, timestamp=960 * seq
+            )
+            receiver._on_packet(packet)
+
+        assert len(receiver._buffers[100]) > 0
+        receiver._last_packet_time[100] = time.monotonic() - 3.0
+        completed = receiver.check_silence()
+        assert len(completed) == 1
+        assert completed[0][0] == 42