Time and Sync

What you’ll learn: how a graph decides when each Flow runs (clocks) and which upstream record it consumes (sync policies) — the two knobs that let cameras, controllers, and slow policies coexist without a shared timestep.

Robots do not run on one global step. Cameras stream, controllers run fast, planners block, policies have variable latency, operators intervene at irregular times. Retriever makes that explicit with two independent decisions per Flow:

Decision	Made by	Question it answers
Clock	@ Rate(...), @ Trigger(...), @ Hybrid(...)	When does this Flow wake up?
Sync policy	sync= on each incoming edge	Which buffered record becomes its input?

Clocks decide when

A Flow fires at its own clock times. There is no shared k that means “the next step” for every Flow.

• Rate(hz=30) — periodic; fire every 1/30 s.
• Trigger("image") — event-driven; fire when the image field arrives on an incoming edge.
• Hybrid(hz=10, "replan") — periodic and event-driven; fire on the timer or immediately on the named field.

from retriever import Latest, Pipeline, Rate, Trigger
from examples.shared.perception_flows import CameraSource, ColorDetector, DisplayFlow

pipe = Pipeline("tutorial.perception")
with pipe:
    camera   = CameraSource(use_real_camera=False) @ Rate(hz=30)
    detector = ColorDetector(min_confidence=0.6)    @ Trigger("image")
    display  = DisplayFlow(display="stdout")         @ Rate(hz=3)

    camera.then(detector, sync=Latest())
    detector.then(display, sync=Latest())

The camera source wakes periodically; the detector wakes when a frame arrives; the display wakes at its own slower rate. The .then(..., sync=...) calls are the reactive graph edges: upstream event history flows forward, and the sync policy decides which record is visible when the downstream Flow wakes. Run it:

retriever run webcam-mock

Building perception pipeline:
  Camera @ Rate(20Hz) -> ColorDetector @ Trigger -> Display @ Rate

✓ Graph created: 3 nodes, 5 edges

Running for 0.1 seconds...
Tip: This run is using mock frames. Use --camera-mode real to require a live webcam path.
------------------------------------------------------------
  Frame 1: 2 objects - [('red_object', '0.95'), ('blue_object', '0.95')]

What happens at a firing time

When a Flow’s clock fires at time t, the runtime samples each incoming edge’s buffer, assembles one typed input object, and calls the synchronous step(...). The rest of the graph may be sleeping, catching up, or firing on other clocks.

# Conceptual loop for ONE Flow, not a global robot timestep.
def on_flow_clock_tick(t):
    image = Latest()(camera_edge_buffer, now=t)   # sync policy samples the edge
    obs = SkillObs(image=image)
    out = detector.step(obs)                        # your Flow.step()
    detection_edge.append((t, out))

Each Flow has its own schedule. Each edge declares how upstream history becomes one aligned input.

Sync policies decide which record

Every edge accumulates a timestamped buffer of upstream outputs. The sync= policy consumes that buffer at firing time t and returns exactly one value for step(...):

Policy	What it returns
Latest()	The most recent value in the buffer.
Window(buffer_size, duration, agg)	An aggregate (mean/last/max/min/first) over the last duration seconds.
Hold(debounce=...)	Zero-order hold of the last value, optionally rate-limited.
Events(buffer_size, duration)	A slice of recent history, for Flows that reason over a stream.

Latest() is right when one fresh value is enough. Window(...) is right when a slower Flow should see a short history summarized at its own clock time — the classic fast-producer, slow-consumer bridge:

from retriever import Flow, Pipeline, Rate, Window, io

sensor   = Sensor()   @ Rate(hz=30)   # fast producer
smoother = Smoother() @ Rate(hz=10)   # mid-rate consumer
printer  = Printer()  @ Rate(hz=1)    # slow observer

# Sample the last 0.5 s of the 30 Hz stream and take its mean.
# buffer_size must cover hz * duration = 30 * 0.5 = 15 samples.
sensor.then(smoother, sync=Window(buffer_size=50, duration=0.5, agg="mean"))
smoother.then(printer, sync=Latest())

retriever run multirate

[t= 1.0s] x_mean=-0.671
[t= 2.0s] x_mean=-0.628
[t= 3.0s] x_mean=-0.530
[t= 4.0s] x_mean=-0.695

The 30 Hz sensor, 10 Hz smoother, and 1 Hz printer never share a step. The Window on the edge is what turns a fast stream into a value the slow printer can consume — and it does so from the buffered records, not from global state.

Why this is the determinism boundary

The clock is where wall-clock nondeterminism lives: exactly which frames land in a window depends on scheduling, so the numbers above shift run to run. A sync policy is a function of the buffered, timestamped records and their defined order. That separation is Retriever’s functional determinism boundary: replay the same ordered input history and every sync samples the same records, so deterministic step() bodies produce the same discrete-event outputs. Live scheduling chooses the history; recording makes that history available to replay and verify.

When timing looks wrong

Symptom	Check first	Why
A Flow never runs	Its clock — Rate, Trigger fields, or Hybrid.	A Flow only wakes when its own clock fires.
A Flow runs on stale data	The edge sync= and buffer size.	Input is sampled from buffered history, not magic global state.
A slow model blocks progress	Split it into its own Flow with a clock/sync boundary.	Latency should be visible in the graph, not hidden in a callback.
A run won’t reproduce	Render the graph, then record/replay inputs.	Debug from explicit clocks, ports, and buffers.

Next: Runtime connects clocks and sync to validation, in-process stepping, backends, and replay.