Media dtypes: image, audio and video as narrowed artifact references

A port may now declare `image`, `audio` or `video`. Each is the artifact
reference the engine already had, narrowed by the `media_type` on it, so a
speech recogniser declares what it eats rather than taking any bytes at all and
finding out. Bytes still never travel as a message and nothing on the wire
stops being JSON: a camera publishes one reference per frame, a microphone one
per chunk, and a reference may carry a `meta` dict nothing here interprets.

Streaming media is therefore an ordinary streaming port — with one change to
what that means. An emission used to journal an item with no payload, so
downstream read whatever was current when the item was claimed; a consumer
slower than its producer saw only the newest chunk and the ones between were
lost. That is right for a training curve and wrong for a second of speech, so
an emission now journals a `kind="emission"` item carrying its values, and the
executor hands them to the nodes reading that message instead of writing them
to state again. The value in state stays the latest, which is what everything
else reads, and the wave is filtered by what actually changed rather than
walking everything reachable. No queue serialization change — the existing
`outputs` field carries it.

Continuous media makes the store's missing GC a real problem, so this closes
it: `sweep_artifacts` runs hourly, keeps every digest a `run_artifact` row
records or a live message holds, spares anything written in the last hour, and
stands aside entirely while a run is in flight, since a node may store a
checkpoint long before it returns the reference to it. That also collects the
orphans a deleted flow has always left behind. `ARTIFACT_GC_INTERVAL_S=0` turns
it off.

Around the edges: `GET /artifacts/{digest}` serves the media type the caller
passes and answers ranged requests, so a browser plays a clip rather than
downloading it; `PUT` spools to disk instead of holding the whole body in
memory, as does `save_artifact` given a path; a Media widget draws whatever its
message points at, and a wall panel may fetch the bytes its own tiles are
showing and nothing else; and a connector gets `save_artifact`, for a device
whose readings are bytes.

What this cannot do is live video: a frame every second or two is a glance, and
the honest answer above that is the camera's own stream, which the widget takes
as a URL and the browser plays from source.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/fluksio/api/deps.py

@@ -15,6 +15,7 @@ from fluksio.core import security
 from fluksio.core.config import settings
 from fluksio.core.db import engine
 from fluksio.flow import panels
+from fluksio.flow.artifacts import is_reference
 from fluksio.flow.controller import FlowController
 from fluksio.flow.dashboards import DashboardStore
 from fluksio.flow.workers import PythonWorkerPool
@@ -78,13 +79,36 @@ def _panel_messages(panel_id: str, request: Request) -> set[str]:
     return panels.messages_for(panel_id, store)
 
 
+def _panel_digests(panel_id: str, request: Request) -> set[str]:
+    """The artifacts this panel's messages are pointing at right now.
+
+    A media tile fetches the bytes its message names, so the messages already
+    bounding the panel bound this too — one step further along, through
+    whatever those messages currently hold.
+    """
+    controller: FlowController | None = getattr(
+        request.app.state, "flow_controller", None
+    )
+    if controller is None:
+        return set()
+    names = _panel_messages(panel_id, request)
+    if not names:
+        return set()
+    found: set[str] = set()
+    for value in controller.state.get_present(sorted(names)).values():
+        if is_reference(value):
+            found.add(str(value["digest"]))
+    return found
+
+
 def _panel_may(payload: dict[str, Any], request: Request) -> None:
     """Refuse anything a wall panel has no business asking for.
 
     A panel credential names the account that approved the pairing, so without
     this it would be that person's session hanging on a wall. What a panel
     genuinely needs is small and worth writing out: the dashboards it was
-    assigned, its own definition, and the messages its own widgets bind to.
+    assigned, its own definition, the messages its own widgets bind to, and the
+    bytes those messages currently point at, for a tile drawing a camera frame.
 
     Publishing is in the list because a panel cannot be strictly read-only — a
     control on a panel is the point of putting one there, and a querying chart
@@ -128,6 +152,13 @@ def _panel_may(payload: dict[str, Any], request: Request) -> None:
         else:
             name = ""
         allowed = bool(name) and name in _panel_messages(panel.id, request)
+    elif method == "GET" and path.startswith(f"{api}/artifacts/"):
+        # The bytes behind a media message a tile on this panel is drawing.
+        # Scoped to what those messages hold *now*, which is exactly what a
+        # live widget asks for — a screen has no business reading an artifact
+        # off an old run because it happens to know the digest.
+        digest = unquote(path[len(f"{api}/artifacts/") :])
+        allowed = "/" not in digest and digest in _panel_digests(panel.id, request)
 
     if not allowed:
         raise HTTPException(

backend/fluksio/api/routes/artifacts.py

@@ -5,19 +5,28 @@ worker cannot — and having one path rather than two is what keeps a flow's
 code the same wherever it runs.
 """
 
+import re
+import tempfile
+from pathlib import Path
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request
-from fastapi.responses import StreamingResponse
+from fastapi.responses import FileResponse
 from jwt.exceptions import InvalidTokenError
 from pydantic import BaseModel
 from sqlmodel import Session
+from starlette.concurrency import run_in_threadpool
 
 from fluksio.api.deps import user_from_token
 from fluksio.core import security
 from fluksio.core.db import engine
 from fluksio.flow.artifacts import ArtifactStore
 
+#: What may be echoed back as a response content type. The caller holds the
+#: reference and passes its media type, so this guards a header rather than
+#: trusting one — anything else is served as bytes.
+_MEDIA_TYPE = re.compile(r"^[\w.+-]+/[\w.+-]+$")
+
 
 def artifact_caller(request: Request) -> str:
     """Who may move artifacts: a signed-in person, or an attached worker.
@@ -71,21 +80,45 @@ async def put_artifact(
     name: str = Query(default=""),
     media_type: str = Query(default=""),
 ) -> Any:
-    """Store the request body and answer with the reference to it."""
+    """Store the request body and answer with the reference to it.
+
+    Spooled to disk as it arrives rather than buffered: a video segment is as
+    legitimate a body here as a checkpoint, and neither should have to fit in
+    memory twice.
+    """
     store = _store(request)
-    body = await request.body()
-    return store.put([body], name=name, media_type=media_type)
+    handle = tempfile.NamedTemporaryFile(dir=store.root, delete=False)
+    try:
+        with handle:
+            async for chunk in request.stream():
+                handle.write(chunk)
+        return await run_in_threadpool(
+            store.put_file, Path(handle.name), media_type, name
+        )
+    finally:
+        Path(handle.name).unlink(missing_ok=True)
 
 
 @router.get("/{digest}")
-def get_artifact(digest: str, request: Request) -> Any:
-    """Stream one artifact back."""
+def get_artifact(
+    digest: str, request: Request, media_type: str = Query(default="")
+) -> Any:
+    """Serve one artifact back.
+
+    The caller passes the media type off the reference it holds, which is what
+    lets a browser play a clip rather than download it; the store itself keeps
+    only bytes. Ranged requests are answered because an audio or video element
+    scrubbing through a file asks for them.
+    """
     store = _store(request)
     path = store.path(digest)
     if path is None:
         raise HTTPException(status_code=404, detail="No such artifact")
-    return StreamingResponse(
-        store.read(digest),
-        media_type="application/octet-stream",
-        headers={"Content-Length": str(path.stat().st_size)},
+    return FileResponse(
+        path,
+        media_type=(
+            media_type if _MEDIA_TYPE.match(media_type) else "application/octet-stream"
+        ),
+        # The digest is the content, so it is also the perfect validator.
+        headers={"ETag": f'"{digest}"'},
     )

backend/fluksio/core/config.py

@@ -113,6 +113,14 @@ class Settings(BaseSettings):
     FLOW_GPUS: int = 0
     # How long the engine's own metrics, events and run records are kept.
     OBS_RETENTION_DAYS: int = 30
+    # How often artifact bytes nothing refers to any more are swept away; 0
+    # never sweeps. A flow streaming media writes one artifact per frame, so
+    # without this the store only grows.
+    ARTIFACT_GC_INTERVAL_S: int = 3600
+    # How long a freshly written artifact is spared, whatever refers to it.
+    # Storing bytes and recording the reference are two steps; this is the
+    # window between them.
+    ARTIFACT_GC_GRACE_S: int = 3600
     # 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

@@ -19,6 +19,7 @@ import hashlib
 import logging
 import os
 import tempfile
+import time
 from collections.abc import Iterable, Iterator
 from pathlib import Path
 from typing import Any
@@ -96,11 +97,13 @@ class ArtifactStore:
             "name": name,
         }
 
-    def put_file(self, path: Path, media_type: str = "") -> dict[str, Any]:
+    def put_file(
+        self, path: Path, media_type: str = "", name: str = ""
+    ) -> dict[str, Any]:
         with path.open("rb") as handle:
             return self.put(
                 iter(lambda: handle.read(CHUNK), b""),
-                name=path.name,
+                name=name or path.name,
                 media_type=media_type,
             )
 
@@ -119,20 +122,29 @@ class ArtifactStore:
             while chunk := handle.read(CHUNK):
                 yield chunk
 
-    def collect(self, keep: set[str]) -> int:
-        """Delete what no run refers to any more. Returns how many went.
+    def collect(self, keep: set[str], grace_s: float = 0.0) -> int:
+        """Delete what nothing refers to any more. Returns how many went.
 
         The caller passes every digest still recorded; anything else in the
         store was produced by a run that has since been pruned, or never got a
         row at all because the run failed between writing and recording.
+
+        ``grace_s`` spares anything written that recently. Storing bytes and
+        recording the reference to them are two steps, and a sweep landing
+        between them would take an artifact its run is about to name — so
+        recent files are left for the next pass, by which time they are either
+        referenced or genuinely orphaned.
         """
         removed = 0
+        cutoff = time.time() - grace_s
         for entry in self.root.glob("*/*"):
             if not entry.is_file():
                 continue
             if DIGEST_PREFIX + entry.name in keep:
                 continue
             try:
+                if grace_s > 0 and entry.stat().st_mtime > cutoff:
+                    continue
                 entry.unlink()
                 removed += 1
             except OSError:

backend/fluksio/flow/connector.py

@@ -17,7 +17,10 @@ What a connector gets from the base class:
   be known by the engine;
 * :meth:`ConnectorNode.write`, the other direction — values arriving on the
   node's input ports, for a connector that commands something rather than only
-  reading it.
+  reading it;
+* :meth:`ConnectorNode.save_artifact`, for a device whose readings are bytes —
+  a camera frame, a recorded clip — which travel as a reference on a media
+  port rather than as the message itself.
 
 The message schemas and the parameter model are the rest of the contract, and
 they are the same ones the built-in nodes use. See ``docs/connectors/`` for the
@@ -37,6 +40,8 @@ from fluksio.flow.nodes import Node
 if TYPE_CHECKING:
     from fastapi import FastAPI
 
+    from fluksio.flow.artifacts import ArtifactStore
+
 logger = logging.getLogger(__name__)
 
 #: Bumped when a change would break connectors written against the old surface.
@@ -81,7 +86,7 @@ class ConnectorNode(Node):
             description="Seconds between polls; 0 polls never.",
         )
 
-    __slots__ = ("config", "_poll_task", "_stop_event", "_last_published")
+    __slots__ = ("config", "_poll_task", "_stop_event", "_last_published", "_artifacts")
 
     def __init__(self, **kwargs: Any) -> None:
         super().__init__(f=self._dispatch, **kwargs)
@@ -89,6 +94,7 @@ class ConnectorNode(Node):
         self._poll_task: asyncio.Task[None] | None = None
         self._stop_event: asyncio.Event | None = None
         self._last_published: dict[str, Any] = {}
+        self._artifacts: ArtifactStore | None = None
 
     def _dispatch(self, params: dict[str, Any], **ports: Any) -> dict[str, Any] | None:
         """The scheduler's entry point. Settings are already on ``self.config``."""
@@ -122,7 +128,30 @@ class ConnectorNode(Node):
     # What the engine drives
     # -------------------------------------------------------------------------
 
+    def save_artifact(
+        self,
+        data: bytes,
+        name: str = "",
+        media_type: str = "application/octet-stream",
+    ) -> dict[str, Any]:
+        """Store bytes and return the reference to publish on a media port.
+
+        A camera frame or a recorded clip is far too big to be a message, so a
+        connector publishing one publishes this instead: the bytes go to the
+        store and the reference names them, which is what an ``image``,
+        ``audio`` or ``video`` port carries.
+
+        Only available once the node has started — the store belongs to the
+        engine, and is handed over then.
+        """
+        if self._artifacts is None:
+            raise RuntimeError(
+                "no artifact store: a connector can only save bytes once it has started"
+            )
+        return self._artifacts.put([data], name=name, media_type=media_type)
+
     async def start(self, app: FastAPI | None = None) -> None:
+        self._artifacts = getattr(app.state, "artifact_store", None) if app else None
         if self.config.poll_interval > 0 and self._stop_event is None:
             self._stop_event = asyncio.Event()
             self._poll_task = self._run_supervised("poll", self._poll_loop)

backend/fluksio/flow/dashboards.py

@@ -63,6 +63,7 @@ WidgetType = Literal[
     "icon",
     "forecast",
     "clock",
+    "media",
     # Input
     "button",
     "switch",
@@ -100,6 +101,10 @@ WIDGET_DTYPES: dict[str, set[str]] = {
     # Either shape a colour can travel as; which of the two this widget means
     # is its ``format``, checked against ``COLOR_DTYPES`` below.
     "color": {"list", "str"},
+    # A camera frame, a clip, a segment. What it draws follows the type it is
+    # bound to; a plain artifact is taken as well, since the bytes may be
+    # anything and the media type on the reference is what says what they are.
+    "media": {"image", "audio", "video", "artifact"},
     # An icon maps weather strings, bool hints and numbers alike, and a clock
     # binds nothing at all, so neither has a row to be held to.
 }

backend/fluksio/flow/executor.py

@@ -398,8 +398,18 @@ class ExecutionService:
             )
 
         replay = item.deliveries > 1
+        emission = item.kind == "emission"
         try:
-            published = pipeline.apply_outputs(node, item.outputs or None)
+            # An emission's values went into state when the node produced them;
+            # this item carries them so its readers get the chunk that caused
+            # the wave rather than whichever is newest by the time they run.
+            # Applying them again would let a mid-node emission overwrite what
+            # the node returned at the end.
+            published = (
+                set(item.outputs)
+                if emission
+                else pipeline.apply_outputs(node, item.outputs or None)
+            )
             pipeline.run_downstream(
                 node,
                 entry_id=item.entry_id,
@@ -408,6 +418,7 @@ class ExecutionService:
                 # an item with no payload is the value already being in state.
                 # Neither can say what changed, so neither filters on it.
                 changed=None if replay or not item.outputs else published,
+                overrides=item.outputs if emission else None,
             )
         finally:
             # Paired, or a cascade that raised — state backend gone, say — is a

backend/fluksio/flow/messages.py

@@ -33,6 +33,13 @@ class DType(str, Enum):
     thirty-megabyte checkpoint never sits in Redis. Inline codecs would only be
     needed for payloads too small to be worth a round trip, and nothing asks
     for that yet.
+
+    ``image``, ``audio`` and ``video`` are that same reference narrowed to a
+    media family, so a speech recogniser declares what it eats rather than
+    taking any bytes at all and finding out. Bytes still never travel as a
+    message: a camera publishes one reference per frame, a microphone one per
+    chunk. A reference may carry a ``meta`` dict — sample rate, dimensions, a
+    sequence number — which nothing here interprets.
     """
 
     FLOAT = "float"
@@ -51,6 +58,12 @@ class DType(str, Enum):
     #: A reference to stored bytes:
     #: ``{"digest": "sha256:…", "size": int, "media_type": str, "name": str}``.
     ARTIFACT = "artifact"
+    #: An artifact reference whose ``media_type`` is ``image/*``.
+    IMAGE = "image"
+    #: An artifact reference whose ``media_type`` is ``audio/*``.
+    AUDIO = "audio"
+    #: An artifact reference whose ``media_type`` is ``video/*``.
+    VIDEO = "video"
 
 
 _JSON_TYPES = (dict, list, str, int, float, bool, type(None))
@@ -65,6 +78,14 @@ _ITEM_TYPES = frozenset(
     {DType.FLOAT, DType.INT, DType.STR, DType.BOOL, DType.JSON, DType.RECORD}
 )
 
+#: The media dtypes, and the ``media_type`` family a reference must declare to
+#: satisfy each. A stream of them is a stream of references, one per chunk.
+MEDIA_FAMILIES = {
+    DType.IMAGE: "image/",
+    DType.AUDIO: "audio/",
+    DType.VIDEO: "video/",
+}
+
 
 #: How deep to look for a non-finite number. Deeper than any payload that
 #: reads well on a canvas, and a bound on a value that refers to itself.
@@ -147,6 +168,11 @@ def _is_artifact(value: Any) -> bool:
     )
 
 
+def _media_family(value: Any) -> str:
+    """The ``media_type`` an artifact reference declares, lowercased."""
+    return str(value.get("media_type") or "").lower()
+
+
 def _matches(dtype: DType, value: Any) -> bool:
     """Whether one value satisfies a scalar or record type."""
     if dtype is DType.BOOL:
@@ -161,6 +187,10 @@ def _matches(dtype: DType, value: Any) -> bool:
         return _is_record(value)
     if dtype is DType.ARTIFACT:
         return _is_artifact(value)
+    if dtype in MEDIA_FAMILIES:
+        return _is_artifact(value) and _media_family(value).startswith(
+            MEDIA_FAMILIES[dtype]
+        )
     return isinstance(value, _JSON_TYPES)
 
 
@@ -243,6 +273,14 @@ class MessageSpec(BaseModel):
         else:
             ok = _matches(self.dtype, value)
         if not ok:
+            if self.dtype in MEDIA_FAMILIES and _is_artifact(value):
+                # It is a reference, just to the wrong kind of bytes — saying so
+                # beats "expected audio, got dict" on a media_type typo.
+                raise TypeError(
+                    f"{where}: expected {self.dtype.value} "
+                    f"({MEDIA_FAMILIES[self.dtype]}*), got an artifact of "
+                    f"'{_media_family(value) or 'no media type'}'"
+                )
             raise TypeError(
                 f"{where}: expected {self.dtype.value}, got {type(value).__name__}"
             )
@@ -263,7 +301,13 @@ class MessageSpec(BaseModel):
             return str(value).lower() in ("true", "1", "yes", "on")
         if self.dtype is DType.STR:
             return value if isinstance(value, str) else json.dumps(value)
-        if self.dtype in (DType.SERIES, DType.RECORD, DType.LIST, DType.ARTIFACT):
+        if self.dtype in (
+            DType.SERIES,
+            DType.RECORD,
+            DType.LIST,
+            DType.ARTIFACT,
+            *MEDIA_FAMILIES,
+        ):
             # A structured payload arriving as text is the same hint a numeric
             # one is; the shape itself is still checked afterwards.
             return json.loads(value) if isinstance(value, str) else value

backend/fluksio/flow/pipeline.py

@@ -928,7 +928,11 @@ class Pipeline:
         return True, outputs
 
     def _execute_node(
-        self, node: Node, state: StateBackend, entry_id: str = ""
+        self,
+        node: Node,
+        state: StateBackend,
+        entry_id: str = "",
+        overrides: dict[str, Any] | None = None,
     ) -> dict[str, Any] | None:
         """Run one node and record its outputs. Never raises."""
         started = time.perf_counter()
@@ -938,6 +942,17 @@ class Pipeline:
             # single bulk read does not, and it was the engine's one global
             # mutex — every node of every cascade queued behind it.
             inputs = state.get_present(list(node.requires))
+            if overrides:
+                # The value this wave is delivering wins over whatever state
+                # holds by now. Not written back: the newest value is still the
+                # one everything else reads.
+                inputs.update(
+                    {
+                        name: value
+                        for name, value in overrides.items()
+                        if name in node.requires
+                    }
+                )
 
             key = ""
             if self.run_cache is not None and node.fingerprint:
@@ -1074,13 +1089,16 @@ class Pipeline:
             return
         self._record_outputs(node, passed, self._state)
         if self._queue is not None:
-            # Journalled with no payload of its own: the value is already in
-            # state, published in the order it was produced. An item carrying
-            # it would re-apply that value whenever it happened to be claimed,
-            # which is how an emission from the middle of a node overwrites the
-            # one it returned at the end. Downstream reads what is current,
-            # which is what "the latest value wins" has always meant here.
-            self._enqueue_cascade(node, None)
+            # Journalled carrying the emitted values, as an ``emission`` item:
+            # the executor hands them to the nodes reading them instead of
+            # writing them to state a second time. That distinction is the
+            # whole of it — re-applying would let a mid-node emission overwrite
+            # the value the node returned at the end, while reading state
+            # instead means a consumer slower than its producer sees only the
+            # newest chunk and the ones between are lost. A frame of video or a
+            # second of speech is worth delivering; the value in state stays
+            # the latest, which is what everything else reads.
+            self._enqueue_cascade(node, passed, kind="emission")
 
     def _observe(self, outcome: NodeOutcome) -> None:
         """Tell the run watching this pipeline, if there is one."""
@@ -1166,6 +1184,7 @@ class Pipeline:
         entry_id: str = "",
         replay: bool = False,
         changed: set[str] | None = None,
+        overrides: dict[str, Any] | None = None,
     ) -> StateBackend:
         """Execute nodes concurrently, scheduling each as its inputs arrive.
 
@@ -1173,6 +1192,10 @@ class Pipeline:
         whose triggering inputs are all untouched is completed without being
         run, so what is downstream of *it* is judged on the same footing. None
         runs everything the subset holds, which is what a manual run means.
+
+        ``overrides`` reaches the nodes that read those names directly, and no
+        further: a value carried by this wave is what its readers should see,
+        while everything past them reads what those readers produced.
         """
         # One view of the graph for the whole wave: a flow replaced halfway
         # through must not have this wave asking the new dependencies about a
@@ -1248,7 +1271,7 @@ class Pipeline:
                         continue
                     submitted.add(n)
                     node_futures[n] = executor.submit(
-                        self._execute_node, n, state, entry_id
+                        self._execute_node, n, state, entry_id, overrides
                     )
 
         def drain(executor: ThreadPoolExecutor) -> None:
@@ -1348,6 +1371,7 @@ class Pipeline:
         entry_id: str = "",
         replay: bool = False,
         changed: set[str] | None = None,
+        overrides: dict[str, Any] | None = None,
     ) -> StateBackend:
         """Run everything downstream of a node that has just published.
 
@@ -1361,6 +1385,10 @@ class Pipeline:
         already read, which is how one node came to publish 619 messages a
         minute off inputs that changed six times. ``None`` means walk
         everything reachable, which is what a manual run wants.
+
+        ``overrides`` carries values to hand to whoever reads them instead of
+        what state holds — an emission delivering the chunk that caused this
+        wave rather than whichever one is newest by the time it runs.
         """
         if changed is not None and not changed:
             # Everything the cascade carried was held back by a rate limit, so
@@ -1376,6 +1404,7 @@ class Pipeline:
             entry_id=entry_id,
             replay=replay,
             changed=changed,
+            overrides=overrides,
         )
 
     def trigger(
@@ -1439,13 +1468,21 @@ class Pipeline:
             return False
 
     def _run_here(
-        self, node: Node, outputs: dict[str, Any] | None, cause: str = "manual"
+        self,
+        node: Node,
+        outputs: dict[str, Any] | None,
+        cause: str = "manual",
+        delivered: bool = False,
     ) -> StateBackend:
         """Run a cascade in this thread, under a run id of its own.
 
         The queued path gets its run id from the journal entry. A run that never
         went through the queue still belongs in the history, so it makes one —
         marked as such, because it is no one's idempotency key.
+
+        ``delivered`` marks an emission the queue could not take: its values are
+        already in state, so they are handed to their readers rather than
+        written again.
         """
         run_id = f"{MANUAL_RUN_PREFIX}{uuid.uuid4().hex[:12]}"
         self._publish(
@@ -1460,13 +1497,16 @@ class Pipeline:
             }
         )
         try:
-            published = self.apply_outputs(node, outputs)
+            published = (
+                set(outputs or {}) if delivered else self.apply_outputs(node, outputs)
+            )
             state = self.run_downstream(
                 node,
                 entry_id=run_id,
                 # No payload means the value is already in state and this is a
                 # wake-up, which has nothing to name as changed.
                 changed=published if outputs else None,
+                overrides=outputs if delivered else None,
             )
         finally:
             # Paired, or a cascade that raised leaves the run open until the
@@ -1564,12 +1604,14 @@ class Pipeline:
             logger.error("Could not defer work for '%s': %s", node.id, exc)
             return False
 
-    def _enqueue_cascade(self, node: Node, outputs: dict[str, Any] | None) -> None:
+    def _enqueue_cascade(
+        self, node: Node, outputs: dict[str, Any] | None, kind: str = "cascade"
+    ) -> None:
         """Journal a trigger, or fall back to running it here if that fails."""
         from fluksio.flow.queue import WorkItem
 
         item = WorkItem(
-            kind="cascade",
+            kind=kind,
             node=node.id,
             flow=node.flow,
             outputs=outputs or {},
@@ -1591,7 +1633,7 @@ class Pipeline:
                 }
             )
         # Losing the value outright would be worse than running it here.
-        self._run_here(node, outputs, cause="external")
+        self._run_here(node, outputs, cause="external", delivered=kind == "emission")
 
     def values(self, flow: str | None = None) -> dict[str, dict[str, Any]]:
         """Last value and timestamp of every message, optionally one flow's.

backend/fluksio/flow/queue.py

@@ -40,12 +40,15 @@ class WorkItem:
     """One unit of journaled work.
 
     :param kind: ``cascade`` replays a node's outputs and runs what is
-        downstream; ``flush`` lets out what a node's rate limits held back;
-        ``run`` is a whole batch run, and carries only its id.
+        downstream; ``emission`` is a value a node published while still
+        running, already in state, carried so its readers get *that* value
+        rather than whichever is newest when they run; ``flush`` lets out what
+        a node's rate limits held back; ``run`` is a whole batch run, and
+        carries only its id.
     :param node: The node the item is about — the source for a cascade, the
         one whose rate limits are released for a flush.
     :param flow: The flow that node belongs to, so gating needs no lookup.
-    :param outputs: What the source node emitted (cascade only).
+    :param outputs: What the source node emitted (cascade and emission).
     :param cause: Where the work came from, for logs and debugging.
     :param not_before: Epoch seconds before which the item must not run.
     :param guard: ``(key, value)`` the target node must still remember for

backend/fluksio/flow/runs.py

@@ -295,6 +295,66 @@ def _from_digest(
     }
 
 
+#: How deep a stored value is walked looking for artifact references. A
+#: reference nested past this is not something any port declares.
+_REF_DEPTH = 8
+#: How many state keys are read back at a time by the sweep.
+_SWEEP_BATCH = 500
+
+
+def _references_in(value: Any, found: set[str], depth: int = 0) -> None:
+    """Every artifact digest inside one stored value."""
+    if is_reference(value):
+        found.add(str(value["digest"]))
+        return
+    if depth >= _REF_DEPTH:
+        return
+    if isinstance(value, dict):
+        items: Any = value.values()
+    elif isinstance(value, (list, tuple)):
+        items = value
+    else:
+        return
+    for item in items:
+        _references_in(item, found, depth + 1)
+
+
+def sweep_artifacts(
+    store: ArtifactStore, state: StateBackend, grace_s: float = 0.0
+) -> int:
+    """Remove artifact bytes nothing refers to any more. Returns how many went.
+
+    Two things refer to an artifact: a run that recorded it, and a message
+    currently holding it. The second is what makes a stream of media
+    affordable — a camera publishing a frame a second replaces the reference
+    each time, so yesterday's frames are unreferenced by definition and the
+    store does not grow without bound. What a node *returns* is recorded
+    against its run and kept; what it emits along the way is not.
+
+    Skipped entirely while anything is running: a node that stores a checkpoint
+    an hour before it returns has neither a row nor a message naming it yet,
+    and a sweep in that window would take the bytes out from under it.
+    """
+    with Session(db_engine) as session:
+        active = session.exec(
+            select(Run.id).where(col(Run.status).in_(("running", "queued"))).limit(1)
+        ).first()
+        if active is not None:
+            logger.debug("Artifact sweep skipped: run %s is in flight", active)
+            return 0
+        keep = set(session.exec(select(col(RunArtifact.digest)).distinct()).all())
+
+    keys = state.keys()
+    for start in range(0, len(keys), _SWEEP_BATCH):
+        for value in state.get_multi(keys[start : start + _SWEEP_BATCH]).values():
+            _references_in(value, keep)
+
+    removed = store.collect(keep, grace_s)
+    if removed:
+        logger.info("Artifact sweep removed %d unreferenced artifacts", removed)
+    return removed
+
+
 def seed_values(
     flow: FlowDef, params: dict[str, Any], seed: int | None = None
 ) -> dict[str, Any]:

backend/fluksio/main.py

@@ -1,5 +1,6 @@
 import asyncio
 import contextlib
+import logging
 from collections.abc import AsyncIterator
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 
@@ -32,13 +33,15 @@ from fluksio.flow.plugins import load_plugins
 from fluksio.flow.queue import MemoryWorkQueue, RedisWorkQueue, WorkQueue
 from fluksio.flow.remote import RemoteWorkerHub
 from fluksio.flow.resources import ResourceAccountant, fair_share_env
-from fluksio.flow.runs import RUN_STATE_TTL, RunService
+from fluksio.flow.runs import RUN_STATE_TTL, RunService, sweep_artifacts
 from fluksio.flow.secrets import init_secrets
 from fluksio.flow.state import MemoryState, RedisState, StateBackend
 from fluksio.flow.store import FlowStore
 from fluksio.flow.watchdog import LoopWatchdog
 from fluksio.flow.workers import PythonWorkerPool
 
+logger = logging.getLogger(__name__)
+
 
 def custom_generate_unique_id(route: APIRoute) -> str:
     return f"{route.tags[0]}-{route.name}"
@@ -80,6 +83,28 @@ def _run_state(namespace: str) -> StateBackend:
     return MemoryState()
 
 
+async def _sweep_artifacts(store: ArtifactStore, controller: FlowController) -> None:
+    """Take unreferenced artifact bytes off the disk, on a slow loop.
+
+    A flow streaming media writes one artifact per frame, so a store nothing
+    prunes only grows. Runs in a thread: it walks a directory and reads state.
+    """
+    interval = settings.ARTIFACT_GC_INTERVAL_S
+    if interval <= 0:
+        return
+    while True:
+        await asyncio.sleep(interval)
+        try:
+            await run_in_threadpool(
+                sweep_artifacts,
+                store,
+                controller.state,
+                settings.ARTIFACT_GC_GRACE_S,
+            )
+        except Exception:
+            logger.exception("Artifact sweep failed")
+
+
 def _mcp_sessions() -> AbstractAsyncContextManager[None]:
     """The MCP session manager's run scope, or nothing when MCP is off."""
     if not settings.MCP_ENABLED:
@@ -181,6 +206,9 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
     metrics_task = asyncio.create_task(
         MetricsCollector(event_bus).run(), name="metrics-collector"
     )
+    gc_task = asyncio.create_task(
+        _sweep_artifacts(artifacts, controller), name="artifact-gc"
+    )
     await controller.start()
     run_service.start()
     # Optional, and off unless someone enrolled this installation: the
@@ -208,6 +236,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
         watchdog_task.cancel()
         alerts_task.cancel()
         metrics_task.cancel()
+        gc_task.cancel()
         enrol_task.cancel()
         # Re-read from app.state: enrolling at runtime replaces this.
         running_cloud = getattr(app.state, "cloud_task", None) or cloud_task

backend/fluksio/sdk/__init__.py

@@ -39,7 +39,20 @@ F = TypeVar("F", bound=Callable[..., Any])
 #: Mirrors :class:`fluksio.flow.messages.DType`. Mirrored rather than imported:
 #: importing it would pull the engine into a research process.
 DTYPES = frozenset(
-    {"float", "int", "str", "bool", "json", "series", "record", "list", "artifact"}
+    {
+        "float",
+        "int",
+        "str",
+        "bool",
+        "json",
+        "series",
+        "record",
+        "list",
+        "artifact",
+        "image",
+        "audio",
+        "video",
+    }
 )
 #: What a list may hold, mirroring ``messages._ITEM_TYPES``.
 ITEM_TYPES = frozenset({"float", "int", "str", "bool", "json", "record"})