Skip to content

zymera.missions_terms

missions_terms

The proven reward-term zoo, ported from zymera v0's reward block (examples/comm_coverage.py step) and examples/cbf.py.

Every term is (prev_world, world, action, ctx) -> (N,) float32 returning an UNSIGNED, UNWEIGHTED magnitude — penalties get their sign from the :class:~zymera.missions.RewardTerm weight, exactly mirroring v0's reward = w_cov·newly + w_conn·conn − w_overlap·overlap − w_coll·coll (so :data:DEFAULT_TERMS carries -4.0 on the collision term).

Parameterized terms are factories returning a term function. Each function (including factory outputs) carries a .requires frozenset attribute naming the :class:~zymera.metrics.StepCtx fields it reads — pass it through to RewardTerm(requires=...) so the env derives those fields. The CBF terms compute from positions directly (requires = ∅): their λ₂ eigendecomposition only compiles into envs that actually use them.

new_coverage 

new_coverage(prev, world, action, ctx)

(N,) — cells of my footprint this step the TEAM had not covered before (v0 newly).

Source code in zymera/missions_terms.py 
49
50
51
52
53
54
@_requires("newly_covered")
def new_coverage(prev, world, action, ctx) -> chex.Array:
    """(N,) — cells of my footprint this step the TEAM had not covered before
    (v0 ``newly``)."""
    del prev, world, action
    return ctx.newly_covered

reach_fraction 

reach_fraction(prev, world, action, ctx)

(N,) — fraction of OTHER agents reachable through the potential comm graph (v0 per-agent connectivity, the conn_cap=None branch).

Source code in zymera/missions_terms.py 
57
58
59
60
61
62
63
@_requires("reach")
def reach_fraction(prev, world, action, ctx) -> chex.Array:
    """(N,) — fraction of OTHER agents reachable through the potential comm
    graph (v0 per-agent ``connectivity``, the ``conn_cap=None`` branch)."""
    del prev, action
    n = world.n_agents
    return ((ctx.reach.sum(-1) - 1) / max(n - 1, 1)).astype(jnp.float32)

capped_giant 

capped_giant(cap)

Factory — (N,) shared min(giant, cap) / cap where giant is the largest-component size (v0 conn_cap branch: cap=N−1 ⇒ "3 connected + 1 roamer" scores like a full clump → no clump incentive).

Source code in zymera/missions_terms.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def capped_giant(cap: int) -> Callable:
    """Factory — (N,) shared ``min(giant, cap) / cap`` where ``giant`` is the
    largest-component size (v0 ``conn_cap`` branch: cap=N−1 ⇒ "3 connected +
    1 roamer" scores like a full clump → no clump incentive)."""
    if cap < 1:
        raise ValueError(f"cap must be >= 1, got {cap}")

    @_requires("reach")
    def fn(prev, world, action, ctx) -> chex.Array:
        del prev, action
        giant = ctx.reach.sum(-1).max()
        val = (jnp.minimum(giant, cap) / cap).astype(jnp.float32)
        return jnp.broadcast_to(val, (world.n_agents,))

    return fn

collision_count 

collision_count(prev, world, action, ctx)

(N,) — number of OTHER agents sharing my cell. Unsigned magnitude: weight it negatively (v0 subtracted w_coll·collisions).

Source code in zymera/missions_terms.py 
83
84
85
86
87
88
@_requires("collisions")
def collision_count(prev, world, action, ctx) -> chex.Array:
    """(N,) — number of OTHER agents sharing my cell. Unsigned magnitude:
    weight it negatively (v0 subtracted ``w_coll·collisions``)."""
    del prev, world, action
    return ctx.collisions

same_step_overlap 

same_step_overlap(prev, world, action, ctx)

(N,) — cells of my footprint this step ANOTHER agent also covers now (v0 anti-redundancy; weight negatively).

Source code in zymera/missions_terms.py 
91
92
93
94
95
96
@_requires("overlap")
def same_step_overlap(prev, world, action, ctx) -> chex.Array:
    """(N,) — cells of my footprint this step ANOTHER agent also covers now
    (v0 anti-redundancy; weight negatively)."""
    del prev, world, action
    return ctx.overlap

cohesion_leash 

cohesion_leash(leash, comm_r)

Factory — (N,) max(nearest-teammate-dist − leash, 0), with the nearest distance clamped at comm_r (an agent can't measure a teammate it can't sense). Soft tether; weight negatively.

Source code in zymera/missions_terms.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def cohesion_leash(leash: float, comm_r: int) -> Callable:
    """Factory — (N,) ``max(nearest-teammate-dist − leash, 0)``, with the
    nearest distance clamped at ``comm_r`` (an agent can't measure a teammate
    it can't sense). Soft tether; weight negatively."""

    @_requires("dist")
    def fn(prev, world, action, ctx) -> chex.Array:
        del prev, action
        off = ~jnp.eye(world.n_agents, dtype=bool)
        d_off = jnp.where(off, ctx.dist, jnp.inf)
        nn = jnp.minimum(d_off.min(-1), float(comm_r))
        return jnp.maximum(nn - leash, 0.0).astype(jnp.float32)

    return fn

degree_floor 

degree_floor(floor, comm_r)

Factory — (N,) max(floor − in-range-neighbour-count, 0). Purely local anti-isolation signal (no λ₂, no global graph); weight negatively.

Source code in zymera/missions_terms.py 
120
121
122
123
124
125
126
127
128
129
130
131
def degree_floor(floor: float, comm_r: int) -> Callable:
    """Factory — (N,) ``max(floor − in-range-neighbour-count, 0)``. Purely
    local anti-isolation signal (no λ₂, no global graph); weight negatively."""

    @_requires("dist")
    def fn(prev, world, action, ctx) -> chex.Array:
        del prev, action
        off = ~jnp.eye(world.n_agents, dtype=bool)
        num_nb = ((ctx.dist <= comm_r) & off).sum(-1).astype(jnp.float32)
        return jnp.maximum(floor - num_nb, 0.0).astype(jnp.float32)

    return fn

phi_nearest_frontier 

phi_nearest_frontier(world)

Φ1 (N,) — NEGATED Chebyshev distance to the nearest uncovered free cell (frontier-seeking). Negated so larger Φ = closer to fresh ground.

Source code in zymera/missions_terms.py 
139
140
141
142
143
144
145
def phi_nearest_frontier(world) -> chex.Array:
    """Φ1 (N,) — NEGATED Chebyshev distance to the nearest uncovered free
    cell (frontier-seeking). Negated so larger Φ = closer to fresh ground."""
    uncovered = ~world.covered & ~world.wall
    return -metrics.dist_to_frontier(
        world.body.position, uncovered, world.grid_h, world.grid_w
    )

phi_field_mean 

phi_field_mean(world)

Φ3 () — NEGATED mean over uncovered free cells of the distance to the nearest agent (team blanket/territory; shared scalar).

Source code in zymera/missions_terms.py 
148
149
150
151
152
153
154
def phi_field_mean(world) -> chex.Array:
    """Φ3 () — NEGATED mean over uncovered free cells of the distance to the
    nearest agent (team blanket/territory; shared scalar)."""
    uncovered = ~world.covered & ~world.wall
    return -metrics.field_mean_dist(
        world.body.position, uncovered, world.grid_h, world.grid_w
    )

pbrs 

pbrs(phi, gamma)

Factory — PBRS combinator F = γ·Φ(world) − Φ(prev), policy-invariant by construction. phi(world) returns (N,) or a shared scalar (broadcast to (N,)). With the NEGATED-distance potentials above this matches v0's w·(γ·(−p1) − (−p0)) exactly; weight positively.

Source code in zymera/missions_terms.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
def pbrs(phi: Callable, gamma: float) -> Callable:
    """Factory — PBRS combinator ``F = γ·Φ(world) − Φ(prev)``, policy-invariant
    by construction. ``phi(world)`` returns ``(N,)`` or a shared scalar
    (broadcast to ``(N,)``). With the NEGATED-distance potentials above this
    matches v0's ``w·(γ·(−p1) − (−p0))`` exactly; weight positively."""

    @_requires()
    def fn(prev, world, action, ctx) -> chex.Array:
        del action, ctx
        f = gamma * phi(world) - phi(prev)
        return jnp.broadcast_to(f, (world.n_agents,)).astype(jnp.float32)

    return fn

cbf_conn 

cbf_conn(alpha, eps, sharp, comm_r)

Factory — (N,) shared connectivity-barrier violation residual on h = λ₂ − eps, divided by N as v0 does (shared team penalty).

Source code in zymera/missions_terms.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def cbf_conn(alpha: float, eps: float, sharp: float, comm_r: int) -> Callable:
    """Factory — (N,) shared connectivity-barrier violation residual on
    ``h = λ₂ − eps``, divided by N as v0 does (shared team penalty)."""

    @_requires()
    def fn(prev, world, action, ctx) -> chex.Array:
        del action, ctx
        n = world.n_agents
        hp = _lambda2(prev.body.position, comm_r, sharp) - eps
        hn = _lambda2(world.body.position, comm_r, sharp) - eps
        res = _cbf_residual(hp, hn, alpha) / n
        return jnp.broadcast_to(res, (n,)).astype(jnp.float32)

    return fn

cbf_coll 

cbf_coll(alpha, dmin)

Factory — (N,) per-agent sum of pairwise collision-barrier violation residuals on h_ij = d_ij − dmin.

Source code in zymera/missions_terms.py 
230
231
232
233
234
235
236
237
238
239
240
241
def cbf_coll(alpha: float, dmin: float) -> Callable:
    """Factory — (N,) per-agent sum of pairwise collision-barrier violation
    residuals on ``h_ij = d_ij − dmin``."""

    @_requires()
    def fn(prev, world, action, ctx) -> chex.Array:
        del action, ctx
        hp = _coll_barriers(prev.body.position, dmin)
        hn = _coll_barriers(world.body.position, dmin)
        return _cbf_residual(hp, hn, alpha).sum(-1).astype(jnp.float32)

    return fn