Streaming Pipelines

Every real-time uRun app is a pipeline. A live video restyle, an avatar, a streaming text-to-image, a webcam-in/captions-out model — they are all the same shape: a chain of stages composed with the | operator and driven continuously by await pipeline.run().

The pipeline builder lives on ctx. Stages chain with |; rank boundaries auto-insert zero-copy IPC transport. For the whole app around the pipeline, see Anatomy of a uRun app.

One primitive: the Stage

There is a single streaming primitive — a Stage: a stream transducer that consumes 0..N input streams and yields an output stream. Everything in the DSL is a specialization of it, distinguished only by input arity and placement — not by separate primitive classes:

• ctx.stream("name") is the I/O endpoint, and it is polymorphic — an input when it is a head (alone or inside a list), an output when it is a tail. The same endpoint can drive a pipeline (ctx.stream("mic") | …) or sink it (… | ctx.stream("video")).
• ctx.pace / ctx.window / ctx.unbatch are standard-library reshape/rate transducers — sugar over the one Stage. A bare generator lifts to a source; an item -> item callable lifts to a map.
• device / rank are placement metadata on a stage, not a separate primitive.
• doc.bind(construct, deps) is a higher-order Stage decorator (a useMemo for stages) — it returns a Stage, so it composes with | like any other.

This is why a pattern the DSL does not yet cover is added as a new Stage rather than worked around in app code — a custom Stage is a one-liner of the same type.

The grammar

A streaming app is a composition of stages:

doc = ctx.doc("control")
pipeline = (
    ctx.stream("mic")                        # inbound stream — the HEAD (when there is one)
    | ctx.window(window=N, stride=S)         # clip the inbound stream into windows
    | ctx.source(make_denoise(models, doc))  # generator / model — reads cheap config INLINE
    | doc.bind(lambda d: ctx.pipe(make_vae(models, d)), [doc.desired.vae.model])  # useMemo
    | ctx.pace(fps=24)                        # smooth to a steady frame rate
    | ctx.unbatch()                           # split a batch/window into individual frames
    | ctx.stream("video", codec="h264")       # outbound media sink
)
await pipeline.run()                          # start, hold while connected, stop cleanly

Not every app uses every stage — but every app is some subset of this chain. The inbound ctx.stream(...) is the head for apps that consume input; generator-driven apps start at ctx.source(...) instead.

The stages

• ctx.stream("in") — an inbound stream as the pipeline head. Browser-produced audio/video/data pipes straight into the next stage. For a codec-less inbound data stream, pass kind="data" explicitly (ctx.stream("frames", kind="data")) so the runtime consumes the browser-produced data segments instead of binding an RTP ingest.
• ctx.window(window=N, stride=S) — clips/buffers the inbound stream into windows. The streaming dual of ctx.pace: where the pacer spaces items out in time, the window groups them into clips for a model that wants windows rather than the raw stream. idle= supplies a pad value when the head has nothing buffered (idle-park).
• ctx.source(fn, device=...) — the generator head. fn(*, stop_event, config) yields items continuously; config is a live dict updated as control changes arrive.
• ctx.pipe(fn, device=...) — a processing stage; fn(item) returns the next item. device="cuda" makes it a GPU stage.
• ctx.tap(fn, ...) — a side-effect stage (logging, metrics); the item passes through unchanged.
• ctx.pace(fps=...) — smooths bursty output to a steady frame rate with adaptive buffering.
• ctx.unbatch() — splits an upstream batch (e.g. a VAE window) into individual frames.
• ctx.realtime(model="personaplex", *, endpoint=None, api_key=None, voice=None, instructions=None, sample_rate=24_000, transcribe=False) — a 1-input speech→speech stage wrapping an OpenAI-compat realtime session. Consumes inbound PCM audio frames, yields assistant PCM audio frames. The uRun-served personaplex model (Moshi) is the default; pass endpoint= + api_key= for an external API. PTS-stamped output. Mockable via MockRealtimeTransport for the CPU harness.
• ctx.mux(*, fps, sample_rate=24_000) — a fan-in A/V fusion stage. Takes a list of two inputs [frames, voice] via the list-head fan-in syntax and fuses them into one PTS-synchronised A/V item (MuxedAV(video, audio, pts)), aligned via a reorder buffer on the assistant-audio master timeline. Pipe the result into ctx.stream("av", codec="h264", audio_codec="opus") to produce one lip-synced track.
• ctx.stream(name, codec=...) — the media sink. Passing codec or bitrate puts ctx.stream in pipeline-sink mode; it owns the encode (float NHWC tensors in [0,1] are auto-converted to encoded frames). Use codec="h264" for video, codec="opus" for audio, codec="jpeg" for a single encoded image on the request/response lane. The app emits a tensor and never touches PIL/numpy itself — the encoder is the primitive.

Inbound-driven apps

When the model consumes a live input stream (a webcam restyle, a meeting avatar, video-in/captions-out), the inbound stream is just the head of the same chain — composed with | like every other stage. There is no special "source-fed-by-inbound" feed API and no consume loop:

pipeline = (
    ctx.stream("mic")                        # inbound audio is the head
    | ctx.window(window=24)                  # group into 24-frame clips for the model
    | ctx.source(make_model(models, doc), device="cuda")
    | ctx.pace(fps=24)
    | ctx.stream("captions")
)
await pipeline.run()

The pipeline idles/parks when there is no input and reanchors on resume, but it stays a single continuous pipeline driven by pipeline.run().

Real inbound audio from the browser uses ctx.stream("mic", codec="opus") — the platform decodes the incoming Opus stream to PCM frames so each stage receives raw float32 samples. A plain ctx.stream("mic") (no codec) is an event/data stream, not audio; specify codec="opus" for mic audio.

Fan-in: a list of streams as the head

Multiple inputs are not a different primitive — pass a list of streams as the head and the stage receives them all (merge / sync / multiplex):

pipeline = (
    [ctx.stream("mic"), ctx.stream("cam")]   # N inputs — fan-in
    | ctx.pipe(merge_av)                      # one stage receives both, synced per tick
    | ctx.source(make_model(models, doc), device="cuda")
    | ctx.pace(fps=24)
    | ctx.stream("video", codec="h264")
)
await pipeline.run()

A full-duplex app (audio in and out) is two streams — one driving the head, one sinking the tail.

Fan-out: implicit tee (no ctx.tee verb)

When you bind a stage to a variable and reference it in more than one downstream, the platform automatically inserts a shared-node tee — both downstreams receive the output:

denoise = ctx.source(make_denoise(models, doc), device=device)

pipeline_video = denoise | ctx.pace(fps=24) | ctx.stream("video", codec="h264")
pipeline_audio = denoise | ctx.realtime() | ctx.stream("audio", codec="opus")

There is no ctx.tee verb — referencing the same stage object twice is the fan-out. This matches the unix-filter model where a named pipe can be read by multiple consumers.

Reactive control: cheap reads inline, structural change with doc.bind

A live doc steers a running pipeline. The canonical pattern is exactly two cleanly separated jobs, with no field-mapping table:

Cheap / live config (prompt, cfg scale, strength) — the stage reads the live doc property inline whenever it needs it. doc.desired.prompt is a reactive accessor over the CRDT; read it at a chunk boundary, zero machinery, no bind:

def make_denoise(models, doc):
    def denoise(*, stop_event, config):
        while not stop_event.is_set():
            prompt = doc.desired.prompt.value      # live read at the chunk boundary
            yield models.denoise(prompt)
    return denoise

Structural config (model, resolution, topology — needs a rebuild) — doc.bind(construct, deps), which is exactly React's useMemo. construct is (doc) -> stage (it builds the stage, reading what it needs from the doc); deps is the list of live doc properties it depends on. bind evaluates construct(doc) once and, when any dep property value changes (shallow compare), re-runs construct → reconstructs the stage → the pipeline drains and swaps it (the hot-reload / iterate path):

doc = ctx.doc("control")
pipeline = (
    ctx.stream("mic") | ctx.window(window=24)
    | ctx.source(make_denoise(models, doc))                                   # cheap reads inline
    | doc.bind(lambda d: ctx.pipe(make_vae(models, d)), [doc.desired.vae.model])  # useMemo: recreate on change
    | ctx.pace(fps=24) | ctx.stream("video", codec="h264")
)
await pipeline.run()

On the browser side, writing the control doc steers the pipeline live:

session.doc('control').set({ desired: { prompt: 'a forest at dawn' } })

Multi-GPU (SPMD) pipelines

uRun declares no topology. You ask for a shape with gpus="type:n"; the model shards itself across the ranks (model-native parallelism via RANK / WORLD_SIZE / LOCAL_RANK), and you place each stage with rank= / device= — placement metadata on the stage, nothing more. When consecutive stages run on different ranks, uRun inserts shared-memory IPC with CUDA IPC handles for zero-copy GPU tensor transfer:

@app.function(gpus="b200:4", deps=DEPS)
async def runtime(ctx: urun.Context):
    device = ctx.device
    models = load_model(device=device)
    doc = ctx.doc("control")
    pipeline = (
        ctx.source(make_denoise(models, doc), device=device, rank=0)
        | doc.bind(lambda d: ctx.pipe(make_vae(models, d), device=device, rank=3), [doc.desired.vae.model])
        | ctx.unbatch(rank=3)
        | ctx.pace(fps=24, rank=3, min_buffer_frames=4)
        | ctx.stream("video", codec="h264", bitrate=8_000_000, fps=24, rank=3)
    )
    with ctx.self_test():
        _drive_one_and_assert_shape(pipeline)
    await pipeline.run()

ctx.group: when the SPMD group is initialized, ctx.group resolves to dist.group.WORLD (usable directly with torch.distributed.all_reduce(..., group=ctx.group)); it is None on a single-GPU app.

Several distinct models (not one sharded model) is just more ctx.pipe stages — a multi-model pipeline, each stage placed on its rank (see ctx.rank / ctx.world_size).

Keeping models warm

Load models through @store.cache(key=...) so a GPU worker that serves many sessions over its lifetime keeps the model resident between sessions:

store = app.store

@store.cache(key="restyle-14b")              # AUTO org+app-scoped (store = app.store)
def load_model(*, device):
    with REPO:                               # de-vendored upstream import (see below)
        from restyle import RestyleModel
    return RestyleModel.from_pretrained(store.path("hf://org/restyle-14b")).to(device)

How many models stay warm is a function of your replica count. store.cache keeps the model resident within each running replica; the size of your warm pool is set by urun app scale --replicas N (see Capacity). Each provisioned replica holds the model warm and consumes credits while it stays provisioned — so scaling replicas up trades idle credit burn for cold-start protection during spikes, and scaling to zero gives up warmth in exchange for consuming nothing while idle. Above the warm floor, the platform scales with demand: queued sessions add replicas up to the function's concurrency limit (@app.function(max_concurrency=), default 2; urun app concurrency overrides it live) and idle replicas drain back to the floor.

See the Context reference for every stage's full signature.

Related:

• Anatomy of a uRun app — the whole canonical shape + app-hygiene rules
• Context reference — source/pipe/tap/pace/unbatch/window/stream/doc/self_test
• Pipeline reference — run(), drive(), fan-in, rank boundaries
• Store — app.store, @store.cache, store.repo, model downloads
• De-vendoring — pull in upstream model code with store.repo
• Strict mode & CPU testing — ctx.self_test as the single readiness check