Store

Store is urun's content-addressed object store — it materializes model weights, keeps loaded models resident across sessions, and shares data across GPUs.

Store gives you one interface for everything a live app touches: model weights from HuggingFace, resident loaded models that survive across sessions and snapshot wakes, and named data shared between functions or sessions. You access it through from urun import store — the same import you used in the getting started guide to download Wan2.1 weights.

Where App defines your functions and Compute provisions your GPUs, Store handles everything those functions read and write.

App-bound store

Inside an app, bind the store to the app and use that handle:

from urun import App

app = App("liveavatar")
store = app.store

app.store is the same global store, with one behavioral change: @store.cache(key=...) cache keys are automatically scoped to your org and app. Write the bare key and uRun resolves the namespace for you:

@store.cache(key="wan22-s2v-14b")     # resolves to "<org>:liveavatar:wan22-s2v-14b"
def load(*, device):
    ...

Two tenants that deploy the same app with the same bare key get distinct cache namespaces, so there is no cross-app or cross-tenant collision in a multi-tenant cluster. The org segment comes from the runtime/deploy context — you never pass it by hand.

Plain from urun import store still works for scripts and one-off jobs that aren't bound to an app — it is the same store without the auto key-scoping.

Downloading models

The most common Store operation is downloading model weights. There are two entry points, and neither of them is store.get() — that call only loads previously store.set() / store.put() content, not URIs.

store.model() downloads and loads a HuggingFace model in one call — auto-detecting the model class and loading it via from_pretrained:

from urun import store

pipe = store.model("hf://Wan-AI/Wan2.1-T2V-1.3B")

store.path() downloads (if needed) and returns the local filesystem path, for anything that needs a path rather than a loaded object — a raw checkpoint you'll torch.load yourself, a subprocess call, or memory-mapped access:

import subprocess
from urun import store

path = store.path("hf://TheBloke/Llama-2-7B-GGUF/llama-2-7b.Q4_K_M.gguf")
subprocess.run(["./llama.cpp/main", "-m", str(path)])

store.model() accepts hf:// URIs only. store.path() is the broader primitive and accepts all of these:

HuggingFace URIs support specific revisions with @:

path = store.path("hf://meta-llama/Llama-2-7b-hf@v1.0")

Named storage

Store works as a key-value interface for sharing data between functions or across runs:

from urun import store

# Store any serializable data under a name
store.set("pipeline_config", config_dict)

# Retrieve it later (same session or a different one)
config = store.get("pipeline_config")

import { createStore } from '@urun-sh/core'

const store = createStore({ baseUrl: 'https://urun.sh', orgId: 'your-org-id' })

// Read named data set by Python functions
const config = await store.get("pipeline_config")

This is useful for passing configuration between functions — a pipeline stage and the app that steers it — or persisting results across sessions. The TypeScript SDK can read any named data written by Python functions through the same key-value interface.

Ephemeral refs

For temporary data that does not need a persistent name, store.put() returns a ref handle:

# put() takes only an object -- no name parameter -- and returns a ref
ref = store.put(large_tensor)

# Retrieve later using the ref
tensor = store.get(ref)

Refs are reference-counted. The data is lazily evicted when no handles remain, making put() ideal for passing large intermediate results between pipeline stages without managing cleanup.

Versioned artifacts with a latest pointer

For artifacts an app revises over time — a world state a session should resume from, a tuned pipeline profile, a published scene — use version-specific keys with an explicit "latest" pointer:

from urun import App, store

app = App("world")
STORE_PREFIX = "world"

@app.function(gpus="h100:1")
async def play(ctx):
    while session_active(ctx):
        state = step_world(ctx)

        if state.scene_changed:
            # Persist this scene under a version-specific key
            scene_key = f"{STORE_PREFIX}_scene_{state.scene_id}"
            store.set(scene_key, state.snapshot())

            # Update "latest" pointer
            store.set(f"{STORE_PREFIX}_scene_latest", scene_key)

When the user comes back — a new session, possibly on a different pod — pick up exactly where they left off using store.lookup(), which follows the pointer automatically (it reads the pointer value, then fetches the target):

state = store.lookup(f"{STORE_PREFIX}_scene_latest")
if state:
    world = restore_world(state)   # resume the last scene
else:
    world = fresh_world()          # first visit

This pattern gives you full control over artifact naming while keeping resumption simple.

Auto PyTorch handling

Store detects PyTorch nn.Module objects and handles state_dict extraction automatically:

import torch.nn as nn

model = MyModel()  # nn.Module subclass

# set() auto-extracts state_dict -- no need to call model.state_dict()
store.set("my_model", model)

# get() with target auto-loads state_dict into the module
store.get("my_model", target=model)

No need to manually call model.state_dict() or model.load_state_dict() — Store handles the round-trip. This composes with the latest-pointer pattern above for module state you version explicitly (an adapter your app hot-swaps mid-session, for example).

Dict-style access

For quick interactive use, Store supports Python's dict syntax:

store["my_key"] = value
value = store["my_key"]
del store["my_key"]

if "my_key" in store:
    # key exists
    ...

Dict access raises KeyError on missing keys. Use store.get() when you want None for missing keys instead.

Caching

All URI downloads are cached on disk with LFU (Least Frequently Used) eviction. On a cache hit — while the entry hasn't been evicted — calling store.path("hf://...") (or store.model(...)) again returns instantly, no re-download.

# First call: downloads from HuggingFace (~508 MB)
path = store.path("hf://Wan-AI/Wan2.1-T2V-1.3B/Wan2.1_VAE.pth")

# Second call: instant, served from disk cache
path = store.path("hf://Wan-AI/Wan2.1-T2V-1.3B/Wan2.1_VAE.pth")

The cache is persistent across process restarts and shared across functions on the same node. Manage it with:

# Check cache usage
stats = store.cache_stats()

# Clear all cached downloads
store.clear_cache()

# Configure cache location and size limit
store.configure_cache(cache_dir="/data/cache", max_size_gb=100)

The default cache size is auto-detected (a fraction of available system memory, capped at 32 GB) rather than a fixed number — use configure_cache(max_size_gb=...) to override it. When the cache fills up, the least frequently used entries are evicted first.

Keeping models loaded with @store.cache

Downloads are cached on disk, but a loaded model object — weights on the GPU, compiled graphs, allocator pools — is the expensive thing. @store.cache(key=...) keeps the loaded object resident so a worker that serves many sessions over its lifetime loads once:

store = app.store

@store.cache(key="wan22-s2v-14b")
def load_model(*, device):
    import torch
    from diffusers import DiffusionPipeline
    return DiffusionPipeline.from_pretrained(
        store.path("hf://org/wan22-s2v-14b"), torch_dtype=torch.bfloat16
    ).to(device)

The loader is just def load(*, device): return model — no lifecycle callbacks. (Warmup, precompile, and validation happen in one place, ctx.self_test, not in the loader.) store.cache plays two roles:

• Weights residency — the cached object lives in a content-addressed, tiered resident store (GPU memory → host memory → local disk → peers → durable storage). It is the 60–90 GB shared thing, deduped across functions and pods.
• Snapshot collaboration — cached weights are excluded from the process snapshot on capture (so the snapshot image is weight-free) and streamed back from the resident tier on restore. Weights live in the warm shared tier, never baked into the per-function image.

This is why the canonical app loads everything through @store.cache loaders and never holds a model that the cache doesn't know about.

Content addressing & dedup

Everything the store holds is content-addressed — identical bytes are stored once, keyed by a strong content hash. On top of that flat guarantee, the store dedups and streams structurally. Both mechanisms are transparent — no app changes, apps just keep calling the same primitives:

• Structural dedup — nested objects split at reuse boundaries: a multi-component model or pipeline is stored as a tree of content-addressed component subtrees plus a tiny manifest, so a text-encoder or VAE shared across a dozen pipelines is stored once, and an adapter is referenced independently of its base. The chunk boundary is a unit that is independently reused or streamed — never smaller than a set of always-co-used tensors. Large opaque blobs with no recognizable layout fall back to content-defined chunking, so near-identical objects still share every unchanged chunk.
• Blockwise weight streaming — weight files are split into per-tensor, content-addressed blocks (a safetensors shard becomes a header plus one block per tensor segment). On restore, only the blocks not already warm in a higher residency tier are fetched — in parallel, delivered in execution order — so fetch overlaps load instead of pulling one opaque blob.

These ride the same residency ladder and the same store.cache exclusion/restream collaboration described above, and they are why snapshot wakes stay fast: the snapshot image is weight-free, and weights stream back block-by-block from wherever they are already resident.

Streaming

Store provides bidirectional streaming for real-time applications. There are two ways to create streams:

Session-scoped streams

Inside an @app.function, use ctx.stream(name) to create a pipeline video sink tied to the current call. Compose stages with the | operator and steer with live inline control.desired.* reads (structural change uses doc.bind):

import urun

@app.function(gpus="h100:1")
async def generate_video(ctx: urun.Context, prompt: str):
    model = load_model()
    control = ctx.doc("control")
    video = ctx.stream("video", codec="h264", fps=30)

    def generate(*, stop_event, config):
        while not stop_event.is_set():
            output = model(prompt=control.desired.prompt.value or prompt, num_frames=17)
            yield output.frames[0]

    pipeline = (
        ctx.source(generate, device="cuda")
        | ctx.pace(fps=30)
        | video
    )

    await pipeline.run()

ctx.stream("video") is sugar for store.stream(f"{ctx.session_id}/video") — the context automatically scopes the stream to the current call. The video sink auto-converts float tensors to encoded frames.

On the TypeScript side, the client connects and receives the stream:

const session = app.generate_video()

const track = session.stream("video")
videoElement.srcObject = new MediaStream([track])

Streams are bidirectional — the client sends state back via CRDT documents, which the stage reads inline as cheap config (control.desired.*):

// Client sends control messages -- the stage reads control.desired.prompt live
session.doc("control").set({ desired: { prompt: "A new scene" } });

Global streams

For streams that are not tied to a specific call — metrics feeds, cross-session broadcasts — use store.stream(name) directly:

from urun import store

# Global stream, not tied to any session
metrics = store.stream("fleet_metrics")
await metrics.emit({"fps": 20.0, "active_sessions": 12})

// Monitor live pipeline health via a session document
const meta = session.doc("meta")

setInterval(() => {
  const snap = meta.get()
  if (snap) updateDashboard(snap)
}, 1000)

// Steer the pipeline via the control document
session.doc("control").set({ desired: { style: "noir" } })

Video streaming

For video output, pass codec and encoding options to ctx.stream() to create a pipeline sink. Build the pipeline with ctx.source, ctx.pipe, and ctx.pace, then chain into the video sink with the | operator:

@app.function(gpus="h100:1")
async def generate_video(ctx: urun.Context, prompt: str):
    model = load_model()
    control = ctx.doc("control")

    # Pipeline sink -- encodes frames to h264 automatically
    video = ctx.stream("video", codec="h264", fps=30)

    def generate(*, stop_event, config):
        while not stop_event.is_set():
            output = model(prompt=control.desired.prompt.value or prompt, num_frames=17)
            yield output.frames[0]

    pipeline = (
        ctx.source(generate, device="cuda")
        | ctx.pace(fps=30)
        | video
    )

    await pipeline.run()

The video sink auto-converts float tensors (NHWC, values in [0, 1]) to uint8 before encoding — no manual conversion needed. On the TypeScript side, session.stream("video") returns a MediaStreamTrack that can be assigned directly to a <video> element:

const session = app.generate_video()
const track = session.stream("video")

// Assign the track to a <video> element
videoEl.srcObject = new MediaStream([track])

No special video APIs on either side — just a pipeline with encoding kwargs on the Python side and a MediaStreamTrack on the TypeScript side.

For a complete walkthrough that builds a streaming app from scratch, see the getting started guide. For the TypeScript consumer side, see Store / StoreOptions.

Why store is global

Store is not scoped to a single app — from urun import store gives you the same global, content-addressed object store regardless of which app you are in. This is intentional:

• Model weights are deduplicated. If two apps both call store.model("hf://meta-llama/Llama-2-7b-hf"), the weights are downloaded and cached once. Content addressing means identical data is never stored twice.
• Artifacts are shared across apps. One function can publish a versioned artifact — a scene, a profile, a tuned adapter — that another app reads, without any coupling between them.
• Streams cross app boundaries. A generation app can emit metrics via store.stream("fleet_metrics") that a dashboard app consumes, without either app knowing about the other.

Session-scoped streams (ctx.stream()) provide isolation where you need it — each call gets its own namespace. Global streams (store.stream()) are for when you want shared access.

Isolation where you do want it comes from app.store: @store.cache(key=...) keys are org+app scoped so two tenants never share a cache namespace, while the underlying content-addressed weights still dedup globally. Scoping lives on the cache key, not on the data.

Walk-Through

This walk-through covers the three most common Store operations in sequence: materializing model weights, keeping the loaded model resident with @store.cache, and persisting session state so the next session resumes where the last one ended. For environment setup, see Getting Started.

from urun import App
from urun.core import Dependencies
import urun

app = App("store-walkthrough")
store = app.store
DEPS = Dependencies(python=["torch>=2.5.0", "diffusers", "transformers"])
STORE_PREFIX = "store_walkthrough"


# 1. Loaded-model residency: weights are content-addressed and shared;
#    the loaded pipeline stays resident across sessions and snapshot wakes.
@store.cache(key="wan21-t2v-1.3b")
def load_pipeline(*, device):
    import torch
    from diffusers import DiffusionPipeline
    return DiffusionPipeline.from_pretrained(
        store.path("hf://Wan-AI/Wan2.1-T2V-1.3B"), torch_dtype=torch.bfloat16
    ).to(device)


@app.function(gpus="h100:1", deps=DEPS, warm=0)
async def generate(ctx: urun.Context):
    pipe = load_pipeline(device="cuda")
    control = ctx.doc("control")
    video = ctx.stream("video", codec="h264", fps=20)

    # 2. Resume: pick up the last published scene, if any
    scene = store.lookup(f"{STORE_PREFIX}_scene_latest")
    prompt = scene["prompt"] if scene else "A quiet street at dawn"

    def frames(*, stop_event, config):
        while not stop_event.is_set():
            out = pipe(prompt=control.desired.prompt.value or prompt, num_frames=17)
            yield out.frames[0]

    # 3. Persist the session's final state for the next session to resume
    try:
        await (ctx.source(frames, device="cuda") | ctx.pace(fps=20) | video).run()
    finally:
        key = f"{STORE_PREFIX}_scene_{ctx.session_id}"
        store.set(key, {"prompt": control.desired.prompt.value or prompt})
        store.set(f"{STORE_PREFIX}_scene_latest", key)

The first session pays the weight materialization once — content-addressed, deduped with every other app that uses the same model. Every later session (and every snapshot wake) finds the pipeline already resident: @store.cache excludes the weights from the snapshot image and streams them back blockwise from the resident tier on restore. And because the session publishes its last state under a "latest" pointer, a returning user picks up exactly where they left off — even on a different pod.

Next: Distribute

Context exposes the rank-aware surface for multi-GPU parallelism — ctx.rank, ctx.world_size, and ctx.is_main_process.

Related:

• App — Defining apps and event handlers
• Compute — GPU specs and warm capacity
• Getting started — Hands-on streaming walkthrough