Technical Blog

Retriever: A Programming Framework for Closed-Loop Robot Agents

A programming framework for building closed-loop robot systems with explicit timing and data handoff, so perception, reasoning, and control can run together at mismatched rates.

Technical write-up Closed-loop robot agents Multi-rate control

Retriever is a programming framework for general-purpose robot agents. It supports closed-loop systems that combine slow VLM reasoning, medium-rate skill execution, and high-rate control in a single program — each module running on its own independent clock, from seconds down to milliseconds. Timing and data handoff are explicit parts of the program, not buried inside callbacks or middleware behavior.

To show what this looks like in practice, we begin with Retriever-0 — a concrete closed-loop agent we built and deployed with the framework.

Retriever-0: a closed-loop manipulation agent

Retriever-0 is a bimanual manipulation agent built entirely inside a single Retriever pipeline. It runs a VLM planner, an execution monitor, a VLA skill policy, and a joint controller in one coordinated closed loop — each module on its own clock, from seconds down to milliseconds.

Retriever-0 is evaluated on two families of long-horizon tasks that require combining partial observability, conditional replanning, and dexterous manipulation:

Retrieval from a deformable bag. The agent reaches into a cloth bag, locates a target object under uncertainty, and extracts it while continuously adapting its grasp strategy as the bag deforms under manipulation.

Spice search and seasoning. The agent searches a set of drawers for a target spice under partial observability, updates its belief based on what it finds, and completes a multi-step seasoning task using retrieved tools. The planner branches conditionally on inspection outcome, so the execution trace varies depending on which drawer holds the target.

A representative trace from the spice search task:

OpenDrawer(TL) → InspectDrawer(TL) → CloseDrawer(TL) → OpenDrawer(TR) → InspectDrawer(TR) → Pick(spice) → Season(food) → CloseDrawer(TR)

These tasks require a VLM planner reasoning over seconds, a VLA skill policy running at ~2Hz, and a joint controller running at ~200Hz — all exchanging information inside one loop. The execution monitor watches skill progress and triggers replanning when needed; the VLA receives active skill commands and emits action chunks for the controller to execute. No module is isolated; the loop is closed.

Task family

Retrieval under occlusion

The agent searches inside a deformable container, updates belief from new observations, and adapts the manipulation strategy as the scene changes.

Task family

Conditional search and use

The agent inspects candidate locations, replans based on what it finds, and completes a multi-step task using the retrieved object.

Canonical multi-rate closed-loop robot agent pipeline with planning, skill, and control modules. — The Retriever-0 pipeline: slow VLM planning (with belief and memory), execution monitoring, medium-rate VLA skill execution, and high-rate control — all running simultaneously at mismatched clocks inside one program.

# Retriever-0 pipeline — adapted from the actual Python code (names simplified; port mappings omitted)
wrist_cams  = WristCameraFlow()    @ Rate(hz=30)   # 4× wrist cameras (left/right, top/bottom)
head_cam    = HeadCameraFlow()     @ Rate(hz=30)   # stereo head camera
goal        = GoalFlow()           @ Rate(hz=1)
belief      = BeliefMemoryFlow()   @ Trigger("inspection_done")
planner     = VLMPlanFlow("gemini") @ Trigger("replan_request")
monitor     = ExecMonitorFlow()    @ Rate(hz=10)
vla         = VLASkillFlow("pi05") @ Rate(hz=2)
ctrl        = ControllerFlow()     @ Rate(hz=200)
teleop      = TeleopFlow()         @ Rate(hz=100)

with Pipeline("Retriever-0") as pipe:
    # goal → planning and monitoring
    goal.then(planner,        sync=Latest())
    goal.then(monitor,        sync=Latest())

    # perception → belief → planning
    wrist_cams.then(belief,   sync=Latest())
    belief.then(planner,      sync=Latest())
    head_cam.then(planner,    sync=Latest())

    # planning ↔ execution monitoring (replanning loop)
    planner.then(monitor,     sync=Latest())  # plan chunks
    monitor.then(planner,     sync=Latest())  # replan requests

    # monitoring ↔ skill execution
    monitor.then(vla,         sync=Latest())  # active skill command
    vla.then(monitor,         sync=Latest())  # progress p ∈ [0,1]

    # operator override
    teleop.then(monitor,      sync=Latest())

    # skill → control (action chunking handled as sync policy)
    wrist_cams.then(vla,      sync=Latest())
    vla.then(ctrl,            sync=Latest())

# Message passing backend supports in-process, Python multi-processing, and Rust-based dora-rs
retriever.init(backend="dora")
pipe.run(duration=300.0)

Each module is a Flow — a Python class with a step() method — attached to a clock that says when it runs: @ Rate(hz=2) fires on a fixed schedule, @ Trigger(...) fires on an event. The pipe.then(a, b, sync=Latest()) calls wire the graph and declare how each edge samples upstream data at runtime. The rest of this page explains the design behind Retriever that makes this kind of pipeline possible.

Problem: Building general-purpose robot agents

General-purpose robot agents are rarely a single policy call. Long-horizon tasks under partial observability usually turn into a larger system: perception updates what the robot sees, memory tracks what matters, planning proposes what to do next, monitoring checks progress, and control keeps the robot moving safely in the loop.

Those pieces also do not run at the same speed. Fast control loops may run at hundreds of hertz, cameras and state estimation at tens of hertz, and large-model reasoning on the scale of seconds. That is not an edge case — it is the normal shape of a serious robot system. Three challenges make this hard to program:

Challenge 01

Feedback is the rule, not the exception

The agent is inherently cyclic — perception updates belief while skills execute, and monitoring can redirect mid-action.

A long-running agent is not a feedforward pipeline. Perception updates belief while a skill is executing; the planner revises its plan while control is running; monitoring can interrupt or redirect mid-action. These feedback cycles are what make the agent adaptive, but they also mean the program is inherently cyclic — and most stacks force you to implement cycles through ad-hoc callback chains or shared mutable state.

Challenge 02

Modules run at fundamentally different rates

A VLM planner takes seconds; a VLA runs at a few Hz; a controller at hundreds of Hz. These are the basic structure, not edge cases.

If the programming model forces a global tick or blocks the controller on the planner, the robot cannot move safely. The rates differ by orders of magnitude, and each module’s compute time varies — a planner call may take 2 seconds or 10 seconds depending on the scene. The stack must let fast modules keep running while slow modules think.

Challenge 03

Coordination hides in the wrong place

Existing middleware operates at the message-passing level — the important temporal behavior is scattered across callbacks and queues.

Nodes, topics, and callbacks give you message routing, but the programmer manually manages buffering, thread safety, stale-data handling, and coordination logic. The result is that the agent’s temporal behavior — the part that matters for correctness — is not visible in the program. It is an emergent property of callback implementations and queue configurations.

Retriever raises the abstraction to the behavior level: the programmer describes what each module computes and when it runs, while timing, synchronization, and data handoff are explicit, declarative parts of the program. The goal is functional determinism — the same inputs and state always produce the same outputs — so that closed-loop robot programs can be replayed, debugged, and reasoned about.

Design desiderata

Retriever treats each module as a temporal building block — a causal stream function with its own clock. Different wirings and clockings of the same modules compose different agent-level temporal behaviors. Three requirements guide the design:

Desideratum 01

Compositional closed-loop graph

Reusable modules with typed inputs and local state, where feedback loops are visible graph structure — not hidden in callbacks.

Perception, memory, planning, skills, and control should stay separate modules that can still be wired into one closed-loop agent. Feedback cycles — monitoring triggering replanning, new observations updating belief — should be explicit graph structure, not ad-hoc callback glue. The graph is a composition of temporal behaviors: each module contributes its own causal stream function, and the wiring defines how those behaviors interact over time.

Desideratum 02

Explicit local time contract

Each module runs on its own clock. Each edge declares how upstream history is sampled — no fake global tick.

Each module should declare when it runs (periodic, event-triggered, or hybrid) and each edge should declare how upstream data is sampled before a module fires. Planners, skill policies, and controllers should be allowed to run in parallel at very different rates, with no fake global tick forcing synchronization. The timing contract is part of the program, not hidden inside middleware behavior.

Desideratum 03

Deterministic, portable execution

Same inputs and state always produce the same outputs. The same graph runs locally or on deployment backends.

A module should declare how it reads upstream data so replay and debugging use the same consumed snapshots as the original run. For fixed inputs and state, the system should recover the same outputs — functional determinism. The same graph should work for local debugging, multiprocess execution, and distributed deployment without changing the authored program.

Closed-loop by design

Many hierarchical robot agents are organized as a one-way stack: plan, select a skill, execute it, and recover only after something goes wrong. Retriever makes the feedback loops explicit in the program. Perception keeps updating belief and memory while execution is underway; planning can revise plan chunks through the execution monitor; and control continues at high rate using the latest valid action chunks while slower modules reason in the background.

PerceptionBelief / memoryPlanningMonitorSkill / control

feedback updates belief, plans, and control

Perception-planning loop

New observations update belief and memory, so the planner reasons over the latest task state instead of a stale prompt snapshot.

Planning-execution loop

The execution monitor connects plan chunks and skill progress completion, allowing replanning and skill switching while execution continues.

One multi-rate program

Slow planning, medium-rate skill inference, and high-rate control run together because clocks and sync policies define the timing contract.

The interactive timeline below shows how these feedback loops play out over continuous time — scrub across to see how slow planning, medium-rate skill inference, and high-rate control overlap:

Interactive timeline

This is one robot program shown as time moving left to right. Each row is a Flow that wakes up on its own clock; arrows are computation, and blocks are outputs that remain usable until replaced.

t = 1.2s · high-rate control keeps flowing between slow policy calls

camera flow

Rate(10Hz)

belief/memory flow

update belief t = 5sExecuting an information-gathering action produces new evidence, so memory updates before planning continues. update belief t = 10sSkill progress completion updates belief for the next skill switch.

planner

compute plan t = 2sThe planner proposes a plan chunk that can be consumed while execution continues. replan t = 7sSkill progress completion opens a revision step, so the planner reads current belief again.

VLA skill policy

infer action t = 4sThe VLA skill emits a time-extended action chunk plus progress feedback. t = 6sThe skill policy refreshes commands using the latest synchronized inputs. t = 8sThe skill policy continues refreshing short action chunks while control runs. t = 10sSkill progress completion selects the next policy step without stopping control. t = 12sThe final visible action chunk carries the controller toward the end of this window.

controller flow

Dots mark Flow starts. Dashed links show one event causing another; they are not regular global ticks.

Different clocks. Camera, memory, planning, skill inference, and control do not run at the same rate.
Reusable outputs. Memory, plan chunks, and action chunks stay valid until a newer block replaces them.
Continuous control. Feedback can update memory, replan, or switch skills while the controller keeps consuming overlapping action chunks.

clocked procedure start computation time trigger / switching edge plan / action chunk high-rate control tick

Concrete design: Flow

Flow is the programming object that implements Retriever’s design choices. It is analogous to nn.Module in PyTorch: a stateful Python class with typed inputs and outputs that you can instantiate, call directly, and compose with others. The key difference is that PyTorch has no notion of when a module should run or how long computation takes. Retriever adds exactly that — each Flow is attached to a clock that determines its schedule, and each edge carries a sync policy that determines how data is sampled across time boundaries.

We model each Flow formally as a causal stream function (CSF) — a stateful, causal map over continuous event time. At each firing time $t$ , the Flow reads a synchronized input snapshot $x_t$ and updates its internal state:

(h_{t^+},\; y_t) \;=\; f(h_{t^-},\; x_t)

There is no global discrete tick — each Flow fires on its own clock, and $t$ is a continuous-time event. Because CSFs satisfy closure — a directed graph of CSFs is itself a CSF — a pipeline of Flows inherits the same causal semantics as each individual module.

Full formalism: CSF definition, Flow binding, and composition

A CSF is a causal transformation from input streams to output streams: $\mathcal{F}: \mathbf{S}(\mathcal{I}) \to \mathbf{S}(\mathcal{O})$ , defined so that the output at time $t$ depends only on the input history up to $t$ :

s_{out}(t) \;=\; \mathcal{F}\bigl(s_{in}(\cdot)\big|_{\leq t}\bigr)(t)

In the local transition $(h_{t^+}, y_t) = f(h_{t^-}, x_t)$ , the terms are: $h_{t^-}$ and $h_{t^+}$ are the internal state immediately before and after the event, $x_t \in \mathcal{I}$ is the synchronized input snapshot assembled by the sync policy $\sigma$ from upstream histories, and $y_t \in \mathcal{O}$ is the emitted output.

To execute a CSF, Retriever binds it to a clock $\mathcal{C}$ as a runtime Flow:

\texttt{Flow}(\mathcal{F}, \mathcal{C}) : \mathbf{S}(\mathcal{I}) \to \mathbf{S}(\mathcal{O})

Cyclic compositions require strict causality: every directed cycle must contain at least one delay or stateful update, so outputs at time $t$ depend only on events strictly before $t$ .

This CSF model is realized through five concrete programming objects:

Flow

The reusable module boundary. Each Flow owns local state and exposes a synchronous step(), so perception, belief, planning, skills, and control compose without sharing global callback state.

Clock

The run condition for one Flow. Clocks can be periodic, event-triggered, or hybrid; there is no single global tick forced across the robot agent.

Sync policy

The per-edge rule for local synchronization. Policies such as Latest() and Window() define the input snapshot consumed at a Flow step.

Pipeline

The closed-loop graph. Feedback from execution to belief and planning is explicit graph structure, not ad-hoc concurrency inside callbacks.

Runtime / HAL

The backend and hardware abstraction layer. It maps Flows, channels, adapters, and devices while preserving clock and synchronization semantics.

The runtime handles scheduling and alignment; the Flow only sees one aligned input per call:

from dataclasses import dataclass
from retriever import Flow, io

@io
@dataclass
class FlowInput:
    observation: dict
    memory: dict
    goal: str

@io
@dataclass
class FlowOutput:
    action: list[float]
    score: float

class ExampleFlow(Flow[FlowInput, FlowOutput]):
    def __init__(self, model):
        self.model = model

    def step(self, inp: FlowInput) -> FlowOutput:
        action, score = self.model(inp.observation, inp.memory, inp.goal)
        return FlowOutput(action=action, score=score)

flow = ExampleFlow(model)
out = flow(FlowInput(observation, memory, goal))  # direct Python call

A Flow is an ordinary Python object. The step() method runs synchronously over one aligned input; flow(inp) works without any runtime infrastructure. When deployed in a graph, the runtime decides when to call each Flow and which upstream snapshot to pass in — no async API required.

Why this should feel familiar

This is intentionally close to ordinary Python and familiar ML/control patterns: a stateful object, a typed input, a typed output, and one synchronous call. Retriever’s addition is not a new async authoring API; it is that schedules, synchronization, and feedback loops live in the graph around the object.

Same pattern, but as a VLA skill flow

@io
@dataclass
class VLAInput:
    camera: CameraData
    state: RobotState
    skill_cmd: SkillCommand

class VLASkillFlow(Flow[VLAInput, ActionChunk]):
    def __init__(self, processor, policy):
        # simplified — the reference code loads processor/policy from model_name: str
        self.processor = processor
        self.policy = policy
        self.last_chunk = None

    def step(self, inp: VLAInput) -> ActionChunk:
        model_inputs = self.processor.prepare(
            image=inp.camera.rgb,
            proprio=self._encode_state(inp.state),
            instruction=inp.skill_cmd.text,
            previous_chunk=self.last_chunk,
        )
        raw_actions = self.policy.sample_actions(model_inputs, horizon=8)
        chunk = self.processor.decode_chunk(raw_actions, reference_state=inp.state)
        self.last_chunk = chunk
        return chunk

Reference code: concrete VLA flow

@io
@dataclass
class VLAInput:
    camera: CameraData
    state: RobotState
    skill_cmd: SkillCommand

class VLASkillFlow(Flow[VLAInput, ActionChunk]):
    def __init__(self, model_name: str):
        self.processor = load_vla_processor(model_name)
        self.policy = load_vla_policy(model_name)
        self.last_chunk = None

    def reset(self):
        self.last_chunk = None

    def step(self, inp: VLAInput) -> ActionChunk:
        model_inputs = self.processor.prepare(
            image=inp.camera.rgb,
            proprio=self._encode_state(inp.state),
            instruction=inp.skill_cmd.text,
            previous_chunk=self.last_chunk,
        )
        raw_actions = self.policy.sample_actions(model_inputs, horizon=8)
        chunk = self.processor.decode_chunk(raw_actions, reference_state=inp.state)
        self.last_chunk = chunk
        return chunk

# Full pipeline with port mappings — showing how output fields route to input fields
wrist_cams = WristCameraFlow()     @ Rate(hz=30)
head_cam   = HeadCameraFlow()      @ Rate(hz=30)
goal       = GoalFlow()            @ Rate(hz=1)
belief     = BeliefMemoryFlow()    @ Trigger("inspection_done")
planner    = VLMPlanFlow("gemini") @ Trigger("replan_request")
monitor    = ExecMonitorFlow()     @ Rate(hz=10)
vla        = VLASkillFlow("pi05")  @ Rate(hz=2)
ctrl       = ControllerFlow()      @ Rate(hz=200)
teleop     = TeleopFlow()          @ Rate(hz=100)

with Pipeline("Retriever-0") as pipe:
    goal.then(planner,  map={"task": "task"},                       sync=Latest())
    goal.then(monitor,  map={"task": "task"},                       sync=Latest())

    wrist_cams.then(belief,  map={"frame": "frames"},              sync=Latest())
    belief.then(planner,     map={"belief": "state"},              sync=Latest())
    head_cam.then(planner,   map={"frame": "frame"},               sync=Latest())

    planner.then(monitor,    map={"result": "planner_result"},     sync=Latest())
    monitor.then(planner,    map={"replan_trigger": "context"},    sync=Latest())

    monitor.then(vla,        map={"current_skill": "prompt"},      sync=Latest())
    vla.then(monitor,        map={"feedback": "policy_feedback"},  sync=Latest())

    wrist_cams.then(vla,     map={"frame": "obs_dict"},            sync=Latest())
    vla.then(ctrl,                                                  sync=Latest())

    teleop.then(monitor,                                            sync=Latest())

retriever.init(backend="dora")
pipe.run(duration=300.0)

Sync policy

Synchronization policies over edges in a temporal dataflow graph. — Each edge declares how data is sampled before a Flow runs, replacing implicit “latest message” behavior with explicit rules such as `Latest()` and `Window()`.

When a Flow wakes, the runtime samples each incoming stream with its declared policy, builds one aligned local input, and then calls the same class:

from retriever import Latest
from retriever.flow import Window
from retriever.flow.types import EventStream

camera_events: EventStream[CameraData]
state_events: EventStream[RobotState]
skill_events: EventStream[SkillCommand]

vla_inp = VLAInput(
    camera=camera_events.sample(Latest(), now=t),
    state=state_events.sample(Latest(), now=t),
    skill_cmd=skill_events.sample(Latest(), now=t),
)

recent_frames = camera_events.sample(
    Window(buffer_size=16, duration=0.5),
    now=t,
)

next_chunk = vla_flow(vla_inp)

Retriever provides a set of primitive sync policies — each one a deterministic rule for sampling upstream history into one input snapshot:

Latest

Sample-and-hold the most recent valid value, with optional staleness bounds.

Window

Collect a bounded buffer of recent events, frames, or state estimates.

Join

Assemble sampled edge values into the single local input record for a Flow.

Per edge

Use different policies on different incoming connections instead of imposing one graph-wide synchronization rule.

How sync policies align inputs

For each edge $u \to v$ , the policy $\sigma$ looks at the upstream history available before that firing time and returns one snapshot value for Flow $v$ . The runtime applies $\sigma$ independently on every incoming edge, assembles the aligned bundle, then calls step() exactly once.

Operational detail: lag, deadlines, and safe fallback

When compute cannot keep up, Retriever makes the fallback rule explicit instead of leaving it to queue timing:

drop or ignore inputs that have become too stale
declare missing or expired inputs
trigger a safe fallback such as replanning, terminating a skill, or handing over to teleoperation
raise a hard error in debug mode when the timing contract is violated

This is the same design logic as sync policies: slow-to-fast and fast-to-slow boundaries need explicit contracts rather than middleware luck.

Temporal chunking

Sync policies handle how a downstream Flow samples its inputs. But when a slow module needs to feed a much faster one, there is a related problem: the fast consumer cannot wait for each slow inference. Temporal chunking addresses this by letting slow producers emit time-extended outputs that remain valid across multiple fast consumer steps.

Slow producers emit reusable chunks so faster consumers can stay on their own clock. In the Retriever-0 pipeline this appears in two ways:

Action chunking (VLA → controller) — the VLA (2Hz) emits an ActionChunk so the controller (200Hz) can run without waiting for each policy inference.
Plan chunking (planner → execution monitor) — the VLM planner proposes bounded-horizon plan chunks asynchronously; the monitor commits them only when skill progress $p \ge 0.9$ is sustained for ~0.5s, preventing mid-skill plan rewrites.

Plan chunking and progress-based skill switching

The pipeline uses two additional mechanisms on top of temporal chunking:

Information-gathering branching. The planner can branch on new evidence, such as IF drawer is empty THEN pick spice ELSE close drawer.
Plan chunking. The planner proposes short bounded-horizon chunks while the current skill is still executing, so VLM reasoning overlaps with action instead of blocking it.
Skill progress prediction. The monitor uses a dense progress score $p \in [0,1]$ to decide when a skill is effectively complete and when it is safe to commit the next plan chunk.

Theory: why this graph model is enough

The Flows, clocks, and sync policies described above are not just engineering conventions — they rest on a formal property that we prove in the paper. Because each Flow is a causal stream function, and a graph of CSFs is itself a CSF (closure), the entire pipeline is a well-defined composition of temporal behaviors. This means different wirings and clockings of the same modules produce different agent-level behaviors, but each composition retains functional determinism: the same inputs and state always yield the same outputs, enabling principled replay and debugging.

Theorem statement

Any finite-memory causal policy over asynchronous observation histories can be represented by a Retriever graph composed of primitive stream operators, explicit clocks, and deterministic synchronization policies.

Consequently, closed-loop robot pipelines can be written as temporal dataflow graphs without losing the agent-level timing semantics needed for deterministic replay and debugging.

From graph to runtime

Retriever separates graph authoring from execution:

Definition (Python) — users write Flow classes, attach clocks and triggers, and declare how edges should sample upstream data.
Compilation (IR) — the authored graph is validated, port mappings and adapters are resolved, and the result becomes a static representation of topology, clocks, and edge policies.
Execution — a backend lowers that IR into workers, channels, buffers, and scheduling logic without changing the authored graph.

The authored graph stays stable while the execution mapping changes. In practice that means the same pipeline can be stepped in-process for debugging, run on local multiprocessing, or lowered onto Dora without rewriting Flow code. The compiled graph becomes explicit runtime data — nodes, clocks, adapters, queue sizes, and topology all inspectable:

Definition

Write Flows, clocks, ports, and edge policies as one agent graph. No runtime placement is implied yet.

Clear semantics

Clocks and sync policies define which input history each Flow consumes, so behavior is a deterministic function of event-time histories.

Runtime mapping

Dispatch the graph to local stepping, multiprocessing, or a supported transport backend while preserving the same consumed-input contract.

Hardware abstraction

Device and transport adapters sit below the graph, so hardware-specific behavior is not treated as part of the algorithm.

Illustrative IR snippet

{
  "version": "1.0.0",
  "metadata": {
    "name": "Agent",
    "validated": true,
    "optimized": false
  },
  "nodes": [
    {
      "id": "camera",
      "type": "CameraSourceFlow",
      "config": {"clock": {"Rate": {"hz": 30}}}
    },
    {
      "id": "vla",
      "type": "VLASkillFlow",
      "config": {"clock": {"Rate": {"hz": 2}}}
    }
  ],
  "edges": [
    {
      "source": {"node": "camera", "port": "rgb"},
      "destination": {"node": "vla", "port": "camera"},
      "adapter": {"latest": {"buffer_size": 1}},
      "qsize": 10
    }
  ],
  "topology": {
    "node_count": 2,
    "edge_count": 1,
    "has_cycle": false
  }
}

Execution modes: in-process, multiprocess, and Dora

The same authored graph can be lowered into several execution modes:

in-process for local debugging and deterministic stepping, where pipe.step() lets you inspect one logical firing at a time
multiprocess when you want isolation between modules on one machine while keeping the same authored graph
Dora-backed execution when you want the same graph on a different runtime substrate with explicit transport underneath

What changes between those modes is the execution mapping underneath the graph, not the authored graph itself. The Python Flow classes, the clocks, and the edgewise sync policies stay fixed; the runtime decides how to place and schedule them.

That is the point of the IR layer: it gives the runtime one explicit object to lower. Backends do not have to rediscover timing intent from callback order or infer data handoff conventions from user code; they receive node definitions, clock declarations, adapters, queue settings, and topology directly.

That separation is why Retriever can support:

pipe.step() for local inspection
backend execution without rewriting Flow code
record/replay workflows for in-process debugging and dataset extraction

The key difference with pub/sub stacks (ROS2, Dora, and similar middleware) is where coordination complexity lives — and whether the system gives you functional determinism: the same inputs and state always produce the same outputs, regardless of wall-clock timing.

In Retriever, the sync policy defines a deterministic input snapshot at each firing. In pub/sub systems, what a node reads depends on arrival order, callback scheduling, and queue state — so the same logical inputs can produce different consumed snapshots across runs.

RetrieverPub/Sub (ROS2, Dora)

Authoring model

Ordinary step() code over one aligned input

Async coordination logic spread across callbacks, timers, and topic state

Time semantics

Deterministic: explicit clocks and sync policies define which input history each module consumes

Arrival-time: consumed inputs depend on callback ordering, queue depth, and scheduler timing

Functional determinism

Same inputs + same state → same outputs; replay recovers the same consumed snapshots

Behavior can vary across runs even with identical inputs due to nondeterministic message ordering

Variable latency

Declared via clocks, triggers, and buffer policies

Manual worker threads, queues, drop logic, and stale-data handling

Illustrative code: Retriever vs. ROS2

The snippets below omit message definitions and model-loading boilerplate. They show where timing, buffering, and launch logic end up.

from dataclasses import dataclass

from retriever import Flow, Pipeline, Rate, Latest, io
from retriever.flow import Trigger

@io
@dataclass
class VLAInput:
  camera: "CameraFrame"
  state: "RobotState"
  active_skill: "SkillCommand"

class PlannerFlow(Flow["BeliefState", "PlanChunk"]):
  def step(self, belief):
      return self.vlm_plan(belief)

class VLASkillFlow(Flow[VLAInput, "ActionChunk"]):
  def step(self, inp):
      return self.vla(inp.camera, inp.state, inp.active_skill)

with Pipeline("Agent") as pipe:
  cam = CameraSourceFlow() @ Rate(hz=30)
  belief = BeliefUpdaterFlow() @ Trigger("observation")
  planner = PlannerFlow("gemini") @ Trigger("state")
  monitor = ExecutionMonitorFlow() @ Trigger("executor_status")
  vla = VLASkillFlow("pi05") @ Rate(hz=2)
  ctrl = ControllerFlow() @ Rate(hz=200)

  pipe.connect(cam, belief, sync=Latest())
  pipe.connect(belief, planner, sync=Latest())
  pipe.connect(planner, monitor, sync=Latest())
  pipe.connect(cam, vla, sync=Latest())
  pipe.connect(monitor, vla, sync=Latest())
  pipe.connect(vla, ctrl, sync=Latest())

if __name__ == "__main__":
  pipe.run(backend="dora", duration=30.0)

Design lineage

Retriever is motivated by classic systems ideas, but the robotics setting changes the contract. Robot agents are cyclic, stateful, physical, and latency-sensitive; their graphs must say not only what data flows, but when data is valid and how slow decisions overlap fast control.

UNIX

Composable processes. Small units become powerful when connected by simple streams; Retriever keeps that compositional instinct while adding clocks, feedback, and typed temporal handoff.

PyTorch

Graphs as program structure. ML frameworks make computation graphs concrete enough to inspect and execute; Retriever uses a graph boundary for stateful robot Flows over time.

Ray

Actors as runtime units. Actor systems are a natural execution model for independent modules; Retriever uses that lesson below the programming model while preserving graph semantics.

ROS

Robotics middleware. Robot systems need practical message passing and device integration; Retriever focuses on the agent-level timing and handoff contract above that substrate.

What we tested

We show three kinds of evidence below: runtime overhead, replay behavior under timing stress, and path-sensitive dynamics where small timing changes can alter the executed computation itself.

Overhead

Message latency benchmark

Does explicit clocks-and-sync structure add much cost? Retriever with a Dora backend stays close to native transport baselines across payload sizes.

Latency benchmark results across payload sizes. — End-to-end message latency versus payload size (log-log). Retriever with Dora backend stays close to native `dora-rs`.

Stress test

Hybrid differentiable physics

What if the executed path itself changes under tiny timing differences? This example makes replay sensitive by mixing continuous dynamics with discrete contact events.

Hybrid physics stress test for asynchronous determinism. — A bouncing-ball system with discrete impact events, used to stress trace sensitivity.

Replay

Path-gradient consistency

Does replay converge to one execution trace or drift across many? Event-time semantics collapse repeated runs onto one trace; arrival-time pub/sub spreads them apart.

Gradient histogram from asynchronous replay runs. — Event-time semantics collapse path gradients to a single trace; arrival-time semantics spread across multiple.

Conclusion

Retriever raises robot agent programming from the message-passing level to the behavior level. Each module is a causal stream function with its own clock; composing them into a graph produces a well-defined composition of temporal behaviors with functional determinism — the same inputs and state always yield the same outputs. Clocks and edgewise synchronization policies are explicit parts of the program, so closed-loop pipelines stay modular and replayable across mismatched rates and runtimes. The experiments show that overhead stays low and replay is substantially more stable than arrival-time pub/sub.