Customize termination¶

cfg.termination_fn is called every step with the pre-step state, the new state, the cfg, and the new time, and returns a scalar bool array. Both states are passed so a termination can resolve events that peak between two decision samples. Once it fires, the episode ends and per-side done fields broadcast that scalar (see In depth → Symmetric core).

Two paths to swap the termination:

1. Pass it as a constructor kwarg — ScenarioConfig(..., termination_fn=MyTermination()). __post_init__ only consults cfg.game.default_termination_fn() when termination_fn is None.
2. dataclasses.replace — cfg = replace(cfg, termination_fn=MyTermination()).

Where defaults come from¶

Every Game subclass declares its default termination via default_termination_fn. ScenarioConfig.__post_init__ calls it when cfg.termination_fn is None. The mapping is:

The protocol¶

Bundled implementations¶

MaxStepsOnly is the universal step-cap termination — it ends the episode when state.step >= params.max_steps. It reads max_steps from the cfg at call time, so it can be constructed without a cfg in scope. NoGame, SunBlocking, and ObservationBlocking use it as their default:

MaxStepsOnly()

LbgEventTermination adds the bandit-breach and guard-catch events on top of the step cap; it's LadyBanditGuard's default and reads breach_radius_m / catch_radius_m plus the optional relative-speed gates breach_speed_mps / catch_speed_mps from its own fields (which LadyBanditGuard.default_termination_fn threads from the game knobs). Each event is evaluated over the whole step by treating the relative motion as a straight line between samples, so a slow encounter that never lands inside the radius at a sample still fires. catch_dwell_steps and breach_dwell_steps turn either event into a hold: the event fires once a bandit has been inside that radius for that many consecutive steps, counted on the environment state, and zero (the default) fires on the first step inside it:

LbgEventTermination(
    breach_radius_m: float,
    catch_radius_m: float = 0.0,
    breach_speed_mps: float = float("inf"),
    catch_speed_mps: float = float("inf"),
    catch_dwell_steps: int = 0,
    breach_dwell_steps: int = 0,
    escape_radius_m: float = 0.0,
    repel_on_empty_tank: bool = False,
)

PursuitEvasion ships PursuitEvasionTermination that fires on max_steps OR capture (relative distance below cfg.game.capture_distance_m).

Worked example A: max-steps only (no breach)¶

    import dataclasses
    from dataclasses import dataclass

    import jax.numpy as jnp

    from orbitalgym import OrbitalGymEnv, SingleAgentView, make_lady_bandit_guard
    from orbitalgym.termination.reference import MaxStepsOnly

    # The framework already ships `MaxStepsOnly` (reads `params.max_steps`
    # at call time, no stored field). Wire it onto the cfg to drop LBG's
    # event-driven termination and keep only the step-cap.
    cfg = make_lady_bandit_guard(seed=0, max_horizon_s=200.0)
    cfg = dataclasses.replace(cfg, termination_fn=MaxStepsOnly())

Worked example B: AND-composite of two existing conditions¶

    @dataclass(frozen=True)
    class AndComposite:
        """Episode ends only when BOTH conditions are met (rare; usually OR)."""

        first: object
        second: object

        def __call__(self, prev_state, state, params, t):
            return jnp.logical_and(
                self.first(prev_state, state, params, t),
                self.second(prev_state, state, params, t),
            )

AndComposite accepts any two callables of the right shape and ANDs their results. Use it sparingly — most game-design reasons want OR composition, which you can write the same way with jnp.logical_or.

Wiring is the usual dataclasses.replace:

    cfg = dataclasses.replace(
        cfg,
        termination_fn=AndComposite(MaxStepsOnly(), MaxStepsOnly()),
    )
    env = OrbitalGymEnv(cfg)

Per-game applicability¶

See also¶

• API → Components → Termination
• In depth → Symmetric core — how termination interacts with the latched episode_done mask.