Recipes — content-addressed pipeline caching#

Analysis over a CaptureGraph dataset is longitudinal: the dataset grows, and the naive "process everything" script redoes nearly all of its work on every run. Recipes fix that. You write decomposed pure functions over CGType values and pretend the filesystem doesn't exist — intermediates are stored for you in a managed scratch, keyed by content hash, and reused whenever the same inputs recur. Because the server's source data is durable, everything the cache holds is recomputable: the cache is pure speedup, never a system of record.

A recipe is an ordinary function. You call it — there is no event loop, no await, no runtime to spin up:

import capturegraph as cg

cg.scratch.configure("~/.cache/capturegraph")


@cg.pure_function("brighten/1")
def brighten(image: cg.Image) -> cg.Image:
    ...  # expensive pixel work
    return result


# First run computes; a later run over the same image returns the cached result.
bright = brighten(photo)

The unit of reuse is the function call: brighten(photo) is memoized on ("brighten/1", hash(photo)). Edit one stage of a pipeline and only that stage and its dependents recompute; add new captures and only the new ones are processed. That is the whole point — recompute only what's new.

Where recipes run#

The same function runs in two places, and nothing about it changes between them:

• Outside the server — a script or notebook over a target you copied off a phone or pulled with sync. Configure a scratch once, cg.load the target, and call your recipes; re-run the script as the dataset grows and only the new captures compute.
• Inside the server — an interceptor. The server opens its own scratch at startup (<root>/.cg-scratch, or the scratch_path in config.toml), so an interceptor just calls the recipe. Pure functions memoize expensive work across requests; ImpureState holds the coordination state a crowd needs, such as a reservation book. From Zero to a Scheduled Target builds one.

Tests get an isolated, auto-deleted scratch from the cg_scratch pytest fixture.

The scratch#

Every file-producing recipe operation runs against a scratch: a garbage-collected, content-addressed directory the library manages. Configure it once per process, by precedence:

1. an explicit cg.scratch.configure(path),
2. the CG_SCRATCH environment variable,
3. otherwise the first file-producing op raises cg.scratch.ScratchNotConfigured with a message naming these three ways — recipes never silently guess a location.

cg.scratch.configure("~/.cache/capturegraph", max_size_gb=50)

Setting overrides (max_size_gb, free_space_ratio) persist into the scratch's settings.toml, which is read fail-soft on later opens. The index carries a layout version: a scratch written by an older layout is migrated in place on open where the change allows, and otherwise emptied and rebuilt, so an upgrade never refuses a scratch or waits for someone to delete one. For tests and one-off scripts, cg.scratch.temporary() is a context manager that installs a throwaway scratch and deletes its tree on exit; the shipped cg_scratch pytest fixture wraps it. configure sets the default every thread shares; temporary overrides it for the calling thread or task only, so a dry run in one server worker never redirects the requests another is serving. A thread you start yourself begins without the override; pmap carries it into its workers.

A scratch is four ideas and nothing else: objects exist, names reference them, a pin keeps an object alive while a running process holds it, and a lock serializes the writers of one state, the computation of one call, or the users of one machine-wide resource.

One further table is bookkeeping over calls rather than an idea of its own: steps lists every step a living process has open right now (below).

<root>/
  objects/<hash>/        immutable value trees
  live/<process>/        that process's in-progress workspaces
  locks/<process>        the process liveness flock
  locks/state/<key>      one flock per ImpureState key
  locks/call/<key>       one flock per memoized call being computed
  locks/resource/<name>  one flock per machine-wide resource in use
  settings.toml
  index.sqlite           tables: objects, names, pins, steps, file_hashes

A cached call result is the name call:<call_key>/<position>; an ImpureState is the name state:<key>, which is kept. Eviction ranks objects rows and never takes one a kept name holds or a living process pins. live/<process>/ is owned by construction — the process that made it is the only one that writes it, and it is reclaimed the moment that process's liveness flock is free, so a crash leaves nothing behind to reason about; the same liveness check retires a dead process's steps rows.

One filesystem underlies all of it, so sealing a result from live/ into objects/ is an atomic rename (an object that already exists — from any call — dedups to the one copy), and two processes sharing a scratch path co-operate through the index and file locks — a shared cache across every analysis job on the machine.

You never name a path in objects/ or move a file there yourself. A body that produces a new file asks its type for one — cg.Image.new("jpeg"), cg.Thumbnail.new(), cg.ManagedDir.new() — writes to it, and returns it:

@cg.pure_function("thumbnail/1")
def thumbnail(image: cg.Image) -> cg.Thumbnail:
    thumb = cg.Thumbnail.new()   # a fresh path in this call's workspace; no file yet
    render(image, thumb)
    return thumb

The extension names one of the type's declared extensions and is inferred when the type declares exactly one. Every call runs with one live workspace, and new allocates inside it, so what the body writes seals with its result; the framework hashes the file and files it flat into objects/ on the way out. The same new works outside any call too — in a script or a test — where the value gets a workspace of its own for as long as it is held. That is what "pretend the filesystem doesn't exist" means in practice: inputs are values, outputs come from new, and nothing else touches a path.

Eviction: utility density#

The cache is bounded by budget = min(max_size_gb, free_space_ratio × (available + current cache)), and the scratch keeps it under budget by evicting entries — but which entries matters, and the choice is the research-relevant part.

Each entry carries a use score: a leaky integrator over its access times (score = prev·e^(−Δt/decay) + (1 − e^(−Δt/recharge))), which estimates the entry's future use rate. Eviction does not rank by score alone. It ranks candidates ascending by utility density:

$$ \text{density} = \frac{\text{effective_score}(now) \times \text{compute_seconds}}{\text{size_bytes}} $$

Read the units: (uses/second) × (compute-seconds/use) ÷ bytes = compute-seconds saved per second, per byte. It is a value density — how much recomputation each stored byte spares the machine per unit time. Evicting the lowest-density entries first is exactly the greedy knapsack heuristic, near-optimal while individual entries are small against the budget. An object's compute_seconds is the measured body wall-time of the costliest call that ever produced it, so a result several recipes share is priced at what losing it would really cost.

With multiple processes sharing one scratch, every job's usage aggregates into one ranking: the scratch becomes a global allocator of disk to compute, continuously maximizing total compute-time saved per byte across the whole machine, with no per-process tuning.

The honest boundaries: this is single-machine only — flock, statvfs, and SQLite WAL are unreliable on network filesystems, so put the scratch on a local disk. And because cross-process access bumps interleave, the scores are order-approximate (elapsed time is clamped to ≥ 0 so a stale bump can never inflate a score); the reordering that causes is bounded and acceptable for a heuristic.

pure_function#

@cg.pure_function(identity=None) memoizes a function over CGType values.

Identity and the stale-serve contract. The identity keys every cache entry, and its two forms have sharply different invalidation behavior:

• A versioned identity — @cg.pure_function("thumbnail/1") — means you own invalidation. The identity never changes with the code, so editing the body without bumping the version keeps serving the old cached result, deliberately. Bump to "thumbnail/2" to invalidate. This is the contract, not a bug — version-pin a stable stage and you control exactly when its cache resets.
• The bare form — @cg.pure_function — digests the compiled body, so any edit changes the identity and busts the cache automatically. Convenient while iterating; use a versioned identity for anything long-lived.

The return annotation is required and load-bearing: it is the schema the result is stored and loaded with. tuple[A, B] returns are supported element-wise.

Hashability of arguments is crisp. A CGType argument hashes as its merkle content hash (a value on disk hashes in place when it is unchanged; an in-memory one is committed first, so both forms of an equal value hit the same key). Plain int/float/bool/str/bytes/None, datetime/date/timedelta, and tuples, lists, dicts, and frozensets of those get canonical tagged encodings (so 1, 1.0, and True never collide; dicts are key-sorted). Each plain type is one registered encoder in capturegraph.recipes.values.arguments. Anything else — a raw NumPy array, say — raises UnhashableArgumentError naming the parameter; wrap it as a CGType first.

A parameter annotation is a schema too. A container a broadcast produced — sessions.values(), captures.map(session_splat) — carries no element type of its own, so it hashes as the type its parameter declares (splats: cg.Array[cg.Splat]), with no re-wrapping at the call. A parameter that declares a bare cg.Array or cg.Map declares no element type either, and its argument hashes as its own type.

Every result is cached. There is no admission threshold: the output is already in live/, so sealing it is an atomic rename plus one index row, and utility-density eviction flushes whatever fails to earn its bytes back. A hit costs one index read, a pin, and a metadata-weight load (about a millisecond), so wrapping even a fast stage is fine; a function that must not be cached is a plain function.

A Missing argument yields Missing. A call with a Missing anywhere in its arguments returns Missing without running or caching, so absence chains through a recipe the way it chains through attribute access: thumbnail(session.photo) is Missing for a session with no photo.

Calling is just calling. A decorated function is an ordinary synchronous callable:

result = thumbnail(image)

That holds everywhere — a script, a server request handler running in a worker thread, a notebook cell. There is nothing to await and no loop to drive.

Claiming a miss. However many callers reach the same miss together — worker threads, uvicorn workers, separate jobs on one machine — the body runs once. A caller that finds no sealed result takes an exclusive flock on locks/call/<key>, looks the key up again under it, and only then runs the body; everyone else blocks on that lock until the first caller seals, then finds the result and hits. The kernel drops the lock on any death, so a crashed claimant frees the key and the next waiter computes. Callers block for as long as the body takes: there is no timeout and no duplicate work.

Progress: the step tree and the exec log#

A body that runs for minutes is visible while it runs. Progress is a tree of steps: a label, a parent, a start time, and optionally a count. Three things open steps, and the first two need nothing from you:

• every memoized call, on a miss, named after its function;
• every map and pmap, with the element count as its total, advancing as each element finishes — on a cache hit as much as on a miss, so a mostly cached walk fills its bar at once;
• a body of your own, with cg.step:

@cg.pure_function("train_splat/1")
def train_splat(scene: ColmapModel) -> Splat:
    with cg.step("Training splat", total=iterations) as training:
        for batch in training.track(batches):
            loss = ...
            training.note(f"loss {loss:.3f}", loss=loss)
    ...

A step opens beneath the ambient one, the way calls nest, and a pmap worker attaches its steps beneath the pmap's. advance and note change the record in memory and nothing else; a flusher writes the dirty steps of the process to the scratch's steps table at most ten times a second, so a training loop advancing thirty thousand times costs the index a few hundred writes. A closed step is deleted at the next flush, and a process that dies first is reclaimed through its liveness lock like any other owner of rows.

from pathlib import Path

scratch = cg.scratch.Scratch(Path("~/.cache/capturegraph"))
for row in scratch.steps():
    print(row.label, row.done, row.total, row.note, row.parent_id)

memo = scratch.memo(thumbnail.key(image))    # what the scratch has for one call
memo.results                                 # holds on its sealed results, or ()
memo.running                                 # the steps beneath its call, or ()

steps() reports only the steps of living processes, so a crash never leaves a phantom "computing" behind. memo is how a display asks about one call rather than the whole machine: func.key(...) is the key the call is served under, and the memo answers with holds on its sealed results, else the tree of steps beneath the call — each with how deep it sits, how long it has run, how far along it is, its note, and the tail of its log. Nothing sealed and nothing running means nobody has computed it. The server's page status is exactly this.

Outside a server, cg.print_progress shows the same tree in the terminal for a block: redrawn in place ten times a second on a terminal, one line per step opening and closing anywhere else.

with cg.print_progress():
    thumbnails = journal.sessions.photo.pmap(session_thumbnail)

The other view is the exec log. A tool a body runs through exec (below) streams its combined stdout and stderr, as they arrive, into exec.log at the root of the call's workspace — outside every result, so it never seals into the object. A watcher that knows the call step's workspace can tail it; the server's pages do exactly this while a page computes. When a call raises, its workspace is released, and the exception carries the log's last lines as a note, so the failure explains itself wherever it is reported.

Global locks#

Some work must not overlap: two splats training at once exhaust the GPU. A resource of the machine gets a named lock, and a body that needs it holds the lock for the block:

@cg.pure_function("train_splat/1")
def train_splat(dataset: PosedImages) -> cg.Splat:
    with cg.lock("gpu"):
        work.exec(["ns-train", "splatfacto", "--data", dataset])
    ...

The lock is an exclusive flock on locks/resource/<name>, so holders serialize across every thread, uvicorn worker and script sharing the scratch, and the kernel frees it on any kind of death: a crashed trainer never wedges the GPU. "gpu" is the one conventional name; any other string is a lock of its own.

The wait is visible. lock opens a step labelled <name> lock beneath the ambient one. While it waits, the step's note says who it is waiting for, read from the holder's own step: waiting for train_splat (pid 4242), held 5m12s. Once acquired the note becomes held and the step stays open for the block, so a computing page or cg.print_progress shows a run waiting for the GPU as exactly that, and shows what is holding it while it does.

Nesting passes through. A lock of a name the context already holds runs its block at once, and a pmap worker of a held block, running in a copy of the caller's context, does the same: holding the GPU and mapping a recipe that also locks it never waits on itself. A thread you start yourself begins without the hold and waits like any other caller.

A lock is scoped to the scratch, which is what the machine's jobs share; two scratch roots on one machine do not exclude each other.

ImpureState — keyed persistent state#

Some things are honestly stateful: a reservation ledger a fleet of server workers bumps, a COLMAP reconstruction refined by every new batch of images. cg.ImpureState gives them exactly one home. Subclass it like a cg.Struct — the annotated fields ARE the persistent state — and declare the operations as ordinary methods. open(key) names one state, and every method call on that handle is a whole serialized read-modify-write:

class ReservationBook(cg.ImpureState):
    held: cg.Map[cg.UserID, cg.Time]  # granted slots, keyed by user

    def reserve(self, user_id: str, slot: cg.Time) -> cg.Bool:
        ...  # grant the slot iff no live lease holds it, recording it on self
        return granted


granted = ReservationBook.open(survey_id).reserve(user_id, slot)

Inside a method, self IS the current state: read its fields, assign its fields. The framework loads the committed state before the method runs and atomically commits self after it returns; the first call for a key seeds the state from the class's field defaults, and a field without a default starts Missing (test with cg.is_missing). Because operations are methods over one class, several operations naturally share one state — a book with reserve and release is one class with two methods, not two functions with two disjoint states.

State is a name over objects/, not a mutable file. Every committed version is a sealed immutable object named by its own merkle; an update seals the new version and atomically repoints the state: name at it. Two consequences fall out for free:

• Crash safety. A crash anywhere mid-call — even between sealing the new object and repointing — leaves the name on the previous version; the half-committed object is just an unreferenced tree the sweeper ranks like any other.
• Free history. Only the object the state: name currently holds is eviction-exempt. A superseded version loses the hold and becomes evictable history — still loadable by hash (S.load(scratch/objects/<hash>)) until density eviction takes it. You get versioned state without a versioning system.

One writer per key. The whole load → method → commit runs under an exclusive flock on locks/state/<key>, so concurrent callers — worker threads, uvicorn workers, separate jobs — serialize their read-modify-writes and never lose an update (this is what lets counters and reservations retire ad-hoc persistence layers). The kernel drops that lock on any death, orderly or not, so a crashed holder's key is simply free; nothing wedges.

Never cached. Every method call runs against the latest state; arguments are not hashed and results are never memoized. Mutability lives only where it is honest — the state — and the pure cache stays immutable (its pins and dedup depend on that).

Reading without writing. open(key).peek() loads the committed state as a plain value — a snapshot, not a live view.

Progressive external tools. Decomposition covers everything decomposable: per-item pure_functions plus cheap pure assembly, where the cache itself is the progressive state. For the rest — sequential/stochastic cores like the COLMAP mapper or a training loop, which refine one artifact in place — the artifact IS the state, and holding it is what the class is for:

class SparseModel(cg.ImpureState):
    model: cg.ManagedDir                    # the whole COLMAP model tree
    passes: cg.Number = cg.Number(0.0)

    def advance(self, database: cg.Blob) -> cg.ManagedDir:
        if cg.is_missing(self.model):
            self.model = cg.ManagedDir.new()
        self.model.exec(["colmap", "mapper", "--database_path", database, "--output_path", "."])
        self.passes = cg.Number(float(self.passes.value) + 1.0)
        return self.model

Inside a method, self is a private working copy: the committed tree is copied into the call's workspace before the method runs, so a directory or file field is the method's own to write in place — the tool refines self.model where it stands — and the commit afterwards snapshots whatever the method left there. The sealed object is never touched.

Each SparseModel.open(target_id).advance(...) call refines that target's model; the identity of the evolving artifact is declared by the key ("the reconstruction of target X"), never inferred from input hashes. No folds, no warm-start hints, no mutable cache entries — those were considered and rejected.

Identity and invalidation. State survives editing a method — the identity is the class's qualified name plus a digest of its schema, not a source digest. Changing the fields (or renaming/moving the class) starts an empty state: the struct IS the format, and a format change is a new identity, not a migration.

Durability equals the scratch's. State lives in the scratch's object store, so a durable cg.scratch.configure root means durable state, and cg.scratch.temporary state is ephemeral by definition.

Mapping and parallelism#

A recipe applies to one value; to run it over a collection, map it. There are two ways, and the choice is explicit — recipes never parallelize behind your back:

• Array.map(fn) / Map.map(fn) run fn over every element (or value, keys preserved) sequentially, results in order, per-element failures isolated as Missing:

thumbs = images.map(thumbnail)

• Array.pmap(fn) / Map.pmap(fn) are the same but across the process's one thread pool, sized to the CPU count; with cg.parallelism(n): caps how many elements a pmap has in flight at once. They parallelize for real only when the bodies release the GIL — NumPy/PyTorch/PIL/ffmpeg/CUDA kernels or a subprocess, which is the dominant analysis workload. A pure-Python CPU body serializes on the interpreter lock, so pmap buys nothing there. A pmap inside a worker runs sequentially: the outer pool is the parallelism, so nested fan-out never oversubscribes.

thumbs = images.pmap(thumbnail)

Beyond that, hashing runs multithreaded inside blake3 for large files, and an external tool run in a directory (below) parallelizes itself however it does. There is deliberately no framework-level fan-out primitive: the mental model stays "a recipe is a function you call," and the parallelism you get is the parallelism you asked for.

Blackbox tools#

Some stages are external tools — colmap, ffmpeg — that write an opaque directory of files. A blackbox recipe gets a fresh directory with cg.ManagedDir.new(), stages its inputs into it, runs the tool there, and returns the result tree (or a subdirectory of it) as a first-class cached cg.ManagedDir:

import capturegraph as cg


@cg.pure_function("colmap_sfm/1")
def colmap_sfm(images: cg.Array[cg.Image]) -> cg.ManagedDir:
    work = cg.ManagedDir.new()                        # a fresh directory in this call
    for index, image in enumerate(images):
        work.stage(image, f"images/{index}.jpg")      # typed value straight in, parents made
    work.exec(["colmap", "automatic_reconstructor", "--workspace_path", "."])
    return work / "sparse"                            # the tree the tool wrote

A ManagedDir is a pathlib.Path, so you join and read it as one; it adds two methods, and they are the whole blackbox vocabulary:

• work.stage(source, rel) — copy a file or tree to rel, creating rel's parents: source is any path or CGType file value. The copy is this directory's own, so a tool may overwrite it in place.
• work.exec(argv, *, env=None, timeout=None, check=True, capture_output=False) — subprocess.run with work as the working directory; path values in argv pass straight through, and a non-zero exit raises CalledProcessError.

Inputs are read-only: a value handed to a recipe lives in the object store or the data root, and stage is what gives the tool a copy it may write.

cg.ManagedDir (an unannotated managed folder) and cg.Blob (a file scalar accepting any extension) are analysis-only types: they let a recipe hold arbitrary tool output, including as fields of an ImpureState — a whole tool directory persists beside its progress counters, no archiving. They never appear in a procedure or target schema (no captured-data or iOS obligation); @cg.procedure rejects a schema holding one, naming the offending position.

Splats — the shipped recipe library#

cg.splats is a recipe library built on everything above: 3D Gaussian splats from posed captures. Each stage is its own memoized recipe, so a scene's splat and its sessions' share the frames and seeds they have in common instead of recomputing them side by side:

@cg.pure_function("session_splat/1")
def session_splat(capture: cg.SceneCapture) -> cg.Splat:
    frames = cg.splats.scene_dataset(capture)
    return cg.splats.train(frames, cg.splats.seed(frames))


@cg.pure_function("scene_splat/1")
def scene_splat(sessions: cg.Map[cg.Date, Session]) -> cg.Splat:
    captures = sessions.values().capture
    frames = cg.Array[cg.splats.PosedImage](
        [frame for session in captures.map(cg.splats.scene_dataset) for frame in session]
    )
    seed = cg.splats.splat_centers(captures.map(session_splat))
    return cg.splats.train(frames, seed, iterations=30_000)

• A dataset is an Array of PosedImage: a cg.Image with the cg.Pose that took it, its intrinsics expressed at the image's own size, and the cg.Depth it measured, Missing where none was. Arrays of frames in one world concatenate.
• scene_dataset(capture) takes a SceneCapture's keyframes with their ARKit poses, resized; sessions relocalized into one world map share a frame.
• seed(frames) back-projects each frame's captured depth into a cg.PointCloud; seed(frames, distance_meters=) casts a grid of pixels a constant distance out instead, for captures that measured no depth. splat_centers(splats) is the centres of trained splats as a cloud.
• train(frames, seed, iterations=, optimize_cameras=) runs nerfstudio's splatfacto under cg.lock("gpu") and reports a Training step with the iteration count, loss and gaussian count as it runs.
• cg.PointCloud.points() and cg.Splat.centers() read the positions and colours of either PLY.

Training runs in a separate interpreter holding torch, gsplat and nerfstudio, named by the CG_GPU_PYTHON environment variable (else found beside the ns-train on the path); the library's own environment never imports them. The result is a cg.Splat, which cg.ui.SplatViewer renders on a page.

Adding a type#

Everything above works on the directory tree a value stores as, so a new type joins hashing, caching, new, and stage by declaring only its on-disk form. Three leaf kinds cover almost every case; declaring one registers its wire name:

class Splat(cg.FileScalar, name="edu.example.splat", extensions=("ply",)): ...

class Calibration(cg.JSONScalar, name="edu.example.calibration"):
    focal_mm: float
    sensor_mm: float

class ColmapModel(cg.DirScalar, name="edu.example.colmap_model"): ...

A FileScalar value is a file (Splat.new("ply") hands a recipe a fresh path to write), a JSONScalar is a frozen dataclass stored as one JSON file, and a DirScalar value is a directory with new, stage, and exec — the same vocabulary cg.ManagedDir has, which is just DirScalar under the managed_dir wire name. Compose them into cg.Struct, cg.Array, and cg.Map as usual. A shape none of those fit subclasses cg.CGType directly and implements load(base) and store(base, value); nothing else in the system needs to know it exists.