Camera permission and review fixes (openpanel C3, C4)

C4: a device's `viewers` (superusers by default, or members) covers the
image and video messages its ports publish. Anyone else gets none of their
values on the socket, in the catalogue or a flow's state, 403 for their
history, for the frames they hold (the last 8 included) and for TURN
credentials, no bytes when asking for them, and `bye: forbidden` for a live
view. A device credential stays bounded by its dashboards. The device page
sets it in its Camera section, and a tile the viewer may not see says
"No access to this camera" and asks the camera for nothing.

C3:
- A volatile upload is held to MAX_FRAME_BYTES (4 MiB, 413 above it, read
  without extra copies), and so is a frame on a device's socket.
- A frame the engine holds in memory is written to the ring directory
  before a local worker is handed it, so load_artifact finds it.
- A socket the bus dropped events for catches up with a snapshot, demand
  included, and a device's credential is checked again per frame, cached by
  the flow store revision and its nonce, so a lost event never keeps an
  unpaired device's frames coming.
- A report is at most 64 KB once its named config_schemas are put in place.
- A device away for 60 s has its true bool messages set to false.
- Refused frames are logged by fixed keys, a frame is named after its
  message, `fluksio serve` waits 5 s for open connections, a tile with a
  stream URL creates no demand, the 8-name cap keeps the newest names.
- Every container's log rotates (json-file, 3 x 10 MB).

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

backend/fluksio/api/deps.py

@@ -177,10 +177,12 @@ def _device_digests(device: DeviceDef, request: Request) -> set[str]:
     controller: FlowController | None = getattr(
         request.app.state, "flow_controller", None
     )
-    if controller is None:
-        return set()
-    names = _device_messages(device, request)
-    if not names:
+    return digests_of(controller, _device_messages(device, request))
+
+
+def digests_of(controller: FlowController | None, names: set[str]) -> set[str]:
+    """The artifacts these messages point at now, and the last few before."""
+    if controller is None or not names:
         return set()
     found: set[str] = set()
     for value in controller.state.get_present(sorted(names)).values():
@@ -191,6 +193,19 @@ def _device_digests(device: DeviceDef, request: Request) -> set[str]:
     return found
 
 
+def hidden_from(user: User, app: Any, *, device: bool = False) -> frozenset[str]:
+    """The camera messages ``user`` may not see (``DeviceDef.viewers``).
+
+    None for a superuser, and none for a device credential, which its
+    dashboards bound instead: a superuser assigned them, camera tiles
+    included. ``device`` says the caller is one.
+    """
+    if device or user.is_superuser:
+        return frozenset()
+    store: DeviceStore | None = getattr(app.state, "device_store", None)
+    return store.restricted() if store is not None else frozenset()
+
+
 def _device_may(payload: dict[str, Any], request: Request) -> None:
     """Refuse anything a wall panel has no business asking for.
 
@@ -520,6 +535,18 @@ def get_current_user(request: Request, session: SessionDep, token: TokenDep) ->
 CurrentUser = Annotated[User, Depends(get_current_user)]
 
 
+def get_hidden(request: Request, user: CurrentUser) -> frozenset[str]:
+    """The camera messages the caller of this request may not see."""
+    device = getattr(request.state, "device", None) is not None
+    return hidden_from(user, request.app, device=device)
+
+
+HiddenDep = Annotated[frozenset[str], Depends(get_hidden)]
+
+#: Why a camera's values, bytes or live view are refused.
+CAMERA_IS_SUPERUSERS = "Only a superuser may see this camera"
+
+
 def get_flow_controller(request: Request) -> FlowController:
     controller: FlowController | None = getattr(
         request.app.state, "flow_controller", None

backend/fluksio/api/routes/artifacts.py

@@ -18,8 +18,11 @@ from sqlmodel import Session
 from starlette.concurrency import run_in_threadpool
 
 from fluksio.api.deps import (
+    CAMERA_IS_SUPERUSERS,
     FlowControllerDep,
+    digests_of,
     get_current_active_superuser,
+    hidden_from,
     user_from_token,
     worker_claims,
 )
@@ -63,6 +66,8 @@ def artifact_caller(request: Request) -> str:
         user = user_from_token(session, token, request)
     if user is None:
         raise HTTPException(status_code=401, detail="Not authenticated")
+    # For the camera check a fetch is held to, which a worker is not.
+    request.state.person = user
     return user.email
 
 
@@ -103,26 +108,30 @@ async def put_artifact(
     watching now, which the oldest of falls out of memory rather than being
     kept. A worker on another host or a device publishing a camera sends this.
     It is read into memory, where the ring holds it, and never touches the
-    disk.
+    disk, so it is held to ``MAX_FRAME_BYTES`` rather than the store's cap.
     """
     store = _store(request)
-    cap = settings.MAX_ARTIFACT_BYTES
+    ring = store.volatile if volatile else None
+    cap = settings.MAX_FRAME_BYTES if ring is not None else settings.MAX_ARTIFACT_BYTES
     declared = request.headers.get("content-length")
     if cap and declared and declared.isdigit() and int(declared) > cap:
         raise HTTPException(
             status_code=413, detail=f"An artifact may be at most {cap} bytes"
         )
-    ring = store.volatile if volatile else None
     if ring is not None:
-        body = bytearray()
+        # The chunks as they came, joined once: the frame is copied into
+        # memory one time, never past the cap.
+        chunks: list[bytes] = []
+        size = 0
         async for chunk in request.stream():
-            body += chunk
-            if cap and len(body) > cap:
+            size += len(chunk)
+            if cap and size > cap:
                 raise HTTPException(
                     status_code=413,
                     detail=f"An artifact may be at most {cap} bytes",
                 )
-        return await run_in_threadpool(ring.put, [bytes(body)], name, media_type)
+            chunks.append(chunk)
+        return await run_in_threadpool(ring.put, chunks, name, media_type)
     handle = tempfile.NamedTemporaryFile(dir=store.root, delete=False)
     written = 0
     try:
@@ -158,6 +167,15 @@ def get_artifact(
     scrubbing through a file asks for them.
     """
     store = _store(request)
+    user = getattr(request.state, "person", None)
+    if user is not None:
+        # What a camera only superusers may see points at, now or in its last
+        # few frames. A worker's node reads anything, as node code can.
+        device = getattr(request.state, "device", None) is not None
+        hidden = hidden_from(user, request.app, device=device)
+        controller = getattr(request.app.state, "flow_controller", None)
+        if hidden and digest in digests_of(controller, set(hidden)):
+            raise HTTPException(status_code=403, detail=CAMERA_IS_SUPERUSERS)
     path = store.path(digest)
     # A live frame is held in memory, not on disk.
     frame = store.volatile.get(digest) if path is None and store.volatile else None

backend/fluksio/api/routes/devices.py

@@ -43,12 +43,14 @@ from datetime import datetime, timedelta
 from typing import Any, cast
 
 import httpx
+import orjson
 import redis
 from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.concurrency import run_in_threadpool
 from pydantic import BaseModel, Field, ValidationError, field_validator
 
 from fluksio.api.deps import (
+    CAMERA_IS_SUPERUSERS,
     CurrentUser,
     DeviceStoreDep,
     get_current_active_superuser,
@@ -120,6 +122,10 @@ HELLO_LENGTH = 64
 #: The most a capability report may weigh. A panel with every header pin
 #: reported is a few kB.
 REPORT_MAX = 16 * 1024
+#: The most it may weigh as kept, with each named ``config_schemas`` entry put
+#: in place: thirty pins naming one schema send it once and keep it thirty
+#: times. The P4 panel's is about 7 kB.
+REPORT_STORED_MAX = 64 * 1024
 
 
 class _Pending(BaseModel):
@@ -690,9 +696,13 @@ async def read_ice(device_id: str, store: DeviceStoreDep, user: CurrentUser) ->
 
     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.
+    Only for whoever may see the device's camera (``DeviceDef.viewers``).
     """
-    if await run_in_threadpool(store.find, device_id) is None:
+    device = await run_in_threadpool(store.find, device_id)
+    if device is None:
         raise _not_found(device_id)
+    if not user.is_superuser and device.viewers != "members":
+        raise HTTPException(status_code=403, detail=CAMERA_IS_SUPERUSERS)
     servers = [IceServer(**server) for server in ice_servers(str(user.id))]
     return IceServers(ice_servers=servers, ttl=ICE_TTL_S)
 
@@ -745,6 +755,12 @@ async def put_report(device_id: str, request: Request, store: DeviceStoreDep) ->
                 include_url=False, include_input=False, include_context=False
             ),
         )
+    if len(orjson.dumps(report.stored())) > REPORT_STORED_MAX:
+        raise HTTPException(
+            status_code=413,
+            detail=f"A report is at most {REPORT_STORED_MAX // 1024} KB once "
+            "its config_schemas are put in place",
+        )
     for part in left_out:
         logger.warning("Device '%s' reported %s, left out", device_id, part)
     first = await run_in_threadpool(store.report, device_id, report)

backend/fluksio/api/routes/flows.py

@@ -23,14 +23,17 @@ from pydantic import BaseModel
 from sqlmodel import Session, col, select
 
 from fluksio.api.deps import (
+    CAMERA_IS_SUPERUSERS,
     CODE_IS_SUPERUSERS,
     CurrentUser,
     FlowControllerDep,
+    HiddenDep,
     SessionDep,
     TokenDep,
     decode_token,
     get_code_author,
     get_current_user,
+    hidden_from,
     user_from_token,
 )
 from fluksio.api.routes.messages import port_source
@@ -279,10 +282,15 @@ def _require_enabled(controller: FlowController, name: str) -> None:
         )
 
 
-def _flow_state(controller: FlowController, name: str) -> FlowStatePublic:
+def _flow_state(
+    controller: FlowController, name: str, hidden: frozenset[str]
+) -> FlowStatePublic:
+    """The flow's values, but those of a camera the caller may not see."""
     return FlowStatePublic(
         values={
-            key: MessageValue(**value) for key, value in controller.values(name).items()
+            key: MessageValue(**value)
+            for key, value in controller.values(name).items()
+            if key not in hidden
         },
         nodes=controller.node_statuses(name),
     )
@@ -892,6 +900,7 @@ async def run_flow(
     request: Request,
     controller: FlowControllerDep,
     user: CurrentUser,
+    hidden: HiddenDep,
 ) -> Any:
     """Run every node of a flow once.
 
@@ -919,14 +928,14 @@ async def run_flow(
             )
         except RunRejected as exc:
             raise HTTPException(status_code=422, detail=str(exc)) from exc
-        return _flow_state(controller, name)
+        return _flow_state(controller, name, hidden)
 
     inputs = {qualify(name, key): value for key, value in body.inputs.items()}
     if controller.store.has_draft(name):
         await run_in_threadpool(controller.run_preview, name, inputs)
     else:
         await run_in_threadpool(controller.run_flow, name, inputs)
-    return _flow_state(controller, name)
+    return _flow_state(controller, name, hidden)
 
 
 @router.post("/{name}/nodes/{node_id}/trigger", response_model=FlowStatePublic)
@@ -935,6 +944,7 @@ async def trigger_node(
     node_id: str,
     body: TriggerRequest,
     controller: FlowControllerDep,
+    hidden: HiddenDep,
 ) -> Any:
     """Feed values into a single node."""
     _require_enabled(controller, name)
@@ -950,7 +960,7 @@ async def trigger_node(
         # The canvas already has the failure from the bus; the person who
         # clicked gets to read it too, instead of a stack trace in the log.
         raise HTTPException(status_code=400, detail=error)
-    return _flow_state(controller, name)
+    return _flow_state(controller, name, hidden)
 
 
 @router.post("/{name}/nodes/{node_id}/cancel", response_model=Message)
@@ -995,10 +1005,10 @@ def acknowledge_node_error(
 
 
 @router.get("/{name}/state", response_model=FlowStatePublic)
-def read_flow_state(name: str, controller: FlowControllerDep) -> Any:
+def read_flow_state(name: str, controller: FlowControllerDep, hidden: HiddenDep) -> Any:
     """The last value seen on every message of this flow."""
     _read_flow(controller, name)
-    return _flow_state(controller, name)
+    return _flow_state(controller, name, hidden)
 
 
 @router.get("/{name}/history/{message}", response_model=MessageHistory)
@@ -1006,6 +1016,7 @@ def read_message_history(
     name: str,
     message: str,
     controller: FlowControllerDep,
+    hidden: HiddenDep,
 ) -> Any:
     """The recent values of one message, for plotting.
 
@@ -1014,6 +1025,8 @@ def read_message_history(
     """
     _read_flow(controller, name)
     key = qualify(name, message)
+    if key in hidden:
+        raise HTTPException(status_code=403, detail=CAMERA_IS_SUPERUSERS)
     return MessageHistory(
         message=key,
         numeric=as_number(controller.state.get(key)) is not None,
@@ -1034,6 +1047,7 @@ def snapshot_payload(
     only: set[str] | None = None,
     device: str = "",
     uplink: str = "",
+    hidden: frozenset[str] = frozenset(),
 ) -> dict[str, Any]:
     """Everything a client needs to catch up, sent the moment it connects.
 
@@ -1052,6 +1066,8 @@ def snapshot_payload(
     A device's socket also says which of its ports have ``demand`` now, and
     which ``uplink`` its frames take: ``binary`` on its own socket, ``http``
     through the portal, whose tunnel carries text only.
+
+    ``hidden`` are the camera messages a person may not see, left out.
     """
     if only is not None:
         # The same keys, emptied rather than dropped: a screen already hanging
@@ -1071,9 +1087,12 @@ def snapshot_payload(
             payload["demand"] = demand.ports(device)
             payload["uplink"] = uplink
         return payload
+    values = controller.values()
     return {
         "type": "snapshot",
-        "values": controller.values(),
+        "values": (
+            {k: v for k, v in values.items() if k not in hidden} if hidden else values
+        ),
         "nodes": [s.model_dump() for s in controller.node_statuses()],
         "issues": [i.model_dump() for i in controller.issues],
         "paused": controller.paused_flows(),
@@ -1096,6 +1115,41 @@ def device_of(token: str, app: Any) -> str:
         return ""
 
 
+class Credential:
+    """Whether a device socket's credential still stands, asked per frame.
+
+    Checked in full only once what it is checked against has moved: the
+    device files, which a device's save, deletion or a revert commits to the
+    flow store, and the nonces in ``devices.json``, which an unpair bumps.
+    Otherwise a frame costs a stat of that file.
+    """
+
+    def __init__(self, token: str, app: Any, device_id: str) -> None:
+        self.token = token
+        self.app = app
+        self.device = device_id
+        self._seen: tuple[int, int] | None = None
+
+    def _key(self) -> tuple[int, int] | None:
+        devices: DeviceStore | None = getattr(self.app.state, "device_store", None)
+        if devices is None:
+            return None
+        try:
+            return devices.flows.revision, devices.nonce(self.device)
+        except (OSError, ValueError):
+            return None
+
+    async def revoked(self) -> bool:
+        key = self._key()
+        if key is not None and key == self._seen:
+            return False
+        if await run_in_threadpool(revoked, self.token, self.app):
+            return True
+        # The key read before the check, so a change meanwhile is checked.
+        self._seen = key
+        return False
+
+
 def revoked(token: str, app: Any) -> bool:
     """Whether a device's credential has stopped standing since its handshake.
 
@@ -1155,6 +1209,21 @@ def event_for_panel(event: dict[str, Any], only: set[str]) -> bool:
     return event.get("type") == "message_value" and str(event.get("name") or "") in only
 
 
+def carried(
+    event: dict[str, Any], only: set[str] | None, hidden: frozenset[str]
+) -> bool:
+    """Whether a socket carries this event on the bus.
+
+    A panel's what it draws (``event_for_panel``), a person's everything but
+    the values of the cameras ``hidden`` from them.
+    """
+    if only is not None:
+        return event_for_panel(event, only)
+    return not (
+        hidden and event.get("type") == "message_value" and event.get("name") in hidden
+    )
+
+
 # How many events one frame may carry. A ceiling rather than a target: the
 # batch is whatever the bus happens to hold, and a client that has been away
 # should not be handed the whole queue in one message.
@@ -1166,14 +1235,19 @@ MAX_FRAME_EVENTS = 64
 MAX_MEDIA_NAMES = 8
 
 
-def wanted_names(frame: str, only: set[str] | None) -> set[str] | None:
+def wanted_names(
+    frame: str, only: set[str] | None, hidden: frozenset[str] = frozenset()
+) -> set[str] | None:
     """What this client is asking to be sent bytes for, or None if it said
     something else.
 
     A client asks by name — the tiles it is drawing — and is answered only for
-    the names a panel credential would have been allowed anyway. Bytes are the
-    expensive thing on this socket, so nothing is pushed until something says
-    it is looking at it.
+    the names a panel credential would have been allowed anyway, and none of
+    the ``hidden`` cameras. Bytes are the expensive thing on this socket, so
+    nothing is pushed until something says it is looking at it.
+
+    Past the cap the names it gave last are kept: a client lists the tile that
+    came on screen most recently last, so one scrolled back to keeps working.
     """
     try:
         payload = orjson.loads(frame)
@@ -1186,15 +1260,16 @@ def wanted_names(frame: str, only: set[str] | None) -> set[str] | None:
         return set()
     asked = {str(name) for name in names}
     if len(asked) > MAX_MEDIA_NAMES:
-        kept = list(dict.fromkeys(str(name) for name in names))[:MAX_MEDIA_NAMES]
+        kept = list(dict.fromkeys(str(name) for name in names))[-MAX_MEDIA_NAMES:]
         logger.warning(
             "A socket asked for live media on %d messages; only %d are served, "
-            "the first it named: %s",
+            "the last it named: %s",
             len(asked),
             MAX_MEDIA_NAMES,
             ", ".join(kept),
         )
         asked = set(kept)
+    asked -= hidden
     return asked if only is None else asked & only
 
 
@@ -1271,28 +1346,47 @@ def parse_frame(data: bytes) -> tuple[dict[str, Any], bytes] | None:
     return header, data[4 + length :]
 
 
+#: Why a device's frame was dropped, by the key it is logged under once per
+#: socket. Fixed, so nothing the device chose reaches the log.
+FRAME_REFUSALS = {
+    "not_a_frame": "not a frame",
+    "not_a_device": "only a device sends frames",
+    "no_engine": "the engine is not running",
+    "not_wired": "it names no message one of the device's ports publishes",
+    "not_image": "the port it names is not wired as an image",
+    "media_type": "its media type is not an image type",
+    "too_large": "it is larger than MAX_FRAME_BYTES",
+    "undeclared": "no flow declares its message",
+    "refused": "its message refused the value",
+}
+
+
 def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str:
     """Store a frame a device sent on its socket and publish it.
 
-    Returns why it was refused, or "" once it is published.
+    Returns the key in ``FRAME_REFUSALS`` it was refused for, or "" once it
+    is published.
 
     The same frame over HTTP is two requests, each authenticated: the upload,
     then the publish. Here the socket was authenticated once, at its
     handshake, and the frame is held and published in one step. Only a
     message one of the device's own ``in`` ports is wired to, as an ``image``,
     and the value is credited to the device node holding that port, as an
-    HTTP publish is.
+    HTTP publish is. The frame is named after that message.
     """
+    cap = settings.MAX_FRAME_BYTES
+    if cap and len(data) > cap:
+        return "too_large"
     parsed = parse_frame(data)
     if parsed is None or parsed[0].get("type") != "frame":
-        return "not a frame"
+        return "not_a_frame"
     header, payload = parsed
     name = str(header.get("name") or "")
     devices: DeviceStore | None = getattr(app.state, "device_store", None)
     store: Any = getattr(app.state, "artifact_store", None)
     controller: FlowController | None = getattr(app.state, "flow_controller", None)
     if devices is None or store is None or controller is None:
-        return "the engine is not running"
+        return "no_engine"
     binding = next(
         (
             b
@@ -1302,24 +1396,19 @@ def ingest_frame(app: Any, device: DeviceDef, data: bytes) -> str:
         None,
     )
     if binding is None:
-        return f"{name!r} is not a message one of its ports publishes"
+        return "not_wired"
     if binding.dtype != "image":
-        return f"its port {binding.port!r} is wired as {binding.dtype}, not image"
+        return "not_image"
     media_type = str(header.get("media_type") or "")
     if not _IMAGE_TYPE.match(media_type):
-        return f"{media_type!r} is not an image type"
-    cap = settings.MAX_ARTIFACT_BYTES
-    if cap and len(payload) > cap:
-        return f"a frame may be at most {cap} bytes"
-    reference = store.put(
-        [payload], name=binding.port, media_type=media_type, volatile=True
-    )
+        return "media_type"
+    reference = store.put([payload], name=name, media_type=media_type, volatile=True)
     try:
         controller.publish_message(name, reference, port_source(devices, device, name))
     except KeyError:
-        return f"no flow declares {name!r}"
-    except TypeError as exc:
-        return str(exc)
+        return "undeclared"
+    except TypeError:
+        return "refused"
     return ""
 
 
@@ -1407,6 +1496,11 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
     viewer = object()
     # Each reason a frame was refused for, logged once per socket.
     refused: set[str] = set()
+    # The cameras a person may not see. Looked at again with each batch, so
+    # a device switched to members, or back, takes effect on open sockets.
+    hidden = hidden_from(user, websocket.app, device=only is not None)
+    # A device's credential, checked again per frame once it may have moved.
+    credential = Credential(token, websocket.app, device) if device else None
 
     await websocket.accept()
 
@@ -1420,11 +1514,36 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
         if controller is None:
             return []
         payload = await run_in_threadpool(
-            snapshot_payload, controller, only, device, "binary"
+            snapshot_payload, controller, only, device, "binary", hidden
         )
         await websocket.send_text(orjson.dumps(payload).decode())
         return list(payload.get("demand", []))
 
+    async def catch_up() -> bool:
+        """Rescope a panel and send the snapshot again, demand included.
+
+        False once a device whose credential stopped standing is closed with
+        1008, which sends it to check: otherwise its frames would still be
+        taken here.
+        """
+        nonlocal only, device_def, wanted
+        if only is not None:
+            if device and await run_in_threadpool(revoked, token, websocket.app):
+                await websocket.close(code=1008)
+                return False
+            # The scope was resolved once, at the handshake. A panel pointed
+            # at another dashboard would otherwise fetch the new document and
+            # then draw tiles nothing ever updates. ``or set()`` because a
+            # panel that was deleted resolves to None, the same as a person's
+            # token — and that would widen this socket to everything on the
+            # bus.
+            only = panel_scope(token, websocket.app) or set()
+            device_def = _device_def(websocket.app, device)
+            wanted &= only
+            demand.want(viewer, wanted)
+        await send_snapshot()
+        return True
+
     told = await send_snapshot()
 
     async with event_bus.subscribe() as queue:
@@ -1477,23 +1596,27 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
                     if isinstance(frame, bytes):
                         # A camera's frame. Refused ones are dropped and the
                         # socket stays open: one bad frame is not a reason to
-                        # take the device's live values away.
+                        # take the device's live values away. A device that
+                        # was unpaired meanwhile is closed instead.
+                        if credential is not None and await credential.revoked():
+                            await websocket.close(code=1008)
+                            return
                         reason = (
                             await run_in_threadpool(
                                 ingest_frame, websocket.app, device_def, frame
                             )
                             if device_def is not None
-                            else "only a device sends frames"
+                            else "not_a_device"
                         )
                         if reason and reason not in refused:
                             refused.add(reason)
                             logger.warning(
                                 "Dropped a frame from %s: %s",
                                 device or "a person's socket",
-                                reason,
+                                FRAME_REFUSALS[reason],
                             )
                     else:
-                        asked = wanted_names(frame, only)
+                        asked = wanted_names(frame, only, hidden)
                         if asked is not None:
                             wanted = asked
                             demand.want(viewer, wanted)
@@ -1506,6 +1629,7 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
                                 devices=getattr(
                                     websocket.app.state, "device_store", None
                                 ),
+                                superuser=user.is_superuser,
                             )
                     receiver = asyncio.create_task(_receive(websocket))
                     if sender not in done:
@@ -1523,48 +1647,46 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
                     except asyncio.QueueEmpty:
                         break
 
+                # A panel this far behind lost events the bus dropped for it:
+                # its demand, a dashboard published or its own unpairing. It
+                # catches up once this batch is out, as after a dashboard.
+                resync = only is not None and event_bus.dropped(queue)
+                if only is None:
+                    fresh = hidden_from(user, websocket.app)
+                    if fresh != hidden:
+                        # A camera was opened or closed to members: the
+                        # snapshot gives back what may now be seen.
+                        hidden = fresh
+                        wanted -= hidden
+                        demand.want(viewer, wanted)
+                        resync = True
+
                 out: list[dict[str, Any]] = []
                 for event in batch:
-                    if event.get("type") == "device_demand":
+                    kind = event.get("type")
+                    if kind == "device_demand":
                         # For the device it names, and for nobody else.
                         if device and event.get("device") == device:
                             await websocket.send_text(
                                 demand_frame(list(event.get("ports") or []))
                             )
                         continue
-                    if only is not None and event.get("type") == "dashboard_changed":
-                        # An unpaired or deleted device is closed with 1008,
-                        # which sends it to check its credential: otherwise
-                        # its frames would still be taken here.
-                        if device and await run_in_threadpool(
-                            revoked, token, websocket.app
-                        ):
-                            await websocket.close(code=1008)
-                            return
-                        # The scope was resolved once, at the handshake. A
-                        # panel pointed at another dashboard would otherwise
-                        # fetch the new document and then draw tiles nothing
-                        # ever updates. ``or set()`` because a panel that was
-                        # deleted resolves to None, the same as a person's
-                        # token — and that would widen this socket to
-                        # everything on the bus.
-                        only = panel_scope(token, websocket.app) or set()
-                        device_def = _device_def(websocket.app, device)
-                        wanted &= only
-                        demand.want(viewer, wanted)
+                    if only is not None and kind == "dashboard_changed":
                         if out:
                             await _send(websocket, out)
                             out = []
-                        await send_snapshot()
-                    if only is not None and not event_for_panel(event, only):
-                        continue
-                    out.append(event)
+                        if not await catch_up():
+                            return
+                    if carried(event, only, hidden):
+                        out.append(event)
                 if out:
                     # The bytes first: a tile that has the frame when the
                     # value lands draws it in one pass rather than two.
                     for frame in media_frames(out, wanted, store):
                         await websocket.send_bytes(frame)
                     await _send(websocket, out)
+                if resync and not await catch_up():
+                    return
         except (WebSocketDisconnect, RuntimeError):
             # A peer that goes away mid-send takes the RuntimeError route
             # ("websocket.send after websocket.close") rather than the clean

backend/fluksio/api/routes/messages.py

@@ -12,7 +12,12 @@ from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.concurrency import run_in_threadpool
 from pydantic import BaseModel
 
-from fluksio.api.deps import FlowControllerDep, get_current_user
+from fluksio.api.deps import (
+    CAMERA_IS_SUPERUSERS,
+    FlowControllerDep,
+    HiddenDep,
+    get_current_user,
+)
 from fluksio.flow.devices import DeviceDef, DeviceStore
 from fluksio.flow.messages import flow_of
 from fluksio.flow.pipeline import ValueSource
@@ -66,9 +71,20 @@ class MessagePoints(BaseModel):
 
 
 @router.get("/", response_model=MessagesPublic)
-async def read_messages(controller: FlowControllerDep) -> Any:
-    """Every message any published flow declares, with its last value."""
+async def read_messages(controller: FlowControllerDep, hidden: HiddenDep) -> Any:
+    """Every message any published flow declares, with its last value.
+
+    A camera the caller may not see is listed without its value, so a tile
+    can still be bound to it.
+    """
     infos = await run_in_threadpool(controller.message_catalog)
+    if hidden:
+        infos = [
+            info.model_copy(update={"value": None, "ts": None})
+            if info.name in hidden
+            else info
+            for info in infos
+        ]
     return MessagesPublic(data=infos, count=len(infos))
 
 
@@ -131,8 +147,12 @@ async def publish_message(
 
 
 @router.get("/{name}/history", response_model=MessagePoints)
-def read_message_history(name: str, controller: FlowControllerDep) -> Any:
+def read_message_history(
+    name: str, controller: FlowControllerDep, hidden: HiddenDep
+) -> Any:
     """The series behind a chart. Numbers and flags — nothing else plots."""
+    if name in hidden:
+        raise HTTPException(status_code=403, detail=CAMERA_IS_SUPERUSERS)
     if controller.pipeline is None:
         return MessagePoints(message=name, numeric=False, points=[])
     series = controller.state.history(name)

backend/fluksio/cli.py

@@ -581,6 +581,9 @@ def cmd_serve(args: argparse.Namespace) -> int:
             # compressing each one again per open socket was a quarter of the
             # engine's time with four cameras.
             ws_per_message_deflate=False,
+            # A device's socket that never finishes closing would otherwise
+            # hold a stop, and a container's restart, for good.
+            timeout_graceful_shutdown=5,
         )
     )
     try:

backend/fluksio/cloud/connector.py

@@ -669,11 +669,11 @@ class CloudConnector:
         """
         from sqlmodel import Session
 
-        from fluksio.api.deps import user_from_token
+        from fluksio.api.deps import hidden_from, user_from_token
         from fluksio.api.routes.flows import (
+            carried,
             demand_frame,
             device_of,
-            event_for_panel,
             panel_scope,
             revoked,
             snapshot_payload,
@@ -719,12 +719,14 @@ class CloudConnector:
 
         await socket.send(_dump({"op": "ws_open_ok", "id": stream_id}))
 
+        # The cameras a person may not see, looked at again per event.
+        hidden = hidden_from(user, self._app, device=only is not None)
         controller = getattr(self._app.state, "flow_controller", None)
         told: list[str] = []
         if controller is not None:
             # A device here sends its frames over HTTP: the tunnel carries
             # text only.
-            snapshot = snapshot_payload(controller, only, device, "http")
+            snapshot = snapshot_payload(controller, only, device, "http", hidden)
             told = list(snapshot.get("demand", []))
             await socket.send(
                 _dump({"op": "ws_msg", "id": stream_id, "text": _dump(snapshot)})
@@ -743,7 +745,7 @@ class CloudConnector:
 
         def heard(text: str) -> None:
             nonlocal wanted
-            asked = wanted_names(text, only)
+            asked = wanted_names(text, only, hidden)
             if asked is not None:
                 wanted = asked
                 demand.want(viewer, asked)
@@ -754,6 +756,7 @@ class CloudConnector:
                     person=str(user.id) if only is None else "",
                     device=device,
                     devices=getattr(self._app.state, "device_store", None),
+                    superuser=user.is_superuser,
                 )
 
         self._listeners[stream_id] = heard
@@ -793,15 +796,37 @@ class CloudConnector:
                         if sender not in done:
                             continue
                     event = sender.result()
-                    if event.get("type") == "device_demand":
+                    # Behind far enough to lose events, a device catches up
+                    # as after a dashboard: the lost one may have been its
+                    # demand or its unpairing. A person whose cameras opened
+                    # or closed is sent what may be seen now.
+                    resync = only is not None and event_bus.dropped(queue)
+                    if only is None:
+                        fresh = hidden_from(user, self._app)
+                        if fresh != hidden:
+                            hidden = fresh
+                            wanted -= hidden
+                            demand.want(viewer, wanted)
+                            resync = True
+                    kind = event.get("type")
+                    if kind == "device_demand":
                         # For the device it names, and for nobody else.
                         if device and event.get("device") == device:
                             text = demand_frame(list(event.get("ports") or []))
                             await socket.send(
                                 _dump({"op": "ws_msg", "id": stream_id, "text": text})
                             )
+                    elif only is not None and kind == "dashboard_changed":
+                        resync = True
+                    elif carried(event, only, hidden):
+                        await socket.send(
+                            _dump(
+                                {"op": "ws_msg", "id": stream_id, "text": _dump(event)}
+                            )
+                        )
+                    if not resync:
                         continue
-                    if only is not None and event.get("type") == "dashboard_changed":
+                    if only is not None:
                         if device and revoked(token, self._app):
                             # Unpaired or deleted: 1008 sends it to check.
                             await socket.send(
@@ -817,24 +842,26 @@ class CloudConnector:
                         only = panel_scope(token, self._app) or set()
                         wanted &= only
                         demand.want(viewer, wanted)
-                        if controller is not None:
-                            catch_up = snapshot_payload(
-                                controller, only, device, "http"
+                    if controller is not None:
+                        catch_up = snapshot_payload(
+                            controller, only, device, "http", hidden
+                        )
+                        await socket.send(
+                            _dump(
+                                {
+                                    "op": "ws_msg",
+                                    "id": stream_id,
+                                    "text": _dump(catch_up),
+                                }
                             )
-                            await socket.send(
-                                _dump(
-                                    {
-                                        "op": "ws_msg",
-                                        "id": stream_id,
-                                        "text": _dump(catch_up),
-                                    }
-                                )
+                        )
+                    if kind == "dashboard_changed" and only is not None:
+                        # After the snapshot, as the device socket sends it.
+                        await socket.send(
+                            _dump(
+                                {"op": "ws_msg", "id": stream_id, "text": _dump(event)}
                             )
-                    elif only is not None and not event_for_panel(event, only):
-                        continue
-                    await socket.send(
-                        _dump({"op": "ws_msg", "id": stream_id, "text": _dump(event)})
-                    )
+                        )
         except asyncio.CancelledError:
             raise
         except Exception as exc:

backend/fluksio/core/config.py

@@ -221,6 +221,11 @@ class Settings(BaseSettings):
     #: Docker's default 64 MB `/dev/shm`: raise `shm_size` with it when
     #: workers publish frames.
     ARTIFACT_VOLATILE_BYTES: int = 48 * 1024 * 1024
+    #: The most one live frame may weigh: a volatile `PUT /artifacts` is
+    #: refused with 413 above it, and a device's frame on its socket is
+    #: dropped. Held in memory whole, which is why it is far below the
+    #: durable store's cap. Keep it under uvicorn's 16 MiB websocket limit.
+    MAX_FRAME_BYTES: int = 4 * 1024 * 1024
     # Without a Redis host the engine keeps its state in memory.
     REDIS_HOST: str | None = None
     REDIS_PORT: int = 6379

backend/fluksio/flow/artifacts.py

@@ -306,6 +306,32 @@ class VolatileStore:
                 return True
         return valid_digest(digest) and self._file(digest).is_file()
 
+    def spill(self, digest: str) -> None:
+        """Write a frame held in memory to the directory, for a worker here.
+
+        A node in this container opens a frame with ``load_artifact``, which
+        looks in the directory: the worker is another process and cannot see
+        this one's memory. Called before a call naming the frame. The
+        directory's timer trims it like a worker's own frames.
+        """
+        with self._lock:
+            data = self._frames.get(digest)
+        if data is None:
+            return
+        target = self._file(digest)
+        if target.exists():
+            return
+        target.parent.mkdir(parents=True, exist_ok=True)
+        handle = tempfile.NamedTemporaryFile(dir=target.parent, delete=False)
+        try:
+            with handle:
+                handle.write(data)
+            os.replace(handle.name, target)
+        except BaseException:
+            with contextlib.suppress(OSError):
+                os.unlink(handle.name)
+            raise
+
     @property
     def size(self) -> int:
         """Bytes held in memory."""

backend/fluksio/flow/devices.py

@@ -68,6 +68,9 @@ PORT_PATTERN = re.compile(r"^[a-z0-9_.]+$")
 #: The node type that stands for a device in a flow.
 DEVICE_TYPE = "device"
 
+#: The dtypes a device's ``viewers`` setting covers: what a camera sends.
+CAMERA_DTYPES = frozenset({"image", "video"})
+
 
 @dataclass(frozen=True)
 class Binding:
@@ -104,6 +107,7 @@ class DeviceBinding(BaseModel):
     port: str
     message: str
     direction: Literal["in", "out"]
+    dtype: str = ""
 
 
 def bindings_of(flow: FlowDef) -> list[Binding]:
@@ -149,6 +153,11 @@ class DeviceDef(BaseModel):
     #: touched rather than pointed at. A fact about the screen, not about any
     #: dashboard on it.
     touch: bool = False
+    #: Who may see what the device's camera sends, its ``image`` and ``video``
+    #: ``in`` messages, their bytes and its live view: superusers only, or
+    #: every member too. A device credential is bounded by its dashboards
+    #: instead, which a superuser assigned.
+    viewers: Literal["superusers", "members"] = "superusers"
     #: Bumped on every save. A save based on an older one is refused.
     version: int = 0
 
@@ -401,6 +410,9 @@ class DeviceStore:
         self._states_seen: (
             tuple[tuple[int, int, int], dict[str, DeviceState]] | None
         ) = None
+        #: The camera messages only superusers may see, at the flow store
+        #: revision they were worked out at. A device save commits there too.
+        self._restricted: tuple[int, frozenset[str]] | None = None
 
     def _file(self, device_id: str) -> Path:
         return self.root / f"{checked(device_id)}.json"
@@ -459,6 +471,30 @@ class DeviceStore:
                 found.append(binding)
         return found
 
+    def restricted(self) -> frozenset[str]:
+        """The messages only a superuser may see: what the cameras of devices
+        whose ``viewers`` are superusers publish, their ``image`` and ``video``
+        ``in`` ports.
+
+        Asked on every value a person's socket carries, so worked out again
+        only once the flow store commits, which every device save does.
+        """
+        revision = self.flows.revision
+        held = self._restricted
+        if held is not None and held[0] == revision:
+            return held[1]
+        wiring = self.wiring()
+        found = frozenset(
+            binding.message
+            for device in self.list()
+            if device.viewers == "superusers"
+            for binding in wiring.get(device.id, ())
+            if binding.direction == "in" and binding.dtype in CAMERA_DTYPES
+        )
+        # Keyed by the revision read before, so a commit meanwhile is seen.
+        self._restricted = (revision, found)
+        return found
+
     def io(self, device_id: str) -> list[DeviceIo]:
         """The ports as the device reads them."""
         return [
@@ -481,6 +517,7 @@ class DeviceStore:
                     port=b.port,
                     message=b.message,
                     direction=b.direction,
+                    dtype=b.dtype,
                 )
                 for b in self.wiring().get(device_id, [])
             ),

backend/fluksio/flow/events.py

@@ -2,7 +2,8 @@
 
 Nodes execute in a thread pool; websocket clients live on the event loop.
 Publishers are therefore thread-safe and never block: a subscriber that cannot
-keep up loses its oldest queued events rather than stalling the engine.
+keep up loses its oldest queued events rather than stalling the engine, and
+can ask whether it did (``dropped``) to catch up another way.
 """
 
 from __future__ import annotations
@@ -49,6 +50,9 @@ class EventBus:
         #: may already have moved past, and a device's credential reaches
         #: these as well as the current one.
         self.digests: dict[str, deque[str]] = {}
+        #: Subscribers that lost an event since they last asked. Touched on
+        #: the event loop only, by the flush and the subscribers.
+        self._dropped: set[asyncio.Queue[dict[str, Any]]] = set()
 
     def bind(self, loop: asyncio.AbstractEventLoop) -> None:
         """Attach the bus to the running event loop (called once at startup)."""
@@ -108,8 +112,20 @@ class EventBus:
                         queue.get_nowait()
                     except asyncio.QueueEmpty:
                         pass
+                    self._dropped.add(queue)
                 queue.put_nowait(event)
 
+    def dropped(self, queue: asyncio.Queue[dict[str, Any]]) -> bool:
+        """Whether this subscriber lost events since it last asked.
+
+        A device's socket that did catches up with a snapshot: what it lost
+        may have been its demand, or the event that tells it it was unpaired.
+        """
+        if queue in self._dropped:
+            self._dropped.discard(queue)
+            return True
+        return False
+
     @asynccontextmanager
     async def subscribe(self) -> AsyncIterator[asyncio.Queue[dict[str, Any]]]:
         """Yield a queue receiving every event published while subscribed."""
@@ -119,6 +135,7 @@ class EventBus:
             yield queue
         finally:
             self._subscribers.discard(queue)
+            self._dropped.discard(queue)
 
 
 event_bus = EventBus()

backend/fluksio/flow/live.py

@@ -172,19 +172,21 @@ class LiveSessions:
         person: str,
         device: str,
         devices: Any,
+        superuser: bool = False,
     ) -> 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.
+        the device whose credential does, one of them "". A person watches a
+        device whose ``viewers`` are members, or any as a ``superuser``. 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)
+            self._from_viewer(message, session_id, endpoint, person, devices, superuser)
         elif message.get("type") == "signal" and device:
             self._from_device(message, session_id, endpoint, device)
 
@@ -197,6 +199,7 @@ class LiveSessions:
         endpoint: Endpoint,
         who: str,
         devices: Any,
+        superuser: bool,
     ) -> None:
         device_id = str(message.get("device") or "")
         action = message.get("action")
@@ -205,7 +208,7 @@ class LiveSessions:
             if session is not None:
                 # Said twice, or somebody else's id: neither gets a second one.
                 return
-            reason = self._refusal(device_id, devices)
+            reason = self._refusal(device_id, devices, superuser)
             if reason:
                 endpoint.deliver(
                     _viewer_message(device_id, session_id, "bye", reason=reason)
@@ -289,8 +292,12 @@ class LiveSessions:
         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 ""."""
+    def _refusal(self, device_id: str, devices: Any, superuser: bool) -> str:
+        """Why this device cannot serve this viewer a live view now, or ""."""
+        if not superuser:
+            found = devices.find(device_id) if devices is not None else None
+            if found is None or found.viewers != "members":
+                return "forbidden"
         if not self._devices.get(device_id):
             return "offline"
         reported = devices.reported_ports(device_id) if devices is not None else None

backend/fluksio/flow/nodes/device.py

@@ -13,6 +13,8 @@ reads back as its ``io`` (see ``fluksio.flow.devices``).
 
 from __future__ import annotations
 
+import logging
+import threading
 from collections.abc import Iterable
 from typing import TYPE_CHECKING, Any
 
@@ -21,10 +23,18 @@ from pydantic import BaseModel, Field
 from fluksio.flow.devices import PORT_PATTERN, presence
 from fluksio.flow.messages import MessageSpec
 from fluksio.flow.nodes.base import Node
+from fluksio.flow.pipeline import ValueSource
 
 if TYPE_CHECKING:
     from fastapi import FastAPI
 
+logger = logging.getLogger(__name__)
+
+#: How long a device is away before its flags read false: the floor before
+#: its node is shown degraded (``controller.HEALTH_FLOOR``), so a reconnect
+#: or a restart of this instance leaves them as they were.
+OFFLINE_GRACE_S = 60.0
+
 
 def _never(**_: Any) -> None:
     """What the node would compute, were it ever run: nothing."""
@@ -45,7 +55,7 @@ class DeviceNode(Node):
             ),
         )
 
-    __slots__ = ("cfg",)
+    __slots__ = ("_away", "cfg")
 
     def __init__(
         self,
@@ -55,6 +65,8 @@ class DeviceNode(Node):
         name: str | None = None,
     ):
         self.cfg = self.Params.model_validate(params or {})
+        #: Counting down from the device going away to its flags reading false.
+        self._away: threading.Timer | None = None
         super().__init__(
             f=_never, requires=requires, provides=provides, params=params, name=name
         )
@@ -79,11 +91,49 @@ class DeviceNode(Node):
     async def stop(self, app: FastAPI | None = None) -> None:
         if self.cfg.device:
             presence.unwatch(self.cfg.device, self._presence)
+        self._stay()
 
     def _presence(self, online: bool) -> None:
         # Degraded rather than down: a panel switched off or asleep is a thing
         # to notice, not a fault of the flow wiring it.
         if online:
+            self._stay()
             self.report_health("ok")
         else:
             self.report_health("degraded", "Offline")
+            self._stay()
+            self._away = threading.Timer(OFFLINE_GRACE_S, self._gone)
+            self._away.daemon = True
+            self._away.start()
+
+    def _stay(self) -> None:
+        if self._away is not None:
+            self._away.cancel()
+            self._away = None
+
+    def _gone(self) -> None:
+        """The device has been away for the grace: its flags read false.
+
+        A ``bool`` it publishes says something is so right now, such as
+        ``person``, and nothing tells a device that went away to say it is
+        no longer so. Credited to this node, as the device's own values are.
+        """
+        pipeline = self._pipeline
+        if presence.online(self.cfg.device) or pipeline is None:
+            return
+        flags = {
+            spec.name: False
+            for spec in self.output_ports
+            if spec.name
+            and spec.dtype.value == "bool"
+            and pipeline.state.get(spec.name) is True
+        }
+        if not flags:
+            return
+        source = ValueSource(
+            kind="node", id=self.id, label=self.cfg.device, detail="offline"
+        )
+        try:
+            pipeline.publish(flags, source)
+        except Exception:
+            logger.exception("Could not reset the flags of '%s'", self.cfg.device)

backend/fluksio/flow/workers.py

@@ -24,13 +24,14 @@ import subprocess
 import sys
 import threading
 import time
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
 from pathlib import Path
 from typing import Any
 
 import orjson
 from fluksio_worker import worker_main as _worker_main
 
+from fluksio.flow.artifacts import is_reference
 from fluksio.flow.events import EventBus
 from fluksio.flow.nodes.base import NodeOutputError
 
@@ -129,6 +130,19 @@ def _remote_class(name: str) -> type[RemoteError]:
     return cls
 
 
+def _digests(value: Any) -> Iterator[str]:
+    """Every artifact digest a node's inputs name, however deep."""
+    if isinstance(value, dict):
+        if is_reference(value):
+            yield str(value["digest"])
+            return
+        for item in value.values():
+            yield from _digests(item)
+    elif isinstance(value, list):
+        for item in value:
+            yield from _digests(item)
+
+
 class _Worker:
     """One subprocess, and the framing of one request/response over its pipes."""
 
@@ -237,10 +251,16 @@ class PythonWorkerPool:
         events: EventBus | None = None,
         env: dict[str, str] | None = None,
         max_rss_mb: int = 0,
+        spill: Callable[[str], None] | None = None,
     ) -> None:
         self.python = python
         self.size = size
         self.events = events
+        # Called with each artifact a call's inputs name, before the call: a
+        # frame the engine holds in memory is written where a worker in this
+        # container finds it (``VolatileStore.spill``), since it is another
+        # process.
+        self.spill = spill
         # A worker holding more than this when its call returns is retired
         # rather than kept warm; 0 keeps them all. A library like JAX keeps what
         # it allocated for as long as the process lives, and warm is the point
@@ -307,6 +327,7 @@ class PythonWorkerPool:
                     events=self.events,
                     env={**self.env, **extra},
                     max_rss_mb=self.max_rss_mb,
+                    spill=self.spill,
                 )
                 child._generation = self._generation
                 child.start()
@@ -593,6 +614,9 @@ class PythonWorkerPool:
         run_id: str = "",
         on_event: Callable[[dict[str, Any]], None] | None = None,
     ) -> Any:
+        if self.spill is not None:
+            for digest in _digests(kwargs):
+                self.spill(digest)
         response = self._request(
             {
                 "op": "run",

backend/fluksio/main.py

@@ -329,6 +329,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
                 # what it needs overrides this; one that says nothing gets a share.
                 **fair_share_env(accountant.cpus, settings.FLOW_MAX_WORKERS),
             },
+            # The engine's own frames are in its memory, which a worker cannot
+            # read: each one a call names is written beside the workers'.
+            spill=volatile.spill if volatile else None,
         )
         pool.start()
         # On a thread: stopping a worker waits up to five seconds on each child,

backend/tests/api/routes/test_camera_access.py

@@ -0,0 +1,228 @@
+"""Who may see a device's camera: superusers only, or every member too.
+
+openpanel ROADMAP C4. ``viewers`` on a device covers its ``image`` and
+``video`` messages, their bytes, frames on the socket and its live view. A
+device credential is bounded by its dashboards instead.
+"""
+
+from typing import Any
+
+import pytest
+from fastapi.testclient import TestClient
+
+from fluksio.core.config import settings
+from fluksio.flow.demand import demand
+from tests.api.routes.test_cameras import (
+    CAMERA_NODE,
+    REPORT,
+    _ticket,
+    frame,
+    wait_for,
+)
+from tests.api.routes.test_devices import (
+    FLOWS,
+    _dashboard_with,
+    _pair,
+    _panels,
+    _publish_flow,
+)
+
+API = settings.API_V1_STR
+PREFIX = f"{API}/devices"
+SOCKET = f"{API}/flows/ws"
+
+
+@pytest.fixture
+def camera(client: TestClient, superuser_token_headers: dict[str, str]):
+    """``cam1`` with a camera wired in ``cams``, and a frame on it."""
+    _panels(client, superuser_token_headers, {"panels": [{"id": "cam1"}]})
+    device = _pair(client, superuser_token_headers, "cam1")
+    reported = client.put(f"{PREFIX}/cam1/report", headers=device, json=REPORT)
+    assert reported.status_code == 200, reported.text
+    _publish_flow(client, superuser_token_headers, "cams", [CAMERA_NODE])
+    controller = client.app.state.flow_controller
+    with client.websocket_connect(SOCKET, headers=device) as ws:
+        ws.receive_json()
+        for index in range(3):
+            ws.send_bytes(frame("cams.camera", b"\xff\xd8 frame %d" % index))
+        wait_for(
+            lambda: (controller.values().get("cams.camera") or {})
+            .get("value", {})
+            .get("size")
+            == len(b"\xff\xd8 frame 2")
+        )
+    controller.publish_message("cams.person", True)
+    yield device
+    client.delete(f"{FLOWS}/cams", headers=superuser_token_headers)
+    client.delete(f"{PREFIX}/cam1", headers=superuser_token_headers)
+
+
+def _viewers(client: TestClient, headers: dict[str, str], viewers: str) -> None:
+    current = next(
+        one
+        for one in client.get(f"{PREFIX}/", headers=headers).json()["data"]
+        if one["id"] == "cam1"
+    )
+    body = {
+        key: current[key] for key in ("id", "title", "dashboards", "touch", "version")
+    }
+    saved = client.put(
+        f"{PREFIX}/cam1", headers=headers, json={**body, "viewers": viewers}
+    )
+    assert saved.status_code == 200, saved.text
+
+
+def _surfaces(client: TestClient, headers: dict[str, str]) -> dict[str, Any]:
+    """What one account gets of the camera on every surface but the socket."""
+    controller = client.app.state.flow_controller
+    digest = controller.values()["cams.camera"]["value"]["digest"]
+    catalog = {
+        info["name"]: info
+        for info in client.get(f"{API}/messages/", headers=headers).json()["data"]
+    }
+    return {
+        "catalog": catalog["cams.camera"]["value"] is not None,
+        "person": catalog["cams.person"]["value"],
+        "history": client.get(
+            f"{API}/messages/cams.camera/history", headers=headers
+        ).status_code,
+        "flow_history": client.get(
+            f"{FLOWS}/cams/history/camera", headers=headers
+        ).status_code,
+        "state": "cams.camera"
+        in client.get(f"{FLOWS}/cams/state", headers=headers).json()["values"],
+        "artifact": client.get(
+            f"{API}/artifacts/{digest}", headers=headers
+        ).status_code,
+        "ice": client.get(f"{PREFIX}/cam1/ice", headers=headers).status_code,
+    }
+
+
+def _live_start(client: TestClient, headers: dict[str, str]) -> str:
+    """The reason a viewer's live view of cam1 ends with right away."""
+    with client.websocket_connect(_ticket(client, headers)) as ws:
+        ws.receive_json()
+        ws.send_json(
+            {"type": "live", "device": "cam1", "session": "s-access", "action": "start"}
+        )
+        while True:
+            message = ws.receive_json()
+            if message.get("type") == "live":
+                return str(message.get("reason"))
+
+
+@pytest.mark.usefixtures("camera")
+def test_a_camera_is_for_superusers_until_opened_to_members(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    normal_user_token_headers: dict[str, str],
+) -> None:
+    member, admin = normal_user_token_headers, superuser_token_headers
+    open_to_all = {
+        "catalog": True,
+        "person": True,
+        "history": 200,
+        "flow_history": 200,
+        "state": True,
+        "artifact": 200,
+        "ice": 200,
+    }
+    assert _surfaces(client, admin) == open_to_all
+    assert _surfaces(client, member) == {
+        "catalog": False,
+        # Not a camera: its bool is anyone's.
+        "person": True,
+        "history": 403,
+        "flow_history": 403,
+        "state": False,
+        "artifact": 403,
+        "ice": 403,
+    }
+    # The frames before the current one too.
+    store = client.app.state.artifact_store
+    older = store.put([b"\xff\xd8 older"], media_type="image/jpeg", volatile=True)
+    client.app.state.flow_controller.publish_message("cams.camera", older)
+    newer = store.put([b"\xff\xd8 newer"], media_type="image/jpeg", volatile=True)
+    client.app.state.flow_controller.publish_message("cams.camera", newer)
+    fetched = client.get(f"{API}/artifacts/{older['digest']}", headers=member)
+    assert fetched.status_code == 403
+    assert _live_start(client, member) == "forbidden"
+    assert _live_start(client, admin) == "offline"
+
+    _viewers(client, admin, "members")
+    assert _surfaces(client, member) == open_to_all
+    assert _live_start(client, member) == "offline"
+
+
+@pytest.mark.usefixtures("camera")
+def test_a_members_socket_carries_no_frames_of_a_closed_camera(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    normal_user_token_headers: dict[str, str],
+    camera: dict[str, str],
+) -> None:
+    """Neither its values, nor the bytes, nor demand on the device; and the
+    socket is given them once the camera is opened."""
+    with client.websocket_connect(_ticket(client, normal_user_token_headers)) as ws:
+        snapshot = ws.receive_json()
+        assert "cams.camera" not in snapshot["values"]
+        assert "cams.person" in snapshot["values"]
+        ws.send_json({"type": "media", "names": ["cams.camera", "cams.person"]})
+        wait_for(lambda: demand.viewed() == {"cams.person"})
+        assert demand.ports("cam1") == []
+
+        with client.websocket_connect(SOCKET, headers=camera) as device:
+            device.receive_json()
+            device.send_bytes(frame("cams.camera", b"\xff\xd8 not for members"))
+            client.app.state.flow_controller.publish_message("cams.person", False)
+            # The person's flag arrives, the frame before it did not.
+            while True:
+                message = ws.receive_json()
+                names = [
+                    event.get("name")
+                    for event in message.get("events", [message])
+                    if event.get("type") == "message_value"
+                ]
+                assert "cams.camera" not in names
+                if "cams.person" in names:
+                    break
+
+        _viewers(client, superuser_token_headers, "members")
+        while True:
+            message = ws.receive_json()
+            if message.get("type") == "snapshot":
+                break
+        assert "cams.camera" in message["values"]
+
+
+@pytest.mark.usefixtures("camera")
+def test_a_wall_panel_shows_the_camera_its_dashboards_carry(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    """Assigned by a superuser, so the panel's own scope is what bounds it."""
+    _dashboard_with(
+        client,
+        superuser_token_headers,
+        "panel_cams",
+        [
+            {
+                "id": "w_cam",
+                "type": "media",
+                "config": {"message": "cams.camera", "dtype": "image"},
+            }
+        ],
+    )
+    _panels(
+        client,
+        superuser_token_headers,
+        {"panels": [{"id": "cam1"}, {"id": "hall", "dashboards": ["panel_cams"]}]},
+    )
+    panel = _pair(client, superuser_token_headers, "hall")
+    try:
+        value = client.app.state.flow_controller.values()["cams.camera"]["value"]
+        fetched = client.get(f"{API}/artifacts/{value['digest']}", headers=panel)
+        assert fetched.status_code == 200
+        with client.websocket_connect(SOCKET, headers=panel) as ws:
+            assert "cams.camera" in ws.receive_json()["values"]
+    finally:
+        client.delete(f"{PREFIX}/hall", headers=superuser_token_headers)

backend/tests/api/routes/test_cameras.py

@@ -244,10 +244,14 @@ def test_a_frame_the_device_may_not_send_is_dropped(
     current = controller.values()["cams.camera"]["value"]
     data = client.app.state.artifact_store.volatile.get(current["digest"])
     assert data == b"\xff\xd8 after"
+    from fluksio.api.routes.flows import FRAME_REFUSALS
+
+    # By fixed keys, never in the words the device chose.
     said = [r.getMessage() for r in caplog.records if "Dropped a frame" in r.message]
-    assert any("'person' is wired as bool" in line for line in said)
-    assert any("not a message one of its ports publishes" in line for line in said)
-    assert any("only a device sends frames" in line for line in said)
+    for key in ("not_image", "not_wired", "media_type", "not_a_frame"):
+        assert f"Dropped a frame from cam1: {FRAME_REFUSALS[key]}" in said
+    assert any(FRAME_REFUSALS["not_a_device"] in line for line in said)
+    assert not any("elsewhere" in line or "text/html" in line for line in said)
 
 
 def test_an_unpaired_device_is_closed_and_its_frames_stop(
@@ -264,6 +268,106 @@ def test_an_unpaired_device_is_closed_and_its_frames_stop(
         assert closed.value.code == 1008
 
 
+def test_a_frame_past_the_cap_is_dropped(
+    client: TestClient, camera, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    controller = client.app.state.flow_controller
+    monkeypatch.setattr(settings, "MAX_FRAME_BYTES", 1000)
+    with client.websocket_connect(SOCKET, headers=camera) as device:
+        device.receive_json()
+        device.send_bytes(frame("cams.camera", b"\xff" * 2000))
+        device.send_bytes(frame("cams.camera", b"\xff\xd8 small"))
+        wait_for(lambda: (controller.values().get("cams.camera") or {}).get("value"))
+    digest = controller.values()["cams.camera"]["value"]["digest"]
+    store = client.app.state.artifact_store
+    assert store.volatile.get(digest) == b"\xff\xd8 small"
+
+
+def test_a_frame_is_named_after_its_message(client: TestClient, camera) -> None:
+    controller = client.app.state.flow_controller
+    with client.websocket_connect(SOCKET, headers=camera) as device:
+        device.receive_json()
+        device.send_bytes(frame("cams.camera", b"\xff\xd8 named"))
+        value = wait_for(
+            lambda: (controller.values().get("cams.camera") or {}).get("value")
+        )
+    assert value["name"] == "cams.camera"
+
+
+def test_an_unpaired_device_is_closed_at_its_next_frame(
+    client: TestClient, camera
+) -> None:
+    """Without the event an unpair sends, which a full queue may have lost."""
+    from starlette.websockets import WebSocketDisconnect
+
+    with client.websocket_connect(SOCKET, headers=camera) as device:
+        device.receive_json()
+        device.send_bytes(frame("cams.camera", b"\xff\xd8 one"))
+        client.app.state.device_store.bump("cam1")
+        device.send_bytes(frame("cams.camera", b"\xff\xd8 two"))
+        with pytest.raises(WebSocketDisconnect) as closed:
+            while True:
+                device.receive_json()
+        assert closed.value.code == 1008
+    store = client.app.state.artifact_store
+    current = client.app.state.flow_controller.values()["cams.camera"]["value"]
+    assert store.volatile.get(current["digest"]) == b"\xff\xd8 one"
+
+
+def test_a_device_that_lost_events_catches_up(
+    client: TestClient, camera, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Its demand and its unpairing come again with a snapshot, after the
+    events it did get."""
+    from starlette.websockets import WebSocketDisconnect
+
+    controller = client.app.state.flow_controller
+    with client.websocket_connect(SOCKET, headers=camera) as device:
+        assert device.receive_json()["demand"] == []
+        # Demand changes while the bus drops what the socket has queued.
+        monkeypatch.setattr(event_bus, "dropped", lambda queue: True)
+        demand.want("someone", {"cams.camera"})
+        try:
+            seen = []
+            while not seen or seen[-1].get("type") != "snapshot":
+                seen.append(device.receive_json())
+            assert seen[-1]["demand"] == ["camera"]
+        finally:
+            demand.drop("someone")
+
+        # And an unpair it never heard of closes it at the next batch.
+        client.app.state.device_store.bump("cam1")
+        controller.publish_message("cams.person", True)
+        with pytest.raises(WebSocketDisconnect) as closed:
+            while True:
+                device.receive_json()
+        assert closed.value.code == 1008
+
+
+def test_a_report_is_bounded_once_its_schemas_are_in_place(
+    client: TestClient, camera
+) -> None:
+    """Thirty pins naming one big schema are thirty copies of it, kept."""
+    from fluksio.api.routes.devices import REPORT_MAX, REPORT_STORED_MAX
+
+    schema = {
+        "type": "object",
+        "properties": {f"p{i}": {"type": "integer"} for i in range(250)},
+    }
+    pins = [
+        {"port": f"gpio.{i}", "direction": "in", "dtype": "bool", "config": "pin"}
+        for i in range(60)
+    ]
+    report = {**REPORT, "config_schemas": {"pin": schema}, "ports": pins}
+    raw = orjson.dumps(report)
+    assert len(raw) < REPORT_MAX
+    answer = client.put(f"{PREFIX}/cam1/report", headers=camera, content=raw)
+    assert answer.status_code == 413, answer.text
+    kept = client.app.state.device_store.state("cam1").report
+    assert len(orjson.dumps(kept)) < REPORT_STORED_MAX
+    assert kept["board"] == "nano"
+
+
 def test_a_screen_may_fetch_the_last_few_frames(client: TestClient, camera) -> None:
     """A remote screen fetching frame by frame is often one behind."""
     controller = client.app.state.flow_controller
@@ -290,7 +394,8 @@ def test_a_viewer_asks_for_a_few_cameras_at_most() -> None:
 
     names = [f"cams.c{i}" for i in range(MAX_MEDIA_NAMES + 3)]
     asked = wanted_names(json.dumps({"type": "media", "names": names}), None)
-    assert asked == set(names[:MAX_MEDIA_NAMES])
+    # The newest, which a page lists last: a tile scrolled back to keeps working.
+    assert asked == set(names[-MAX_MEDIA_NAMES:])
 
 
 def test_viewers_through_the_portal_count_too(

backend/tests/api/routes/test_flows.py

@@ -1,5 +1,6 @@
 import asyncio
 from datetime import UTC, datetime
+from types import SimpleNamespace
 
 import pytest
 from fastapi.testclient import TestClient
@@ -519,7 +520,9 @@ async def test_the_live_socket_ends_when_its_client_goes(
     because only SIGKILL ended the process. This returning is what ends that.
     """
     monkeypatch.setattr(
-        flows_route, "user_from_token", lambda session, token, **_: object()
+        flows_route,
+        "user_from_token",
+        lambda session, token, **_: SimpleNamespace(id="someone", is_superuser=True),
     )
 
     class Peer:

backend/tests/api/routes/test_live.py

@@ -337,6 +337,7 @@ def test_a_full_outbox_ends_the_session_instead_of_dropping() -> None:
             person="someone",
             device="",
             devices=Devices(),
+            superuser=True,
         )
         assert registry.count("cam9") == 1
         # The viewer never reads, and the device trickles on.

backend/tests/flow/test_device_node.py

@@ -3,6 +3,8 @@
 import asyncio
 from pathlib import Path
 
+import pytest
+
 from fluksio.flow.alerts import Alert, AlertManager, AlertsConfig, Channel, Rule
 from fluksio.flow.controller import HEALTH_FLOOR, FlowController
 from fluksio.flow.devices import presence
@@ -165,3 +167,41 @@ def test_a_device_away_after_a_restart_alerts_nobody(tmp_path: Path) -> None:
     ]
     assert sent == []
     presence.forget("wall")
+
+
+def test_a_device_gone_for_the_grace_reads_false(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A flag such as `person` stops saying someone is there once the device
+    has been away for the grace, and a device back in time keeps it."""
+    from fluksio.flow.nodes import device as device_node
+
+    monkeypatch.setattr(device_node, "OFFLINE_GRACE_S", 0.1)
+    controller = _controller(tmp_path)
+
+    def bell() -> object:
+        return controller.values("hall")["hall.bell"]["value"]
+
+    async def scenario() -> list[object]:
+        await controller.reload()
+        presence.opened("wall")
+        controller.publish_message("hall.bell", True)
+        seen = [bell()]
+        # Back within the grace: nothing changes.
+        presence.closed("wall")
+        presence.opened("wall")
+        await asyncio.sleep(0.25)
+        seen.append(bell())
+        # Away for good.
+        presence.closed("wall")
+        await asyncio.sleep(0.05)
+        seen.append(bell())
+        await asyncio.sleep(0.25)
+        seen.append(bell())
+        # What reads it hears so too.
+        seen.append(controller.values("hall")["hall.rung"]["value"])
+        await controller._teardown()
+        presence.forget("wall")
+        return seen
+
+    assert asyncio.run(scenario()) == [True, True, True, False, False]

backend/tests/flow/test_event_bus.py

@@ -69,6 +69,25 @@ def test_a_slow_subscriber_still_loses_its_oldest() -> None:
     assert oldest == "n50"
 
 
+def test_a_subscriber_that_lost_events_is_told_once() -> None:
+    """What a device's socket catches up by: it may have lost its demand."""
+
+    async def run() -> tuple[bool, bool, bool]:
+        bus = EventBus()
+        bus.bind(asyncio.get_running_loop())
+        async with bus.subscribe() as kept, bus.subscribe() as behind:
+            for i in range(QUEUE_SIZE):
+                bus.publish(_event(i))
+            await asyncio.sleep(0.01)
+            while not kept.empty():
+                kept.get_nowait()
+            bus.publish(_event(QUEUE_SIZE))
+            await asyncio.sleep(0.01)
+            return bus.dropped(kept), bus.dropped(behind), bus.dropped(behind)
+
+    assert asyncio.run(run()) == (False, True, False)
+
+
 def test_nothing_is_buffered_with_nobody_listening() -> None:
     """An engine with no browser open holds no events."""
 

backend/tests/flow/test_workers.py

@@ -6,9 +6,9 @@ import time
 from collections.abc import Iterator
 
 import pytest
-from fluksio_worker.worker_main import ARTIFACT_DIR_ENV
+from fluksio_worker.worker_main import ARTIFACT_DIR_ENV, ARTIFACT_VOLATILE_DIR_ENV
 
-from fluksio.flow.artifacts import ArtifactStore
+from fluksio.flow.artifacts import ArtifactStore, VolatileStore
 from fluksio.flow.controller import EmitSink, FlowController
 from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.nodes import Node
@@ -473,6 +473,43 @@ def test_a_node_saves_and_loads_an_artifact(tmp_path):
         pool.stop()
 
 
+def test_a_node_opens_a_camera_frame_held_in_memory(tmp_path):
+    """A frame from a device's socket is in the engine's memory only, which a
+    worker cannot read: the pool writes it out for the call naming it."""
+    store = ArtifactStore(tmp_path / "artifacts")
+    store.volatile = VolatileStore(tmp_path / "ring", 1024 * 1024)
+    pool = PythonWorkerPool(
+        python=sys.executable,
+        size=1,
+        env={
+            ARTIFACT_DIR_ENV: str(store.root),
+            ARTIFACT_VOLATILE_DIR_ENV: str(store.volatile.root),
+        },
+        spill=store.volatile.spill,
+    )
+    pool.start()
+    try:
+        frame = store.put([b"\xff\xd8 a frame"], media_type="image/jpeg", volatile=True)
+        assert not store.volatile._file(frame["digest"]).exists()
+        loaded = pool.run(
+            "demo",
+            "look",
+            "import fluksio\n"
+            "def process(snapshot, earlier):\n"
+            "    with open(fluksio.load_artifact(snapshot), 'rb') as f:\n"
+            "        return {'bytes': f.read().decode('latin-1')}\n",
+            # Nested, as a node handing on a list of frames would get them.
+            {"snapshot": frame, "earlier": [frame]},
+            "demo.look",
+            timeout=10,
+        )
+        assert loaded == {"bytes": "\xff\xd8 a frame"}
+        # On the ring's directory, which the engine trims, never the volume.
+        assert store.path(frame["digest"]) is None
+    finally:
+        pool.stop()
+
+
 def test_a_node_passes_audio_to_the_next_one(tmp_path):
     # The media path end to end: one node writes a clip and declares what kind
     # of bytes it is, the next opens it. Only the reference travels.

backend/tests/test_artifacts.py

@@ -240,6 +240,55 @@ def test_an_upload_may_ask_for_the_ring(
         store.volatile = previous
 
 
+def test_a_ring_never_holds_more_than_its_bound_and_one_frame(tmp_path):
+    """Frames of every size, the bound never passed by more than the newest."""
+    ring = VolatileStore(tmp_path / "ring", limit_bytes=10_000)
+    sizes = [700, 9_000, 30, 12_000, 4_000, 4_000, 2_500, 9_999, 1, 6_000] * 3
+    for index, size in enumerate(sizes):
+        ring.put([bytes([index % 256]) * size])
+        assert ring.size <= ring.limit_bytes or len(ring._frames) == 1
+        assert ring.size <= ring.limit_bytes + size
+
+
+def test_a_volatile_upload_is_held_to_the_frame_cap(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A frame is held in memory whole, so it is refused past a few MB before
+    any of it is read, and a chunked one as soon as it passes the cap."""
+    store = client.app.state.artifact_store
+    previous = store.volatile
+    store.volatile = VolatileStore(
+        store.root.parent / "test-ring", limit_bytes=1_000_000
+    )
+    monkeypatch.setattr(settings, "MAX_FRAME_BYTES", 1000)
+    url = f"{settings.API_V1_STR}/artifacts?volatile=1&media_type=image%2Fjpeg"
+    try:
+        declared = client.put(url, headers=superuser_token_headers, content=b"x" * 1001)
+        assert declared.status_code == 413
+
+        def chunks():
+            for _ in range(5):
+                yield b"x" * 400
+
+        chunked = client.put(url, headers=superuser_token_headers, content=chunks())
+        assert chunked.status_code == 413
+        assert store.volatile.size == 0
+
+        fits = client.put(url, headers=superuser_token_headers, content=b"x" * 1000)
+        assert fits.status_code == 200, fits.text
+        # The durable store keeps its own, larger cap.
+        kept = client.put(
+            f"{settings.API_V1_STR}/artifacts",
+            headers=superuser_token_headers,
+            content=b"y" * 5000,
+        )
+        assert kept.status_code == 200, kept.text
+    finally:
+        store.volatile = previous
+
+
 def test_a_type_that_can_run_script_is_served_sandboxed(
     client: TestClient, superuser_token_headers: dict[str, str]
 ) -> None:

docker/compose.dev.yml

@@ -2,12 +2,21 @@
 # published host ports, hot reload and the mail/browser test helpers.
 # Never used in production — the Makefile only layers it for `make dev*`.
 
+# Every container's log rotates: json-file, at most 3 files of 10 MB each.
+# Docker's default keeps one file that grows until the host's disk is full.
+x-logging: &logging
+  driver: json-file
+  options:
+    max-size: "10m"
+    max-file: "3"
+
 services:
 
   # Local Traefik. Services stay reachable both on their published ports and
   # on http://api.localhost / http://app.localhost via the routers in
   # compose.yml. Dashboard at http://localhost:8090.
   proxy:
+    logging: *logging
     # A patch tag, so a host holding an older cached 3.6 pulls this one. 3.6.25
     # knows --accesslog.fields.queryparameters, which older 3.6 refuses.
     image: traefik:3.6.25
@@ -61,6 +70,7 @@ services:
   # them, and 1883 in particular tends to be taken by the real broker on a
   # machine that runs this alongside anything else.
   mosquitto:
+    logging: *logging
     image: eclipse-mosquitto:2
     restart: "no"
     # The config the image ships for exactly this: listener on 1883, anonymous
@@ -68,6 +78,7 @@ services:
     command: mosquitto -c /mosquitto-no-auth.conf
 
   influxdb:
+    logging: *logging
     image: influxdb:2.7
     restart: "no"
     # Dev credentials, and no volume: the container comes up initialised and a
@@ -151,6 +162,7 @@ services:
       API_DOCS_ENABLED: "true"
 
   mailcatcher:
+    logging: *logging
     image: schickling/mailcatcher
     ports:
       - "1080:1080"
@@ -162,6 +174,7 @@ services:
     restart: "no"
 
   playwright:
+    logging: *logging
     build:
       context: ..
       dockerfile: frontend/Dockerfile.playwright

docker/compose.traefik.yml

@@ -1,5 +1,14 @@
+# Every container's log rotates: json-file, at most 3 files of 10 MB each.
+# Docker's default keeps one file that grows until the host's disk is full.
+x-logging: &logging
+  driver: json-file
+  options:
+    max-size: "10m"
+    max-file: "3"
+
 services:
   traefik:
+    logging: *logging
     # A patch tag, so a host holding an older cached 3.6 pulls this one. 3.6.25
     # knows --accesslog.fields.queryparameters, which older 3.6 refuses.
     image: traefik:3.6.25

docker/compose.yml

@@ -6,11 +6,20 @@
 # Paths are relative to this directory, which compose uses as the project
 # directory (build contexts therefore point at `..`, the repo root).
 
+# Every container's log rotates: json-file, at most 3 files of 10 MB each.
+# Docker's default keeps one file that grows until the host's disk is full.
+x-logging: &logging
+  driver: json-file
+  options:
+    max-size: "10m"
+    max-file: "3"
+
 services:
 
   # Umami's database, and nothing else's. The engine keeps its own state in
   # SQLite on the flow-data volume, which is why this is behind a profile now.
   db:
+    logging: *logging
     image: postgres:18
     container_name: fluksio-db
     restart: always
@@ -36,6 +45,7 @@ services:
       - POSTGRES_DB=${POSTGRES_DB:-app}
 
   redis:
+    logging: *logging
     image: redis:8.2-alpine
     container_name: fluksio-redis
     restart: always
@@ -57,6 +67,7 @@ services:
       - app-redis-data:/data
 
   adminer:
+    logging: *logging
     image: adminer
     container_name: fluksio-adminer
     restart: always
@@ -73,6 +84,7 @@ services:
       - ADMINER_DESIGN=pepa-linha-dark
 
   backend:
+    logging: *logging
     image: '${DOCKER_IMAGE_BACKEND?Variable not set}:${TAG-latest}'
     container_name: fluksio-api
     restart: always
@@ -177,6 +189,7 @@ services:
   # Static documentation site (zensical), served at docs.${DOMAIN}. Content and
   # config live in ../docs/ and ../zensical.toml; `make docs-serve` previews it.
   docs:
+    logging: *logging
     container_name: fluksio-docs
     restart: always
     security_opt:
@@ -211,6 +224,7 @@ services:
   # authority: on a machine that runs anything besides this stack, that is a
   # deliberate operator decision. `make up` opts in; the dev stacks do not.
   autoheal:
+    logging: *logging
     image: willfarrell/autoheal:latest
     container_name: fluksio-autoheal
     profiles: ["autoheal"]
@@ -236,6 +250,7 @@ services:
   # goes in through a config file rather than the command line, which `ps`
   # shows. See DEPLOY.md for ports, DNS and certificates.
   coturn:
+    logging: *logging
     image: coturn/coturn:4.6.3
     container_name: fluksio-coturn
     profiles: ["turn"]
@@ -282,6 +297,7 @@ services:
   # `analytics` profile, so a plain `make up` never starts it; bring it up with
   # `make umami`, which also provisions its database.
   umami:
+    logging: *logging
     container_name: fluksio-umami
     # Pinned by digest so a bump is a deliberate step: this is an internet-facing
     # third-party dashboard, not a library.

docs/code/api.md

@@ -174,7 +174,10 @@ curl "$FLUKSIO/artifacts/<digest>" \
 The upload returns a reference with the content digest. Identical bytes are
 stored once. Node code uses `save_artifact` and `load_artifact` instead.
 `?volatile=true` holds the bytes in the engine's memory for a live frame, where
-newer frames push them out, instead of storing them.
+newer frames push them out, instead of storing them. Such a frame is at most
+`MAX_FRAME_BYTES` (4 MB), and a larger one is refused with 413. A camera only
+superusers may see answers 403 to anyone else for the frames its messages
+hold, the last 8 included (see [Who may watch a camera](../interface/dashboards.md#who-may-watch-a-camera)).
 
 A superuser can remove stored bytes with `DELETE /artifacts/<digest>`. While
 a run, a current message or a flow input's initial value still references the
@@ -200,8 +203,8 @@ on its own, so this is for reclaiming space sooner.
 | `GET` | `/devices/pair/{code}/device` | what is waiting on a code, before it is approved |
 | `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 |
+| `PUT` | `/devices/{id}/report` | what the device has, from its own credential only; at most 16 KB, and 64 KB once its named `config_schemas` are put in place (413 above either), 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; 403 for whoever may not watch its camera, and 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.
@@ -254,7 +257,7 @@ The socket sends a snapshot followed by node, value, health, and run events.
 
 A client asks for the bytes of the media it shows with
 `{"type": "media", "names": [...]}`, which replaces what it asked for before,
-for at most 8 messages. Each frame held in memory then arrives as a binary
+for at most 8 messages, the last 8 it names when it names more. Each frame held in memory then arrives as a binary
 message before the value naming it: four bytes of header length (big endian),
 a JSON header `{"type": "media", "name", "digest", "media_type", "ts"}`, then
 the bytes. Messages are sent without per-message compression.
@@ -264,17 +267,26 @@ A device's socket is bounded to its own messages. Its snapshot carries
 `"uplink"`, `"binary"` on its own socket and `"http"` through the portal. When
 that set changes it is sent `{"type": "demand", "ports": [...]}`. A port has
 demand while a socket asks for the media of the message it is wired to, or a
-node of a running flow reads that message.
+node of a running flow reads that message. A socket that falls so far behind
+that events are dropped for it is sent a fresh snapshot, demand included, and
+is closed with 1008 if it was unpaired meanwhile.
+
+A person who may not watch a camera (see
+[Who may watch a camera](../interface/dashboards.md#who-may-watch-a-camera))
+gets none of its values in the snapshot or after it, and no frames when asking
+for them. Switching the camera's setting sends that person's open sockets a
+fresh snapshot.
 
 With a `binary` uplink the device sends a camera frame as a binary message in
 the same layout, with the header `{"type": "frame", "name": "<message>",
 "media_type": "image/jpeg", "ts": <device ms>}`. The message has to be one of
-its `in` ports wired as `image`. The frame is held in memory and published
-in one step, credited to its device node. A frame that does not qualify is
-logged and dropped and the socket stays open. With `http` it uploads with
-`PUT /artifacts?volatile=true&name=<message>` and publishes with
-`POST /messages/<message>`. A device that is unpaired or deleted while
-connected is closed with 1008.
+its `in` ports wired as `image`, and the whole message at most
+`MAX_FRAME_BYTES`. The frame is held in memory and published in one step,
+named after the message and credited to its device node. A frame that does not
+qualify is dropped, the reason logged once per socket, and the socket stays
+open. With `http` it uploads with `PUT /artifacts?volatile=true&name=<message>`
+and publishes with `POST /messages/<message>`. A device that is unpaired or
+deleted while connected is closed with 1008, at the latest with its next frame.
 
 ### Live view
 
@@ -294,10 +306,11 @@ the viewer. In order:
 | 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
+A superuser may watch any device, anyone else a device whose camera is open
+to members, and `start` is answered `bye` with `forbidden` otherwise. 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.

docs/interface/dashboards.md

@@ -552,12 +552,20 @@ lights up on the canvas. `out` messages are read only. The node is shown
 degraded, *Offline*, in amber while the device is not connected: a warning on
 Home, not a fault, since a panel switched off or asleep breaks nothing.
 
+A `bool` the device publishes, such as a camera's `person`, says something is
+so right now, and a device that went away cannot say it no longer is. Once the
+device has been away for 60 seconds, the same wait as before its node shows
+*Offline*, each of its `bool` messages that reads true is set to false by its
+device node. A device back within that wait keeps its values, so a restart of
+either side changes nothing. Its other values keep the last one it sent.
+
 ### What a device reports
 
 A paired device says what it has with `PUT /devices/{id}/report`, from its own
 credential only. openpanel sends it after pairing and once per boot. The report
-replaces the one before and is at most 16 KB. It is read leniently, so a newer
-firmware still reports to an older instance:
+replaces the one before and is at most 16 KB, and 64 KB as kept, once named
+`config_schemas` are put in place. It is read leniently, so a newer firmware
+still reports to an older instance:
 
 - keys this version does not know are dropped
 - a value it does not know for a display `shape` or a port `direction` reads as
@@ -635,6 +643,29 @@ 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.
 
+#### Who may watch a camera
+
+**Who may watch** in the Camera section of the device's page decides who sees
+what its camera sends, its `image` and `video` messages: **Superusers**, the
+default, or **All members**. A member of the portal is a member here. For
+anyone it leaves out:
+
+- the messages' values are left out of the live socket, `GET /messages/` and a
+  flow's state, and their history answers 403
+- the frames those messages hold, the last 8 included, answer 403, and a
+  socket asking for their bytes gets none
+- a live view is refused with `forbidden`, and so are TURN credentials
+- a dashboard tile or the device's preview says *No access to this camera*
+
+A wall panel shows the camera when one of its dashboards carries it, whoever
+watches the wall: a superuser assigned it the dashboards.
+
+The setting covers the camera's own messages, not what a flow makes of them.
+Anyone who edits flows can wire a node that passes a picture on under another
+name, and node code reads whatever it is handed. How long a kept `snapshot`
+stays and a light on the device that shows the camera is on are not part of
+it yet.
+
 #### Live view
 
 A media tile showing a device's `camera`, and the preview on the device's

docs/reference/configuration.md

@@ -169,14 +169,18 @@ MIG entries are passed on as written.
 | `ARTIFACT_GC_GRACE_S` | `3600` | minimum age before cleanup |
 | `ARTIFACT_VOLATILE_DIR` | auto-detected | where local workers write live frames, under `/dev/shm` or the temporary directory |
 | `ARTIFACT_VOLATILE_BYTES` | `50331648` (48 MiB) | live-media ring size, in the engine's memory and again in that directory; `0` stores volatile saves durably |
+| `MAX_FRAME_BYTES` | `4194304` (4 MiB) | the largest live frame: a volatile upload past it gets 413, a device's frame on its socket is dropped; keep it under 16 MiB, the websocket's limit |
 
 Cleanup removes durable artifacts no longer referenced by recorded runs or
 current messages, and pauses while a run is active. Volatile artifacts live in
 a bounded ring in the engine's memory, and the oldest disappear as it fills.
 Recorded run outputs are copied into durable storage. A worker in the engine's
 container writes its frames to `ARTIFACT_VOLATILE_DIR` instead, which the
-engine trims to the same size every few seconds. If workers publish frames
-under Docker, raise `shm_size` with the ring size.
+engine trims to the same size every few seconds. A frame the engine holds is
+written there too before a node in this container is handed it, so
+`load_artifact` finds it. If workers publish or read frames under Docker,
+raise `shm_size` with the ring size. The ring never holds more than its size
+plus one frame.
 
 ## Agents
 

frontend/src/client/schemas.gen.ts

@@ -933,6 +933,11 @@ export const DeviceBindingSchema = {
             type: 'string',
             enum: ['in', 'out'],
             title: 'Direction'
+        },
+        dtype: {
+            type: 'string',
+            title: 'Dtype',
+            default: ''
         }
     },
     type: 'object',
@@ -976,6 +981,12 @@ export const DeviceDefSchema = {
             title: 'Touch',
             default: false
         },
+        viewers: {
+            type: 'string',
+            enum: ['superusers', 'members'],
+            title: 'Viewers',
+            default: 'superusers'
+        },
         version: {
             type: 'integer',
             title: 'Version',
@@ -1011,6 +1022,12 @@ export const DeviceDocumentSchema = {
             title: 'Touch',
             default: false
         },
+        viewers: {
+            type: 'string',
+            enum: ['superusers', 'members'],
+            title: 'Viewers',
+            default: 'superusers'
+        },
         version: {
             type: 'integer',
             title: 'Version',
@@ -1090,6 +1107,12 @@ export const DevicePublicSchema = {
             title: 'Touch',
             default: false
         },
+        viewers: {
+            type: 'string',
+            enum: ['superusers', 'members'],
+            title: 'Viewers',
+            default: 'superusers'
+        },
         version: {
             type: 'integer',
             title: 'Version',

frontend/src/client/sdk.gen.ts

@@ -271,7 +271,7 @@ export class ArtifactsService {
      * watching now, which the oldest of falls out of memory rather than being
      * kept. A worker on another host or a device publishing a camera sends this.
      * It is read into memory, where the ring holds it, and never touches the
-     * disk.
+     * disk, so it is held to ``MAX_FRAME_BYTES`` rather than the store's cap.
      * @param data The data for the request.
      * @param data.name
      * @param data.mediaType
@@ -1015,6 +1015,7 @@ export class DevicesService {
      *
      * 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.
+     * Only for whoever may see the device's camera (``DeviceDef.viewers``).
      * @param data The data for the request.
      * @param data.deviceId
      * @returns IceServers Successful Response
@@ -1958,6 +1959,9 @@ export class MessagesService {
     /**
      * Read Messages
      * Every message any published flow declares, with its last value.
+     *
+     * A camera the caller may not see is listed without its value, so a tile
+     * can still be bound to it.
      * @returns MessagesPublic Successful Response
      * @throws ApiError
      */

frontend/src/client/types.gen.ts

@@ -238,6 +238,7 @@ export type DeviceBinding = {
     port: string;
     message: string;
     direction: 'in' | 'out';
+    dtype?: string;
 };
 
 export type direction = 'in' | 'out';
@@ -254,9 +255,12 @@ export type DeviceDef = {
     title?: string;
     dashboards?: Array<(string)>;
     touch?: boolean;
+    viewers?: 'superusers' | 'members';
     version?: number;
 };
 
+export type viewers = 'superusers' | 'members';
+
 /**
  * A device as it reads itself: what it shows, and what its ports carry.
  *
@@ -268,6 +272,7 @@ export type DeviceDocument = {
     title?: string;
     dashboards?: Array<(string)>;
     touch?: boolean;
+    viewers?: 'superusers' | 'members';
     version?: number;
     io?: Array<DeviceIo>;
     bindings?: Array<DeviceBinding>;
@@ -293,6 +298,7 @@ export type DevicePublic = {
     title?: string;
     dashboards?: Array<(string)>;
     touch?: boolean;
+    viewers?: 'superusers' | 'members';
     version?: number;
     io?: Array<DeviceIo>;
     bindings?: Array<DeviceBinding>;

frontend/src/components/Dashboard/MediaWidget.tsx

@@ -6,7 +6,7 @@ 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 { useCameraHidden, useLiveDevice, useLiveView } from "./useLiveView"
 import type { WidgetProps } from "./widgets"
 
 /**
@@ -66,10 +66,15 @@ export function MediaView({
   const ref = isRef(value) ? value : null
   const kind = (ref?.media_type ?? dtype).split("/")[0]
   const playing = kind === "audio" || kind === "video"
-  const camera = useLiveDevice(stream ? "" : message)
+  // Only superusers may see this camera: nothing is asked of it.
+  const hidden = useCameraHidden(message)
+  const camera = useLiveDevice(stream || hidden ? "" : message)
   const live = useLiveView(camera)
   const onAir = live.phase === "playing"
-  const active = (shown || playing) && !onAir
+  // A stream URL draws the picture itself, all but sound: the message's
+  // bytes would only be demand on a camera nothing shows.
+  const draws = (!stream || kind === "audio") && !hidden
+  const active = (shown || playing) && !onAir && draws
   const url = useArtifactUrl(
     active ? ref : null,
     active ? message || undefined : undefined,
@@ -116,6 +121,13 @@ export function MediaView({
     if (!message && !stream) {
       return <p className="text-muted-foreground">Pick a message.</p>
     }
+    if (hidden) {
+      return (
+        <p className="text-muted-foreground" data-testid="media-no-access">
+          No access to this camera
+        </p>
+      )
+    }
     if (!ref) {
       return <p className="text-muted-foreground">Nothing published yet.</p>
     }

frontend/src/components/Dashboard/useLiveView.ts

@@ -1,13 +1,18 @@
 import { useQuery } from "@tanstack/react-query"
 import { useCallback, useEffect, useRef, useState } from "react"
 
-import { LIVE_PORT, reportedPorts } from "@/components/Devices/ports"
+import {
+  cameraMessages,
+  LIVE_PORT,
+  reportedPorts,
+} from "@/components/Devices/ports"
 import { devicesQueryOptions } from "@/components/Devices/queries"
 import {
   type LiveSignal,
   listenLive,
   sendLive,
 } from "@/components/Flow/useFlowSocket"
+import useAuth from "@/hooks/useAuth"
 import { appRoute } from "@/lib/portal"
 
 /** How long a live view gets from the tap to its first frame. */
@@ -45,6 +50,29 @@ export function useLiveDevice(message: string): string {
   return device?.id ?? ""
 }
 
+/**
+ * Whether this viewer may not see the camera publishing `message`: a
+ * device's image or video while its `viewers` are superusers, for anyone
+ * else. The server holds the frames back too, this only says why the tile
+ * stays empty.
+ *
+ * A wall panel shows what its dashboards carry, which a superuser assigned,
+ * and cannot read the device list anyway.
+ */
+export function useCameraHidden(message: string): boolean {
+  const { user } = useAuth()
+  const enabled =
+    Boolean(message) &&
+    !appRoute().startsWith("/panel") &&
+    user?.is_superuser === false
+  const { data } = useQuery({ ...devicesQueryOptions(), enabled })
+  if (!enabled) return false
+  return (data?.data ?? []).some(
+    (device) =>
+      device.viewers !== "members" && cameraMessages(device).includes(message),
+  )
+}
+
 /**
  * A live view of one device's camera: the browser's own `RTCPeerConnection`
  * answering the device's offer, signalled over the page's socket.

frontend/src/components/Devices/DeviceDetail.tsx

@@ -44,6 +44,7 @@ import {
   type DeviceReport,
   deviceNodeOf,
   freeNodeId,
+  hasCamera,
   LIVE_PORT,
   type ReportedPort,
   reportedPorts,
@@ -141,7 +142,7 @@ export function DeviceDetail({ id }: { id: string }) {
           <section className={cn(CARD, "grid gap-4")}>
             <Basics device={device} canEdit={canEdit} onChange={write} />
           </section>
-          <Camera device={device} />
+          <Camera device={device} canEdit={canEdit} onChange={write} />
           <section className={CARD}>
             <Dashboards device={device} canEdit={canEdit} onChange={write} />
           </section>
@@ -219,33 +220,81 @@ function Basics({
   )
 }
 
+/** Who may see a device's camera, as the setting reads. */
+const VIEWERS = {
+  superusers: "Superusers",
+  members: "All members",
+} as const
+
 /**
- * What the device's camera sees, once a flow wires its `camera` port.
+ * Who may watch the device's camera, and what it sees once a flow wires its
+ * `camera` port.
  *
  * Watching is what turns the camera on: it sends frames only while this
  * preview, a tile or a flow reading them wants them.
  */
-function Camera({ device }: { device: DevicePublic }) {
+function Camera({
+  device,
+  canEdit,
+  onChange,
+}: {
+  device: DevicePublic
+  canEdit: boolean
+  onChange: (changes: Partial<DeviceDef>) => void
+}) {
+  if (!hasCamera(device)) return null
   const bound = (device.bindings ?? []).find(
     (binding) => binding.port === "camera" && binding.direction === "in",
   )
-  if (!bound) return null
   return (
     <section className={CARD}>
       <PanelSection
         title="Camera"
         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"
-          data-testid="device-camera"
-        >
-          <MediaView
-            message={bound.message}
-            fit="contain"
-            label={`${device.title || device.id} camera`}
-          />
+        <div className="flex items-center gap-2 text-sm">
+          <span className="flex-1">Who may watch</span>
+          <InfoTip label="Who may watch">
+            Its pictures, their history and its live view: superusers only, or
+            every member of this instance. A wall panel shows the camera when
+            one of its dashboards carries it. Whoever edits flows can still pass
+            the pictures on through a node.
+          </InfoTip>
+          <Select
+            value={device.viewers ?? "superusers"}
+            disabled={!canEdit}
+            onValueChange={(viewers) =>
+              onChange({ viewers: viewers as DeviceDef["viewers"] })
+            }
+          >
+            <SelectTrigger
+              className="w-40"
+              aria-label="Who may watch"
+              data-testid="device-viewers"
+            >
+              <SelectValue />
+            </SelectTrigger>
+            <SelectContent>
+              {Object.entries(VIEWERS).map(([value, label]) => (
+                <SelectItem key={value} value={value}>
+                  {label}
+                </SelectItem>
+              ))}
+            </SelectContent>
+          </Select>
         </div>
+        {bound ? (
+          <div
+            className="flex aspect-video w-full overflow-hidden rounded-md bg-muted text-center text-sm"
+            data-testid="device-camera"
+          >
+            <MediaView
+              message={bound.message}
+              fit="contain"
+              label={`${device.title || device.id} camera`}
+            />
+          </div>
+        ) : null}
       </PanelSection>
     </section>
   )

frontend/src/components/Devices/ports.ts

@@ -63,6 +63,30 @@ export function screensOf(devices: DevicePublic[] = []) {
   })
 }
 
+/** What a device's `viewers` setting covers: what a camera sends. */
+const CAMERA_DTYPES: readonly string[] = ["image", "video"]
+
+/** The messages a device's camera publishes, its image and video `in` ports. */
+export const cameraMessages = (device: DevicePublic) =>
+  (device.bindings ?? [])
+    .filter(
+      (binding) =>
+        binding.direction === "in" &&
+        CAMERA_DTYPES.includes(binding.dtype ?? ""),
+    )
+    .map((binding) => binding.message)
+
+/**
+ * Whether a device has a camera to say who may watch: one it reports, or an
+ * image or video port wired in a flow.
+ */
+export const hasCamera = (device: DevicePublic) =>
+  cameraMessages(device).length > 0 ||
+  (reportedPorts(device) ?? []).some(
+    (port) =>
+      port.direction === "in" && CAMERA_DTYPES.includes(port.dtype ?? ""),
+  )
+
 /** Whether a reported port may be wired in `direction`. `live` never is. */
 export const wiresAs = (port: ReportedPort, direction: "in" | "out") =>
   port.port !== LIVE_PORT &&