Winning-region safety-game shielding¶

MASA's winning-region shields prevent safety violations by restricting execution to actions that keep the system inside a computed winning region. The region is computed once, when the wrapper is constructed, from a finite transition model and a deterministic finite automaton (DFA) recognizing bad prefixes of a safety property.

There are two interfaces to the same safety analysis:

• PreemptiveLTLShield exposes the permitted actions before the policy chooses. An unsafe choice is rejected before the environment advances.
• PostposedLTLShield checks the policy's proposal and replaces it only when necessary. Safe proposals are executed unchanged.

These are environment wrappers, not learning algorithms. They do not optimize reward, learn a dynamics model, or guarantee eventual task completion.

Important

Winning-region safety does not require deterministic dynamics or deterministic action replacement. Every positive-probability successor is treated as possible. Random replacement is safe when it selects only permitted actions.

The masa.deterministic_shield package name distinguishes this support-based, zero-risk construction from MASA's risk-budget-based probabilistic shielding; it does not imply that every shielded trajectory is deterministic.

For risk budgets and projection onto safe action distributions, see Probabilistic Shielding. Sampling uniformly among winning actions is not the probabilistic-shielding algorithm: no probability of violation is budgeted.

The safety game¶

Let the base environment be a finite, fully observed MDP with states \(S\), actions \(A\), transition probabilities \(P(s' \mid s,a)\), and a fixed state-labelling function \(L\).

A safety DFA has states \(Q\), initial state \(q_0\), transition function \(\delta\), and bad-prefix accepting states \(F\). In MASA's safety monitor, accepting means a violation, not success. Supply a correct DFA for the intended safety property; these wrappers do not compile LTL strings or solve general liveness objectives.

The product state \(x=(q,s)\) records both the physical state and the monitor's memory. The game interpretation is:

1. the controller chooses an action;
2. the environment may realize any successor in that action's transition support.

An explicit second player does not need to be implemented in the simulator.

Monitor timing and state encoding¶

LTLSafetyEnv.reset() consumes the initial state's labels. Its step() consumes the labels of the newly returned state. Therefore \(q\) in the current product observation has already consumed \(L(s)\), and the shield uses

The initial product state is

For discrete observations, MASA encodes the product state as

q_index * n_states + state

where q_index is the monitor's internal index for a DFA state, not necessarily the numeric value or name of that DFA state.

The shield's winning_region and safe_actions arrays use this same encoding.

Computing the winning region¶

Initialize the candidate region to the non-rejecting product states:

Then repeatedly remove states for which there is no enabled action whose entire successor support remains inside the current candidate region:

Because the product is finite and the sequence only shrinks, this converges to a greatest fixed point \(W\).

For each product state, the permissive safe-action set is

An action with no positive transition mass is disabled, not vacuously safe.

A state with no enabled winning action is losing.

No probability cutoff is used: even a very rare unsafe successor disqualifies an action.

Why this is stronger than one-step checking¶

Suppose start has two actions:

• one keeps the system at start;
• one enters trap.

trap is not currently unsafe, but its only action leads to bad.

A one-step safety filter may allow the transition to trap. Winning-region shielding does not: trap is removed from the fixed point, so the action entering it is also removed from the safe-action mask at start.

Why the restriction preserves safety¶

Starting in \(W\), every permitted action has all modeled successors in \(W\). Applying that property inductively keeps every supported execution inside \(W\), and therefore outside the rejecting DFA states.

Both the preemptive and postposed interfaces enforce the same invariant. Their difference is only when the safe-action set is consulted.

This is a model-relative guarantee. It requires:

• the real successor support to be contained in the modeled support;
• the deterministic labelling function to describe the real states correctly;
• the DFA to encode the intended safety property.

A conservative model can remove genuinely safe actions. An incomplete model can invalidate the guarantee.

Getting started¶

Use this wrapper order:

finite base environment
    -> LabelledEnv
    -> LTLSafetyEnv(obs_type="discrete")
    -> PreemptiveLTLShield OR PostposedLTLShield
    -> optional observation transforms / time limit / auto-reset / vectorization

Place the shield directly outside LTLSafetyEnv.

The base environment must expose:

• a finite transition model;
• zero-based Discrete state IDs;
• zero-based Discrete action IDs.

The shield keeps the action space and the monitor's discrete product-observation space unchanged.

The following setup uses the colour-bomb grid world and the safety property

import numpy as np

from masa.common.constraints.ltl_safety import LTLSafetyEnv
from masa.common.labelled_env import LabelledEnv
from masa.common.ltl import Atom, DFA
from masa.deterministic_shield import (
    PreemptiveLTLShield,
    PostposedLTLShield,
    random_safe,
)
from masa.envs.tabular.colour_bomb_grid_world import (
    ColourBombGridWorld,
    label_fn,
)


def make_ltl_env():
    dfa = DFA([0, 1], initial=0, accepting=[1])
    dfa.add_edge(0, 1, Atom("bomb"))

    base = ColourBombGridWorld(slip_prob=0.0)
    return LTLSafetyEnv(
        LabelledEnv(base, label_fn),
        dfa=dfa,
        obs_type="discrete",
    )

Preemptive shielding¶

Preemptive shielding exposes the safe set before action selection.

env = PreemptiveLTLShield(make_ltl_env())
policy_rng = np.random.default_rng(42)

try:
    obs, info = env.reset(seed=17)

    for _ in range(50):
        permitted = np.flatnonzero(info["action_mask"])
        action = int(policy_rng.choice(permitted))

        obs, reward, terminated, truncated, info = env.step(action)

        if terminated or truncated:
            break
finally:
    env.close()

The policy may use either

info["action_mask"]

or

env.action_masks()

to obtain the current safe-action mask.

action_masks() returns a copy.

If the caller submits an unsafe action, PreemptiveLTLShield.step() raises ValueError before the environment is stepped. It does not silently replace the action, and the caller may retry with a safe choice.

Exposing a mask does not automatically make a reinforcement-learning algorithm mask-aware. A preemptive learner should apply the mask during:

• exploration;
• exploitation;
• maximization over next-state actions in value targets.

For policy-gradient methods, mask and normalize the action distribution before sampling and before computing the corresponding log-probability.

Postposed shielding¶

Postposed shielding lets the policy propose an action normally.

If the proposal is safe, it is executed unchanged. If it is unsafe, the configured replacement strategy chooses another safe action.

env = PostposedLTLShield(
    make_ltl_env(),
    replacement=random_safe(seed=23),
)
policy_rng = np.random.default_rng(42)

try:
    obs, info = env.reset(seed=17)

    for _ in range(50):
        proposal = int(policy_rng.integers(env.action_space.n))

        obs, reward, terminated, truncated, info = env.step(proposal)

        if info["shield_intervened"]:
            print(
                info["shield_proposed_action"],
                "->",
                info["shield_executed_action"],
            )

        if terminated or truncated:
            break
finally:
    env.close()

Every replacement is checked against the original safe-action mask before the wrapped environment is stepped.

DeterministicLTLShield remains a compatibility alias for PostposedLTLShield, including the default lowest-index replacement behavior.

Replacement strategies¶

Replacement strategies live in

masa/deterministic_shield/replacement_strategies.py

and are also exported from masa.deterministic_shield.

Random safe replacement¶

random_safe(seed) owns its own NumPy random generator.

replacement = random_safe(seed=42)

Important details:

• env.reset(seed=...) does not seed or rewind the replacement generator;
• the same replacement seed and the same sequence of intervention masks reproduce the same replacement sequence;
• construct a fresh random_safe(seed) to restart the replacement sequence;
• use a separate replacement callback for each environment;
• safe proposals do not consume replacement random numbers.

Random replacement changes only the selection among already-safe actions. It does not introduce a non-zero acceptable probability of violating the property.

Highest-score replacement¶

highest_score(score_fn) can use Q-values, policy logits, or another priority vector.

replacement = highest_score(lambda product_obs: q_values[product_obs])

score_fn must return one score per action. Scores for safe actions must be finite. Unsafe entries are ignored.

The callback is invoked only when replacement is required.

Custom replacement¶

A custom callback has signature

replacement(product_observation, proposed_action, safe_mask) -> action

The mask passed to the callback is a copy. The returned action is validated against the shield's internal safe set.

A custom callback should not step, reset, or otherwise mutate the wrapped environment.

Public interface¶

Both wrappers expose the same synthesized safety information:

After a successful step, info contains:

action_mask
shield_intervened
shield_proposed_action
shield_executed_action

For the preemptive wrapper, the proposed and executed actions are identical and shield_intervened is false.

Episode boundaries¶

After true termination, the returned action_mask is all false.

After truncation without termination, the returned mask still describes the final observation. This can be useful for bootstrapping a masked value target.

In both cases, another call to step() or action_masks() requires reset().

If an outer auto-reset wrapper replaces the final observation, use that wrapper's final-transition interface rather than combining a reset observation with the previous episode's mask.

Postposed shielding and learning semantics¶

For postposed learning it is important to distinguish the policy's proposal from the action actually executed by the underlying environment.

If the shielded wrapper itself is the learner's environment, the proposal is the learner's action and replacement is part of the environment dynamics.

If instead you are modelling the underlying unshielded environment, the executed action is the relevant action.

Do not overwrite an on-policy sampled action with the replacement while keeping the log-probability of the original proposal.

Failure modes and limits¶

Losing initial state¶

If reset produces a product state outside the winning region, the shield raises RuntimeError.

There is no "least unsafe" fallback.

Model mismatch¶

At runtime, the shield checks whether the observed product transition is present in the modeled support of the executed action.

This check occurs after the real environment step. It can diagnose model disagreement, but cannot undo a transition that has already happened.

The safety guarantee therefore requires the modeled support to cover all real successors.

Fixed dynamics and labels¶

The transition model, deterministic labelling function, and DFA must remain fixed after shield synthesis.

Reconstruct the shield if any of them changes.

Infinite-horizon safety¶

The fixed-point solver reasons about infinite continuation.

It does not solve safety only until a time limit, and it does not implement finite-trace LTL semantics.

Terminal states therefore need explicit modeled dynamics, normally absorbing.

Safety only¶

Remaining in the winning region does not imply:

• reward optimality;
• eventual goal completion;
• food collection;
• fairness;
• satisfaction of arbitrary liveness properties;
• general LTL synthesis.

This component is specifically a safety-game shield for bad-prefix properties.

Model representation and cost¶

read_support() prefers MASA's successor dictionaries when available.

Otherwise it reads a dense transition kernel using

P[next_state, current_state, action]

Synthesis uses whether each transition probability is positive. There is no risk threshold.

Internally, if

• \(S\) is the number of base states,
• \(A\) is the number of actions,
• \(Q\) is the number of DFA states,
• \(K\) is the maximum successor-row width,

then the padded support has shape

(S, A, K)

and product targets have shape

(Q, S, K)

There is no dense product transition tensor.

The current vectorized fixed-point loop has operation count on the order of

for \(I\) fixed-point iterations.

Synthesis happens when the wrapper is constructed, not on every step.

Mini PacMan tutorial¶

The companion tutorial notebook uses

where ghost is Mini PacMan's existing collision label.

It compares:

• mask-aware preemptive action selection;
• postposed random replacement.

It is deliberately a small shielding-interface example rather than a training benchmark.

Open the Mini PacMan deterministic-shielding notebook

API reference¶

PreemptiveLTLShield(env: LTLSafetyEnv)

33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

def __init__(self, env: LTLSafetyEnv):
    if not isinstance(env, LTLSafetyEnv):
        raise TypeError("Place the shield directly outside LTLSafetyEnv.")
    if env._obs_type != "discrete":
        raise TypeError("LTLSafetyEnv must use obs_type='discrete'.")
    for space in (env._orig_obs_space, env.action_space):
        if not isinstance(space, spaces.Discrete) or space.start != 0:
            raise TypeError("Base states and actions must be zero-based Discrete.")
    super().__init__(env)
    self._state: int | None = None
    self._n_states = int(env._orig_obs_space.n)
    n_actions = int(env.action_space.n)

    dfa = env._constraint.get_dfa()
    q_index = env._automaton_states_idx
    rejecting = np.zeros(len(q_index), dtype=bool)
    for q in dfa.accepting:
        rejecting[q_index[q]] = True
    if dfa.initial in dfa.accepting:
        raise ValueError("The DFA already rejects the empty prefix.")

    labels = [set(env.label_fn(s)) for s in range(self._n_states)]
    next_q = np.empty((len(q_index), self._n_states), dtype=np.intp)
    for q, i in q_index.items():
        for s, label in enumerate(labels):
            next_q[i, s] = q_index[dfa.transition(q, label)]

    successors, self._support = read_support(
        env.unwrapped, self._n_states, n_actions
    )
    # The live monitor already consumed L(s); its next update consumes L(s').
    self._targets = next_q[:, successors] * self._n_states + successors
    self._initial_states = (
        next_q[q_index[dfa.initial]] * self._n_states
        + np.arange(self._n_states)
    )
    self.winning_region, self.safe_actions = winning_region(
        self._targets, self._support, rejecting
    )
    self.winning_region.flags.writeable = False
    self.safe_actions.flags.writeable = False

reset(*, seed=None, options=None)

101
102
103
104
105
106
107
108

def reset(self, *, seed=None, options=None):
    self._state = None
    obs, info = self.env.reset(seed=seed, options=options)
    state = self._check_state(obs)
    if state != self._initial_states[state % self._n_states]:
        raise RuntimeError("Reset labels/DFA disagree with the synthesized model.")
    self._state = state
    return obs, {**info, "action_mask": self.action_masks()}

step(action)

16
17
18

def step(self, action):
    proposed = self._validate_proposal(action)
    return self._step_safe(proposed, proposed)

action_masks() -> np.ndarray

97
98
99

def action_masks(self) -> np.ndarray:
    """A copy of the current safe-action mask; requires an active episode."""
    return self.safe_actions[self._require_state()].copy()

close()

150
151
152

def close(self):
    self._state = None
    return self.env.close()

PostposedLTLShield(env: LTLSafetyEnv, *, replacement: Replacement | None = None)

26
27
28
29
30
31

def __init__(self, env: LTLSafetyEnv, *, replacement: Replacement | None = None):
    if replacement is not None and not callable(replacement):
        raise TypeError("replacement must be callable or None.")
    super().__init__(env)
    self._replacement = replacement
    self._fallback = self.safe_actions.argmax(axis=1)

reset(*, seed=None, options=None)

101
102
103
104
105
106
107
108

def reset(self, *, seed=None, options=None):
    self._state = None
    obs, info = self.env.reset(seed=seed, options=options)
    state = self._check_state(obs)
    if state != self._initial_states[state % self._n_states]:
        raise RuntimeError("Reset labels/DFA disagree with the synthesized model.")
    self._state = state
    return obs, {**info, "action_mask": self.action_masks()}

step(action)

33
34
35
36
37
38
39
40
41
42

def step(self, action):
    proposed = self._validate_proposal(action)
    state = self._require_state()
    if self.safe_actions[state, proposed]:
        executed = proposed
    elif self._replacement is None:
        executed = int(self._fallback[state])
    else:
        executed = self._replacement(state, proposed, self.action_masks())
    return self._step_safe(proposed, executed)

action_masks() -> np.ndarray

97
98
99

def action_masks(self) -> np.ndarray:
    """A copy of the current safe-action mask; requires an active episode."""
    return self.safe_actions[self._require_state()].copy()

close()

150
151
152

def close(self):
    self._state = None
    return self.env.close()

random_safe(seed: int | None = None) -> Replacement

20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

def random_safe(seed: int | None = None) -> Replacement:
    """Sample uniformly from the current safe actions when replacement is needed.

    One private RNG is created per selector, not per call. The same seed and
    sequence of masks reproduce the same replacements. Use a separate selector
    for each environment to avoid sharing a random stream between environments.

    The RNG is independent of the environment's RNG: env.reset(seed=...) neither
    seeds nor rewinds this selector. Recreate random_safe(seed) to restart it.
    Safe proposals do not call the selector and therefore do not advance its RNG.
    """
    rng = np.random.default_rng(seed)

    def select(state: int, proposed: int, mask: np.ndarray) -> int:
        return int(rng.choice(_candidates(mask)))

    return select

highest_score(score_fn: Callable[[int], ndarray]) -> Replacement

39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

def highest_score(score_fn: Callable[[int], np.ndarray]) -> Replacement:
    """Maximize score_fn(product_observation)[action] over safe actions.

    Use Q-values, policy logits or a priority vector. Ties choose the lowest
    action index. The rule is deterministic when score_fn is deterministic.
    It is called only when intervention is needed. Its result must have one
    entry per action and finite values for safe actions; unsafe entries are ignored.
    """
    if not callable(score_fn):
        raise TypeError("score_fn must be callable.")

    def select(state: int, proposed: int, mask: np.ndarray) -> int:
        actions = _candidates(mask)
        scores = np.asarray(score_fn(state), dtype=np.float64)
        if scores.shape != mask.shape:
            raise ValueError("score_fn must return one score per action.")
        if not np.isfinite(scores[actions]).all():
            raise ValueError("Scores of safe actions must be finite.")
        return int(actions[np.argmax(scores[actions])])

    return select

read_support(base: Env, n_states: int, n_actions: int) -> tuple[np.ndarray, np.ndarray]

 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

def read_support(
    base: gym.Env, n_states: int, n_actions: int
) -> tuple[np.ndarray, np.ndarray]:
    """Read MASA dynamics as successors[s, k] and support[s, a, k].

    Prefer the sparse dictionary. Dense MASA kernels use P[next, current, action].
    Missing/zero-mass actions are disabled, not vacuously safe.
    """
    if getattr(base, "has_successor_states_dict", False):
        successor_dict, probability_dict = base.get_successor_states_dict()
        transition = None
    elif getattr(base, "has_transition_matrix", False):
        transition = np.asarray(base.get_transition_matrix())
        if transition.shape != (n_states, n_states, n_actions):
            raise ValueError("Expected transition shape (next_state, state, action).")
    else:
        raise ValueError("The base environment must expose a finite transition model.")

    rows = []
    for s in range(n_states):
        if transition is None:
            ids = np.asarray(successor_dict.get(s, []))
            if ids.ndim != 1 or (ids.size and ids.dtype.kind not in "iu"):
                raise ValueError(f"Successors of state {s} must be integer IDs.")
            if np.any((ids < 0) | (ids >= n_states)):
                raise ValueError(f"Successor of state {s} is out of range.")
            ids = ids.astype(np.intp)
            zero = np.zeros(ids.size)
            p = np.asarray(
                [probability_dict.get((s, a), zero) for a in range(n_actions)],
                dtype=np.float64,
            )
        else:
            ids = np.arange(n_states, dtype=np.intp)
            p = np.asarray(transition[:, s, :].T, dtype=np.float64)

        if p.shape != (n_actions, ids.size):
            raise ValueError(f"Probability vectors for state {s} have wrong shape.")
        if not np.all(np.isfinite(p)) or np.any(p < 0):
            raise ValueError(f"Invalid transition probabilities at state {s}.")
        mass = p.sum(axis=1)
        if not np.all((mass == 0) | np.isclose(mass, 1, rtol=1e-6, atol=1e-8)):
            raise ValueError(f"Each action at state {s} must have mass 0 or 1.")

        # No probability cutoff: even arbitrarily rare outcomes must be safe.
        positive = p > 0
        keep = positive.any(axis=0)
        rows.append((ids[keep], positive[:, keep]))

    width = max(1, max(len(ids) for ids, _ in rows))
    successors = np.zeros((n_states, width), dtype=np.intp)
    support = np.zeros((n_states, n_actions, width), dtype=bool)
    for s, (ids, positive) in enumerate(rows):
        successors[s, :len(ids)] = ids
        support[s, :, :len(ids)] = positive
    return successors, support

winning_region(targets: ndarray, support: ndarray, rejecting: ndarray) -> tuple[np.ndarray, np.ndarray]

 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

def winning_region(
    targets: np.ndarray, support: np.ndarray, rejecting: np.ndarray
) -> tuple[np.ndarray, np.ndarray]:
    """Greatest fixed point; all possible successors must remain winning.

    targets[q, s, k] is a flattened product-state ID.
    support[s, a, k] says whether that successor is possible under action a.
    rejecting[q] marks a bad-prefix DFA state. Padding has support=False.
    Returns winning[product_state] and allowed[product_state, action].
    """
    n_dfa, n_states, _ = targets.shape
    n_actions = support.shape[1]
    winning = np.broadcast_to(~rejecting[:, None], (n_dfa, n_states)).copy()
    enabled = support.any(axis=2)
    allowed = np.zeros((n_dfa, n_states, n_actions), dtype=bool)

    while True:
        flat = winning.ravel()
        for q in range(n_dfa):
            successor_wins = flat[targets[q]]
            escapes = (support & ~successor_wins[:, None, :]).any(axis=2)
            allowed[q] = winning[q, :, None] & enabled & ~escapes
        updated = allowed.any(axis=2)
        if np.array_equal(updated, winning):
            return winning.ravel(), allowed.reshape(-1, n_actions)
        winning = updated