Live view over WebRTC (openpanel L1, server and browser)

A device reporting a `live` port streams its camera to the browser's own
RTCPeerConnection. The flows socket carries the signalling as text, so it
crosses the portal: `live` messages on a person's socket, `signal` on a
device's (start, sdp, candidate, bye, busy). The server routes by session,
holds a device to two viewers, ends a session when either socket closes or
no answer comes within 10 s, and gives each socket an outbox of its own
that ends a session rather than drop a message. TURN credentials are
minted per session in coturn's REST scheme (STUN_URLS, TURN_URLS,
TURN_SECRET), the viewer's with the device's offer; GET /devices/{id}/ice
hands them out too. The tunnel keeps a stream's early frames instead of
dropping them past eight and forgets them when the stream or link goes.

A media tile on a device's camera, and the Devices tab's preview, play
live when tapped, pause frames meanwhile, and fall back to frames with a
short note on busy, failure or no picture within 10 s. coturn is an
optional compose service behind the `turn` profile.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0138N7k5PT4dcR1iEqXaaQ9k

.env.example

@@ -67,3 +67,10 @@ SIGNUP_ENABLED=false
 
 # Local time for schedules; set from the root .env by scripts/setup.sh.
 TZ=UTC
+
+# A camera's live view between networks: STUN and TURN servers, comma
+# separated, and the secret shared with the TURN server (coturn's
+# static-auth-secret). Empty is direct connections only.
+STUN_URLS=
+TURN_URLS=
+TURN_SECRET=

backend/fluksio/api/routes/devices.py

@@ -71,6 +71,7 @@ from fluksio.flow.devices import (
     read_report,
 )
 from fluksio.flow.events import event_bus
+from fluksio.flow.live import ICE_TTL_S, ice_servers
 from fluksio.flow.state import CONNECT_TIMEOUT_S, SOCKET_TIMEOUT_S
 from fluksio.flow.store import StaleVersion
 from fluksio.models import Message, User
@@ -387,6 +388,20 @@ class DevicesPublic(BaseModel):
     frontend_host: str = ""
 
 
+class IceServer(BaseModel):
+    """One STUN or TURN server, as `RTCPeerConnection` takes it."""
+
+    urls: list[str]
+    username: str | None = None
+    credential: str | None = None
+
+
+class IceServers(BaseModel):
+    ice_servers: list[IceServer] = Field(default_factory=list)
+    #: Seconds the TURN credentials stay good.
+    ttl: int = 0
+
+
 def _describe(request: Request) -> str:
     """A line naming the device behind a pairing request.
 
@@ -665,6 +680,23 @@ async def read_device(device_id: str, store: DeviceStoreDep) -> Any:
     return await run_in_threadpool(_document, device, store)
 
 
+@router.get(
+    "/{device_id}/ice",
+    response_model=IceServers,
+    response_model_exclude_none=True,
+)
+async def read_ice(device_id: str, store: DeviceStoreDep, user: CurrentUser) -> Any:
+    """Where a browser looks for a way to this device's live view.
+
+    STUN and TURN servers, the TURN ones with credentials of the caller's own
+    that expire. Empty when none are set, which is enough on one network.
+    """
+    if await run_in_threadpool(store.find, device_id) is None:
+        raise _not_found(device_id)
+    servers = [IceServer(**server) for server in ice_servers(str(user.id))]
+    return IceServers(ice_servers=servers, ttl=ICE_TTL_S)
+
+
 panels_router.add_api_route(
     "/{device_id}",
     read_device,

backend/fluksio/api/routes/flows.py

@@ -37,6 +37,7 @@ from fluksio.api.routes.messages import port_source
 from fluksio.api.socket_auth import TICKET_TTL_S, issue_ticket, socket_token
 from fluksio.core.config import settings
 from fluksio.core.db import engine
+from fluksio.flow import live as live_view
 from fluksio.flow.artifacts import is_reference
 from fluksio.flow.controller import NODE_TYPES, FlowController, follow_header
 from fluksio.flow.dashboards import DashboardStore
@@ -1369,6 +1370,9 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
     [...]}``) and is pushed those frames. A device also sends its camera's
     frames here, as binary frames in the layout it is pushed them in, and is
     told when its on-demand ports have demand.
+
+    A live view's signalling rides here too (``fluksio.flow.live``): ``live``
+    messages on a person's socket, ``signal`` messages on a device's.
     """
     token = socket_token(websocket.headers, ticket, token)
 
@@ -1425,6 +1429,9 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
 
     async with event_bus.subscribe() as queue:
         receiver = asyncio.create_task(_receive(websocket))
+        # Where a live view this socket is one end of reaches it. Its own
+        # outbox rather than the bus's queue, which drops when full.
+        endpoint = live_view.Endpoint()
         if device:
             presence.opened(device)
             # Demand that moved between the snapshot and the subscription
@@ -1432,12 +1439,23 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
             ports = demand.ports(device)
             if ports != told:
                 await websocket.send_text(demand_frame(ports))
+            live_view.sessions.attach(device, endpoint)
+        outbox = asyncio.create_task(endpoint.next())
         try:
             while True:
                 sender = asyncio.create_task(queue.get())
                 done, _ = await asyncio.wait(
-                    {sender, receiver}, return_when=asyncio.FIRST_COMPLETED
+                    {sender, receiver, outbox}, return_when=asyncio.FIRST_COMPLETED
                 )
+                if outbox in done:
+                    # Signalling, for this socket alone.
+                    if sender not in done:
+                        sender.cancel()
+                    text = outbox.result()
+                    outbox = asyncio.create_task(endpoint.next())
+                    await websocket.send_text(text)
+                    if receiver not in done and sender not in done:
+                        continue
                 if receiver in done:
                     if sender not in done:
                         # Nobody will ever read this one: the next turn makes
@@ -1479,11 +1497,21 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
                         if asked is not None:
                             wanted = asked
                             demand.want(viewer, wanted)
+                        elif (signal := live_view.parse(frame)) is not None:
+                            live_view.sessions.heard(
+                                signal,
+                                endpoint,
+                                person=str(user.id) if only is None else "",
+                                device=device,
+                                devices=getattr(
+                                    websocket.app.state, "device_store", None
+                                ),
+                            )
                     receiver = asyncio.create_task(_receive(websocket))
                     if sender not in done:
                         continue
-                # `sender` is done from here: `FIRST_COMPLETED` returns with at
-                # least one of the two, so `receiver` missing means this one is.
+                # `sender` is done from here: the turns where it is not have
+                # gone round again above.
 
                 # Everything the bus has right now, not just the one event
                 # that woke this: a cascade puts a dozen in at once, and one
@@ -1544,7 +1572,9 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
             pass
         finally:
             receiver.cancel()
+            outbox.cancel()
             demand.drop(viewer)
+            live_view.sessions.detach(endpoint)
             if device:
                 presence.closed(device)
 

backend/fluksio/cloud/connector.py

@@ -73,7 +73,9 @@ ALLOWED_PATHS = frozenset(
 #: reference to a task that was still running.
 MAX_IN_FLIGHT = 256
 #: How many of a client's frames are kept for a stream not listening yet.
-MAX_EARLY = 8
+#: One past it closes the stream rather than losing a frame: a live view's
+#: candidates are worth nothing with one missing.
+MAX_EARLY = 64
 #: How often to look for a config that appeared while the engine was up.
 #: Enrolment is something a person does and then waits on, so this is the
 #: delay they sit through — short enough not to be worth a restart.
@@ -299,7 +301,20 @@ class CloudConnector:
                             # A browser speaks as soon as the hub accepts it,
                             # before this end has checked its credential.
                             early = self._early.setdefault(stream, [])
-                            early[:] = [*early[-(MAX_EARLY - 1) :], text]
+                            if len(early) < MAX_EARLY:
+                                early.append(text)
+                            else:
+                                logger.warning(
+                                    "Closing stream %s: %d frames before it listened",
+                                    stream,
+                                    MAX_EARLY,
+                                )
+                                self._cancel_stream(stream)
+                                await socket.send(
+                                    _dump(
+                                        {"op": "ws_close", "id": stream, "code": 1011}
+                                    )
+                                )
                     elif op == "ws_close":
                         self._cancel_stream(str(frame.get("id") or ""))
                     elif op == "member_removed":
@@ -321,6 +336,7 @@ class CloudConnector:
                     task.cancel()
                 self._calls.clear()
                 self._streams.clear()
+                self._early.clear()
         return attached
 
     @staticmethod
@@ -631,12 +647,16 @@ class CloudConnector:
                 "Refusing stream %s: %d already open", stream_id, len(self._streams)
             )
             return
-        existing = self._streams.pop(stream_id, None)
-        if existing is not None:
-            existing.cancel()
+        self._cancel_stream(stream_id)
         task = asyncio.create_task(self._stream(socket, frame))
         self._streams[stream_id] = task
-        task.add_done_callback(lambda _t: self._streams.pop(stream_id, None))
+        task.add_done_callback(lambda done: self._stream_done(stream_id, done))
+
+    def _stream_done(self, stream_id: str, task: asyncio.Task[None]) -> None:
+        # Its own entry only: an id the hub reused belongs to the newer task.
+        if self._streams.get(stream_id) is task:
+            del self._streams[stream_id]
+            self._early.pop(stream_id, None)
 
     async def _stream(self, socket: Any, frame: dict[str, Any]) -> None:
         """Feed the engine's live events down the tunnel.
@@ -661,6 +681,7 @@ class CloudConnector:
         )
         from fluksio.api.socket_auth import redeem_ticket
         from fluksio.core.db import engine
+        from fluksio.flow import live as live_view
         from fluksio.flow.demand import demand
         from fluksio.flow.devices import presence
         from fluksio.flow.events import event_bus
@@ -714,6 +735,11 @@ class CloudConnector:
         # are fetched one by one through the portal.
         viewer = ("tunnel", stream_id)
         wanted: set[str] = set()
+        # Where a live view this stream is one end of reaches it. Its
+        # signalling is text, so it crosses, on an outbox of its own.
+        endpoint = live_view.Endpoint()
+        outbox: asyncio.Task[str] | None = None
+        sender: asyncio.Task[dict[str, Any]] | None = None
 
         def heard(text: str) -> None:
             nonlocal wanted
@@ -721,6 +747,14 @@ class CloudConnector:
             if asked is not None:
                 wanted = asked
                 demand.want(viewer, asked)
+            elif (signal := live_view.parse(text)) is not None:
+                live_view.sessions.heard(
+                    signal,
+                    endpoint,
+                    person=str(user.id) if only is None else "",
+                    device=device,
+                    devices=getattr(self._app.state, "device_store", None),
+                )
 
         self._listeners[stream_id] = heard
         for text in self._early.pop(stream_id, []):
@@ -728,8 +762,10 @@ class CloudConnector:
 
         if device:
             presence.opened(device)
+            live_view.sessions.attach(device, endpoint)
         try:
             async with event_bus.subscribe() as queue:
+                outbox = asyncio.create_task(endpoint.next())
                 if device and (ports := demand.ports(device)) != told:
                     await socket.send(
                         _dump(
@@ -741,7 +777,22 @@ class CloudConnector:
                         )
                     )
                 while True:
-                    event = await queue.get()
+                    sender = asyncio.create_task(queue.get())
+                    done, _ = await asyncio.wait(
+                        {sender, outbox}, return_when=asyncio.FIRST_COMPLETED
+                    )
+                    if outbox in done:
+                        # Signalling, for this stream alone.
+                        if sender not in done:
+                            sender.cancel()
+                        text = outbox.result()
+                        outbox = asyncio.create_task(endpoint.next())
+                        await socket.send(
+                            _dump({"op": "ws_msg", "id": stream_id, "text": text})
+                        )
+                        if sender not in done:
+                            continue
+                    event = sender.result()
                     if event.get("type") == "device_demand":
                         # For the device it names, and for nobody else.
                         if device and event.get("device") == device:
@@ -793,9 +844,13 @@ class CloudConnector:
                     _dump({"op": "ws_close", "id": stream_id, "code": 1011})
                 )
         finally:
+            for task in (outbox, sender):
+                if task is not None:
+                    task.cancel()
             self._listeners.pop(stream_id, None)
             self._early.pop(stream_id, None)
             demand.drop(viewer)
+            live_view.sessions.detach(endpoint)
             if device:
                 presence.closed(device)
 
@@ -867,6 +922,9 @@ class CloudConnector:
                 )
 
     def _cancel_stream(self, stream_id: str) -> None:
+        # The frames it was holding go with it, including those of a stream
+        # cancelled before it ever ran its own cleanup.
+        self._early.pop(stream_id, None)
         task = self._streams.pop(stream_id, None)
         if task is not None:
             task.cancel()

backend/fluksio/core/config.py

@@ -229,6 +229,18 @@ class Settings(BaseSettings):
         list[AnyUrl] | str, BeforeValidator(parse_cors)
     ] = []
 
+    #: Where a live view's two ends look for a way to each other, comma
+    #: separated. None by default: on one network the peers find each other
+    #: without, and a third-party server would learn who watches what. coturn
+    #: serves STUN as well, so its `stun:` address goes here.
+    STUN_URLS: Annotated[list[str] | str, BeforeValidator(parse_cors)] = []
+    #: The relay a live view falls back to when the two ends cannot reach each
+    #: other, as `turn:` and `turns:` URLs. Used only with `TURN_SECRET`, the
+    #: secret the TURN server shares (coturn's `static-auth-secret`), from
+    #: which each viewer and device gets credentials of its own.
+    TURN_URLS: Annotated[list[str] | str, BeforeValidator(parse_cors)] = []
+    TURN_SECRET: str = ""
+
     @model_validator(mode="before")
     @classmethod
     def _derive_data_paths(cls, data: Any) -> Any:

backend/fluksio/flow/live.py

@@ -0,0 +1,365 @@
+"""Live view: WebRTC signalling between a viewer's socket and a device's.
+
+A dashboard tile asks a device's camera for H.264 over WebRTC. The video goes
+from the device to the browser directly, or through a TURN relay, never
+through here: this only carries the few messages that set a connection up.
+Both ends already hold a ``/flows/ws`` socket, which also crosses the portal
+tunnel because it carries text, so signalling rides on it.
+
+A person's socket speaks ``{"type": "live", "device", "session", "action"}``
+and a device's ``{"type": "signal", "session", "action"}``. ``start`` from
+the viewer is handed to the device with ICE servers of its own, the device
+offers (``sdp``), which reaches the viewer with ICE servers of the viewer's
+own, the viewer answers (``sdp``), both trickle ``candidate``, and ``bye``
+from either side ends it. A device past its cap answers ``busy``, and so does
+this end once a device has ``MAX_SESSIONS``.
+
+Signalling never rides the event bus, which drops a slow socket's oldest
+events: one lost candidate is a connection that never comes up. Each socket
+is an :class:`Endpoint` with an outbox of its own that its loop drains, so
+only that socket writes to its connection, and a full outbox ends the session
+instead of losing a message. One process holds every socket, so one registry
+is the whole picture.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import hmac
+import logging
+import re
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+import orjson
+
+from fluksio.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+#: The port a device reports when it can stream live. Signalled, never wired.
+LIVE_PORT = "live"
+#: Live viewers per device. The NANO encodes once for all of them, and S3
+#: measured two at 720p; past that a tile stays on frames.
+MAX_SESSIONS = 2
+#: How long a session has to get from ``start`` to the viewer's answer.
+ANSWER_TIMEOUT_S = 10.0
+#: How long TURN credentials stay good. Minted per session, for whoever is
+#: one end of it. coturn checks them on every refresh, so a view relayed
+#: through it ends this long after it started.
+ICE_TTL_S = 3600
+#: Signalling messages one socket may have waiting. A session trickles a
+#: dozen candidates at most, so a socket this far behind is stuck.
+OUTBOX = 64
+
+_SESSION = re.compile(r"^[A-Za-z0-9_-]{8,64}$")
+#: An offer is one to three kilobytes, a candidate a line.
+MAX_SDP = 16 * 1024
+MAX_CANDIDATE = 1024
+#: Which actions are passed on once a session stands.
+_RELAYED = ("sdp", "candidate")
+
+
+def ice_servers(who: str) -> list[dict[str, Any]]:
+    """The STUN and TURN servers for one end, with TURN credentials of its own.
+
+    coturn's REST scheme (``use-auth-secret``): the username is
+    ``<expiry>:<who>`` and the password the base64 HMAC-SHA1 of it under the
+    shared secret, so coturn checks it without asking anyone.
+    """
+    servers: list[dict[str, Any]] = []
+    if settings.STUN_URLS:
+        servers.append({"urls": list(settings.STUN_URLS)})
+    if settings.TURN_URLS and settings.TURN_SECRET:
+        username = f"{int(time.time()) + ICE_TTL_S}:{who}"
+        digest = hmac.new(
+            settings.TURN_SECRET.encode(), username.encode(), hashlib.sha1
+        ).digest()
+        servers.append(
+            {
+                "urls": list(settings.TURN_URLS),
+                "username": username,
+                "credential": base64.b64encode(digest).decode(),
+            }
+        )
+    return servers
+
+
+def parse(frame: str) -> dict[str, Any] | None:
+    """A signalling message a socket sent, or None for anything else."""
+    try:
+        payload = orjson.loads(frame)
+    except orjson.JSONDecodeError:
+        return None
+    if isinstance(payload, dict) and payload.get("type") in ("live", "signal"):
+        return payload
+    return None
+
+
+class Endpoint:
+    """One socket signalling can reach, and what is waiting to go out on it."""
+
+    def __init__(self) -> None:
+        self._outbox: asyncio.Queue[str] = asyncio.Queue(maxsize=OUTBOX)
+
+    def deliver(self, message: dict[str, Any]) -> bool:
+        """Put a message in the outbox. False when it is full."""
+        try:
+            self._outbox.put_nowait(orjson.dumps(message).decode())
+        except asyncio.QueueFull:
+            return False
+        return True
+
+    async def next(self) -> str:
+        """The next message for this socket to send, as text."""
+        return await self._outbox.get()
+
+
+@dataclass(eq=False)
+class Session:
+    id: str
+    device: str
+    viewer: Endpoint
+    camera: Endpoint
+    #: Who is watching, whose TURN credentials the viewer is handed.
+    who: str
+    offered: bool = False
+    answered: bool = False
+    started: float = field(default_factory=time.monotonic)
+    timer: asyncio.TimerHandle | None = None
+
+
+class LiveSessions:
+    """Every live view being set up or running, and the devices that serve them.
+
+    Called only from the event loop, by the sockets themselves.
+    """
+
+    def __init__(self) -> None:
+        self._sessions: dict[str, Session] = {}
+        #: Each device's open sockets, newest last. A device reconnecting can
+        #: hold two for a moment; a new session goes to the newer one.
+        self._devices: dict[str, list[Endpoint]] = {}
+
+    def attach(self, device_id: str, endpoint: Endpoint) -> None:
+        """A device's socket opened."""
+        self._devices.setdefault(device_id, []).append(endpoint)
+
+    def detach(self, endpoint: Endpoint) -> None:
+        """A socket closed: every session it was one end of is over."""
+        for device_id, endpoints in list(self._devices.items()):
+            if endpoint in endpoints:
+                endpoints.remove(endpoint)
+                if not endpoints:
+                    del self._devices[device_id]
+        for session in list(self._sessions.values()):
+            if session.viewer is endpoint:
+                self._end(session, "closed", tell_viewer=False)
+            elif session.camera is endpoint:
+                self._end(session, "offline", tell_camera=False)
+
+    def count(self, device_id: str) -> int:
+        return sum(1 for s in self._sessions.values() if s.device == device_id)
+
+    def heard(
+        self,
+        message: dict[str, Any],
+        endpoint: Endpoint,
+        *,
+        person: str,
+        device: str,
+        devices: Any,
+    ) -> None:
+        """One signalling message from a socket.
+
+        ``person`` is the id of the account holding the socket and ``device``
+        the device whose credential does, one of them "". Any person may watch
+        any device, camera privacy being deferred. A device speaks only for
+        the sessions it was asked to serve on this socket.
+        """
+        session_id = str(message.get("session") or "")
+        if not _SESSION.match(session_id):
+            return
+        if message.get("type") == "live" and person:
+            self._from_viewer(message, session_id, endpoint, person, devices)
+        elif message.get("type") == "signal" and device:
+            self._from_device(message, session_id, endpoint, device)
+
+    # -- the two directions ---------------------------------------------------
+
+    def _from_viewer(
+        self,
+        message: dict[str, Any],
+        session_id: str,
+        endpoint: Endpoint,
+        who: str,
+        devices: Any,
+    ) -> None:
+        device_id = str(message.get("device") or "")
+        action = message.get("action")
+        session = self._sessions.get(session_id)
+        if action == "start":
+            if session is not None:
+                # Said twice, or somebody else's id: neither gets a second one.
+                return
+            reason = self._refusal(device_id, devices)
+            if reason:
+                endpoint.deliver(
+                    _viewer_message(device_id, session_id, "bye", reason=reason)
+                )
+                return
+            if self.count(device_id) >= MAX_SESSIONS:
+                endpoint.deliver(_viewer_message(device_id, session_id, "busy"))
+                return
+            session = Session(
+                session_id, device_id, endpoint, self._devices[device_id][-1], who
+            )
+            session.timer = asyncio.get_running_loop().call_later(
+                ANSWER_TIMEOUT_S, self._expire, session
+            )
+            self._sessions[session_id] = session
+            logger.info("Live view %s on %s asked for", session_id, device_id)
+            self._to_camera(
+                session,
+                {
+                    "action": "start",
+                    "ice_servers": ice_servers(f"device-{device_id}"),
+                },
+            )
+            return
+        if (
+            session is None
+            or session.viewer is not endpoint
+            or session.device != device_id
+        ):
+            return
+        if action == "bye":
+            self._end(session, "closed", tell_viewer=False)
+            return
+        relayed = _relayed(message, action)
+        if relayed is None:
+            return
+        if action == "sdp" and session.offered and not session.answered:
+            session.answered = True
+            if session.timer is not None:
+                session.timer.cancel()
+        self._to_camera(session, relayed)
+
+    def _from_device(
+        self,
+        message: dict[str, Any],
+        session_id: str,
+        endpoint: Endpoint,
+        device_id: str,
+    ) -> None:
+        session = self._sessions.get(session_id)
+        if (
+            session is None
+            or session.device != device_id
+            or session.camera is not endpoint
+        ):
+            return
+        action = message.get("action")
+        if action in ("bye", "busy"):
+            reason = str(message.get("reason") or "")[:64] or "closed"
+            self._end(session, reason, tell_camera=False, action=str(action))
+            return
+        relayed = _relayed(message, action)
+        if relayed is None:
+            return
+        if action == "sdp" and not session.offered:
+            # The viewer builds its connection on the offer, so its ICE
+            # servers come with it: minted only for a session that got a slot.
+            session.offered = True
+            relayed["ice_servers"] = ice_servers(session.who)
+        self._to_viewer(session, relayed)
+
+    # -- delivering and ending ------------------------------------------------
+
+    def _to_camera(self, session: Session, message: dict[str, Any]) -> None:
+        sent = {"type": "signal", "session": session.id, **message}
+        if not session.camera.deliver(sent):
+            self._end(session, "overflow", tell_camera=False)
+
+    def _to_viewer(self, session: Session, message: dict[str, Any]) -> None:
+        sent = _viewer_message(session.device, session.id, **message)
+        if not session.viewer.deliver(sent):
+            self._end(session, "overflow", tell_viewer=False)
+
+    def _refusal(self, device_id: str, devices: Any) -> str:
+        """Why this device cannot serve a live view now, or ""."""
+        if not self._devices.get(device_id):
+            return "offline"
+        reported = devices.reported_ports(device_id) if devices is not None else None
+        if LIVE_PORT not in (reported or {}):
+            return "unsupported"
+        return ""
+
+    def _expire(self, session: Session) -> None:
+        if self._sessions.get(session.id) is session and not session.answered:
+            self._end(session, "timeout")
+
+    def _end(
+        self,
+        session: Session,
+        reason: str,
+        *,
+        tell_viewer: bool = True,
+        tell_camera: bool = True,
+        action: str = "bye",
+    ) -> None:
+        if self._sessions.pop(session.id, None) is not session:
+            return
+        if session.timer is not None:
+            session.timer.cancel()
+        # Told once, without a retry: a side whose outbox is full is stuck,
+        # and its own timeout or a closing socket ends it there.
+        if tell_viewer:
+            session.viewer.deliver(
+                _viewer_message(session.device, session.id, action, reason=reason)
+            )
+        if tell_camera:
+            session.camera.deliver(
+                {
+                    "type": "signal",
+                    "session": session.id,
+                    "action": "bye",
+                    "reason": reason,
+                }
+            )
+        logger.info(
+            "Live view %s on %s ended after %.1f s: %s",
+            session.id,
+            session.device,
+            time.monotonic() - session.started,
+            reason,
+        )
+
+
+def _relayed(message: dict[str, Any], action: Any) -> dict[str, Any] | None:
+    """What of an ``sdp`` or ``candidate`` message is passed on, or None."""
+    if action not in _RELAYED:
+        return None
+    value = message.get(action)
+    limit = MAX_SDP if action == "sdp" else MAX_CANDIDATE
+    if not isinstance(value, str) or not value or len(value) > limit:
+        return None
+    return {"action": action, action: value}
+
+
+def _viewer_message(
+    device: str, session: str, action: str, **rest: Any
+) -> dict[str, Any]:
+    return {
+        "type": "live",
+        "device": device,
+        "session": session,
+        "action": action,
+        **rest,
+    }
+
+
+sessions = LiveSessions()

backend/tests/api/routes/test_live.py

@@ -0,0 +1,441 @@
+"""Live view: WebRTC signalling relayed between a viewer's socket and a device's.
+
+openpanel ROADMAP L1. The server routes by session, holds a device to two
+viewers, ends a session when either socket goes or no answer comes, and hands
+out TURN credentials in coturn's REST scheme.
+"""
+
+import asyncio
+import base64
+import hashlib
+import hmac
+import json
+import time
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+from fluksio.api.routes import oauth
+from fluksio.cloud.connector import CloudConnector
+from fluksio.core.config import settings
+from fluksio.flow import live as live_view
+from tests.api.routes.test_cameras import REPORT, wait_for
+from tests.api.routes.test_devices import _pair, _panels
+
+PREFIX = f"{settings.API_V1_STR}/devices"
+SOCKET = f"{settings.API_V1_STR}/flows/ws"
+LIVE_REPORT = {
+    **REPORT,
+    "ports": [
+        *REPORT["ports"],
+        {"port": "live", "direction": "in", "dtype": "video", "on_demand": True},
+    ],
+}
+
+
+@pytest.fixture(scope="module")
+def cams(client: TestClient, superuser_token_headers: dict[str, str]):
+    """``cam1`` reports a live port, ``cam2`` does not. Yields their headers."""
+    for bucket in [b for b in oauth._hits if b.startswith("pair:")]:
+        del oauth._hits[bucket]
+    _panels(
+        client, superuser_token_headers, {"panels": [{"id": "cam1"}, {"id": "cam2"}]}
+    )
+    cam1 = _pair(client, superuser_token_headers, "cam1")
+    cam2 = _pair(client, superuser_token_headers, "cam2")
+    for headers, report in ((cam1, LIVE_REPORT), (cam2, REPORT)):
+        device = "cam1" if headers is cam1 else "cam2"
+        put = client.put(f"{PREFIX}/{device}/report", headers=headers, json=report)
+        assert put.status_code == 200, put.text
+    yield cam1, cam2
+    for device in ("cam1", "cam2"):
+        client.delete(f"{PREFIX}/{device}", headers=superuser_token_headers)
+
+
+def _viewer(client: TestClient, headers: dict[str, str]) -> Any:
+    ticket = client.post(f"{settings.API_V1_STR}/flows/ws-ticket", headers=headers)
+    return client.websocket_connect(f"{SOCKET}?ticket={ticket.json()['ticket']}")
+
+
+def _next(ws: Any, kind: str) -> dict[str, Any]:
+    """The next message of this type on a socket, skipping the rest."""
+    while True:
+        message = ws.receive_json()
+        if message.get("type") == kind:
+            return message
+
+
+def _start(ws: Any, session: str, device: str = "cam1") -> None:
+    ws.send_json(
+        {"type": "live", "device": device, "session": session, "action": "start"}
+    )
+
+
+def test_signalling_is_relayed_both_ways(
+    client: TestClient, superuser_token_headers: dict[str, str], cams
+) -> None:
+    """start, the device's offer, the answer, candidates both ways, bye."""
+    cam1, _ = cams
+    with (
+        client.websocket_connect(SOCKET, headers=cam1) as device,
+        _viewer(client, superuser_token_headers) as viewer,
+    ):
+        device.receive_json()
+        _start(viewer, "session-one")
+        start = _next(device, "signal")
+        assert start["action"] == "start"
+        assert start["session"] == "session-one"
+        assert start["ice_servers"] == []
+
+        device.send_json(
+            {
+                "type": "signal",
+                "session": "session-one",
+                "action": "sdp",
+                "sdp": "v=0 offer",
+            }
+        )
+        assert _next(viewer, "live") == {
+            "type": "live",
+            "device": "cam1",
+            "session": "session-one",
+            "action": "sdp",
+            "sdp": "v=0 offer",
+            "ice_servers": [],
+        }
+        viewer.send_json(
+            {
+                "type": "live",
+                "device": "cam1",
+                "session": "session-one",
+                "action": "sdp",
+                "sdp": "v=0 answer",
+            }
+        )
+        assert _next(device, "signal")["sdp"] == "v=0 answer"
+        viewer.send_json(
+            {
+                "type": "live",
+                "device": "cam1",
+                "session": "session-one",
+                "action": "candidate",
+                "candidate": "candidate:1 1 udp 1 10.0.0.2 5000 typ host",
+            }
+        )
+        assert _next(device, "signal")["candidate"].startswith("candidate:1 ")
+        device.send_json(
+            {
+                "type": "signal",
+                "session": "session-one",
+                "action": "candidate",
+                "candidate": "candidate:2 1 udp 1 10.0.0.3 5000 typ host",
+            }
+        )
+        assert _next(viewer, "live")["candidate"].startswith("candidate:2 ")
+
+        viewer.send_json(
+            {
+                "type": "live",
+                "device": "cam1",
+                "session": "session-one",
+                "action": "bye",
+            }
+        )
+        bye = _next(device, "signal")
+        assert (bye["action"], bye["session"]) == ("bye", "session-one")
+        assert live_view.sessions.count("cam1") == 0
+
+
+def test_a_third_viewer_is_busy_and_closing_ends_the_sessions(
+    client: TestClient, superuser_token_headers: dict[str, str], cams
+) -> None:
+    cam1, _ = cams
+    with client.websocket_connect(SOCKET, headers=cam1) as device:
+        device.receive_json()
+        with _viewer(client, superuser_token_headers) as viewer:
+            _start(viewer, "session-a")
+            _start(viewer, "session-b")
+            _start(viewer, "session-c")
+            busy = _next(viewer, "live")
+            assert (busy["session"], busy["action"]) == ("session-c", "busy")
+            asked = {_next(device, "signal")["session"] for _ in range(2)}
+            assert asked == {"session-a", "session-b"}
+        # The viewer's socket closed: the device hears bye for both.
+        ended = {_next(device, "signal")["session"] for _ in range(2)}
+        assert ended == {"session-a", "session-b"}
+        wait_for(lambda: live_view.sessions.count("cam1") == 0)
+
+    with _viewer(client, superuser_token_headers) as viewer:
+        with client.websocket_connect(SOCKET, headers=cam1) as device:
+            device.receive_json()
+            _start(viewer, "session-d")
+            _next(device, "signal")
+            # The device answers busy itself, past its own cap.
+            device.send_json(
+                {"type": "signal", "session": "session-d", "action": "busy"}
+            )
+            assert _next(viewer, "live")["action"] == "busy"
+
+            _start(viewer, "session-e")
+            _next(device, "signal")
+        # The device's socket closed.
+        bye = _next(viewer, "live")
+        assert (bye["session"], bye["action"], bye["reason"]) == (
+            "session-e",
+            "bye",
+            "offline",
+        )
+
+
+def test_a_session_without_an_answer_times_out(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    cams,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(live_view, "ANSWER_TIMEOUT_S", 0.2)
+    cam1, _ = cams
+    with (
+        client.websocket_connect(SOCKET, headers=cam1) as device,
+        _viewer(client, superuser_token_headers) as viewer,
+    ):
+        device.receive_json()
+        _start(viewer, "session-slow")
+        _next(device, "signal")
+        bye = _next(viewer, "live")
+        assert (bye["action"], bye["reason"]) == ("bye", "timeout")
+        told = _next(device, "signal")
+        assert (told["action"], told["reason"]) == ("bye", "timeout")
+
+
+def test_only_a_person_starts_and_only_the_device_speaks_for_its_session(
+    client: TestClient, superuser_token_headers: dict[str, str], cams
+) -> None:
+    cam1, cam2 = cams
+    with _viewer(client, superuser_token_headers) as viewer:
+        # Not connected, then connected without a live port.
+        _start(viewer, "session-off")
+        assert _next(viewer, "live")["reason"] == "offline"
+        with client.websocket_connect(SOCKET, headers=cam2) as other:
+            other.receive_json()
+            _start(viewer, "session-old", device="cam2")
+            assert _next(viewer, "live")["reason"] == "unsupported"
+
+            with client.websocket_connect(SOCKET, headers=cam1) as device:
+                device.receive_json()
+                # A device's credential cannot start one.
+                _start(other, "session-dev")
+                _start(viewer, "session-ok")
+                assert _next(device, "signal")["session"] == "session-ok"
+                assert live_view.sessions.count("cam1") == 1
+
+                # Another device cannot speak for it.
+                other.send_json(
+                    {
+                        "type": "signal",
+                        "session": "session-ok",
+                        "action": "sdp",
+                        "sdp": "x",
+                    }
+                )
+                time.sleep(0.1)
+                device.send_json(
+                    {
+                        "type": "signal",
+                        "session": "session-ok",
+                        "action": "sdp",
+                        "sdp": "y",
+                    }
+                )
+                assert _next(viewer, "live")["sdp"] == "y"
+
+                # Nor can another viewer, for a session it does not hold.
+                with _viewer(client, superuser_token_headers) as stranger:
+                    stranger.send_json(
+                        {
+                            "type": "live",
+                            "device": "cam1",
+                            "session": "session-ok",
+                            "action": "bye",
+                        }
+                    )
+                    time.sleep(0.1)
+                assert live_view.sessions.count("cam1") == 1
+
+
+def test_ice_servers_carry_coturn_credentials(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    cams,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """``<expiry>:<who>``, and the base64 HMAC-SHA1 of it under the secret."""
+    monkeypatch.setattr(settings, "STUN_URLS", ["stun:turn.example.com:3478"])
+    monkeypatch.setattr(
+        settings,
+        "TURN_URLS",
+        ["turn:turn.example.com:3478", "turns:turn.example.com:5349?transport=tcp"],
+    )
+    monkeypatch.setattr(settings, "TURN_SECRET", "s3cret")
+    cam1, _ = cams
+
+    def check(server: dict[str, Any], who: str) -> None:
+        expiry, _, name = server["username"].partition(":")
+        assert name == who
+        assert 0 < int(expiry) - time.time() <= live_view.ICE_TTL_S
+        mac = hmac.new(b"s3cret", server["username"].encode(), hashlib.sha1)
+        assert server["credential"] == base64.b64encode(mac.digest()).decode()
+
+    answer = client.get(f"{PREFIX}/cam1/ice", headers=superuser_token_headers)
+    assert answer.status_code == 200, answer.text
+    stun, turn = answer.json()["ice_servers"]
+    assert stun == {"urls": ["stun:turn.example.com:3478"]}
+    me = client.get(f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers)
+    check(turn, me.json()["id"])
+
+    assert (
+        client.get(f"{PREFIX}/nope/ice", headers=superuser_token_headers).status_code
+        == 404
+    )
+    # A device gets its servers in `start`, not here.
+    assert client.get(f"{PREFIX}/cam1/ice", headers=cam1).status_code == 403
+
+    with (
+        client.websocket_connect(SOCKET, headers=cam1) as device,
+        _viewer(client, superuser_token_headers) as viewer,
+    ):
+        device.receive_json()
+        _start(viewer, "session-ice")
+        check(_next(device, "signal")["ice_servers"][1], "device-cam1")
+        # The viewer's come with the offer, minted for this session's user.
+        device.send_json(
+            {"type": "signal", "session": "session-ice", "action": "sdp", "sdp": "v=0"}
+        )
+        check(_next(viewer, "live")["ice_servers"][1], me.json()["id"])
+
+
+def test_a_full_outbox_ends_the_session_instead_of_dropping() -> None:
+    """A socket that stops reading loses its session, never a message."""
+
+    class Devices:
+        def reported_ports(self, _device: str) -> dict:
+            return {"live": {}}
+
+    async def run() -> None:
+        registry = live_view.LiveSessions()
+        camera, viewer = live_view.Endpoint(), live_view.Endpoint()
+        registry.attach("cam9", camera)
+        registry.heard(
+            {
+                "type": "live",
+                "device": "cam9",
+                "session": "s-overflow",
+                "action": "start",
+            },
+            viewer,
+            person="someone",
+            device="",
+            devices=Devices(),
+        )
+        assert registry.count("cam9") == 1
+        # The viewer never reads, and the device trickles on.
+        for index in range(live_view.OUTBOX + 1):
+            registry.heard(
+                {
+                    "type": "signal",
+                    "session": "s-overflow",
+                    "action": "candidate",
+                    "candidate": f"candidate:{index} 1 udp 1 10.0.0.9 5000 typ host",
+                },
+                camera,
+                person="",
+                device="cam9",
+                devices=None,
+            )
+        assert registry.count("cam9") == 0
+        start, bye = [json.loads(await camera.next()) for _ in range(2)]
+        assert start["action"] == "start"
+        assert (bye["action"], bye["reason"]) == ("bye", "overflow")
+
+    asyncio.run(run())
+
+
+def test_the_tunnel_carries_signalling(
+    client: TestClient, superuser_token_headers: dict[str, str], cams
+) -> None:
+    """A viewer through the portal, a device on this network, and back."""
+    cam1, _ = cams
+    ticket = client.post(
+        f"{settings.API_V1_STR}/flows/ws-ticket", headers=superuser_token_headers
+    ).json()["ticket"]
+    connector = CloudConnector(client.app)
+    sent: list[dict[str, Any]] = []
+
+    class Socket:
+        async def send(self, text: str) -> None:
+            sent.append(json.loads(text))
+
+    def said() -> list[dict[str, Any]]:
+        return [
+            json.loads(frame["text"])
+            for frame in sent
+            if frame["op"] == "ws_msg" and '"live"' in frame["text"]
+        ]
+
+    frame = {"id": "t1", "path": "/api/v1/flows/ws", "query": f"ticket={ticket}"}
+    assert client.portal is not None
+    with client.websocket_connect(SOCKET, headers=cam1) as device:
+        device.receive_json()
+        stream = client.portal.start_task_soon(connector._stream, Socket(), frame)
+        try:
+            wait_for(lambda: "t1" in connector._listeners)
+            hear = connector._listeners["t1"]
+            # The stream subscribes to the bus once it has said its snapshot.
+            wait_for(lambda: any(f["op"] == "ws_msg" for f in sent))
+            time.sleep(0.1)
+            client.portal.call(
+                hear,
+                json.dumps(
+                    {
+                        "type": "live",
+                        "device": "cam1",
+                        "session": "tunnel-1",
+                        "action": "start",
+                    }
+                ),
+            )
+            assert _next(device, "signal")["session"] == "tunnel-1"
+            device.send_json(
+                {"type": "signal", "session": "tunnel-1", "action": "sdp", "sdp": "v=0"}
+            )
+            offer = wait_for(lambda: said())[0]
+            assert (offer["session"], offer["sdp"]) == ("tunnel-1", "v=0")
+        finally:
+            stream.cancel()
+            wait_for(stream.done)
+        # The stream went: the device hears bye.
+        assert _next(device, "signal")["action"] == "bye"
+
+
+def test_frames_held_for_a_stream_go_with_it(client: TestClient) -> None:
+    """A stream refused before it listened leaves nothing behind."""
+    connector = CloudConnector(client.app)
+    sent: list[str] = []
+
+    class Socket:
+        async def send(self, text: str) -> None:
+            sent.append(text)
+
+    frame = {"id": "t2", "path": "/api/v1/flows/ws", "query": "ticket=spent"}
+    assert client.portal is not None
+
+    def open_and_speak() -> None:
+        connector._start_stream(Socket(), frame)
+        # What the link does with a frame for a stream not listening yet,
+        # here before the stream has even started to run.
+        connector._early["t2"] = ['{"type": "media", "names": []}']
+
+    client.portal.call(open_and_speak)
+    wait_for(lambda: "t2" not in connector._streams)
+    assert "t2" not in connector._early

docker/compose.yml

@@ -226,6 +226,57 @@ services:
     volumes:
       - /var/run/docker.sock:/var/run/docker.sock:ro
 
+  # ── TURN relay for live view (optional) ────────────────────────────
+  # coturn, for a camera's live view between a viewer and a device that
+  # cannot reach each other directly. It serves STUN too. Behind the `turn`
+  # profile, so a plain `make up` never starts it. On the host's network,
+  # because a relay port range through Docker's port mapping is a proxy
+  # process per port. Credentials are minted by the backend from the same
+  # TURN_SECRET (coturn's REST scheme), so nothing is stored here. The secret
+  # goes in through a config file rather than the command line, which `ps`
+  # shows. See DEPLOY.md for ports, DNS and certificates.
+  coturn:
+    image: coturn/coturn:4.6.3
+    container_name: fluksio-coturn
+    profiles: ["turn"]
+    restart: always
+    network_mode: host
+    security_opt:
+      - no-new-privileges:true
+    environment:
+      # Not `?`: compose reads every service, so a stack without the
+      # profile would refuse to start. Checked when coturn starts instead.
+      - TURN_SECRET=${TURN_SECRET:-}
+      - TURN_REALM=turn.${DOMAIN?Variable not set}
+      - TURN_MIN_PORT=${TURN_MIN_PORT:-49160}
+      - TURN_MAX_PORT=${TURN_MAX_PORT:-49200}
+    volumes:
+      # fullchain.pem and privkey.pem for turn.${DOMAIN}. Without them coturn
+      # runs without TLS on 5349.
+      - ${TURN_CERT_DIR:-./turn-certs}:/certs:ro
+    entrypoint: ["sh", "-c"]
+    command:
+      - |
+        [ -n "$$TURN_SECRET" ] || { echo "TURN_SECRET is empty" >&2; exit 1; }
+        cat > /tmp/turnserver.conf <<EOF
+        static-auth-secret=$$TURN_SECRET
+        EOF
+        exec turnserver -c /tmp/turnserver.conf --log-file=stdout --no-cli \
+          --listening-port=3478 --tls-listening-port=5349 \
+          --min-port=$$TURN_MIN_PORT --max-port=$$TURN_MAX_PORT \
+          --use-auth-secret --realm=$$TURN_REALM --fingerprint \
+          --cert=/certs/fullchain.pem --pkey=/certs/privkey.pem \
+          --no-multicast-peers \
+          --denied-peer-ip=0.0.0.0-0.255.255.255 \
+          --denied-peer-ip=10.0.0.0-10.255.255.255 \
+          --denied-peer-ip=100.64.0.0-100.127.255.255 \
+          --denied-peer-ip=127.0.0.0-127.255.255.255 \
+          --denied-peer-ip=169.254.0.0-169.254.255.255 \
+          --denied-peer-ip=172.16.0.0-172.31.255.255 \
+          --denied-peer-ip=192.168.0.0-192.168.255.255 \
+          --denied-peer-ip=fc00::-fdff:ffff:ffff:ffff:ffff:ffff:ffff:ffff \
+          --denied-peer-ip=fe80::-febf:ffff:ffff:ffff:ffff:ffff:ffff:ffff
+
   # ── Site analytics (optional) ──────────────────────────────────────
   # Self-hosted, cookieless Umami on analytics.${DOMAIN}. Gated behind the
   # `analytics` profile, so a plain `make up` never starts it; bring it up with

docs/code/api.md

@@ -201,6 +201,7 @@ on its own, so this is for reclaiming space sooner.
 | `POST` | `/devices/{id}/pair` | adopt by `{code}`, or by `{request}` from the pending list for a request from your network |
 | `POST` | `/devices/{id}/unpair` | revoke the device credential, one the portal issued included |
 | `PUT` | `/devices/{id}/report` | what the device has, from its own credential only; at most 16 KB, replaces the last report, read leniently (see [What a device reports](../interface/dashboards.md#what-a-device-reports)) |
+| `GET` | `/devices/{id}/ice` | STUN and TURN servers for a live view of the device, as `{ice_servers: [{urls, username?, credential?}], ttl}`, with TURN credentials of the caller's own; not for a device credential |
 
 `/panels/pair`, `/panels/pair/{code}` and `GET /panels/{id}` answer as their
 `/devices/` counterparts, for firmware that has not updated yet.
@@ -275,6 +276,37 @@ logged and dropped and the socket stays open. With `http` it uploads with
 `POST /messages/<message>`. A device that is unpaired or deleted while
 connected is closed with 1008.
 
+### Live view
+
+A device that reports a `live` port streams its camera over WebRTC. The
+video goes peer to peer, or through a TURN relay, and the socket carries only
+the signalling, as text, so it crosses the portal too. A person's socket
+sends and hears `{"type": "live", "device": "<id>", "session": "<id>",
+"action": ...}`, a device's socket `{"type": "signal", "session": "<id>",
+"action": ...}`. A session id is 8 to 64 letters, digits, `-` or `_`, chosen by
+the viewer. In order:
+
+| Step | Viewer's socket | Device's socket |
+|---|---|---|
+| 1 | sends `start` | hears `start` with `ice_servers` minted for the device |
+| 2 | hears `sdp` (the offer) with `ice_servers` minted for the viewer | sends `sdp` with its offer |
+| 3 | sends `sdp` with its answer | hears `sdp` |
+| 4 | sends and hears `candidate`, the bare `candidate:...` line | hears and sends `candidate` |
+| 5 | sends `bye`, or hears `bye` or `busy` with a `reason` | hears `bye` with a `reason`, or sends `bye` or `busy` |
+
+Any signed-in person may watch any device. A device speaks only for the
+sessions it was handed on that socket. The server answers `busy` itself once a
+device has 2 sessions, and `bye` with `offline` or `unsupported` when the
+device is not connected or reports no `live` port. A session the viewer has not
+answered within 10 seconds ends with `timeout`, and either socket closing ends
+its sessions with `closed` or `offline`. A socket that falls 64 signalling
+messages behind ends the session with `overflow` instead of losing one.
+`ice_servers` follow `RTCIceServer`: `{urls: [...]}` for STUN,
+`{urls, username, credential}` for TURN, in coturn's REST scheme with a
+username of `<expiry>:<who>`, good for an hour. A device that sends its camera's
+frames on the same socket pauses them while a session negotiates, so its
+offer and candidates are not queued behind frames.
+
 ## Errors and client generation
 
 | Code | Meaning |

docs/interface/dashboards.md

@@ -208,7 +208,8 @@ at, and stops when it scrolls away, the tab is hidden or the tile closes. A
 clip keeps playing either way. One page is pushed frames for at most 8
 messages at once. A ninth tile fetches its frames one by one and does not turn
 a camera on. A [device's camera](#a-devices-camera) sends only while a tile like
-this wants it.
+this wants it. A tile on a device's camera that streams live shows **Live**:
+tap it for [live video](#live-view).
 
 For continuous video, set **Live stream** to a browser-compatible camera URL.
 The browser must be able to reach that URL; enrollment does not expose the
@@ -600,7 +601,7 @@ keeps the switch as it was set.
 
 ### A device's camera
 
-A camera device (openpanel on a NANO) reports four ports:
+A camera device (openpanel on a NANO) reports these ports:
 
 | Port | Type | Settings | What it sends |
 |---|---|---|---|
@@ -608,6 +609,7 @@ A camera device (openpanel on a NANO) reports four ports:
 | `person` | `bool` | `hold_s` (10) | whether someone is in view, held for `hold_s` after they leave |
 | `detections` | `json` | | `{count, boxes: [{score, x0, y0, x1, y1}]}`, boxes scaled 0 to 1, on change and at most once a second |
 | `snapshot` | `image` | | one stored frame each time `person` turns true |
+| `live` | `video`, on demand | | H.264 over WebRTC while a [live view](#live-view) runs; never wired in a flow |
 
 `camera` sends only while it has demand: while a page wants its frames (a
 dashboard tile, the preview on the device's page, a screen through the
@@ -633,6 +635,24 @@ A device may fetch the last 8 artifacts each of its messages pointed at, not
 only the current one, so a screen fetching a camera frame by frame through the
 portal is not refused a frame the camera has just replaced.
 
+#### Live view
+
+A media tile showing a device's `camera`, and the preview on the device's
+page, offer **Live** while the device is online and reports `live`. Tap the
+tile and the browser connects to the device with WebRTC, without a plugin,
+and plays its video in the tile under a **Live** badge. Tap again to stop. It
+stops by itself when the tile scrolls away, the tab is hidden or the page
+closes. While it plays the tile asks for no frames, so the camera encodes one
+picture rather than two, and frames come back when it ends.
+
+A device serves 2 live viewers. A third tile, a failed connection or no
+picture within 10 seconds stays on frames and says so briefly. On one network
+the two ends find each other directly. Between networks they need STUN and,
+behind strict NATs, a TURN relay: set `STUN_URLS`, `TURN_URLS` and
+`TURN_SECRET` (see [Configuration](../reference/configuration.md#live-view)).
+The server only passes on the few messages that set a connection up, the
+video never crosses it. A wall panel's own screen does not offer live view.
+
 ### Before device nodes
 
 Before device nodes a device carried its ports itself. On its first start an

docs/reference/configuration.md

@@ -189,6 +189,24 @@ under Docker, raise `shm_size` with the ring size.
 
 See [Agents over MCP](../code/agents.md).
 
+## Live view
+
+A camera's [live view](../interface/dashboards.md#live-view) connects the
+browser and the device directly. These name the servers that help when the two
+are on different networks. Unset, only direct connections are tried, and no
+third party learns who watches what.
+
+| Variable | Default | Purpose |
+|---|---|---|
+| `STUN_URLS` | unset | STUN servers, comma separated, such as `stun:turn.example.com:3478` |
+| `TURN_URLS` | unset | TURN relays, comma separated, such as `turn:turn.example.com:3478,turns:turn.example.com:5349?transport=tcp` |
+| `TURN_SECRET` | unset | the secret shared with the TURN server (coturn's `static-auth-secret`); TURN is used only with it |
+
+Each viewer and device gets TURN credentials of its own per live view, good for
+an hour, which coturn checks against the shared secret. A view relayed through
+TURN therefore ends after an hour. coturn also answers STUN, so one server
+covers both.
+
 ## Mail and monitoring
 
 Password-reset mail requires `SMTP_HOST` and `EMAILS_FROM_EMAIL`.

frontend/src/client/schemas.gen.ts

@@ -2006,6 +2006,63 @@ export const HistoryPointSchema = {
     description: 'One numeric value a message carried, and when.'
 } as const;
 
+export const IceServerSchema = {
+    properties: {
+        urls: {
+            items: {
+                type: 'string'
+            },
+            type: 'array',
+            title: 'Urls'
+        },
+        username: {
+            anyOf: [
+                {
+                    type: 'string'
+                },
+                {
+                    type: 'null'
+                }
+            ],
+            title: 'Username'
+        },
+        credential: {
+            anyOf: [
+                {
+                    type: 'string'
+                },
+                {
+                    type: 'null'
+                }
+            ],
+            title: 'Credential'
+        }
+    },
+    type: 'object',
+    required: ['urls'],
+    title: 'IceServer',
+    description: 'One STUN or TURN server, as `RTCPeerConnection` takes it.'
+} as const;
+
+export const IceServersSchema = {
+    properties: {
+        ice_servers: {
+            items: {
+                '$ref': '#/components/schemas/IceServer'
+            },
+            type: 'array',
+            title: 'Ice Servers'
+        },
+        ttl: {
+            type: 'integer',
+            title: 'Ttl',
+            default: 0
+        }
+    },
+    type: 'object',
+    title: 'IceServers'
+} as const;
+
 export const ImportArtifactSchema = {
     properties: {
         name: {

frontend/src/client/types.gen.ts

@@ -686,6 +686,20 @@ export type HTTPValidationError = {
     detail?: Array<ValidationError>;
 };
 
+/**
+ * One STUN or TURN server, as `RTCPeerConnection` takes it.
+ */
+export type IceServer = {
+    urls: Array<(string)>;
+    username?: (string | null);
+    credential?: (string | null);
+};
+
+export type IceServers = {
+    ice_servers?: Array<IceServer>;
+    ttl?: number;
+};
+
 export type ImportArtifact = {
     name: string;
     filename?: (string | null);
@@ -1879,6 +1893,12 @@ export type DevicesDeleteDeviceData = {
 
 export type DevicesDeleteDeviceResponse = (Message);
 
+export type DevicesReadIceData = {
+    deviceId: string;
+};
+
+export type DevicesReadIceResponse = (IceServers);
+
 export type DevicesPutReportData = {
     deviceId: string;
 };

frontend/src/components/Dashboard/MediaWidget.tsx

@@ -1,10 +1,12 @@
-import { useRef } from "react"
+import { LoaderCircle, Radio, X } from "lucide-react"
+import { type ReactNode, useEffect, useRef } from "react"
 
 import { useOnScreen } from "@/hooks/useOnScreen"
 import { isRef, useArtifactUrl } from "@/lib/media"
 import { cn } from "@/lib/utils"
 import { useBoundValue } from "./dataContext"
 import { config, text } from "./ui/core/config"
+import { useLiveDevice, useLiveView } from "./useLiveView"
 import type { WidgetProps } from "./widgets"
 
 /**
@@ -37,6 +39,10 @@ export function MediaWidget({ widget }: WidgetProps) {
  * looked at: a camera is only sent while somebody watches it, and a tile
  * scrolled away or behind another tab is nobody. A clip keeps playing either
  * way.
+ *
+ * A camera whose device streams live is tapped for it. Frames pause while
+ * the live view plays, so the device encodes one picture rather than two,
+ * and come back when it ends.
  */
 export function MediaView({
   message,
@@ -55,12 +61,15 @@ export function MediaView({
 }) {
   const box = useRef<HTMLDivElement>(null)
   const shown = useOnScreen(box)
-  const live = useBoundValue(message || undefined)
-  const value = live?.value
+  const bound = useBoundValue(message || undefined)
+  const value = bound?.value
   const ref = isRef(value) ? value : null
   const kind = (ref?.media_type ?? dtype).split("/")[0]
   const playing = kind === "audio" || kind === "video"
-  const active = shown || playing
+  const camera = useLiveDevice(stream ? "" : message)
+  const live = useLiveView(camera)
+  const onAir = live.phase === "playing"
+  const active = (shown || playing) && !onAir
   const url = useArtifactUrl(
     active ? ref : null,
     active ? message || undefined : undefined,
@@ -69,12 +78,24 @@ export function MediaView({
   const objectFit = fit === "contain" ? "object-contain" : "object-cover"
   const alt = label || ref?.name || message
 
+  // Scrolled away, or the tab hidden: nobody is watching.
+  const { stop } = live
+  useEffect(() => {
+    if (!shown) stop()
+  }, [shown, stop])
+
   return (
     <div
       ref={box}
       className="flex min-h-0 w-full flex-1 flex-col justify-center"
     >
-      {body()}
+      {camera ? (
+        <LiveFrame live={live} label={alt} fit={objectFit}>
+          {body()}
+        </LiveFrame>
+      ) : (
+        body()
+      )}
     </div>
   )
 
@@ -141,3 +162,95 @@ export function MediaView({
     )
   }
 }
+
+/**
+ * A camera tile that streams live when tapped, and stops when tapped again.
+ *
+ * The whole tile is the control, so it is as easy to hit on a wall panel as
+ * on a desk. The video waits under the frames until its first picture, then
+ * takes their place.
+ */
+function LiveFrame({
+  live,
+  label,
+  fit,
+  children,
+}: {
+  live: ReturnType<typeof useLiveView>
+  label: string
+  fit: string
+  children: ReactNode
+}) {
+  const { phase, stream, note } = live
+  const onAir = phase === "playing"
+  const chip =
+    "pointer-events-none absolute z-20 flex items-center gap-1 rounded-full border border-border px-2.5 py-1 text-xs font-medium shadow-e2"
+  const frosted = "bg-card/80 text-foreground backdrop-blur-md"
+
+  return (
+    <button
+      type="button"
+      onClick={phase === "off" ? live.start : live.stop}
+      aria-label={
+        phase === "off" ? `Watch ${label} live` : `Stop watching ${label} live`
+      }
+      aria-pressed={phase !== "off"}
+      data-testid="media-live"
+      data-live={phase}
+      className="relative isolate flex min-h-0 w-full flex-1 cursor-pointer flex-col justify-center rounded-md text-center focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring"
+    >
+      <span
+        className={cn(
+          "relative z-10 flex min-h-0 w-full flex-1 flex-col justify-center",
+          onAir && "invisible",
+        )}
+      >
+        {children}
+      </span>
+      {stream ? (
+        <video
+          ref={(element) => {
+            if (element && element.srcObject !== stream)
+              element.srcObject = stream
+          }}
+          autoPlay
+          muted
+          playsInline
+          onPlaying={live.played}
+          className={cn("absolute inset-0 h-full w-full rounded-md", fit)}
+        />
+      ) : null}
+      {onAir ? (
+        <span
+          className={cn(
+            chip,
+            "right-2 top-2 border-transparent bg-primary text-primary-foreground",
+          )}
+        >
+          Live
+          <X className="size-3.5" aria-hidden />
+        </span>
+      ) : (
+        <span className={cn(chip, frosted, "right-2 top-2")}>
+          {phase === "connecting" ? (
+            <LoaderCircle
+              className="size-3.5 motion-safe:animate-spin"
+              aria-hidden
+            />
+          ) : (
+            <Radio className="size-3.5" aria-hidden />
+          )}
+          {phase === "connecting" ? "Connecting" : "Live"}
+        </span>
+      )}
+      {note ? (
+        <output
+          className={cn(chip, frosted, "bottom-2 left-2 text-muted-foreground")}
+          data-testid="media-live-note"
+        >
+          {note}
+        </output>
+      ) : null}
+    </button>
+  )
+}

frontend/src/components/Dashboard/useLiveView.ts

@@ -0,0 +1,153 @@
+import { useQuery } from "@tanstack/react-query"
+import { useCallback, useEffect, useRef, useState } from "react"
+
+import { LIVE_PORT, reportedPorts } from "@/components/Devices/ports"
+import { devicesQueryOptions } from "@/components/Devices/queries"
+import {
+  type LiveSignal,
+  listenLive,
+  sendLive,
+} from "@/components/Flow/useFlowSocket"
+import { appRoute } from "@/lib/portal"
+
+/** How long a live view gets from the tap to its first frame. */
+const FIRST_FRAME_MS = 10_000
+/** How long the note after a fallback stays. */
+const NOTE_MS = 4000
+
+const BUSY = "Live view busy, showing frames"
+const FAILED = "Live view unavailable, showing frames"
+
+export type LivePhase = "off" | "connecting" | "playing"
+
+/**
+ * The device whose camera publishes `message` and can stream it live, or "".
+ *
+ * A device offers it by reporting a `live` port, and only while it is online.
+ * Read off the device list, which a wall panel's credential cannot read, so a
+ * panel never offers it.
+ */
+export function useLiveDevice(message: string): string {
+  const enabled = Boolean(message) && !appRoute().startsWith("/panel")
+  const { data } = useQuery({ ...devicesQueryOptions(), enabled })
+  if (!enabled) return ""
+  const device = data?.data?.find(
+    (one) =>
+      one.online &&
+      (one.bindings ?? []).some(
+        (binding) =>
+          binding.port === "camera" &&
+          binding.direction === "in" &&
+          binding.message === message,
+      ) &&
+      (reportedPorts(one) ?? []).some((port) => port.port === LIVE_PORT),
+  )
+  return device?.id ?? ""
+}
+
+/**
+ * A live view of one device's camera: the browser's own `RTCPeerConnection`
+ * answering the device's offer, signalled over the page's socket.
+ *
+ * Busy, a failure or no first frame within ten seconds ends it with a short
+ * note, and the tile goes on showing frames. Leaving says `bye`.
+ */
+export function useLiveView(device: string) {
+  const [phase, setPhase] = useState<LivePhase>("off")
+  const [stream, setStream] = useState<MediaStream | null>(null)
+  const [note, setNote] = useState("")
+  const running = useRef<{ stop: () => void; played: () => void } | null>(null)
+
+  const stop = useCallback(() => running.current?.stop(), [])
+  const played = useCallback(() => running.current?.played(), [])
+
+  const start = useCallback(() => {
+    if (!device || running.current) return
+    // Not `randomUUID`, which a page on plain HTTP on a LAN does not have.
+    const session = Array.from(
+      crypto.getRandomValues(new Uint8Array(12)),
+      (byte) => byte.toString(16).padStart(2, "0"),
+    ).join("")
+    let pc: RTCPeerConnection | null = null
+    let over = false
+    // The device's candidates that came before its offer was applied.
+    const early: string[] = []
+
+    const end = (why: string, tell: boolean) => {
+      if (over) return
+      over = true
+      clearTimeout(timer)
+      unlisten()
+      if (tell) sendLive({ device, session, action: "bye" })
+      pc?.close()
+      running.current = null
+      setStream(null)
+      setPhase("off")
+      setNote(why)
+    }
+    const add = (candidate: string) =>
+      pc
+        ?.addIceCandidate({ candidate, sdpMLineIndex: 0 })
+        .catch(() => undefined)
+
+    const answer = async (offer: LiveSignal) => {
+      const peer = new RTCPeerConnection({
+        iceServers: offer.ice_servers ?? [],
+      })
+      pc = peer
+      peer.ontrack = (event) =>
+        setStream(event.streams[0] ?? new MediaStream([event.track]))
+      peer.onicecandidate = (event) => {
+        // The bare `candidate:` line, which is what the device reads.
+        const candidate = event.candidate?.candidate
+        if (candidate)
+          sendLive({ device, session, action: "candidate", candidate })
+      }
+      peer.onconnectionstatechange = () => {
+        if (peer.connectionState === "failed") end(FAILED, true)
+      }
+      await peer.setRemoteDescription({ type: "offer", sdp: offer.sdp })
+      const reply = await peer.createAnswer()
+      await peer.setLocalDescription(reply)
+      sendLive({ device, session, action: "sdp", sdp: reply.sdp })
+      for (const candidate of early.splice(0)) await add(candidate)
+    }
+
+    const unlisten = listenLive(session, (message) => {
+      if (over) return
+      if (message.action === "busy") end(BUSY, false)
+      else if (message.action === "bye") end(FAILED, false)
+      else if (message.action === "sdp" && message.sdp && !pc)
+        answer(message).catch(() => end(FAILED, true))
+      else if (message.action === "candidate" && message.candidate) {
+        if (pc?.remoteDescription) add(message.candidate)
+        else early.push(message.candidate)
+      }
+    })
+    const timer = setTimeout(() => end(FAILED, true), FIRST_FRAME_MS)
+    running.current = {
+      stop: () => end("", true),
+      played: () => {
+        clearTimeout(timer)
+        setPhase("playing")
+      },
+    }
+    setNote("")
+    setPhase("connecting")
+    if (!sendLive({ device, session, action: "start" })) end(FAILED, false)
+  }, [device])
+
+  // A tile that goes away, or is pointed at another device, ends its view.
+  useEffect(() => {
+    if (!device) return
+    return () => running.current?.stop()
+  }, [device])
+
+  useEffect(() => {
+    if (!note) return
+    const timer = setTimeout(() => setNote(""), NOTE_MS)
+    return () => clearTimeout(timer)
+  }, [note])
+
+  return { phase, stream, note, start, stop, played }
+}

frontend/src/components/Devices/DeviceDetail.tsx

@@ -44,6 +44,7 @@ import {
   type DeviceReport,
   deviceNodeOf,
   freeNodeId,
+  LIVE_PORT,
   type ReportedPort,
   reportedPorts,
   reportOf,
@@ -233,7 +234,7 @@ function Camera({ device }: { device: DevicePublic }) {
     <section className={CARD}>
       <PanelSection
         title="Camera"
-        help="Live while this preview is on screen. The camera sends frames only while someone watches them here or on a dashboard, or a running flow reads them."
+        help="Frames while this preview is on screen, which the camera sends only while someone watches them here or on a dashboard, or a running flow reads them. Tap it for live video when the device streams live."
       >
         <div
           className="flex aspect-video w-full overflow-hidden rounded-md bg-muted text-center text-sm"
@@ -748,7 +749,11 @@ function Capabilities({
                     </span>
                   ) : null}
                 </span>
-                {canEdit && said && wired.length === 0 ? (
+                {said?.port === LIVE_PORT ? (
+                  <span className="shrink-0 text-xs text-muted-foreground">
+                    Live view
+                  </span>
+                ) : canEdit && said && wired.length === 0 ? (
                   <UseInFlow device={device} port={said} size="sm" />
                 ) : null}
               </li>

frontend/src/components/Devices/ports.ts

@@ -3,6 +3,12 @@ import type { DevicePublic, DType, MessageSpec, NodeDef_Input } from "@/client"
 /** The node type a device is in a flow. */
 export const DEVICE_TYPE = "device"
 
+/**
+ * The port a device reports when its camera can stream live over WebRTC.
+ * Signalled on the socket, never wired in a flow.
+ */
+export const LIVE_PORT = "live"
+
 /** One port as a device reports it. */
 export type ReportedPort = {
   port: string
@@ -57,8 +63,9 @@ export function screensOf(devices: DevicePublic[] = []) {
   })
 }
 
-/** Whether a reported port may be wired in `direction`. */
+/** Whether a reported port may be wired in `direction`. `live` never is. */
 export const wiresAs = (port: ReportedPort, direction: "in" | "out") =>
+  port.port !== LIVE_PORT &&
   (port.modes?.length ? port.modes : [port.direction]).includes(direction)
 
 /** What the server accepts as a port name. */

frontend/src/components/Flow/CommandPalette.tsx

@@ -2,7 +2,11 @@ import { useQuery } from "@tanstack/react-query"
 import { useNavigate } from "@tanstack/react-router"
 
 import type { DevicePublic, FlowSummary, NodeTypeInfo } from "@/client"
-import { type ReportedPort, reportedPorts } from "@/components/Devices/ports"
+import {
+  LIVE_PORT,
+  type ReportedPort,
+  reportedPorts,
+} from "@/components/Devices/ports"
 import { devicesQueryOptions } from "@/components/Devices/queries"
 import {
   CommandDialog,
@@ -58,7 +62,9 @@ export function CommandPalette({
   // A shortcut per port a device says it has, "Hallway camera". A device that
   // has not said is added as a Device node and its ports typed in.
   const capabilities = (devicesData?.data ?? []).flatMap((device) =>
-    (reportedPorts(device) ?? []).map((port) => ({ device, port })),
+    (reportedPorts(device) ?? [])
+      .filter((port) => port.port !== LIVE_PORT)
+      .map((port) => ({ device, port })),
   )
 
   const close = (action: () => void) => {

frontend/src/components/Flow/useFlowSocket.ts

@@ -134,6 +134,23 @@ type FlowEvent =
       names: string[]
       ts?: number
     }
+  | LiveSignal
+
+/**
+ * A live view's signalling, both ways on this socket (`fluksio.flow.live`).
+ * The page sends `start`, `sdp`, `candidate` and `bye`, and hears `sdp` (the
+ * device's offer, with the ICE servers to use), `candidate`, `bye` and `busy`.
+ */
+export type LiveSignal = {
+  type: "live"
+  device: string
+  session: string
+  action: "start" | "sdp" | "candidate" | "bye" | "busy"
+  sdp?: string
+  candidate?: string
+  reason?: string
+  ice_servers?: RTCIceServer[]
+}
 
 async function socketUrl(): Promise<string> {
   const base = String(OpenAPI.BASE || window.location.origin)
@@ -176,6 +193,42 @@ let client: QueryClient | null = null
 const authHandlers = new Set<() => void>()
 /** Floor re-reads still owed, per node, so a recovery can cancel its own. */
 const healthTimers = new Map<string, ReturnType<typeof setTimeout>>()
+/** What each live session on this page does with what it hears. */
+const liveListeners = new Map<string, (message: LiveSignal) => void>()
+
+/** Send a live view's signalling. False while the socket is not open. */
+export function sendLive(message: Omit<LiveSignal, "type">): boolean {
+  if (socket?.readyState !== WebSocket.OPEN) return false
+  socket.send(JSON.stringify({ type: "live", ...message }))
+  return true
+}
+
+/**
+ * Hear one live session's signalling, until the returned function is called.
+ * A socket that drops ends every session on it, so each hears `bye` then.
+ */
+export function listenLive(
+  session: string,
+  listener: (message: LiveSignal) => void,
+): () => void {
+  liveListeners.set(session, listener)
+  return () => {
+    if (liveListeners.get(session) === listener) liveListeners.delete(session)
+  }
+}
+
+/** The socket went: so did every live session riding on it. */
+function endLive() {
+  for (const [session, listener] of [...liveListeners]) {
+    listener({
+      type: "live",
+      device: "",
+      session,
+      action: "bye",
+      reason: "socket",
+    })
+  }
+}
 
 /** Everything a node's health shows up in, which is two unrelated keys. */
 function invalidateHealth(flow: string | undefined) {
@@ -436,6 +489,9 @@ function open(url: string) {
         // live value, so nothing here goes through the live store.
         client?.invalidateQueries({ queryKey: runKeys.all })
         break
+      case "live":
+        liveListeners.get(message.session)?.(message)
+        break
       case "run_metric":
         // A batch of readings was written. Only what draws them is refetched —
         // the run's own detail (its metric names and curve hang below that
@@ -480,6 +536,7 @@ function open(url: string) {
     socket = null
     liveStore.setConnected(false)
     connectionStore.setSocketOpen(false)
+    endLive()
     if (watchers === 0) return
     if (event.code === 1008 && refused()) return
     // 1013 is the portal saying the instance is not attached — the one
@@ -504,6 +561,7 @@ function release(onAuthFailure?: () => void) {
   const ws = socket
   socket = null
   ws?.close()
+  endLive()
   liveStore.setConnected(false)
   // Its `onclose` is dropped by the guard above, so say so here instead.
   connectionStore.setSocketOpen(false)