Skip to content

Verification

The analysis tools answer questions about recorded history — what happened, why, and did your tests cover the program. Verification answers a different question: does a property hold across every reachable state, not just the states your tests happened to visit?

from pyrung import Bool, Program, rung, latch, reset
from pyrung.core.analysis import never, Proven

EstopOK = Bool(external=True)
Start   = Bool(external=True)
Running = Bool()

with Program(strict=False) as logic:
    with rung(Start, EstopOK):
        latch(Running)
    with rung(~EstopOK):
        reset(Running)

result = never(logic, Running, ~EstopOK)
assert isinstance(result, Proven)

never() proves a bad state is unreachable — never(logic, Running, ~EstopOK) means "the motor is never running without the e-stop being OK." It exhaustively explores every reachable state via BFS over the compiled replay kernel. If the property holds everywhere, you get Proven. If not, Counterexample with a trace you can replay on a real PLC.

always() is the general form — always(logic, Or(~Running, EstopOK)) is equivalent but reads less naturally for safety properties. Use always() when the property isn't a "bad state" pattern (e.g., always(logic, lambda s: s["X"] + s["Y"] < 100)).

Condition syntax

never() and always() accept the same condition expressions as rung() and when():

never(logic, Running, ~EstopOK)           # bad-state: "never running without e-stop"
always(logic, ~Running, EstopOK)          # implicit AND (same property, always form)
always(logic, lambda s: s["X"] + s["Y"] < 100)  # callable fallback

Condition expressions are preferred — the verifier extracts referenced tags and automatically restricts input enumeration to the upstream cone. Callable predicates work but don't get auto-scoping.

Result types

from pyrung.core.analysis import Proven, Counterexample, Intractable

result = never(logic, Running, ~EstopOK)

if isinstance(result, Proven):
    print(f"Holds across {result.states_explored} states")

elif isinstance(result, Counterexample):
    # Replay the trace on a real PLC
    with PLC(logic, dt=0.010) as plc:
        for step in result.trace:
            plc.patch(step.inputs)
            for _ in range(step.scans):
                plc.step()
    # The violation is now visible in plc state

elif isinstance(result, Intractable):
    print(result.reason)  # "unbounded domain on Pressure"
    print(result.tags)    # ["Pressure"] — add choices or min/max
    for hint in result.hints:
        print(hint)       # "Pressure: 65536 values (Int, no choices/min/max)"

Intractable means the state space is too large. The hints list describes the largest dimensions — which tags are blowing up the state space and why. The fix is usually adding choices or min/max metadata to the unbounded tags — the same metadata you'd declare anyway for Data View dropdowns and static validation.

Scoping

With condition expressions, scope is derived automatically from the referenced tags. Override with scope= when needed:

never(logic, Running, ~EstopOK, scope=["Running", "EstopOK"])

Scoping restricts input enumeration to the upstream cone of the named tags — the verifier only explores inputs that can actually influence the property.

How it works

The verifier classifies every tag into one of three roles:

  • Combinational — OTE-only writes, derived from inputs each scan. Not a state dimension.
  • Stateful — latch/reset, timer/counter, copy, calc. Tracked in the visited set.
  • Nondeterministic — external inputs. Enumerated at each state. InputBlock tags (e.g., x[1]), receive() destination tags, and semantic tags mapped to input banks via TagMap are all automatically nondeterministic — no external=True annotation required.

Value domains come from the expression tree: comparison literals in conditions, choices metadata, min/max bounds. A tag compared against == 1 and == 2 gets domain {1, 2, unmatched} — three values instead of 65K.

Don't-care pruning skips inputs that are masked by the current state. And(StateBit, Input) with StateBit=False means Input doesn't matter — the verifier skips it entirely.

run_function and run_enabled_function are opaque to the verifier — it cannot introspect or symbolically execute user-provided Python functions. Output tags are tracked as state-producing writes, but their value domains must come from tag metadata. Without choices= or min=/max= on the output tags, always() returns Intractable:

# Intractable — unbounded output domain
result = Int("Result")
run_function(my_func, outs={"r": result})

# Verifiable — domain declared via metadata
result = Int("Result", min=0, max=100)
run_function(my_func, outs={"r": result})

Timer and counter Done bits use a three-valued abstraction: False, Pending (accumulating), and True (done). The verifier fast-forwards through accumulation rather than stepping one tick at a time.

By default, always() checks predicates on every reachable state, including transient states where a timer is still accumulating. For timer-gated alarm properties, pass settled=True to evaluate predicates only after pending timers/counters have settled:

# Without settled: Counterexample from the transient state before the timer fires
result = never(logic, Cmd, ~Fb, ~Alarm)
assert isinstance(result, Counterexample)

# With settled: evaluates after timers fire — the alarm is reachable
result = never(logic, Cmd, ~Fb, ~Alarm, settled=True)
assert isinstance(result, Proven)

settled=True only suppresses transient violations where settlement produces an alternate state. If the property is violated in a non-timer state, or if settlement diverges, the violation is still reported.

Debugging with journals

Pass journal=True to get a per-tag decision trail showing how the verifier classified, absorbed, or elided each tag:

result = always(logic, condition, journal=True)
print(result.journal)

The Journal object is a mapping from tag name to TagEntry. Each entry records the tag's final outcome (e.g. "stateful", "elided:concrete", "excluded:comparison_only"), its domain, and the chain of Decision objects from each pass that touched it. Useful for diagnosing why the verifier returned an unexpected result — especially when optimized and unoptimized runs disagree.

Available on all three result types (Proven, Counterexample, Intractable). When journal=False (default), result.journal is None and there is no overhead.

Splitting coupled state

When independent zones share a single coupling tag (a mode selector, an auto/manual latch), the verifier can't factor them apart — every zone's inputs look dependent because they all read the shared tag. The state space is the full cross-product.

split_at promotes a stateful tag to a nondeterministic input so the verifier explores it at all values independently. The remaining inputs become separable:

result = always(logic, condition, split_at=["AutoMode"])
states = reachable_states(logic, split_at=["AutoMode"])

The verifier explores each independent group separately, then composes results via delta merge — O(shared × sum of groups) kernel evaluations instead of O(product of all inputs).

Split candidates must be Bool, Done-paired, or choices= tags. External inputs (already nondeterministic), rise()/fall() targets, and unbounded-domain tags are rejected with a clear error.

Because splitting over-approximates (it explores values the tag might never reach in practice), a Counterexample from a split run includes a caveat. Proven results are sound — if the property holds across all split-tag values, it holds across all reachable values too.

split_at is also available in __lock__:

__lock__ = {
    "split_at": ["AutoMode"],
}

When the verifier returns Intractable, the hints list will suggest split_at candidates it detected — stateful tags whose promotion would enable factoring.

Fault coverage

The harness knows every device coupling. Harness.couplings() iterates them as Coupling dataclasses so you can automate fault coverage without maintaining a manual device list:

from pyrung import Coupling, Harness, PLC

plc = PLC(logic, dt=0.001)
harness = Harness(plc)
harness.install()

for coupling in harness.couplings():
    print(coupling.en_name, "→", coupling.fb_name)

Each Coupling has en_name, fb_name, physical, and trigger_value (None for plain bool links, the matched value for link="Tag:value" triggers).

Fault coverage decomposes into two passes over the same coupling list:

Structural coverage — does a path from the fault to an alarm exist at all? always() answers this exhaustively. Batch all conditions into a single call — the verifier shares work across properties:

from pyrung.core.analysis import always, Proven, Counterexample

couplings = list(harness.couplings())
conditions = [
    Or(~plc.tags[c.en_name], plc.tags[c.fb_name], AlarmExtent != 0)
    for c in couplings
]
results = always(logic, conditions, settled=True)

for coupling, result in zip(couplings, results):
    assert isinstance(result, Proven), f"{coupling.fb_name}: no alarm path"

Each condition reads: "in every reachable state, either the enable is off, the feedback is healthy, or the alarm caught it." A Counterexample means there exists a reachable state where the feedback has failed and no alarm fired — a structural detection gap. settled=True suppresses transient violations during timer accumulation — see How it works above.

always() uses a three-valued timer abstraction (False/Pending/True) that collapses accumulator state to make BFS tractable. But it's timing-blind by design — it answers "can the alarm fire?" not "does it fire in time?"

Timing coverage — does the fault timer trip fast enough under real timing? Force-based tests answer this:

for coupling in harness.couplings():
    plc2 = PLC(logic, dt=0.001)
    h2 = Harness(plc2)
    h2.install()
    plc2.force(coupling.en_name, True)
    plc2.run_for(1.5)
    plc2.force(coupling.fb_name, False)
    plc2.run_for(6.0)
    with plc2:
        assert AlarmExtent.value != 0, f"{coupling.fb_name}: too slow"

This catches fault timers that exist structurally but are too slow — the alarm path exists but takes longer than the machine can safely tolerate.

Run always() first — there's no point testing timing on a coupling that never reaches an alarm. Then run the force-based tests for timing validation on the ones that passed. See examples/fault_coverage.py for a complete working example.

Lock files

The lock file captures your program's full reachable behavior as a committed artifact — same mental model as uv.lock or package-lock.json.

pyrung lock my_program        # compute reachable states, write pyrung.lock
pyrung check my_program       # recompute, diff against pyrung.lock, exit 1 if changed

The lock projects to tags marked lock=True by default — the outputs that define your program's observable behavior. Programs using Click TagMap get this automatically (output-mapped tags are stamped lock=True). Programs without TagMap need explicit lock=True on output tags, or use __lock__ or --project:

pyrung lock my_program --project Running MotorOut StatusLight

Tags with choices= metadata get their labels in the lock file instead of raw integers — "FAST" instead of 2. Tags with band= metadata collapse numeric values into categorical labels — see Bands below.

__lock__ — per-module projection override

For programs where the lock=True default misses something (a pivot that matters behaviorally) or includes something cosmetic, define __lock__ at module level:

__lock__ = {
    "include": ["AlarmExtent", "BatchCount"],
    "exclude": ["Sts_DisplayText"],
}
  • include adds tags the default misses.
  • exclude drops tags the default includes.
  • joint declares joint multi-flip input groups (see below).
  • exclusive declares mutually exclusive input groups (see below).
  • All keys are optional. Most programs won't need __lock__ at all.
  • --project on the CLI still overrides everything for one-off checks.

Common patterns:

# Lock down the operator-facing interface too
dv = logic.dataview()
__lock__ = {
    "include": list(dv.public().tags),
}

# Lock Modbus registers
__lock__ = {
    "include": list(dv.contains("Modbus").tags),
}

Bands — collapsing numeric values into categories

The band attribute maps ranges of concrete values to categorical labels in the lock file output. Band metadata is purely a post-processing reduction — it is not used during BFS exploration or always():

AlarmExtent = Int(lock=True, band={"ZERO": 0, "POSITIVE": "> 0"})

In the lock file, AlarmExtent=3 becomes AlarmExtent="POSITIVE". The lock captures which band the value falls into, not the exact number — so adding a new alarm source doesn't change the lock file as long as the alarm extent stays positive.

Band predicates:

Predicate Matches
0 Exact value 0
"*" Any value (wildcard)
">= 100" Comparison (supports ==, !=, >, >=, <, <=)
"0..10" Inclusive range

Predicates are checked in declaration order — the first match wins. Use "*" as a catch-all last entry.

Joint inputs — multi-flip combinations

Single-flip BFS only changes one input per successor state. The joint key in __lock__ declares groups whose members are additionally explored jointly — the BFS generates multi-flip combinations on top of the normal single-flip successors.

__lock__ = {
    "joint": {
        "panel": ["SwitchA", "SwitchB"],
    },
}

Also available programmatically via joint_inputs= on reachable_states() and always().

It's generally unwise to ship logic that depends on multiple inputs changing in the exact same scan cycle — real-world I/O doesn't change atomically. If the verifier only finds a property violation via a multi-flip path, that's usually a signal the logic needs fixing, not that single-flip is too conservative.

Exclusive inputs — at most one True

The exclusive key in __lock__ declares groups where at most one member is True at a time. The BFS only explores all-False and one-hot (single-True) assignments, pruning multi-hot combinations that can't happen in practice. This reduces the state space for selector switches, mode buttons, and similar mutually exclusive physical inputs.

__lock__ = {
    "exclusive": {
        "mode": ["Manual", "Auto", "Step"],
    },
}

Also available programmatically via exclusive_inputs= on reachable_states() and always().

The verifier auto-detects some exclusive patterns (encoder-style one-hot families). Use the exclusive key when auto-detection doesn't cover your case — typically for inputs that are directly read in conditions rather than routed through an encoder tag.

Three levels of lock

Lock everything — full state space equality. For purely cosmetic refactoring (renaming tags, reordering rungs that don't interact). Any behavioral change is flagged.

states = reachable_states(logic)  # default: lock=True tags

Lock I/O — project to inputs and terminals only. For restructuring internal logic where pivots can change freely.

dv = logic.dataview()
states = reachable_states(logic, project=sorted(dv.terminals().tags))

Lock a subset — scope to specific tags. "I'm changing the diverter logic, but motor control shouldn't be affected."

dv = logic.dataview()
motor_tags = sorted(dv.upstream("Running", "Conv_Motor").tags)
states = reachable_states(logic, scope=["Running", "Conv_Motor"], project=motor_tags)

Diffing

from pyrung.core.analysis import reachable_states, diff_states

before = reachable_states(original, project=["Running", "MotorOut"])
after  = reachable_states(refactored, project=["Running", "MotorOut"])
diff   = diff_states(before, after)

assert not diff.added and not diff.removed  # behavioral equivalence

In a PR, the lock file diff tells the story. States omit false values — each entry reads as "what's ON":

  "reachable": [
    {},
    {"Conv_Motor": true, "Running": true},
+   {"Conv_Motor": true}
  ]

Reviewer sees: "Conv_Motor can now be on while Running is off." Either intentional (regenerate with pyrung lock) or a bug.

Programmatic use

from pyrung.core.analysis.prove import check_lock, write_lock, program_hash

# Write
states = reachable_states(logic, project=["Running"])
write_lock(Path("pyrung.lock"), states, ["Running"], program_hash(logic))

# Check
diff = check_lock(logic, Path("pyrung.lock"))
assert diff is None  # None means no change

CLI reference

pyrung lock <module>              # write pyrung.lock
pyrung lock <module> -o out.lock  # custom output path
pyrung lock <module> --project Running MotorOut  # explicit projection
pyrung lock <module> --depth-budget 100          # allow more abstract BFS work
pyrung lock <module> --profile out.prof          # write cProfile stats

pyrung check <module>             # diff against pyrung.lock, exit 1 on change
pyrung check <module> --lock custom.lock         # custom lock path
pyrung check <module> --profile out.prof         # write cProfile stats

--profile dumps cProfile stats even on KeyboardInterrupt. Analyze with pstats.Stats("out.prof") or uvx snakeviz out.prof. --depth-budget is an abstract BFS work budget, not a concrete scan cap. Hidden-event acceleration can cover more concrete PLC scans than the budget value.

The <module> argument is a Python module path (e.g., my_program or examples.conveyor). The module must contain a Program instance.

Next steps

  • Analysis — program structure, diagnosis, cause/effect, test coverage
  • Physical Annotations — declare device behavior, autoharness
  • Testing — forces as fixtures, forking, monitors, breakpoints
  • Runner Guide — execution methods, history, time travel