diff --git a/CLAUDE.md b/CLAUDE.md
index fd816f1..7e08c90 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -152,7 +152,8 @@ Backends: `spi_flash.c` (fmc100), `spi_flash_hisfc350.c` (V1-era parts),
   flash path. Keep installer-transient environment here; persistent board policy
   belongs to firmware/device profiles.
 - **Power** (`src/defib/power/`) — `routeros` (MikroTik PoE, default), `vectis`,
-  `rack`, chosen by `DEFIB_POWER_TYPE`.
+  `rack`, `tasmota` (smart plug `/cm?cmnd=` API), `http` (relay driven by two GET
+  URLs), chosen by `DEFIB_POWER_TYPE`.
 - **TUI** (`src/defib/tui/`) — Textual UI, including the Flash Doctor screen.
 - **Web** (`web/`) — WebSerial browser UI: standalone HTML/JS, no build step,
   deployed to GitHub Pages. `src/defib/web/` is an empty placeholder package —
diff --git a/README.md b/README.md
index 82f4ad2..82dcb4d 100644
--- a/README.md
+++ b/README.md
@@ -206,8 +206,8 @@ Handles both `tftpboot` and `tftp` U-Boot commands transparently.
 ## Automated Power Cycling
 
 Defib can automatically power-cycle devices, eliminating manual intervention
-for recovery loops and research workflows.  Two backends are supported,
-selected via `DEFIB_POWER_TYPE`:
+for recovery loops and research workflows.  The backend is selected via
+`DEFIB_POWER_TYPE`:
 
 ### MikroTik RouterOS PoE switch (default)
 
@@ -267,6 +267,47 @@ socat -,raw,echo=0 TCP:172.17.32.17:35240
 > window can close before the round-trip completes; running Vectis
 > on the same host as defib (or close to it on a LAN) is recommended.
 
+### Tasmota smart plug
+
+Any plug running [Tasmota](https://tasmota.github.io/) (or a firmware that
+speaks the same `/cm?cmnd=` HTTP API) in front of the camera's power supply.
+Each switch is confirmed from the plug's reply, so a cycle that did not
+happen is an error rather than a silent no-op:
+
+```bash
+export DEFIB_POWER_TYPE=tasmota
+export DEFIB_TASMOTA_HOST=192.168.1.50
+export DEFIB_TASMOTA_RELAY=1               # optional, multi-relay devices only
+export DEFIB_TASMOTA_USER=admin            # optional, if a web password is set
+export DEFIB_TASMOTA_PASSWORD=secret
+
+defib burn -c gk7205v500 -f u-boot-gk7205v500-nand.bin -p /dev/ttyUSB0 --power-cycle -t
+```
+
+On a mains plug, leave the default 3 s off time alone: a camera brick can
+hold enough charge to ride through a shorter gap.
+
+### Generic HTTP relay
+
+Any relay that turns on and off with a plain GET request — for example a
+small local bridge in front of a cloud-only smart switch:
+
+```bash
+export DEFIB_POWER_TYPE=http
+export DEFIB_HTTP_POWER_ON_URL=http://127.0.0.1:8090/relay/on
+export DEFIB_HTTP_POWER_OFF_URL=http://127.0.0.1:8090/relay/off
+export DEFIB_HTTP_POWER_TIMEOUT=10         # optional, seconds per request
+
+defib agent upload -c hi3516ev300 -p /dev/ttyUSB0 --power-cycle
+```
+
+The cycle is timed on the host, so it inherits the relay's latency.
+`agent upload --power-cycle` copes with that by starting the handshake while
+the camera is still off and re-cycling once if the bootrom does not answer.
+
+Both are single-outlet controllers: `--poe-port` does not apply, and like
+Vectis they do not work with `defib restore`.
+
 The `-t` flag auto-detects the post-boot mode:
 - **Normal U-Boot shell** (e.g. hi3516ev300): a two-way serial terminal — your
   keystrokes go to the board and its output comes back. Ctrl-C exits the
diff --git a/src/defib/cli/app.py b/src/defib/cli/app.py
index 5fb6300..837779d 100644
--- a/src/defib/cli/app.py
+++ b/src/defib/cli/app.py
@@ -2,12 +2,18 @@
 
 from __future__ import annotations
 
-from typing import Any
+from collections.abc import Callable, Coroutine
+from typing import TYPE_CHECKING, Any
 
 import typer
 
 from defib.install import layout as _install_layout
 
+if TYPE_CHECKING:
+    from defib.power.base import PowerController
+    from defib.recovery.events import HandshakeResult
+    from defib.transport.base import Transport
+
 # Compatibility aliases for existing private imports. Install implementation
 # lives in defib.install.layout; CLI code does not own these helpers.
 _NAND_LAYOUT = _install_layout.NAND_LAYOUT
@@ -29,7 +35,7 @@ def burn(
     port: str = typer.Option("/dev/ttyUSB0", "-p", "--port", help="Serial device (/dev/ttyUSB0), tcp://host:port, rfc2217://host:port, or socket:///path"),
     send_break: bool = typer.Option(False, "-b", "--break", help="Send Ctrl-C after upload"),
     terminal: bool = typer.Option(False, "-t", "--terminal", help="Open serial terminal after upload"),
-    power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via PoE (needs DEFIB_POE_* env vars)"),
+    power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via the controller selected by DEFIB_POWER_TYPE (default routeros, needs DEFIB_POE_* env vars)"),
     poe_port_override: str = typer.Option("", "--poe-port", help="Explicit MikroTik ether port (e.g. ether3) — overrides comment-based auto-discovery. Requires --power-cycle."),
     output: str = typer.Option("human", "--output", help="Output mode: human, json, quiet"),
     debug: bool = typer.Option(False, "-d", "--debug", help="Enable debug logging"),
@@ -1076,16 +1082,18 @@ def agent_upload(
     output: str = typer.Option("human", "--output", help="Output mode: human, json"),
     file: str | None = typer.Option(None, "-f", "--file", help="CV6xx composite boot file (GSL+DDR+U-Boot); required for CV6xx, ignored for other protocols"),
     power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via configured controller (DEFIB_POWER_TYPE)"),
+    poe_port_override: str = typer.Option("", "--poe-port", help="Explicit MikroTik ether port (e.g. ether3) — overrides comment-based auto-discovery. Requires --power-cycle."),
 ) -> None:
     """Upload flash agent to device via boot protocol (requires power-cycle)."""
     import asyncio
-    asyncio.run(_agent_upload_async(chip, port, output, file, power_cycle))
+    asyncio.run(_agent_upload_async(chip, port, output, file, power_cycle, poe_port_override))
 
 
 async def _agent_upload_async(
     chip: str, port: str, output: str,
     composite_path: str | None = None,
     power_cycle: bool = False,
+    poe_port_override: str = "",
 ) -> None:
     import json as json_mod
 
@@ -1095,11 +1103,12 @@ async def _agent_upload_async(
         FlashAgentClient, agent_binary_help, get_agent_binary,
     )
     from defib.firmware import get_cached_path
+    from defib.power.base import PowerControllerError
     from defib.profiles.loader import load_profile
     from defib.protocol.hisilicon_cv6xx import HiSiliconCV6xx
     from defib.protocol.hisilicon_standard import HiSiliconStandard
     from defib.protocol.registry import find_protocol
-    from defib.recovery.events import ProgressEvent
+    from defib.recovery.events import ProgressEvent, Stage
     from defib.transport.serial_platform import (
         create_transport, normalize_port_name,
     )
@@ -1164,12 +1173,31 @@ async def _agent_upload_async(
         spl_data = cached_fw.read_bytes()
         spl_source = f"full U-Boot ({len(spl_data)} bytes — boundary auto-detected)"
 
+    power = None
+    power_port = ""
+    if power_cycle:
+        from defib.power.factory import power_controller_from_env
+        try:
+            power = power_controller_from_env()
+            power_port = await _resolve_power_port(power, port, poe_port_override)
+        except Exception as e:
+            if power is not None:
+                await power.close()
+            if output == "json":
+                print(json_mod.dumps({"event": "error", "message": str(e)}))
+            else:
+                console.print(f"[red]Power controller error:[/red] {e}")
+            raise typer.Exit(1)
+
     if output == "human":
         console.print(f"Agent: [cyan]{agent_path.name}[/cyan] ({len(agent_data)} bytes)")
         console.print(f"SPL: {spl_source}")
-        console.print("\n[yellow]Power-cycle the camera now![/yellow]\n")
+        if power is None:
+            console.print("\n[yellow]Power-cycle the camera now![/yellow]\n")
 
     transport = await create_transport(normalize_port_name(port))
+    if power is not None:
+        _attach_power_transport(power, transport)
     protocol = HiSiliconStandard()
     protocol.set_profile(profile)
 
@@ -1180,7 +1208,32 @@ def on_progress(e: ProgressEvent) -> None:
             elif output == "json":
                 print(json_mod.dumps({"event": "progress", "message": e.message}), flush=True)
 
-    hs = await protocol.handshake(transport, on_progress)
+    if power is not None:
+        # The bootrom's catch window is short, so flood 0xAA from the start.
+        protocol.set_continuous_ack(True)
+
+        def on_power_log(message: str) -> None:
+            on_progress(ProgressEvent(
+                stage=Stage.POWER_CYCLE, bytes_sent=0, bytes_total=1, message=message,
+            ))
+
+        try:
+            hs = await _power_cycle_into_handshake(
+                power, power_port, transport,
+                lambda: protocol.handshake(transport, on_progress),
+                on_power_log,
+            )
+        except PowerControllerError as e:
+            if output == "json":
+                print(json_mod.dumps({"event": "error", "message": f"Power cycle failed: {e}"}))
+            else:
+                console.print(f"[red]Power cycle failed:[/red] {e}")
+            await transport.close()
+            raise typer.Exit(1)
+        finally:
+            await power.close()
+    else:
+        hs = await protocol.handshake(transport, on_progress)
     if not hs.success:
         if output == "json":
             print(json_mod.dumps({"event": "error", "message": "Handshake failed"}))
@@ -1235,6 +1288,112 @@ def on_progress(e: ProgressEvent) -> None:
     await transport.close()
 
 
+async def _resolve_power_port(
+    power: PowerController, serial_port: str, poe_port_override: str,
+) -> str:
+    """The controller port that powers the camera on ``serial_port``.
+
+    Only RouterOS addresses ports by name: ``--poe-port`` wins, otherwise
+    the interface whose comment names the device (``/dev/uart-<NAME>`` ->
+    ``<NAME>``).  Single-outlet controllers ignore the port, so ``""``.
+    """
+    from pathlib import Path
+
+    from defib.power.routeros import RouterOSController
+
+    if not isinstance(power, RouterOSController):
+        return ""
+    if poe_port_override:
+        return poe_port_override
+    label = Path(serial_port).name.removeprefix("uart-")
+    return await power.find_port_by_comment(label)
+
+
+def _attach_power_transport(power: PowerController, transport: Transport) -> None:
+    """Let Vectis pulse reset over the UART connection defib already holds."""
+    from defib.power.vectis import VectisController
+    from defib.transport.rfc2217 import Rfc2217Transport
+
+    if isinstance(power, VectisController) and isinstance(transport, Rfc2217Transport):
+        power.attach_transport(transport)
+
+
+async def _power_cycle_into_handshake(
+    power: PowerController,
+    power_port: str,
+    transport: Transport,
+    start_handshake: Callable[[], Coroutine[Any, Any, HandshakeResult]],
+    log: Callable[[str], None],
+    off_duration: float = 3.0,
+    handshake_timeout: float = 15.0,
+    attempts: int = 2,
+    proactive: bool = False,
+) -> HandshakeResult:
+    """Power the device off, start the handshake, then power it back on.
+
+    The handshake starts while the device is still off: the running OS
+    is dead by then, so nothing on the line can be mistaken for bootrom
+    markers, and the blaster is already on the wire when the bootrom's
+    catch window opens.  This holds however long the controller takes
+    to answer ``power_on`` — cloud-backed relays can lag by seconds.
+
+    Pulse-only controllers (``supports_independent_power`` False) cannot
+    hold the device off, so the order follows RecoverySession: a
+    ``proactive`` handshake (V500, CV6xx: the magic must already be on the
+    wire when the bootrom wakes) starts before the pulse; a reactive one
+    (HiSilicon standard, which counts 0x20 markers that a running OS could
+    also print) starts after it.
+
+    A handshake that does not complete within ``handshake_timeout`` is
+    retried with a fresh power cycle, up to ``attempts`` times in all.
+    Power-controller failures propagate as ``PowerControllerError``.
+    """
+    import asyncio
+
+    from defib.recovery.events import HandshakeResult
+
+    hs = HandshakeResult(success=False, message="no attempt made")
+    for attempt in range(1, attempts + 1):
+        suffix = f" (attempt {attempt}/{attempts})" if attempt > 1 else ""
+        if power.supports_independent_power:
+            log(f"Powering off via {power.name()}{suffix}...")
+            await power.power_off(power_port)
+            await asyncio.sleep(off_duration)
+        pulse_first = not power.supports_independent_power and not proactive
+        if pulse_first:
+            log(f"Power-cycling via {power.name()}{suffix}...")
+            await power.power_cycle(power_port)
+        await transport.flush_input()
+        task = asyncio.create_task(start_handshake())
+        # Yield so the first frames reach the wire before power returns.
+        await asyncio.sleep(0.05)
+        try:
+            if power.supports_independent_power:
+                log("Powering on...")
+                await power.power_on(power_port)
+            elif not pulse_first:
+                log(f"Power-cycling via {power.name()}{suffix}...")
+                await power.power_cycle(power_port)
+        except BaseException:
+            task.cancel()
+            try:
+                await task
+            except BaseException:
+                pass
+            raise
+        try:
+            hs = await asyncio.wait_for(task, timeout=handshake_timeout)
+        except asyncio.TimeoutError:
+            hs = HandshakeResult(
+                success=False,
+                message=f"no bootrom response within {handshake_timeout:.0f}s of power-on",
+            )
+        if hs.success:
+            return hs
+        log(f"Handshake failed: {hs.message}")
+    return hs
+
+
 async def _agent_upload_cv6xx(
     *,
     chip: str,
@@ -2159,7 +2318,7 @@ def install(
         help="Explicit U-Boot artifact override",
     ),
     port: str = typer.Option("/dev/ttyUSB0", "-p", "--port", help="Serial device (/dev/ttyUSB0), tcp://host:port, rfc2217://host:port, or socket:///path"),
-    power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via PoE"),
+    power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via the controller selected by DEFIB_POWER_TYPE"),
     poe_port_override: str = typer.Option("", "--poe-port", help="Explicit MikroTik ether port (e.g. ether3) — overrides comment-based auto-discovery. Requires --power-cycle."),
     nic: str = typer.Option("", "--nic", help="Network interface for TFTP (auto-detect if empty)"),
     host_ip: str = typer.Option("192.168.1.10", "--host-ip", help="IP to assign to host NIC for TFTP"),
@@ -2266,7 +2425,7 @@ def restore(
     host_ip: str = typer.Option("", "--host-ip", help="Host IP for TFTP (auto-detect if empty)"),
     device_ip: str = typer.Option("", "--device-ip", help="Device IP in U-Boot"),
     nic: str = typer.Option("", "--nic", help="Network interface for TFTP"),
-    power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via PoE"),
+    power_cycle: bool = typer.Option(False, "--power-cycle", help="Auto power-cycle via the controller selected by DEFIB_POWER_TYPE"),
     poe_port_override: str = typer.Option("", "--poe-port", help="Explicit MikroTik ether port (e.g. ether3) — overrides comment-based auto-discovery. Requires --power-cycle."),
     tftp_via: str = typer.Option(
         "auto", "--tftp-via",
diff --git a/src/defib/power/base.py b/src/defib/power/base.py
index fb1638f..3831b63 100644
--- a/src/defib/power/base.py
+++ b/src/defib/power/base.py
@@ -17,6 +17,11 @@ class PowerController(ABC):
     smart PDU outlet, relay board).
     """
 
+    #: Whether ``power_off`` and ``power_on`` work on their own.  Controllers
+    #: that can only pulse (Vectis) set this False and support only
+    #: ``power_cycle``.
+    supports_independent_power: bool = True
+
     @classmethod
     @abstractmethod
     def name(cls) -> str:
diff --git a/src/defib/power/factory.py b/src/defib/power/factory.py
index 5b1b4b8..12308b1 100644
--- a/src/defib/power/factory.py
+++ b/src/defib/power/factory.py
@@ -16,6 +16,10 @@ def power_controller_from_env() -> PowerController:
       configured via ``DEFIB_VECTIS_*``.
     - ``DEFIB_POWER_TYPE=rack``: rack pod HTTP API
       (``~/git/rack`` ESP32-S3 spinoff), configured via ``DEFIB_RACK_*``.
+    - ``DEFIB_POWER_TYPE=tasmota``: Tasmota smart plug ``/cm?cmnd=`` API,
+      configured via ``DEFIB_TASMOTA_*``.
+    - ``DEFIB_POWER_TYPE=http``: generic relay driven by two GET URLs,
+      configured via ``DEFIB_HTTP_POWER_*``.
 
     Raises:
         PowerControllerError: if the type is unknown or required env
@@ -31,7 +35,13 @@ def power_controller_from_env() -> PowerController:
     if kind == "rack":
         from defib.power.rack import RackController
         return RackController.from_env()
+    if kind == "tasmota":
+        from defib.power.tasmota import TasmotaController
+        return TasmotaController.from_env()
+    if kind == "http":
+        from defib.power.http import HttpRelayController
+        return HttpRelayController.from_env()
     raise PowerControllerError(
         f"Unknown DEFIB_POWER_TYPE: {kind!r} "
-        "(expected 'routeros', 'vectis', or 'rack')"
+        "(expected 'routeros', 'vectis', 'rack', 'tasmota', or 'http')"
     )
diff --git a/src/defib/power/http.py b/src/defib/power/http.py
new file mode 100644
index 0000000..1d6fc6b
--- /dev/null
+++ b/src/defib/power/http.py
@@ -0,0 +1,133 @@
+"""Generic HTTP relay power controller.
+
+For any relay that exposes "turn on" and "turn off" as plain GET URLs —
+small bridge services in front of cloud-only smart switches (e.g. a local
+eWeLink/Sonoff bridge at ``http://127.0.0.1:8090/relay/on``), ESPHome
+web-server buttons, homegrown relay boards.
+
+Single-port, like :class:`~defib.power.rack.RackController`: the ``port``
+argument is ignored.  Power cycling is timed on the host, so it inherits
+whatever latency the relay's backend adds.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import logging
+import os
+import urllib.error
+import urllib.parse
+import urllib.request
+
+from defib.power.base import PowerController, PowerControllerError
+
+logger = logging.getLogger(__name__)
+
+
+class HttpRelayController(PowerController):
+    """Drives power by GETting one URL for on and another for off."""
+
+    def __init__(self, on_url: str, off_url: str, timeout: float = 10.0) -> None:
+        self._on_url = on_url
+        self._off_url = off_url
+        self._timeout = timeout
+
+    @classmethod
+    def name(cls) -> str:
+        return "HTTP relay"
+
+    @classmethod
+    def from_env(cls) -> HttpRelayController:
+        """Create from ``DEFIB_HTTP_POWER_*`` environment variables.
+
+        Required:
+            DEFIB_HTTP_POWER_ON_URL: GET this to switch power on.
+            DEFIB_HTTP_POWER_OFF_URL: GET this to switch power off.
+        Optional:
+            DEFIB_HTTP_POWER_TIMEOUT: per-request timeout in seconds
+                (default 10).
+        """
+        on_url = os.environ.get("DEFIB_HTTP_POWER_ON_URL")
+        off_url = os.environ.get("DEFIB_HTTP_POWER_OFF_URL")
+        if not on_url or not off_url:
+            raise PowerControllerError(
+                "DEFIB_HTTP_POWER_ON_URL and DEFIB_HTTP_POWER_OFF_URL env vars "
+                "required for HTTP relay power control"
+            )
+        timeout_env = os.environ.get("DEFIB_HTTP_POWER_TIMEOUT", "10")
+        try:
+            timeout = float(timeout_env)
+        except ValueError:
+            raise PowerControllerError(
+                f"DEFIB_HTTP_POWER_TIMEOUT must be a number, got {timeout_env!r}"
+            ) from None
+        return cls(on_url=on_url, off_url=off_url, timeout=timeout)
+
+    async def power_off(self, port: str) -> None:
+        await asyncio.to_thread(self._get_sync, self._off_url)
+
+    async def power_on(self, port: str) -> None:
+        await asyncio.to_thread(self._get_sync, self._on_url)
+
+    async def close(self) -> None:
+        # Stateless HTTP — nothing to release.
+        return None
+
+    def _get_sync(self, url: str) -> None:
+        shown = _redact(url)
+        logger.info("http relay GET %s", shown)
+        try:
+            with urllib.request.urlopen(_request(url), timeout=self._timeout) as resp:
+                resp.read()
+        except urllib.error.HTTPError as e:
+            detail = _redact_text(e.read().decode("utf-8", "replace")[:300], url)
+            raise PowerControllerError(
+                f"HTTP relay {e.code} on GET {shown}: {detail}"
+            ) from e
+        except (urllib.error.URLError, TimeoutError, OSError) as e:
+            raise PowerControllerError(
+                f"HTTP relay unreachable at {shown}: {_redact_text(str(e), url)}"
+            ) from e
+
+
+def _request(url: str) -> urllib.request.Request:
+    """A GET for ``url``, sending any ``user:pass@`` as HTTP Basic auth.
+
+    urllib does not do that itself: it would try to connect to a host
+    literally named ``user:pass@host``.
+    """
+    parts = urllib.parse.urlsplit(url)
+    if parts.username is None:
+        return urllib.request.Request(url)
+    netloc = parts.hostname or ""
+    if parts.port is not None:
+        netloc = f"{netloc}:{parts.port}"
+    bare = urllib.parse.urlunsplit(parts._replace(netloc=netloc))
+    creds = f"{urllib.parse.unquote(parts.username)}:{urllib.parse.unquote(parts.password or '')}"
+    token = base64.b64encode(creds.encode()).decode()
+    return urllib.request.Request(bare, headers={"Authorization": f"Basic {token}"})
+
+
+def _redact(url: str) -> str:
+    """The URL without user info or query, which may carry credentials."""
+    parts = urllib.parse.urlsplit(url)
+    host = parts.hostname or ""
+    if parts.port is not None:
+        host = f"{host}:{parts.port}"
+    query = "?..." if parts.query else ""
+    return f"{parts.scheme}://{host}{parts.path}{query}"
+
+
+def _redact_text(text: str, url: str) -> str:
+    """Strip the URL's secrets from a message that may echo them."""
+    parts = urllib.parse.urlsplit(url)
+    secrets = [parts.password, parts.query]
+    secrets += [value for _, value in urllib.parse.parse_qsl(parts.query)]
+    # Longest first, so a query value never leaves part of the full query.
+    for secret in sorted(filter(None, secrets), key=len, reverse=True):
+        # Skip trivially short values ("1", "on"): blanking them would
+        # mangle the message without hiding anything worth hiding.
+        if len(secret) >= 3:
+            text = text.replace(secret, "...")
+    return text
diff --git a/src/defib/power/tasmota.py b/src/defib/power/tasmota.py
new file mode 100644
index 0000000..6a8f41f
--- /dev/null
+++ b/src/defib/power/tasmota.py
@@ -0,0 +1,158 @@
+"""Tasmota smart-plug power controller.
+
+Drives a single relay on a Tasmota (or API-compatible) smart plug via its
+``/cm?cmnd=`` HTTP API.  Typical use is a mains plug feeding a camera's
+power brick, or an inline USB relay.
+
+Like :class:`~defib.power.rack.RackController`, each plug owns exactly
+one camera, so the ``port`` argument is ignored.  Pass ``""`` from the CLI.
+
+Power cycling is the base class's off / wait / on, with each switch
+confirmed by the plug's reply.  A ``Backlog Power OFF; Delay N; Power ON``
+looks tempting (timed on the plug), but its reply is ``{}`` whatever the
+commands do, so it cannot be verified without polling — and on Tasmota
+14.3 any command sent during the ``Delay`` cuts the delay short: the off
+window shrank from ~4.5 s to ~2 s and the camera behind a mains brick
+never booted.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import urllib.error
+import urllib.parse
+import urllib.request
+
+from defib.power.base import PowerController, PowerControllerError
+
+logger = logging.getLogger(__name__)
+
+class TasmotaController(PowerController):
+    """Drives one relay of a Tasmota plug over HTTP."""
+
+    def __init__(
+        self,
+        host: str,
+        relay: int | None = None,
+        user: str | None = None,
+        password: str | None = None,
+        timeout: float = 10.0,
+    ) -> None:
+        self._host = host
+        self._relay = relay
+        self._user = user
+        self._password = password
+        self._timeout = timeout
+
+    @classmethod
+    def name(cls) -> str:
+        return "Tasmota plug"
+
+    @classmethod
+    def from_env(cls) -> TasmotaController:
+        """Create from ``DEFIB_TASMOTA_*`` environment variables.
+
+        Required:
+            DEFIB_TASMOTA_HOST: plug IP or hostname (e.g. ``10.216.128.80``).
+        Optional:
+            DEFIB_TASMOTA_RELAY: relay index on multi-relay devices
+                (``Power<N>``); omit for single-relay plugs.
+            DEFIB_TASMOTA_USER / DEFIB_TASMOTA_PASSWORD: web credentials,
+                when the plug has a web password set.
+        """
+        host = os.environ.get("DEFIB_TASMOTA_HOST")
+        if not host:
+            raise PowerControllerError(
+                "DEFIB_TASMOTA_HOST env var required for Tasmota power control"
+            )
+        relay_env = os.environ.get("DEFIB_TASMOTA_RELAY")
+        try:
+            relay = int(relay_env) if relay_env else None
+        except ValueError:
+            raise PowerControllerError(
+                f"DEFIB_TASMOTA_RELAY must be an integer, got {relay_env!r}"
+            ) from None
+        return cls(
+            host=host,
+            relay=relay,
+            user=os.environ.get("DEFIB_TASMOTA_USER") or None,
+            password=os.environ.get("DEFIB_TASMOTA_PASSWORD") or None,
+        )
+
+    @property
+    def _power_cmd(self) -> str:
+        return f"Power{self._relay}" if self._relay is not None else "Power"
+
+    async def power_off(self, port: str) -> None:
+        await self._set("OFF")
+
+    async def power_on(self, port: str) -> None:
+        await self._set("ON")
+
+    async def close(self) -> None:
+        # Stateless HTTP — nothing to release.
+        return None
+
+    async def _set(self, want: str) -> None:
+        reply = await self._cmnd(f"{self._power_cmd} {want}")
+        got = self._extract_state(reply)
+        if got != want:
+            raise PowerControllerError(
+                f"Tasmota {self._host}: asked for {want}, plug reports {got!r}"
+            )
+
+    async def _state(self) -> str | None:
+        return self._extract_state(await self._cmnd(self._power_cmd))
+
+    def _extract_state(self, reply: dict[str, object]) -> str | None:
+        # Single-relay plugs answer {"POWER": "ON"}; multi-relay ones
+        # answer {"POWER2": "ON"}.  "Power1" may come back as "POWER".
+        keys = [self._power_cmd.upper()]
+        if self._relay in (None, 1):
+            keys += ["POWER", "POWER1"]
+        for key in keys:
+            value = reply.get(key)
+            if isinstance(value, str):
+                return value.upper()
+        return None
+
+    async def _cmnd(self, command: str) -> dict[str, object]:
+        params = {"cmnd": command}
+        if self._user is not None:
+            params["user"] = self._user
+        if self._password is not None:
+            params["password"] = self._password
+        url = f"http://{self._host}/cm?{urllib.parse.urlencode(params)}"
+        logger.info("tasmota GET %s cmnd=%r", self._host, command)
+        return await asyncio.to_thread(self._get_sync, url, self._timeout)
+
+    def _get_sync(self, url: str, timeout: float) -> dict[str, object]:
+        try:
+            with urllib.request.urlopen(url, timeout=timeout) as resp:
+                payload = resp.read()
+        except urllib.error.HTTPError as e:
+            raise PowerControllerError(
+                f"Tasmota {self._host}: HTTP {e.code}"
+            ) from e
+        except (urllib.error.URLError, TimeoutError, OSError) as e:
+            raise PowerControllerError(
+                f"Tasmota unreachable at {self._host}: {e}"
+            ) from e
+        try:
+            result = json.loads(payload)
+        except json.JSONDecodeError as e:
+            raise PowerControllerError(
+                f"Tasmota {self._host}: non-JSON reply {payload[:100]!r}"
+            ) from e
+        if not isinstance(result, dict):
+            raise PowerControllerError(
+                f"Tasmota {self._host}: unexpected reply {result!r}"
+            )
+        if "WARNING" in result:
+            # Tasmota answers {"WARNING": "Need user=<username>&password=..."}
+            # with HTTP 200 when the web password is wrong or missing.
+            raise PowerControllerError(f"Tasmota {self._host}: {result['WARNING']}")
+        return result
diff --git a/src/defib/power/vectis.py b/src/defib/power/vectis.py
index a186ee6..c8b63f9 100644
--- a/src/defib/power/vectis.py
+++ b/src/defib/power/vectis.py
@@ -52,6 +52,8 @@ class VectisController(PowerController):
       suitable for driving a recovery flow.
     """
 
+    supports_independent_power = False
+
     def __init__(
         self,
         host: str,
diff --git a/src/defib/protocol/hisilicon_v500.py b/src/defib/protocol/hisilicon_v500.py
index 48d3944..7e6c1b4 100644
--- a/src/defib/protocol/hisilicon_v500.py
+++ b/src/defib/protocol/hisilicon_v500.py
@@ -25,7 +25,7 @@
     RecoveryResult,
     Stage,
 )
-from defib.transport.base import Transport, TransportTimeout
+from defib.transport.base import Transport, TransportError, TransportTimeout
 
 logger = logging.getLogger(__name__)
 
@@ -35,6 +35,8 @@
 ])
 
 HANDSHAKE_TIMEOUT = 20.0  # seconds
+HANDSHAKE_BURST_FRAMES = 8  # 112 B, ~10 ms at 115200 baud
+HANDSHAKE_REPLY_LEN = 14
 CHUNK_ACK_TIMEOUT = 4.0   # seconds
 MAX_NAK_RETRIES = 10
 BOOT_LOAD_ADDR = 0x41000000
@@ -74,25 +76,57 @@ async def handshake(
         handshake_frame = append_crc(
             V500_HANDSHAKE_MAGIC + b"\x00\x00\x00\x00\x00\x00\x00\x00"
         )
+        # The bootrom only listens for a few tens of ms after reset before
+        # it falls through to flash boot, so the line must never go idle:
+        # one 14-byte frame is ~1.2 ms on the wire, and waiting 100 ms for
+        # a reply after each one left the window uncovered ~99% of the time.
+        burst = handshake_frame * HANDSHAKE_BURST_FRAMES
 
+        buffer = bytearray()
         while True:
-            await transport.write(handshake_frame)
+            await transport.write(burst)
+            # Read only when something has arrived (bytes_waiting() is
+            # non-blocking on every transport).  An unconditional timed read
+            # would reprogram the serial port's timeout — a tcsetattr — on
+            # every pass.  The host TX queue paces the writes; whatever is
+            # still queued when the reply lands is drained below.
             try:
-                response = await transport.read(14, timeout=0.1)
-                if response.startswith(b"\xbd\x00") and len(response) >= 12:
-                    chip_id = struct.unpack(">I", response[8:12])[0]
-                    self._chip_id = chip_id
-                    _emit(on_progress, ProgressEvent(
-                        stage=Stage.HANDSHAKE, bytes_sent=1, bytes_total=1,
-                        message=f"Detected SoC: {hex(chip_id)}",
-                    ))
-                    return HandshakeResult(
-                        success=True,
-                        chip_id=chip_id,
-                        message=f"Detected SoC: {hex(chip_id)}",
-                    )
+                waiting = await transport.bytes_waiting()
+                if waiting > 0:
+                    buffer += await transport.read(waiting, timeout=0.01)
             except TransportTimeout:
-                continue
+                pass
+
+            # The reply can land anywhere in the stream — after boot noise
+            # from a still-running OS, or mid-way through a burst.
+            idx = buffer.find(b"\xbd\x00")
+            if idx != -1 and len(buffer) - idx >= HANDSHAKE_REPLY_LEN:
+                chip_id = struct.unpack(">I", buffer[idx + 8:idx + 12])[0]
+                self._chip_id = chip_id
+                # The rest of the burst is still in flight and the bootrom
+                # answers each frame: drain our TX queue, let the replies
+                # settle and drop them so they are not mistaken for ACKs
+                # during the HEAD stage.
+                try:
+                    await transport.flush_output()
+                except TransportError:
+                    pass  # best effort: the settle delay still covers it
+                await asyncio.sleep(0.1)
+                await transport.flush_input()
+                _emit(on_progress, ProgressEvent(
+                    stage=Stage.HANDSHAKE, bytes_sent=1, bytes_total=1,
+                    message=f"Detected SoC: {hex(chip_id)}",
+                ))
+                return HandshakeResult(
+                    success=True,
+                    chip_id=chip_id,
+                    message=f"Detected SoC: {hex(chip_id)}",
+                )
+            if idx == -1:
+                # Keep a trailing 0xBD: it may be the start of a reply.
+                del buffer[:max(0, len(buffer) - 1)]
+            else:
+                del buffer[:idx]
 
     async def _send_frame_wait_ack(
         self,
diff --git a/src/defib/transport/socket.py b/src/defib/transport/socket.py
index 86bc7b3..a669b1a 100644
--- a/src/defib/transport/socket.py
+++ b/src/defib/transport/socket.py
@@ -116,6 +116,18 @@ async def flush_output(self) -> None:
         pass  # sendall already ensures data is sent
 
     async def bytes_waiting(self) -> int:
+        # read() goes straight to the socket, so _buf alone only ever holds
+        # bytes pulled in here or pushed back by unread().  Pull whatever the
+        # kernel already has (the socket is non-blocking) so callers that
+        # poll before reading see data that has actually arrived.
+        try:
+            data = self._sock.recv(65536)
+        except (BlockingIOError, InterruptedError):
+            data = None
+        except OSError:
+            data = None  # let the next read() report the error
+        if data:
+            self._buf += data
         return len(self._buf)
 
     async def close(self) -> None:
diff --git a/tests/test_agent_upload_power.py b/tests/test_agent_upload_power.py
new file mode 100644
index 0000000..13473f4
--- /dev/null
+++ b/tests/test_agent_upload_power.py
@@ -0,0 +1,161 @@
+"""Tests for agent upload's power-cycle-into-handshake sequencing."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from defib.cli.app import _power_cycle_into_handshake
+from defib.power.base import PowerController, PowerControllerError
+from defib.recovery.events import HandshakeResult
+from defib.transport.mock import MockTransport
+
+
+class RecordingPower(PowerController):
+    def __init__(self, events: list[str], fail_on: str | None = None) -> None:
+        self.events = events
+        self.fail_on = fail_on
+
+    @classmethod
+    def name(cls) -> str:
+        return "recording"
+
+    async def power_off(self, port: str) -> None:
+        self.events.append(f"off{port}")
+        if self.fail_on == "off":
+            raise PowerControllerError("relay offline")
+
+    async def power_on(self, port: str) -> None:
+        self.events.append(f"on{port}")
+        if self.fail_on == "on":
+            raise PowerControllerError("relay offline")
+
+    async def close(self) -> None:
+        return None
+
+
+async def test_handshake_starts_before_power_on() -> None:
+    events: list[str] = []
+
+    async def handshake() -> HandshakeResult:
+        events.append("handshake")
+        await asyncio.sleep(0)
+        return HandshakeResult(success=True)
+
+    hs = await _power_cycle_into_handshake(
+        RecordingPower(events), "ether3", MockTransport(), handshake, lambda _m: None,
+        off_duration=0.0,
+    )
+    assert hs.success
+    assert events == ["offether3", "handshake", "onether3"]
+
+
+async def test_retries_with_fresh_cycle_on_timeout() -> None:
+    events: list[str] = []
+    results = iter([None, HandshakeResult(success=True)])
+
+    async def handshake() -> HandshakeResult:
+        events.append("handshake")
+        result = next(results)
+        if result is None:
+            await asyncio.sleep(10)  # never answers; times out
+        assert result is not None
+        return result
+
+    logs: list[str] = []
+    hs = await _power_cycle_into_handshake(
+        RecordingPower(events), "ether3", MockTransport(), handshake, logs.append,
+        off_duration=0.0, handshake_timeout=0.05,
+    )
+    assert hs.success
+    assert events == ["offether3", "handshake", "onether3"] * 2
+    assert any("attempt 2/2" in m for m in logs)
+
+
+async def test_gives_up_after_attempts() -> None:
+    async def handshake() -> HandshakeResult:
+        return HandshakeResult(success=False, message="bad marker")
+
+    hs = await _power_cycle_into_handshake(
+        RecordingPower([]), "", MockTransport(), handshake, lambda _m: None,
+        off_duration=0.0, attempts=2,
+    )
+    assert not hs.success
+    assert hs.message == "bad marker"
+
+
+async def test_power_on_failure_cancels_handshake() -> None:
+    cancelled = asyncio.Event()
+
+    async def handshake() -> HandshakeResult:
+        try:
+            await asyncio.sleep(10)
+        except asyncio.CancelledError:
+            cancelled.set()
+            raise
+        return HandshakeResult(success=True)
+
+    with pytest.raises(PowerControllerError):
+        await _power_cycle_into_handshake(
+            RecordingPower([], fail_on="on"), "", MockTransport(), handshake,
+            lambda _m: None, off_duration=0.0,
+        )
+    assert cancelled.is_set()
+
+
+class PulseOnlyPower(RecordingPower):
+    supports_independent_power = False
+
+    async def power_off(self, port: str) -> None:
+        raise PowerControllerError("pulse only")
+
+    async def power_on(self, port: str) -> None:
+        raise PowerControllerError("pulse only")
+
+    async def power_cycle(self, port: str, off_duration: float = 3.0) -> None:
+        self.events.append("pulse")
+
+
+@pytest.mark.parametrize(("proactive", "order"), [
+    (False, ["pulse", "handshake"]),   # reactive: let the OS die first
+    (True, ["handshake", "pulse"]),    # proactive: magic on the wire first
+])
+async def test_pulse_only_controller(proactive: bool, order: list[str]) -> None:
+    events: list[str] = []
+
+    async def handshake() -> HandshakeResult:
+        events.append("handshake")
+        await asyncio.sleep(0)
+        return HandshakeResult(success=True)
+
+    hs = await _power_cycle_into_handshake(
+        PulseOnlyPower(events), "", MockTransport(), handshake, lambda _m: None,
+        off_duration=0.0, proactive=proactive,
+    )
+    assert hs.success
+    assert events == order
+
+
+class TestResolvePowerPort:
+    async def test_single_outlet_controllers_take_no_port(self) -> None:
+        from defib.cli.app import _resolve_power_port
+
+        assert await _resolve_power_port(RecordingPower([]), "/dev/uart-CAM", "ether9") == ""
+
+    async def test_routeros_override_and_discovery(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        from defib.cli.app import _resolve_power_port
+        from defib.power.routeros import RouterOSController
+
+        searched: list[str] = []
+
+        async def find(self: RouterOSController, search: str) -> str:
+            searched.append(search)
+            return "ether4"
+
+        monkeypatch.setattr(RouterOSController, "find_port_by_comment", find)
+        ctrl = RouterOSController.__new__(RouterOSController)
+        assert await _resolve_power_port(ctrl, "/dev/uart-IVG85HG50PYA-S", "ether3") == "ether3"
+        assert searched == []
+        assert await _resolve_power_port(ctrl, "/dev/uart-IVG85HG50PYA-S", "") == "ether4"
+        assert searched == ["IVG85HG50PYA-S"]
diff --git a/tests/test_power_http.py b/tests/test_power_http.py
new file mode 100644
index 0000000..604dd42
--- /dev/null
+++ b/tests/test_power_http.py
@@ -0,0 +1,162 @@
+"""Tests for the generic HTTP relay power controller."""
+
+from __future__ import annotations
+
+import io
+import urllib.error
+from typing import Any
+
+import pytest
+
+from defib.power import http as http_mod
+from defib.power.base import PowerControllerError
+from defib.power.factory import power_controller_from_env
+from defib.power.http import HttpRelayController
+
+ON = "http://127.0.0.1:8090/relay/on"
+OFF = "http://127.0.0.1:8090/relay/off"
+
+
+class FakeResponse(io.BytesIO):
+    def __enter__(self) -> "FakeResponse":
+        return self
+
+    def __exit__(self, *exc_info: object) -> None:
+        self.close()
+
+
+@pytest.fixture
+def calls(monkeypatch: pytest.MonkeyPatch) -> list[str]:
+    seen: list[str] = []
+
+    def fake(req: Any, timeout: float | None = None) -> FakeResponse:
+        seen.append(req.full_url)
+        return FakeResponse(b'{"state":"on"}')
+
+    monkeypatch.setattr(http_mod.urllib.request, "urlopen", fake)
+    return seen
+
+
+class TestFromEnv:
+    @pytest.mark.parametrize("missing", ["DEFIB_HTTP_POWER_ON_URL", "DEFIB_HTTP_POWER_OFF_URL"])
+    def test_missing_url(self, monkeypatch: pytest.MonkeyPatch, missing: str) -> None:
+        monkeypatch.setenv("DEFIB_HTTP_POWER_ON_URL", ON)
+        monkeypatch.setenv("DEFIB_HTTP_POWER_OFF_URL", OFF)
+        monkeypatch.delenv(missing)
+        with pytest.raises(PowerControllerError, match="DEFIB_HTTP_POWER_ON_URL"):
+            HttpRelayController.from_env()
+
+    def test_timeout(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("DEFIB_HTTP_POWER_ON_URL", ON)
+        monkeypatch.setenv("DEFIB_HTTP_POWER_OFF_URL", OFF)
+        monkeypatch.setenv("DEFIB_HTTP_POWER_TIMEOUT", "2.5")
+        assert HttpRelayController.from_env()._timeout == 2.5
+        monkeypatch.setenv("DEFIB_HTTP_POWER_TIMEOUT", "soon")
+        with pytest.raises(PowerControllerError, match="TIMEOUT"):
+            HttpRelayController.from_env()
+
+    def test_factory_dispatches_http(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("DEFIB_POWER_TYPE", "http")
+        monkeypatch.setenv("DEFIB_HTTP_POWER_ON_URL", ON)
+        monkeypatch.setenv("DEFIB_HTTP_POWER_OFF_URL", OFF)
+        monkeypatch.delenv("DEFIB_HTTP_POWER_TIMEOUT", raising=False)
+        assert isinstance(power_controller_from_env(), HttpRelayController)
+
+
+class TestPowerOps:
+    async def test_on_off(self, calls: list[str]) -> None:
+        ctrl = HttpRelayController(on_url=ON, off_url=OFF)
+        await ctrl.power_off("ignored")
+        await ctrl.power_on("ignored")
+        assert calls == [OFF, ON]
+
+    async def test_cycle_is_host_timed(
+        self, calls: list[str], monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        sleeps: list[float] = []
+
+        async def fake_sleep(seconds: float) -> None:
+            sleeps.append(seconds)
+
+        monkeypatch.setattr("defib.power.base.asyncio.sleep", fake_sleep)
+        ctrl = HttpRelayController(on_url=ON, off_url=OFF)
+        await ctrl.power_cycle("", off_duration=2.0)
+        assert calls == [OFF, ON]
+        assert sleeps == [2.0]
+
+    async def test_http_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        def fail(url: str, timeout: float | None = None) -> Any:  # noqa: ANN401
+            raise urllib.error.HTTPError(url, 502, "Bad Gateway", {}, io.BytesIO(b"cloud down"))  # type: ignore[arg-type]
+
+        monkeypatch.setattr(http_mod.urllib.request, "urlopen", fail)
+        with pytest.raises(PowerControllerError, match="502.*cloud down"):
+            await HttpRelayController(on_url=ON, off_url=OFF).power_on("")
+
+    async def test_unreachable(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        def fail(url: str, timeout: float | None = None) -> Any:  # noqa: ANN401
+            raise urllib.error.URLError("connection refused")
+
+        monkeypatch.setattr(http_mod.urllib.request, "urlopen", fail)
+        with pytest.raises(PowerControllerError, match="unreachable"):
+            await HttpRelayController(on_url=ON, off_url=OFF).power_off("")
+
+
+class TestRedaction:
+    SECRET_ON = "http://admin:hunter2@relay.local:8080/relay/on?token=s3cret"
+
+    async def test_logs_hide_credentials(
+        self, calls: list[str], caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        import logging
+
+        caplog.set_level(logging.INFO, logger="defib.power.http")
+        await HttpRelayController(on_url=self.SECRET_ON, off_url=OFF).power_on("")
+        # The query still goes out; the user info moves into a header.
+        assert calls == ["http://relay.local:8080/relay/on?token=s3cret"]
+        assert "hunter2" not in caplog.text
+        assert "s3cret" not in caplog.text
+        assert "http://relay.local:8080/relay/on?..." in caplog.text
+
+    async def test_errors_hide_credentials(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        def fail(url: str, timeout: float | None = None) -> Any:  # noqa: ANN401
+            raise urllib.error.HTTPError(
+                url.full_url, 401, "Unauthorized", {}, io.BytesIO(b"bad token s3cret"),  # type: ignore[attr-defined, arg-type]
+            )
+
+        monkeypatch.setattr(http_mod.urllib.request, "urlopen", fail)
+        with pytest.raises(PowerControllerError) as exc:
+            await HttpRelayController(on_url=self.SECRET_ON, off_url=OFF).power_on("")
+        assert "hunter2" not in str(exc.value)
+        assert "s3cret" not in str(exc.value)
+        assert "401" in str(exc.value)
+
+    async def test_unreachable_message_hides_credentials(
+        self, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        def fail(url: str, timeout: float | None = None) -> Any:  # noqa: ANN401
+            raise urllib.error.URLError(f"cannot reach {url}")
+
+        monkeypatch.setattr(http_mod.urllib.request, "urlopen", fail)
+        with pytest.raises(PowerControllerError) as exc:
+            await HttpRelayController(on_url=self.SECRET_ON, off_url=OFF).power_on("")
+        assert "hunter2" not in str(exc.value)
+        assert "s3cret" not in str(exc.value)
+
+
+class TestBasicAuth:
+    async def test_userinfo_becomes_basic_auth(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        import base64
+
+        seen: list[Any] = []
+
+        def fake(req: Any, timeout: float | None = None) -> FakeResponse:
+            seen.append(req)
+            return FakeResponse(b"ok")
+
+        monkeypatch.setattr(http_mod.urllib.request, "urlopen", fake)
+        url = "http://admin:p%40ss@relay.local/relay/off"
+        await HttpRelayController(on_url=ON, off_url=url).power_off("")
+        req = seen[0]
+        assert req.full_url == "http://relay.local/relay/off"
+        expected = base64.b64encode(b"admin:p@ss").decode()
+        assert req.get_header("Authorization") == f"Basic {expected}"
diff --git a/tests/test_power_tasmota.py b/tests/test_power_tasmota.py
new file mode 100644
index 0000000..1a45f66
--- /dev/null
+++ b/tests/test_power_tasmota.py
@@ -0,0 +1,167 @@
+"""Tests for the Tasmota smart-plug power controller."""
+
+from __future__ import annotations
+
+import io
+import json
+import urllib.error
+import urllib.parse
+from typing import Any
+
+import pytest
+
+from defib.power import tasmota as tasmota_mod
+from defib.power.base import PowerControllerError
+from defib.power.factory import power_controller_from_env
+from defib.power.tasmota import TasmotaController
+
+
+class FakeResponse(io.BytesIO):
+    def __enter__(self) -> "FakeResponse":
+        return self
+
+    def __exit__(self, *exc_info: object) -> None:
+        self.close()
+
+
+class FakePlug:
+    """Answers /cm?cmnd= like a single-relay Tasmota plug."""
+
+    def __init__(self, relay_key: str = "POWER") -> None:
+        self.cmnds: list[str] = []
+        self.queries: list[dict[str, list[str]]] = []
+        self.state = "ON"
+        self.relay_key = relay_key
+        self.reply_override: bytes | None = None
+
+    def __call__(self, url: str, timeout: float | None = None) -> FakeResponse:
+        parsed = urllib.parse.urlparse(url)
+        assert parsed.path == "/cm"
+        query = urllib.parse.parse_qs(parsed.query)
+        self.queries.append(query)
+        cmnd = query["cmnd"][0]
+        self.cmnds.append(cmnd)
+        if self.reply_override is not None:
+            return FakeResponse(self.reply_override)
+        parts = cmnd.split()
+        if len(parts) == 2:
+            self.state = parts[1]
+        return FakeResponse(json.dumps({self.relay_key: self.state}).encode())
+
+
+@pytest.fixture
+def plug(monkeypatch: pytest.MonkeyPatch) -> FakePlug:
+    fake = FakePlug()
+    monkeypatch.setattr(tasmota_mod.urllib.request, "urlopen", fake)
+    return fake
+
+
+class TestFromEnv:
+    def test_missing_host(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.delenv("DEFIB_TASMOTA_HOST", raising=False)
+        with pytest.raises(PowerControllerError, match="DEFIB_TASMOTA_HOST"):
+            TasmotaController.from_env()
+
+    def test_defaults(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("DEFIB_TASMOTA_HOST", "10.216.128.80")
+        for var in ("DEFIB_TASMOTA_RELAY", "DEFIB_TASMOTA_USER", "DEFIB_TASMOTA_PASSWORD"):
+            monkeypatch.delenv(var, raising=False)
+        ctrl = TasmotaController.from_env()
+        assert ctrl._host == "10.216.128.80"
+        assert ctrl._relay is None
+        assert ctrl._user is None
+
+    def test_bad_relay(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("DEFIB_TASMOTA_HOST", "x")
+        monkeypatch.setenv("DEFIB_TASMOTA_RELAY", "two")
+        with pytest.raises(PowerControllerError, match="DEFIB_TASMOTA_RELAY"):
+            TasmotaController.from_env()
+
+    def test_factory_dispatches_tasmota(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setenv("DEFIB_POWER_TYPE", "tasmota")
+        monkeypatch.setenv("DEFIB_TASMOTA_HOST", "10.0.0.5")
+        assert isinstance(power_controller_from_env(), TasmotaController)
+
+
+class TestPowerOps:
+    async def test_power_on_off(self, plug: FakePlug) -> None:
+        ctrl = TasmotaController(host="10.0.0.5")
+        await ctrl.power_off("ignored")
+        await ctrl.power_on("ignored")
+        assert plug.cmnds == ["Power OFF", "Power ON"]
+
+    async def test_relay_index(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        fake = FakePlug(relay_key="POWER2")
+        monkeypatch.setattr(tasmota_mod.urllib.request, "urlopen", fake)
+        ctrl = TasmotaController(host="x", relay=2)
+        await ctrl.power_off("")
+        assert fake.cmnds == ["Power2 OFF"]
+
+    async def test_credentials_in_query(self, plug: FakePlug) -> None:
+        ctrl = TasmotaController(host="x", user="admin", password="s3cret")
+        await ctrl.power_on("")
+        assert plug.queries[0]["user"] == ["admin"]
+        assert plug.queries[0]["password"] == ["s3cret"]
+
+    async def test_state_mismatch_raises(self, plug: FakePlug) -> None:
+        plug.reply_override = b'{"POWER":"ON"}'
+        ctrl = TasmotaController(host="x")
+        with pytest.raises(PowerControllerError, match="asked for OFF"):
+            await ctrl.power_off("")
+
+    async def test_auth_warning_raises(self, plug: FakePlug) -> None:
+        plug.reply_override = b'{"WARNING":"Need user=<username>&password=<password>"}'
+        ctrl = TasmotaController(host="x")
+        with pytest.raises(PowerControllerError, match="Need user"):
+            await ctrl.power_on("")
+
+    async def test_non_json_raises(self, plug: FakePlug) -> None:
+        plug.reply_override = b"<html>"
+        ctrl = TasmotaController(host="x")
+        with pytest.raises(PowerControllerError, match="non-JSON"):
+            await ctrl.power_on("")
+
+
+class TestPowerCycle:
+    async def test_cycle_is_verified_off_then_on(
+        self, plug: FakePlug, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        sleeps: list[float] = []
+
+        async def fake_sleep(seconds: float) -> None:
+            sleeps.append(seconds)
+
+        monkeypatch.setattr("defib.power.base.asyncio.sleep", fake_sleep)
+        await TasmotaController(host="x").power_cycle("", off_duration=3.0)
+        # No Backlog: commands sent during its Delay cut the delay short.
+        assert plug.cmnds == ["Power OFF", "Power ON"]
+        assert sleeps == [3.0]
+
+    async def test_cycle_fails_if_plug_does_not_switch_off(
+        self, plug: FakePlug, monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        async def fake_sleep(seconds: float) -> None:
+            return None
+
+        monkeypatch.setattr("defib.power.base.asyncio.sleep", fake_sleep)
+        plug.reply_override = b'{"POWER":"ON"}'
+        with pytest.raises(PowerControllerError, match="asked for OFF"):
+            await TasmotaController(host="x").power_cycle("", off_duration=3.0)
+
+
+class TestErrorMapping:
+    async def test_url_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        def fail(url: str, timeout: float | None = None) -> Any:  # noqa: ANN401
+            raise urllib.error.URLError("unreachable")
+
+        monkeypatch.setattr(tasmota_mod.urllib.request, "urlopen", fail)
+        with pytest.raises(PowerControllerError, match="unreachable"):
+            await TasmotaController(host="x").power_on("")
+
+    async def test_http_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        def fail(url: str, timeout: float | None = None) -> Any:  # noqa: ANN401
+            raise urllib.error.HTTPError(url, 401, "Unauthorized", {}, io.BytesIO())  # type: ignore[arg-type]
+
+        monkeypatch.setattr(tasmota_mod.urllib.request, "urlopen", fail)
+        with pytest.raises(PowerControllerError, match="HTTP 401"):
+            await TasmotaController(host="x").power_on("")
diff --git a/tests/test_power_vectis.py b/tests/test_power_vectis.py
index c86d757..447c5ca 100644
--- a/tests/test_power_vectis.py
+++ b/tests/test_power_vectis.py
@@ -147,3 +147,12 @@ async def test_close_is_idempotent(self) -> None:
         # should be a safe no-op.
         await ctrl.close()
         await ctrl.close()
+
+
+def test_vectis_is_pulse_only() -> None:
+    """Callers must not sequence Vectis with separate off/on calls."""
+    from defib.power.base import PowerController
+    from defib.power.vectis import VectisController
+
+    assert VectisController.supports_independent_power is False
+    assert PowerController.supports_independent_power is True
diff --git a/tests/test_protocol_v500.py b/tests/test_protocol_v500.py
index 35eef4b..6930e43 100644
--- a/tests/test_protocol_v500.py
+++ b/tests/test_protocol_v500.py
@@ -52,6 +52,87 @@ async def test_handshake_reports_chip_id(self):
         assert result.chip_id == 0xAABBCCDD
 
 
+class _ScriptedRx(MockTransport):
+    """Delivers one scripted RX chunk after each write, like a live UART."""
+
+    def __init__(self, *chunks: bytes) -> None:
+        super().__init__(flush_clears_buffer=True)
+        self._script = list(chunks)
+
+    async def write(self, data: bytes) -> None:
+        await super().write(data)
+        if self._script:
+            self.enqueue_rx(self._script.pop(0))
+
+
+class TestV500HandshakeCatch:
+    REPLY = b"\xbd\x00" + b"\x00" * 6 + struct.pack(">I", 0x72050500) + b"\x00\x00"
+
+    @pytest.mark.asyncio
+    async def test_reply_after_boot_noise(self):
+        transport = _ScriptedRx(b"", b"Starting kernel ...\r\n" + self.REPLY)
+        result = await HiSiliconV500().handshake(transport)
+        assert result.success
+        assert result.chip_id == 0x72050500
+
+    @pytest.mark.asyncio
+    async def test_reply_split_across_reads(self):
+        transport = _ScriptedRx(b"\x00\x01", self.REPLY[:1], self.REPLY[1:9], self.REPLY[9:])
+        result = await HiSiliconV500().handshake(transport)
+        assert result.chip_id == 0x72050500
+
+    @pytest.mark.asyncio
+    async def test_line_never_idles_between_polls(self):
+        """Every write is a multi-frame burst, so the bootrom's short listen
+        window cannot fall into a gap between frames."""
+        transport = _ScriptedRx(b"", b"", self.REPLY)
+        await HiSiliconV500().handshake(transport)
+        assert len(transport.tx_log) == 3
+        for burst in transport.tx_log:
+            assert len(burst) >= 8 * 14
+            assert burst[:4] == b"\xbd\x00\xff\x01"
+
+    @pytest.mark.asyncio
+    async def test_stale_replies_flushed(self):
+        """Replies to the rest of the burst must not linger as fake ACKs."""
+        transport = _ScriptedRx(self.REPLY * 3)
+        await HiSiliconV500().handshake(transport)
+        assert await transport.bytes_waiting() == 0
+
+
+class TestV500HandshakeOverSocket:
+    @pytest.mark.asyncio
+    async def test_reply_reaches_handshake_over_socket(self):
+        """tcp:// and socket:// transports only count already-received bytes
+        in bytes_waiting(), so the handshake must read, not poll."""
+        import asyncio
+        import socket
+
+        from defib.transport.socket import SocketTransport
+
+        ours, bootrom = socket.socketpair()
+        bootrom.setblocking(False)
+        reply = b"\xbd\x00" + b"\x00" * 6 + struct.pack(">I", 0x72050510) + b"\x00\x00"
+        loop = asyncio.get_running_loop()
+
+        async def fake_bootrom() -> None:
+            await loop.sock_recv(bootrom, 4096)  # first burst arrives
+            await loop.sock_sendall(bootrom, reply)
+            while True:  # keep swallowing the rest of the flood
+                if not await loop.sock_recv(bootrom, 4096):
+                    return
+
+        peer = asyncio.create_task(fake_bootrom())
+        transport = SocketTransport(ours)
+        try:
+            result = await asyncio.wait_for(HiSiliconV500().handshake(transport), 5)
+        finally:
+            await transport.close()
+            bootrom.close()
+            peer.cancel()
+        assert result.chip_id == 0x72050510
+
+
 class TestV500FirmwareTransfer:
     @pytest.mark.asyncio
     async def test_send_firmware_with_acks(self):
diff --git a/tests/test_transport_socket.py b/tests/test_transport_socket.py
index 51a34b2..cb1709d 100644
--- a/tests/test_transport_socket.py
+++ b/tests/test_transport_socket.py
@@ -95,3 +95,26 @@ async def test_flush_on_empty_buffer_is_noop(self) -> None:
             await transport.close()
             server.close()
             listener.close()
+
+
+async def test_bytes_waiting_sees_data_not_yet_read() -> None:
+    """Callers poll bytes_waiting() before read(); it must count bytes that
+    are sitting in the kernel, not only ones read() already buffered."""
+    import socket
+
+    from defib.transport.socket import SocketTransport
+
+    ours, peer = socket.socketpair()
+    transport = SocketTransport(ours)
+    try:
+        assert await transport.bytes_waiting() == 0
+        peer.sendall(b"\xbd\x00hello")
+        for _ in range(100):
+            if await transport.bytes_waiting():
+                break
+        assert await transport.bytes_waiting() == 7
+        assert await transport.read(7, timeout=1.0) == b"\xbd\x00hello"
+        assert await transport.bytes_waiting() == 0
+    finally:
+        await transport.close()
+        peer.close()