Push a frame instead of storing and fetching it

The rate the media dtypes could carry was one frame every second or two: each
was a file on the data volume, an event on the socket, and a request back for
the bytes. This closes both halves of that, and they are one feature.

`save_artifact(..., volatile=True)` writes to a `VolatileStore` — the same
content-addressed store, in `/dev/shm`, bounded by size with the oldest falling
out (`ARTIFACT_VOLATILE_BYTES`, 48 MB under the container's raised `shm_size`).
Nothing sweeps it: a frame nobody kept is not worth walking the store to find.
`ArtifactStore.path` falls through to it, which is what lets a volatile frame be
an ordinary reference everywhere else — the dtype check, a panel's digest scope,
`load_artifact` in a node, and the widget's own fetch all work on one unchanged.
`adopt` copies one into the store when a run records it, so "returned media is
kept, emitted media is not" stays true.

The bytes then go down the flows websocket as a length-prefixed binary frame,
sent just ahead of the `message_value` naming them, so a tile has the frame when
it hears the value moved. Nothing is pushed unasked: a client names the messages
it is drawing (`{"type":"media","names":[…]}`), a panel's list is intersected
with the scope it already had, and only the newest frame per name in a batch is
sent — a client that fell behind is not handed frames it would draw over. The
tunnel relays text only, so a screen reached through a portal falls back to
fetching, which is why the rate table now has two rows.

Around the edges: the remote worker's fetch cache is bounded at last
(`FLUKSIO_ARTIFACT_CACHE_BYTES`), since content addressing means nothing in it
ever expires and a media stream fills it with chunks nothing asks for twice; a
port carrying an image draws the frame in the node panel rather than only
saying `image/png · frame.png · 1.79kB`; and an edge chip says that much instead
of a line of hash. The media screenshot stops waiting for `networkidle` — a
camera is a socket that never goes quiet, which is the point of it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YC4u66vjzW54fnHu5Juhh9

backend/fluksio/api/routes/artifacts.py

@@ -80,6 +80,7 @@ async def put_artifact(
     request: Request,
     name: str = Query(default=""),
     media_type: str = Query(default=""),
+    volatile: bool = Query(default=False),
 ) -> Any:
     """Store the request body and answer with the reference to it.
 
@@ -87,6 +88,10 @@ async def put_artifact(
     legitimate a body here as a checkpoint, and neither should have to fit in
     memory twice. Capped, because nothing else here was: any account, and any
     worker credential, could otherwise fill the data volume.
+
+    ``volatile`` puts it in the ring instead of the store — a frame a screen is
+    watching now, which the oldest of falls out of memory rather than being
+    kept. A worker on another host publishing a camera sends this.
     """
     store = _store(request)
     cap = settings.MAX_ARTIFACT_BYTES
@@ -112,7 +117,7 @@ async def put_artifact(
                 # as long as the upload lasts.
                 await run_in_threadpool(handle.write, chunk)
         return await run_in_threadpool(
-            store.put_file, Path(handle.name), media_type, name
+            store.put_file, Path(handle.name), media_type, name, volatile
         )
     finally:
         Path(handle.name).unlink(missing_ok=True)

backend/fluksio/api/routes/flows.py

@@ -28,6 +28,7 @@ from fluksio.api.deps import (
     user_from_token,
 )
 from fluksio.core.db import engine
+from fluksio.flow.artifacts import is_reference
 from fluksio.flow.controller import FlowController
 from fluksio.flow.dashboards import DashboardStore
 from fluksio.flow.events import event_bus
@@ -931,6 +932,88 @@ def event_for_panel(event: dict[str, Any], only: set[str]) -> bool:
 # should not be handed the whole queue in one message.
 MAX_FRAME_EVENTS = 64
 
+#: How many message names one socket may ask for bytes on. A screen draws a
+#: handful of tiles; this is only here so a client cannot ask for the whole
+#: namespace and be served every frame of it.
+MAX_MEDIA_NAMES = 32
+
+
+def wanted_names(
+    frame: str, only: set[str] | None, previous: set[str]
+) -> 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.
+    """
+    try:
+        payload = orjson.loads(frame)
+    except orjson.JSONDecodeError:
+        return None
+    if not isinstance(payload, dict) or payload.get("type") != "media":
+        return None
+    names = payload.get("names")
+    if not isinstance(names, list):
+        return set()
+    asked = {str(name) for name in names[:MAX_MEDIA_NAMES]}
+    return asked if only is None else asked & only
+
+
+def media_frames(
+    events: list[dict[str, Any]], wanted: set[str], store: Any
+) -> list[bytes]:
+    """The bytes behind the media values in this batch, one frame each.
+
+    Referenced-then-fetched costs a round trip per frame, which is what keeps
+    the message plane at a glance rather than a view. Pushing the bytes down
+    the socket that already carries the event closes that, and only for the
+    frames a client said it was drawing.
+
+    Only what the ring holds: the durable store is what a fetch is for, and a
+    checkpoint has no business being pushed at anyone. The newest value per
+    name wins, so a client that fell behind is not handed a backlog of frames
+    it would only draw over.
+    """
+    if not wanted or store is None or getattr(store, "volatile", None) is None:
+        return []
+
+    newest: dict[str, dict[str, Any]] = {}
+    for event in events:
+        if event.get("type") != "message_value":
+            continue
+        name = str(event.get("name") or "")
+        if name not in wanted or not is_reference(event.get("value")):
+            continue
+        newest[name] = event
+
+    frames: list[bytes] = []
+    for name, event in newest.items():
+        value = event["value"]
+        digest = str(value.get("digest") or "")
+        path = store.volatile.path(digest)
+        if path is None:
+            continue
+        try:
+            payload = path.read_bytes()
+        except OSError:
+            continue
+        header = orjson.dumps(
+            {
+                "type": "media",
+                "name": name,
+                "digest": digest,
+                "media_type": str(value.get("media_type") or ""),
+                "ts": event.get("ts"),
+            }
+        )
+        # Length-prefixed, so one frame carries both halves and the reader
+        # never has to guess where the JSON stops.
+        frames.append(len(header).to_bytes(4, "big") + header + payload)
+    return frames
+
 
 async def _send(websocket: WebSocket, events: list[dict[str, Any]]) -> None:
     """One event, or a batch of them under `events`.
@@ -962,12 +1045,16 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None:
         await websocket.close(code=1008)
         return
     only = panel_scope(token, websocket.app)
+    # Nothing until a client says it is drawing something: bytes are what this
+    # socket cannot afford to send speculatively.
+    wanted: set[str] = set()
 
     await websocket.accept()
 
     controller: FlowController | None = getattr(
         websocket.app.state, "flow_controller", None
     )
+    store = getattr(websocket.app.state, "artifact_store", None)
 
     async def send_snapshot() -> None:
         if controller is not None:
@@ -989,7 +1076,10 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None:
                     # stream — a keepalive, or anything else it decides to
                     # say, used to be read as the client going away and cost
                     # it every live update from then on.
-                    receiver.exception()
+                    if receiver.exception() is None:
+                        asked = wanted_names(receiver.result(), only, wanted)
+                        if asked is not None:
+                            wanted = asked
                     receiver = asyncio.create_task(websocket.receive_text())
                     if sender not in done:
                         continue
@@ -1017,6 +1107,7 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None:
                         # token — and that would widen this socket to
                         # everything on the bus.
                         only = panel_scope(token, websocket.app) or set()
+                        wanted &= only
                         if out:
                             await _send(websocket, out)
                             out = []
@@ -1025,6 +1116,10 @@ async def flow_events(websocket: WebSocket, token: str = "") -> None:
                         continue
                     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)
         except (WebSocketDisconnect, RuntimeError):
             # A peer that goes away mid-send takes the RuntimeError route