Runtime

What you’ll learn: the path from an authored graph to a running one — validate() compiles a Pipeline to an IR, step() advances one debugger-friendly graph tick, and run(backend=...) schedules the same IR on multiprocessing or Dora. Given the same ordered timestamped input history, deterministic Flows produce the same discrete-event output history.

A Pipeline is the executable graph. You normally wire Flows with the fluent source.then(target, sync=...) form; the examples on this page use the explicit connect(source, target, sync=...) equivalent because runtime validation and IR examples benefit from naming both endpoints. After wiring, the same object exposes four surfaces:

from retriever import Latest, Pipeline, Rate, Trigger, connect

with Pipeline("agent") as pipe:
    camera   = CameraSource(id=0) @ Rate(hz=30)
    detector = ColorDetector()    @ Trigger("image")
    connect(camera, detector, sync=Latest())

ir = pipe.validate()                 # 1. graph -> validated IR
pipe.visualize("artifacts/agent.html")  # 2. render structure, clocks, ports, edges
pipe.step(dt=0.1)                    # 3. run in-process (breakpoints work)
pipe.run(backend="dora")            # 4. deploy asynchronously

validate() compiles the graph to an IR

validate() checks the wiring — ports exist, source and destination types match, clocks and adapters are well-formed — and returns an IR: the validated logic graph, with every node’s clock and every edge’s adapter made explicit. The IR is the artifact backends execute, visualizers render, and Hub modules load.

import json
from retriever import Flow, Latest, Pipeline, Rate, Trigger, connect, io
from retriever.ir.core import IR
from retriever.error import IRError

@io
class Data:
    value: int

class SourceFlow(Flow[None, Data]):
    def step(self, _): return Data(value=42)

class ProcessFlow(Flow[Data, Data]):
    def step(self, input: Data): return Data(value=input.value * 2)

class SinkFlow(Flow[Data, None]):
    def step(self, input: Data): return None

with Pipeline("pipeline") as pipe:
    source  = SourceFlow()  @ Rate(hz=10)
    process = ProcessFlow() @ Trigger("value")
    sink    = SinkFlow()    @ Trigger("value")
    connect(source, process, sync=Latest())
    connect(process, sink, sync=Latest())

ir = pipe.validate()
print("version:", ir.version, "| nodes:", ir.topology.node_count, "| edges:", ir.topology.edge_count)
for n in ir.nodes:
    print(f"  node {n.id:12s} clock={list(n.config['clock'])[0]:8s} outputs={n.outputs}")
for e in ir.edges:
    print(f"  edge {e.source.node} -> {e.destination.node}  adapter={e.adapter}")

# The IR serializes to JSON and reloads...
reloaded = IR.from_json(ir.to_json())
print("round-trip ok:", reloaded.topology.node_count == ir.topology.node_count)

# ...and loaded IR is validated too. Tamper with it: duplicate a node id.
bad = json.loads(ir.to_json())
bad["nodes"].append(dict(bad["nodes"][0]))
try:
    IR.from_json(json.dumps(bad))
except IRError as exc:
    print("rejected tampered IR:", exc)

version: 1.0.0 | nodes: 3 | edges: 2
  node SourceFlow   clock=Rate     outputs={'value': 'int'}
  node ProcessFlow  clock=Trigger  outputs={'value': 'int'}
  node SinkFlow     clock=Trigger  outputs={}
  edge SourceFlow -> ProcessFlow  adapter={'Latest': {'buffer_size': 1}}
  edge ProcessFlow -> SinkFlow  adapter={'Latest': {'buffer_size': 1}}
round-trip ok: True
rejected tampered IR: [IR_VAL_INVALID]: Duplicate node id 'SourceFlow'

retriever run ir-validation   # prints the full IR as JSON

IR.from_json(...) re-runs structural validation on load, so IR read from disk or a Hub module is not trusted blindly — duplicate node ids, dangling edge or adjacency references, and unknown ports fail with a named node or edge before a backend tries to execute the graph.

Local step() is a debugger-friendly graph tick

pipe.step(dt=...) advances the graph in one process: it samples each edge, calls each Flow.step(), and publishes outputs. This is intentionally a local debug tick, not a full backend scheduler: Rate(...) flows execute once per step() call, while Trigger(...) flows still wait for new inputs. Breakpoints inside step() fire in your own process, and the trace is reproducible. Use an async backend when you want wall-clock Rate(...) scheduling.

from retriever import Flow, Latest, Pipeline, Rate, Trigger, connect, io

@io
class Value:
    n: int

class Ramp(Flow[None, Value]):
    def reset(self): self.k = 0
    def step(self, _): self.k += 1; return Value(n=self.k)

class Double(Flow[Value, Value]):
    def step(self, v): return Value(n=v.n * 2)

class Recorder(Flow[Value, None]):
    def reset(self): self.trace = []
    def step(self, v): self.trace.append(v.n); return None

def run_once():
    recorder = Recorder()
    with Pipeline("determinism") as pipe:
        ramp = Ramp()   @ Rate(hz=10)
        dbl  = Double() @ Trigger("n")
        rec  = recorder @ Trigger("n")
        connect(ramp, dbl, sync=Latest())
        connect(dbl, rec, sync=Latest())
    for _ in range(5):
        pipe.step(dt=0.1)
    pipe.close_stepper()
    return recorder.trace

print("run A:", run_once())
print("run B:", run_once())

run A: [2, 4, 6, 8, 10]
run B: [2, 4, 6, 8, 10]

Two independent runs produce the identical trace because they use the same fixed input history, deterministic Flow.step() bodies, and the same graph state. This is Retriever’s functional determinism boundary: fix the ordered timestamped input history and the discrete-event output history is fixed too.

The contract, precisely. Fix a graph with deterministic Flow.step() bodies and sync policies, fixed initial states, strictly causal cycles, and the same ordered timestamped external input history; replay stochastic model calls from logged seeds or logged outputs. Then the discrete-event output history is uniquely determined. Latest(now=t) selects the event with the greatest timestamp satisfying timestamp <= t; without now, it selects the greatest timestamp in the buffer. Buffers retain timestamp order even when a transport drains fan-in queues in another arrival order. Events with equal timestamps currently preserve insertion order, so scheduler-independent replay requires that tie order to be part of the recorded input history (or that producers attach a deterministic sequence key).

Latest() uses event time, not append order

Suppose a transport appends these records out of order:

append (0.03, "newer")
append (0.01, "older")

Retriever orders the buffer by timestamp before sampling:

Sample	Result	Reason
Latest()	"newer"	0.03 is the greatest timestamp.
Latest(now=0.02)	"older"	0.01 is the greatest timestamp at or before 0.02.
Latest(now=0.005)	raises IndexError	No event is available at that logical time.

Boundaries worth knowing:

• step(now=...) is exact relative to the supplied ordered input history. step(dt=...) keeps logical deltas stable but anchors the first tick at the stepper’s starting time.
• Determinism concerns the discrete event path, not bitwise GPU numerics.
• A live async run (multiprocessing, dora) does not replay its wall-clock scheduling. Its scheduling determines the captured input history; replay reproduces that history, not the original process schedule.
• Equal timestamps require a fixed tie order. The runtime resolves ties as last-inserted on every sampling path; use monotonic timestamps or sequence keys when multiple producers can emit at exactly the same time.
• An event stamped after the sampling tick is not yet visible: the field reads as absent that tick and the event is delivered once now passes its timestamp. On distributed transports this also absorbs producer/consumer clock skew instead of erroring.

The idea has deep roots: Kahn process networks showed deterministic processes over ordered channels compose deterministically (Kahn 1974), and functional reactive programming made time-indexed streams a first-class semantics (Elliott & Hudak 1997, Wan & Hudak 2000). The Retriever paper states this property as the functional determinism theorem and proves it as trace determinism.

retriever run stepper   # the shipped in-process stepper, with a --fail-at breakpoint

Same IR, different backend

The graph you debug in-process is the graph you deploy. Only the backend changes:

Backend	What it does	Use for
in-process	One process; step() / run() in your interpreter.	Debugging, tests, deterministic replay.
multiprocessing	One OS process per Flow, local channels.	Local async execution and parallelism.
dora	Compiles to a Dora dataflow across processes/machines.	Deployment; distributed, high-rate control.

pipe.run(backend="multiprocessing", duration=3)
pipe.run(backend="dora", duration=3)

retriever run multirate         # multiprocessing
retriever run webcam-dora   # dora

Record and replay turn runs into evidence

Recording preserves the consumed, ordered input history. Replaying that history through deterministic Flows reconstructs the same discrete-event outputs; stochastic model calls must come from logged seeds or logged outputs.

retriever run record-replay     # record a run, then replay it step-for-step
retriever run incident-replay   # replay a captured incident and diff the diagnosis

Record with pipe.record(...) (in-process) to MCAP or RRD; replay drives the same graph from the recording instead of live inputs. See Debug and Visualize for the full record/replay walkthrough.