Requirements Checker¶

Overview¶

A requirement is a function you register on the RuntimeManager. It is called with the model, the last simulation state, and (at end of trial) the full telemetry parquet file. Mojo calls each registered function in order, collects the results, and writes them to disk.

The function signature is:

1
2
3
4
5
6
7
8

def my_check(
    mojo_model: MojoModel,
    state: MjState,
    df: MojoDataFrame | None,
) -> tuple[bool | None, str]:
    """Every requirement takes `(mojo_model, state, df)` and returns `(passed, message)`."""
    # passed: True = satisfied, False = violated, None = no verdict yet
    return True, "always passes"

• mojo_model: the MojoModel for the trial, giving access to user_data, distributions, named values, etc.
• state: the last MjState seen before the trial ended (access state.data.qpos, state.data.time, etc.)
• df: the full trial telemetry as a MojoDataFrame (a Polars DataFrame with Mojo's signal-aware namespace). None when the function is called during a live step check (see Live checks).
• Return value: (passed, message): the verdict and a human-readable description. passed is tri-state:
  • True: the requirement is satisfied
  • False: the requirement is violated
  • None: no verdict yet. During live checks this is a no-op (nothing is recorded); a check that is still None at end of trial is recorded as failed

Completion states¶

Requirements affect how a trial's Completion is recorded:

Registering requirements¶

There are two equivalent approaches to defining a requirement: using a decorator and direct registration.

Both styles accept the same keyword arguments:

Decorator style¶

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

@rm.requirement()  # name inferred: "check_settle_time"
def check_settle_time(
    mojo_model: MojoModel,
    state: MjState,
    df: MojoDataFrame | None,
) -> tuple[bool, str]:
    """Passes when the simulation ran for at least one second."""
    return state.data.time > 1.0, f"final sim time = {state.data.time:.3f} s"

@rm.requirement("avg_speed_ok")  # explicit name
def check_avg_speed(
    mojo_model: MojoModel,
    state: MjState,
    df: MojoDataFrame | None,
) -> tuple[bool, str]:
    """Passes when the average forward speed met the per-trial target."""
    target = mojo_model.get_user_data(BumperData).min_speed
    assert df is not None
    avg = df["Bodies/Robotxvelp:x"].mean()
    assert isinstance(avg, float)
    return avg >= target, f"avg speed = {avg:.3f} m/s (target >= {target})"

The decorator returns the original function unchanged, so the function remains callable on its own for testing.

Direct registration¶

Given an end-of-trial check:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

def check_max_height(
    mojo_model: MojoModel,
    state: MjState,
    df: MojoDataFrame | None,
) -> tuple[bool, str]:
    """Passes when the robot's peak height stayed under 2 m for the whole trial."""
    assert df is not None  # end-of-trial checks always receive the telemetry frame
    max_z = df["Bodies/Robot/xpos:z"].max()
    assert isinstance(max_z, float)
    return max_z < 2.0, f"max z = {max_z:.3f} m"

register it with add_requirement:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

runtime_manager: RuntimeManager

runtime_manager.add_requirement(
    check_max_height
)  # name inferred: "check_max_height"
runtime_manager.add_requirement(
    check_max_height, name="max_height_ok"
)  # explicit name

with runtime_manager as rm:
    for _ in range(10_000):
        rm.step(state)

# the requirement is checked here upon leaving the with block!

Live checks¶

By default (every=None), requirements are evaluated once at end of trial. Set every=N to also run the check inside step() every N steps:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

@rm.requirement(every=100)
def upright(
    mojo_model: MojoModel,
    state: MjState,
    df: MojoDataFrame | None,
) -> tuple[bool, str]:
    """Passes while the base stays above 5 cm."""
    # df is None during live checks; use state directly
    z_up = float(state.data.qpos[2])
    return z_up > 0.05, f"base height = {z_up:.3f} m"

• every=N: evaluate every N steps. Useful for expensive checks or when only coarse monitoring is needed
• df is always None during live calls; use state for real-time values or cache key values in your mojo_model.user_data object
• Returning None records the undetermined verdict for that sim time (so last_passed() can tell "evaluated, no verdict yet" apart from "not evaluated at this exact step"), but it never posts telemetry, triggers the sticky-failure rule below, or triggers terminate_on_fail/terminate_on_pass. Use it when the check has no verdict yet (e.g. "reached the goal" before the goal is reachable), so an unfinished objective is not mistaken for a violation
• Unless post_result=False, each True/False verdict is recorded in the telemetry parquet under the column Requirements/{name}:result (1.0 = passed, 0.0 = failed), making it plottable in the Dojo alongside other signals. The last known verdict is re-posted on non-evaluation and no-verdict steps so the signal is continuous

Reacting to Live Results in the Loop¶

Live results are cached by simulation time, so your control code can read the outcome of the most recent evaluation at no additional cost:

1
2
3
4
5
6
7

for _ in range(10_000):
    rm.step(state)
    # returns the cached live result at the current sim time, or None if
    # there is no verdict yet. Pass the function itself (returned
    # unchanged by the decorator) instead of retyping its name.
    if rm.last_passed(upright, state) is False:
        apply_recovery(state)

Early termination¶

On Failure¶

Add terminate_on_fail=True to stop the simulation as soon as a live check fails:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

@rm.requirement(every=50, terminate_on_fail=True)
def upright(
    mojo_model: MojoModel,
    state: MjState,
    df: MojoDataFrame | None,
) -> tuple[bool, str]:
    """Terminates the trial if the robot falls over."""
    return float(state.data.qpos[2]) > 0.05, "robot fell over"

try:
    for _ in range(10_000):
        rm.step(state)
except RequirementTerminated as e:
    print(f"stopped early: {e}")

When the check fails, RuntimeManager.step() raises RequirementTerminated (a subclass of SimulationStopped, so existing handlers keep working), which unwinds the simulation loop. The trial is then marked Completion.TERMINATED. End-of-trial requirement evaluation still runs, so all requirements (including terminating ones) appear in requirements.json. If several terminating requirements fail on the same step, all of them are evaluated and their messages are combined into the single exception.

On Success¶

The mirror image: add terminate_on_pass=True to stop the simulation as soon as a live check passes, without spending the rest of the simulation budget. Combine it with None verdicts so the waiting period is not counted as a failure:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

@rm.requirement(every=25, terminate_on_pass=True)
def reached_goal(
    mojo_model: MojoModel,
    state: MjState,
    df: MojoDataFrame | None,
) -> tuple[bool | None, str]:
    """Ends the trial successfully once the base gets within 10 cm of the goal."""
    handoff = mojo_model.get_user_data(BumperData)

    goal = np.asarray(handoff.goal_position)
    dist = float(np.linalg.norm(handoff.base.rt_xipos(state) - goal))

    if dist < 0.1:
        return True, f"goal reached at t={state.data.time:.2f} s"

    # None: no verdict yet; being far from the goal mid-run is not a failure
    return None, f"distance to goal = {dist:.2f} m"

try:
    for _ in range(100_000):
        rm.step(state)
except RequirementSatisfied as e:
    print(f"success, ending early: {e}")

When the check passes, RuntimeManager.step() raises RequirementSatisfied (also a subclass of SimulationStopped). Unlike terminate_on_fail=True, this is a normal completion: end-of-trial evaluation runs as usual and the trial is marked SUCCESS (or FAILURE if some other requirement failed), not TERMINATED.

If a failing terminator and a passing one fire on the same step, the failure wins and the trial terminates as a failure.

Latching Results¶

Sometimes you may have an computationally expensive requirement which is costly to keep checking after it has already passed or failed. The latch_on_pass and latch_on_fail arguments lock in a verdict the first time the check produces it. The function is never called again afterward (neither live nor at end of trial); the locked-in result is replayed for free. Each can be controlled independently.

They are not symmetric, though:

• latch_on_fail: Any live failure already dooms the trial for that requirement (see the note above), so locking it in changes nothing about the trial's outcome, only how many more times fn gets called afterward. Because of this, latch_on_fail defaults to True so live requirements latch on failure automatically. Pass latch_on_fail=False to opt back out (e.g. if you want the function to keep running after a failure purely for its own logging or side effects).
• latch_on_pass: Without this flag, a requirement that passes once but returns fail on a later evaluation is still failed for the trial, no matter what passed beforehand. With latch_on_pass, the check is never run again once it passes, so that later fail is never produced in the first place, and the requirement stays passed. This is the only way to get "once passed, stays passed" behavior without ending the whole simulation via terminate_on_pass. Because it can hide a real later violation, latch_on_pass defaults to False and must be opted into explicitly.

A requirement latching logs a single debug line at the moment it locks in ("requirement 'x' latched pass/fail at t=..."), not on every subsequent step that replays the cached verdict.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# latch_on_fail=True is the default; spelled out here for symmetry with
# latch_on_pass, which must be opted into explicitly
@rm.requirement(every=23, latch_on_pass=True, latch_on_fail=True)
def check_expensive_requirement(
    mojo_model: MojoModel,
    state: MjState,
    df: MojoDataFrame | None,
) -> tuple[bool | None, str]:
    """Runs a very expensive function."""
    value = very_expensive_function(state)

    if value < 2.7128:
        # check_expensive_requirement will always return this value
        return False, "Expensive function was below Euler's number"
    elif value > 2 * 3.1415:
        # check_expensive_requirement will always return this value
        return True, "Expensive function was above 2 pi!"

    # check_expensive_requirement will be run again in 23 timesteps
    return None, "Requirement neither passed nor failed"

Accessing results¶

In-process¶

After the with runtime_manager block exits, the results are available directly on the manager:

1
2

for result in rm.requirement_results:
    print(result.name, result.passed, result.message)

Per-Trial JSON¶

After each trial, Mojo writes a requirements.json to the trial directory:

1
2
3
4

[
  {"name": "max_height_ok", "passed": true,  "message": "max z = 1.73 m"},
  {"name": "avg_speed_ok",  "passed": false, "message": "avg speed = 0.41 m/s (target >= 0.5)"}
]

Runtime Report¶

The standard Mojo job runner aggregates results across all trials. The generated MOJO_RUNTIME_REPORT.md includes a per-requirement table and pass/fail counters

Dojo¶

The Dojo monitor separates successes, requirement failures, and errors in its chips, progress bar, and timeline chart. This provides a graphical way to quickly identify trials with failed requirements.

• Monitor provides a way to view the job requirements at a glance via the top level summary, callouts for the individual trials, and a timeline chart to indicate if any failures are time based
• Mosaic provides a tabular view for a particular trial in the Trial Execution expander which includes any of the logs provided for verdicts

Using user_data in checks¶

Because the model argument is the full MojoModel, you can access your handoff data to make requirements parametric per-trial. Define your handoff model:

1
2
3
4
5
6
7

class BumperData(UserData):
    """Handoff data written during `generate()` and read back inside requirement checks."""

    min_speed: float = 0.5
    min_ground_clearance: float = 0.02
    goal_position: tuple[float, float, float] = (0.0, 0.0, 0.0)
    base: Body

then read it back with get_user_data inside the check:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

@rm.requirement("clearance_ok")
def check_clearance(
    mojo_model: MojoModel,
    state: MjState,
    df: MojoDataFrame | None,
) -> tuple[bool, str]:
    """Passes if the chassis never dipped below the generated clearance threshold."""
    min_clearance = mojo_model.get_user_data(BumperData).min_ground_clearance
    assert df is not None
    lowest_z = df["Bodies/Chassis/xpos:z"].min()
    assert isinstance(lowest_z, float)
    return (
        lowest_z >= min_clearance,
        f"min z = {lowest_z:.4f} (threshold = {min_clearance})",
    )

This makes the same check automatically adapt to each trial's generated configuration.

Using state for final-state checks¶

End-of-trial requirements receive the last state the simulation reached, which is useful for checking where things ended up without needing the full telemetry:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

@rm.requirement("reached_goal")
def check_goal(
    mojo_model: MojoModel,
    state: MjState,
    df: MojoDataFrame | None,
) -> tuple[bool, str]:
    """Passes if the base finished within 10 cm of the goal position."""
    handoff = mojo_model.get_user_data(BumperData)

    goal = np.asarray(handoff.goal_position)
    dist = float(np.linalg.norm(handoff.base.rt_xipos(state) - goal))

    return dist < 0.1, f"final distance to goal = {dist:.3f} m"