Merge tag 'trace-rv-v7.1' of git://git.kernel.org/pub/scm/linux/kernel/git/trace/linux-trace

Pull runtime verification updates from Steven Rostedt:

 - Refactor da_monitor header to share handlers across monitor types

   No functional changes, only less code duplication.

 - Add Hybrid Automata model class

   Add a new model class that extends deterministic automata by adding
   constraints on transitions and states. Those constraints can take
   into account wall-clock time and as such allow RV monitor to make
   assertions on real time. Add documentation and code generation
   scripts.

 - Add stall monitor as hybrid automaton example

   Add a monitor that triggers a violation when a task is stalling as an
   example of automaton working with real time variables.

 - Convert the opid monitor to a hybrid automaton

   The opid monitor can be heavily simplified if written as a hybrid
   automaton: instead of tracking preempt and interrupt enable/disable
   events, it can just run constraints on the preemption/interrupt
   states when events like wakeup and need_resched verify.

 - Add support for per-object monitors in DA/HA

   Allow writing deterministic and hybrid automata monitors for generic
   objects (e.g. any struct), by exploiting a hash table where objects
   are saved. This allows to track more than just tasks in RV. For
   instance it will be used to track deadline entities in deadline
   monitors.

 - Add deadline tracepoints and move some deadline utilities

   Prepare the ground for deadline monitors by defining events and
   exporting helpers.

 - Add nomiss deadline monitor

   Add first example of deadline monitor asserting all entities complete
   before their deadline.

 - Improve rvgen error handling

   Introduce AutomataError exception class and better handle expected
   exceptions while showing a backtrace for unexpected ones.

 - Improve python code quality in rvgen

   Refactor the rvgen generation scripts to align with python best
   practices: use f-strings instead of %, use len() instead of
   __len__(), remove semicolons, use context managers for file
   operations, fix whitespace violations, extract magic strings into
   constants, remove unused imports and methods.

 - Fix small bugs in rvgen

   The generator scripts presented some corner case bugs: logical error
   in validating what a correct dot file looks like, fix an isinstance()
   check, enforce a dot file has an initial state, fix type annotations
   and typos in comments.

 - rvgen refactoring

   Refactor automata.py to use iterator-based parsing and handle
   required arguments directly in argparse.

 - Allow epoll in rtapp-sleep monitor

   The epoll_wait call is now rt-friendly so it should be allowed in the
   sleep monitor as a valid sleep method.

* tag 'trace-rv-v7.1' of git://git.kernel.org/pub/scm/linux/kernel/git/trace/linux-trace: (32 commits)
  rv: Allow epoll in rtapp-sleep monitor
  rv/rvgen: fix _fill_states() return type annotation
  rv/rvgen: fix unbound loop variable warning
  rv/rvgen: enforce presence of initial state
  rv/rvgen: extract node marker string to class constant
  rv/rvgen: fix isinstance check in Variable.expand()
  rv/rvgen: make monitor arguments required in rvgen
  rv/rvgen: remove unused __get_main_name method
  rv/rvgen: remove unused sys import from dot2c
  rv/rvgen: refactor automata.py to use iterator-based parsing
  rv/rvgen: use class constant for init marker
  rv/rvgen: fix DOT file validation logic error
  rv/rvgen: fix PEP 8 whitespace violations
  rv/rvgen: fix typos in automata and generator docstring and comments
  rv/rvgen: use context managers for file operations
  rv/rvgen: remove unnecessary semicolons
  rv/rvgen: replace __len__() calls with len()
  rv/rvgen: replace % string formatting with f-strings
  rv/rvgen: remove bare except clauses in generator
  rv/rvgen: introduce AutomataError exception class
  ...

14 files changed, 980 insertions, 243 deletions

diff --git a/tools/verification/models/deadline/nomiss.dot b/tools/verification/models/deadline/nomiss.dot
new file mode 100644
index 000000000000..fd1ea6bf2509
--- /dev/null
+++ b/tools/verification/models/deadline/nomiss.dot
@@ -0,0 +1,41 @@
+digraph state_automaton {
+	center = true;
+	size = "7,11";
+	{node [shape = circle] "idle"};
+	{node [shape = plaintext, style=invis, label=""] "__init_ready"};
+	{node [shape = doublecircle] "ready"};
+	{node [shape = circle] "ready"};
+	{node [shape = circle] "running"};
+	{node [shape = circle] "sleeping"};
+	{node [shape = circle] "throttled"};
+	"__init_ready" -> "ready";
+	"idle" [label = "idle"];
+	"idle" -> "idle" [ label = "dl_server_idle" ];
+	"idle" -> "ready" [ label = "dl_replenish;reset(clk)" ];
+	"idle" -> "running" [ label = "sched_switch_in" ];
+	"idle" -> "sleeping" [ label = "dl_server_stop" ];
+	"idle" -> "throttled" [ label = "dl_throttle" ];
+	"ready" [label = "ready\nclk < DEADLINE_NS()", color = green3];
+	"ready" -> "idle" [ label = "dl_server_idle" ];
+	"ready" -> "ready" [ label = "sched_wakeup\ndl_replenish;reset(clk)" ];
+	"ready" -> "running" [ label = "sched_switch_in" ];
+	"ready" -> "sleeping" [ label = "dl_server_stop" ];
+	"ready" -> "throttled" [ label = "dl_throttle;is_defer == 1" ];
+	"running" [label = "running\nclk < DEADLINE_NS()"];
+	"running" -> "idle" [ label = "dl_server_idle" ];
+	"running" -> "running" [ label = "dl_replenish;reset(clk)\nsched_switch_in\nsched_wakeup" ];
+	"running" -> "sleeping" [ label = "sched_switch_suspend\ndl_server_stop" ];
+	"running" -> "throttled" [ label = "dl_throttle" ];
+	"sleeping" [label = "sleeping"];
+	"sleeping" -> "ready" [ label = "sched_wakeup\ndl_replenish;reset(clk)" ];
+	"sleeping" -> "running" [ label = "sched_switch_in" ];
+	"sleeping" -> "sleeping" [ label = "dl_server_stop\ndl_server_idle" ];
+	"sleeping" -> "throttled" [ label = "dl_throttle;is_constr_dl == 1 || is_defer == 1" ];
+	"throttled" [label = "throttled"];
+	"throttled" -> "ready" [ label = "dl_replenish;reset(clk)" ];
+	"throttled" -> "throttled" [ label = "sched_switch_suspend\nsched_wakeup\ndl_server_idle\ndl_throttle" ];
+	{ rank = min ;
+		"__init_ready";
+		"ready";
+	}
+}
diff --git a/tools/verification/models/rtapp/sleep.ltl b/tools/verification/models/rtapp/sleep.ltl
index 6379bbeb6212..6f26c4810f78 100644
--- a/tools/verification/models/rtapp/sleep.ltl
+++ b/tools/verification/models/rtapp/sleep.ltl
@@ -5,6 +5,7 @@ RT_FRIENDLY_SLEEP = (RT_VALID_SLEEP_REASON or KERNEL_THREAD)
 
 RT_VALID_SLEEP_REASON = FUTEX_WAIT
                      or RT_FRIENDLY_NANOSLEEP
+                     or EPOLL_WAIT
 
 RT_FRIENDLY_NANOSLEEP = CLOCK_NANOSLEEP
                     and NANOSLEEP_TIMER_ABSTIME
diff --git a/tools/verification/models/sched/opid.dot b/tools/verification/models/sched/opid.dot
index 840052f6952b..511051fce430 100644
--- a/tools/verification/models/sched/opid.dot
+++ b/tools/verification/models/sched/opid.dot
@@ -1,35 +1,13 @@
 digraph state_automaton {
 	center = true;
 	size = "7,11";
-	{node [shape = plaintext, style=invis, label=""] "__init_disabled"};
-	{node [shape = circle] "disabled"};
-	{node [shape = doublecircle] "enabled"};
-	{node [shape = circle] "enabled"};
-	{node [shape = circle] "in_irq"};
-	{node [shape = circle] "irq_disabled"};
-	{node [shape = circle] "preempt_disabled"};
-	"__init_disabled" -> "disabled";
-	"disabled" [label = "disabled"];
-	"disabled" -> "disabled" [ label = "sched_need_resched\nsched_waking\nirq_entry" ];
-	"disabled" -> "irq_disabled" [ label = "preempt_enable" ];
-	"disabled" -> "preempt_disabled" [ label = "irq_enable" ];
-	"enabled" [label = "enabled", color = green3];
-	"enabled" -> "enabled" [ label = "preempt_enable" ];
-	"enabled" -> "irq_disabled" [ label = "irq_disable" ];
-	"enabled" -> "preempt_disabled" [ label = "preempt_disable" ];
-	"in_irq" [label = "in_irq"];
-	"in_irq" -> "enabled" [ label = "irq_enable" ];
-	"in_irq" -> "in_irq" [ label = "sched_need_resched\nsched_waking\nirq_entry" ];
-	"irq_disabled" [label = "irq_disabled"];
-	"irq_disabled" -> "disabled" [ label = "preempt_disable" ];
-	"irq_disabled" -> "enabled" [ label = "irq_enable" ];
-	"irq_disabled" -> "in_irq" [ label = "irq_entry" ];
-	"irq_disabled" -> "irq_disabled" [ label = "sched_need_resched" ];
-	"preempt_disabled" [label = "preempt_disabled"];
-	"preempt_disabled" -> "disabled" [ label = "irq_disable" ];
-	"preempt_disabled" -> "enabled" [ label = "preempt_enable" ];
+	{node [shape = plaintext, style=invis, label=""] "__init_any"};
+	{node [shape = doublecircle] "any"};
+	"__init_any" -> "any";
+	"any" [label = "any", color = green3];
+	"any" -> "any" [ label = "sched_need_resched;irq_off == 1\nsched_waking;irq_off == 1 && preempt_off == 1" ];
 	{ rank = min ;
-		"__init_disabled";
-		"disabled";
+		"__init_any";
+		"any";
 	}
 }
diff --git a/tools/verification/models/stall.dot b/tools/verification/models/stall.dot
new file mode 100644
index 000000000000..50077d1dff74
--- /dev/null
+++ b/tools/verification/models/stall.dot
@@ -0,0 +1,22 @@
+digraph state_automaton {
+	center = true;
+	size = "7,11";
+	{node [shape = circle] "enqueued"};
+	{node [shape = plaintext, style=invis, label=""] "__init_dequeued"};
+	{node [shape = doublecircle] "dequeued"};
+	{node [shape = circle] "running"};
+	"__init_dequeued" -> "dequeued";
+	"enqueued" [label = "enqueued\nclk < threshold_jiffies"];
+	"running" [label = "running"];
+	"dequeued" [label = "dequeued", color = green3];
+	"running" -> "running" [ label = "sched_switch_in\nsched_wakeup" ];
+	"enqueued" -> "enqueued" [ label = "sched_wakeup" ];
+	"enqueued" -> "running" [ label = "sched_switch_in" ];
+	"running" -> "dequeued" [ label = "sched_switch_wait" ];
+	"dequeued" -> "enqueued" [ label = "sched_wakeup;reset(clk)" ];
+	"running" -> "enqueued" [ label = "sched_switch_preempt;reset(clk)" ];
+	{ rank = min ;
+		"__init_dequeued";
+		"dequeued";
+	}
+}
diff --git a/tools/verification/rvgen/__main__.py b/tools/verification/rvgen/__main__.py
index fa6fc1f4de2f..3be7f85fe37b 100644
--- a/tools/verification/rvgen/__main__.py
+++ b/tools/verification/rvgen/__main__.py
@@ -9,10 +9,11 @@
 #   Documentation/trace/rv/da_monitor_synthesis.rst
 
 if __name__ == '__main__':
-    from rvgen.dot2k import dot2k
+    from rvgen.dot2k import da2k, ha2k
     from rvgen.generator import Monitor
     from rvgen.container import Container
     from rvgen.ltl2k import ltl2k
+    from rvgen.automata import AutomataError
     import argparse
     import sys
 
@@ -28,10 +29,11 @@ if __name__ == '__main__':
     monitor_parser.add_argument('-n', "--model_name", dest="model_name")
     monitor_parser.add_argument("-p", "--parent", dest="parent",
                                 required=False, help="Create a monitor nested to parent")
-    monitor_parser.add_argument('-c', "--class", dest="monitor_class",
-                                help="Monitor class, either \"da\" or \"ltl\"")
-    monitor_parser.add_argument('-s', "--spec", dest="spec", help="Monitor specification file")
-    monitor_parser.add_argument('-t', "--monitor_type", dest="monitor_type",
+    monitor_parser.add_argument('-c', "--class", dest="monitor_class", required=True,
+                                help="Monitor class, either \"da\", \"ha\" or \"ltl\"")
+    monitor_parser.add_argument('-s', "--spec", dest="spec", required=True,
+                                help="Monitor specification file")
+    monitor_parser.add_argument('-t', "--monitor_type", dest="monitor_type", required=True,
                                 help=f"Available options: {', '.join(Monitor.monitor_types.keys())}")
 
     container_parser = subparsers.add_parser("container")
@@ -41,9 +43,11 @@ if __name__ == '__main__':
 
     try:
         if params.subcmd == "monitor":
-            print("Opening and parsing the specification file %s" % params.spec)
+            print(f"Opening and parsing the specification file {params.spec}")
             if params.monitor_class == "da":
-                monitor = dot2k(params.spec, params.monitor_type, vars(params))
+                monitor = da2k(params.spec, params.monitor_type, vars(params))
+            elif params.monitor_class == "ha":
+                monitor = ha2k(params.spec, params.monitor_type, vars(params))
             elif params.monitor_class == "ltl":
                 monitor = ltl2k(params.spec, params.monitor_type, vars(params))
             else:
@@ -51,16 +55,15 @@ if __name__ == '__main__':
                 sys.exit(1)
         else:
             monitor = Container(vars(params))
-    except Exception as e:
-        print('Error: '+ str(e))
-        print("Sorry : :-(")
+    except AutomataError as e:
+        print(f"There was an error processing {params.spec}: {e}", file=sys.stderr)
         sys.exit(1)
 
-    print("Writing the monitor into the directory %s" % monitor.name)
+    print(f"Writing the monitor into the directory {monitor.name}")
     monitor.print_files()
     print("Almost done, checklist")
     if params.subcmd == "monitor":
-        print("  - Edit the %s/%s.c to add the instrumentation" % (monitor.name, monitor.name))
+        print(f"  - Edit the {monitor.name}/{monitor.name}.c to add the instrumentation")
         print(monitor.fill_tracepoint_tooltip())
     print(monitor.fill_makefile_tooltip())
     print(monitor.fill_kconfig_tooltip())
diff --git a/tools/verification/rvgen/dot2c b/tools/verification/rvgen/dot2c
index bf0c67c5b66c..1012becc7fab 100644
--- a/tools/verification/rvgen/dot2c
+++ b/tools/verification/rvgen/dot2c
@@ -16,7 +16,6 @@
 if __name__ == '__main__':
     from rvgen import dot2c
     import argparse
-    import sys
 
     parser = argparse.ArgumentParser(description='dot2c: converts a .dot file into a C structure')
     parser.add_argument('dot_file',  help='The dot file to be converted')
diff --git a/tools/verification/rvgen/rvgen/automata.py b/tools/verification/rvgen/rvgen/automata.py
index 3f06aef8d4fd..b9f8149f7118 100644
--- a/tools/verification/rvgen/rvgen/automata.py
+++ b/tools/verification/rvgen/rvgen/automata.py
@@ -3,112 +3,182 @@
 #
 # Copyright (C) 2019-2022 Red Hat, Inc. Daniel Bristot de Oliveira <bristot@kernel.org>
 #
-# Automata object: parse an automata in dot file digraph format into a python object
+# Automata class: parse an automaton in dot file digraph format into a python object
 #
 # For further information, see:
 #   Documentation/trace/rv/deterministic_automata.rst
 
 import ntpath
+import re
+from typing import Iterator
+from itertools import islice
+
+class _ConstraintKey:
+    """Base class for constraint keys."""
+
+class _StateConstraintKey(_ConstraintKey, int):
+    """Key for a state constraint. Under the hood just state_id."""
+    def __new__(cls, state_id: int):
+        return super().__new__(cls, state_id)
+
+class _EventConstraintKey(_ConstraintKey, tuple):
+    """Key for an event constraint. Under the hood just tuple(state_id,event_id)."""
+    def __new__(cls, state_id: int, event_id: int):
+        return super().__new__(cls, (state_id, event_id))
+
+class AutomataError(Exception):
+    """Exception raised for errors in automata parsing and validation.
+
+    Raised when DOT file processing fails due to invalid format, I/O errors,
+    or malformed automaton definitions.
+    """
 
 class Automata:
-    """Automata class: Reads a dot file and part it as an automata.
+    """Automata class: Reads a dot file and parses it as an automaton.
+
+    It supports both deterministic and hybrid automata.
 
     Attributes:
         dot_file: A dot file with an state_automaton definition.
     """
 
     invalid_state_str = "INVALID_STATE"
+    init_marker = "__init_"
+    node_marker = "{node"
+    # val can be numerical, uppercase (constant or macro), lowercase (parameter or function)
+    # only numerical values should have units
+    constraint_rule = re.compile(r"""
+        ^
+        (?P<env>[a-zA-Z_][a-zA-Z0-9_]+)  # C-like identifier for the env var
+        (?P<op>[!<=>]{1,2})              # operator
+        (?P<val>
+            [0-9]+ |                     # numerical value
+            [A-Z_]+\(\) |                # macro
+            [A-Z_]+ |                    # constant
+            [a-z_]+\(\) |                # function
+            [a-z_]+                      # parameter
+        )
+        (?P<unit>[a-z]{1,2})?            # optional unit for numerical values
+        """, re.VERBOSE)
+    constraint_reset = re.compile(r"^reset\((?P<env>[a-zA-Z_][a-zA-Z0-9_]+)\)")
 
     def __init__(self, file_path, model_name=None):
         self.__dot_path = file_path
         self.name = model_name or self.__get_model_name()
         self.__dot_lines = self.__open_dot()
         self.states, self.initial_state, self.final_states = self.__get_state_variables()
-        self.events = self.__get_event_variables()
-        self.function = self.__create_matrix()
+        self.env_types = {}
+        self.env_stored = set()
+        self.constraint_vars = set()
+        self.self_loop_reset_events = set()
+        self.events, self.envs = self.__get_event_variables()
+        self.function, self.constraints = self.__create_matrix()
         self.events_start, self.events_start_run = self.__store_init_events()
+        self.env_stored = sorted(self.env_stored)
+        self.constraint_vars = sorted(self.constraint_vars)
+        self.self_loop_reset_events = sorted(self.self_loop_reset_events)
 
     def __get_model_name(self) -> str:
         basename = ntpath.basename(self.__dot_path)
         if not basename.endswith(".dot") and not basename.endswith(".gv"):
             print("not a dot file")
-            raise Exception("not a dot file: %s" % self.__dot_path)
+            raise AutomataError(f"not a dot file: {self.__dot_path}")
 
         model_name = ntpath.splitext(basename)[0]
-        if model_name.__len__() == 0:
-            raise Exception("not a dot file: %s" % self.__dot_path)
+        if not model_name:
+            raise AutomataError(f"not a dot file: {self.__dot_path}")
 
         return model_name
 
     def __open_dot(self) -> list[str]:
-        cursor = 0
         dot_lines = []
         try:
-            dot_file = open(self.__dot_path)
-        except:
-            raise Exception("Cannot open the file: %s" % self.__dot_path)
+            with open(self.__dot_path) as dot_file:
+                dot_lines = dot_file.readlines()
+        except OSError as exc:
+            raise AutomataError(exc.strerror) from exc
 
-        dot_lines = dot_file.read().splitlines()
-        dot_file.close()
+        if not dot_lines:
+            raise AutomataError(f"{self.__dot_path} is empty")
 
         # checking the first line:
-        line = dot_lines[cursor].split()
+        line = dot_lines[0].split()
+
+        if len(line) < 2 or line[0] != "digraph" or line[1] != "state_automaton":
+            raise AutomataError(f"Not a valid .dot format: {self.__dot_path}")
 
-        if (line[0] != "digraph") and (line[1] != "state_automaton"):
-            raise Exception("Not a valid .dot format: %s" % self.__dot_path)
-        else:
-            cursor += 1
         return dot_lines
 
     def __get_cursor_begin_states(self) -> int:
-        cursor = 0
-        while self.__dot_lines[cursor].split()[0] != "{node":
-            cursor += 1
-        return cursor
+        for cursor, line in enumerate(self.__dot_lines):
+            split_line = line.split()
+
+            if len(split_line) and split_line[0] == self.node_marker:
+                return cursor
+
+        raise AutomataError("Could not find a beginning state")
 
     def __get_cursor_begin_events(self) -> int:
-        cursor = 0
-        while self.__dot_lines[cursor].split()[0] != "{node":
-            cursor += 1
-        while self.__dot_lines[cursor].split()[0] == "{node":
-            cursor += 1
-        # skip initial state transition
-        cursor += 1
+        state = 0
+        cursor = 0 # make pyright happy
+
+        for cursor, line in enumerate(self.__dot_lines):
+            line = line.split()
+            if not line:
+                continue
+
+            if state == 0:
+                if line[0] == self.node_marker:
+                    state = 1
+            elif line[0] != self.node_marker:
+                break
+        else:
+            raise AutomataError("Could not find beginning event")
+
+        cursor += 1 # skip initial state transition
+        if cursor == len(self.__dot_lines):
+            raise AutomataError("Dot file ended after event beginning")
+
         return cursor
 
     def __get_state_variables(self) -> tuple[list[str], str, list[str]]:
         # wait for node declaration
         states = []
         final_states = []
+        initial_state = ""
 
         has_final_states = False
         cursor = self.__get_cursor_begin_states()
 
         # process nodes
-        while self.__dot_lines[cursor].split()[0] == "{node":
-            line = self.__dot_lines[cursor].split()
-            raw_state = line[-1]
+        for line in islice(self.__dot_lines, cursor, None):
+            split_line = line.split()
+            if not split_line or split_line[0] != self.node_marker:
+                break
+
+            raw_state = split_line[-1]
 
             #  "enabled_fired"}; -> enabled_fired
-            state = raw_state.replace('"', '').replace('};', '').replace(',','_')
-            if state[0:7] == "__init_":
-                initial_state = state[7:]
+            state = raw_state.replace('"', '').replace('};', '').replace(',', '_')
+            if state.startswith(self.init_marker):
+                initial_state = state[len(self.init_marker):]
             else:
                 states.append(state)
-                if "doublecircle" in self.__dot_lines[cursor]:
+                if "doublecircle" in line:
                     final_states.append(state)
                     has_final_states = True
 
-                if "ellipse" in self.__dot_lines[cursor]:
+                if "ellipse" in line:
                     final_states.append(state)
                     has_final_states = True
 
-            cursor += 1
+        if not initial_state:
+            raise AutomataError("The automaton doesn't have an initial state")
 
         states = sorted(set(states))
         states.remove(initial_state)
 
-        # Insert the initial state at the bein og the states
+        # Insert the initial state at the beginning of the states
         states.insert(0, initial_state)
 
         if not has_final_states:
@@ -116,31 +186,95 @@ class Automata:
 
         return states, initial_state, final_states
 
-    def __get_event_variables(self) -> list[str]:
+    def __get_event_variables(self) -> tuple[list[str], list[str]]:
+        events: list[str] = []
+        envs: list[str] = []
         # here we are at the begin of transitions, take a note, we will return later.
         cursor = self.__get_cursor_begin_events()
 
-        events = []
-        while self.__dot_lines[cursor].lstrip()[0] == '"':
+        for line in map(str.lstrip, islice(self.__dot_lines, cursor, None)):
+            if not line.startswith('"'):
+                break
+
             # transitions have the format:
             # "all_fired" -> "both_fired" [ label = "disable_irq" ];
             #  ------------ event is here ------------^^^^^
-            if self.__dot_lines[cursor].split()[1] == "->":
-                line = self.__dot_lines[cursor].split()
-                event = line[-2].replace('"','')
+            split_line = line.split()
+            if len(split_line) > 1 and split_line[1] == "->":
+                event = "".join(split_line[split_line.index("label") + 2:-1]).replace('"', '')
 
-                # when a transition has more than one lables, they are like this
+                # when a transition has more than one label, they are like this
                 # "local_irq_enable\nhw_local_irq_enable_n"
                 # so split them.
 
-                event = event.replace("\\n", " ")
-                for i in event.split():
-                    events.append(i)
-            cursor += 1
+                for i in event.split("\\n"):
+                    # if the event contains a constraint (hybrid automata),
+                    # it will be separated by a ";":
+                    # "sched_switch;x<1000;reset(x)"
+                    ev, *constr = i.split(";")
+                    if constr:
+                        if len(constr) > 2:
+                            raise AutomataError("Only 1 constraint and 1 reset are supported")
+                        envs += self.__extract_env_var(constr)
+                    events.append(ev)
+            else:
+                # state labels have the format:
+                # "enable_fired" [label = "enable_fired\ncondition"];
+                #  ----- label is here -----^^^^^
+                # label and node name must be the same, condition is optional
+                state = line.split("label")[1].split('"')[1]
+                _, *constr = state.split("\\n")
+                if constr:
+                    if len(constr) > 1:
+                        raise AutomataError("Only 1 constraint is supported in the state")
+                    envs += self.__extract_env_var([constr[0].replace(" ", "")])
+
+        return sorted(set(events)), sorted(set(envs))
 
-        return sorted(set(events))
+    def _split_constraint_expr(self, constr: list[str]) -> Iterator[tuple[str,
+                                                                          str | None]]:
+        """
+        Get a list of strings of the type constr1 && constr2 and returns a list of
+        constraints and separators: [[constr1,"&&"],[constr2,None]]
+        """
+        exprs = []
+        seps = []
+        for c in constr:
+            while "&&" in c or "||" in c:
+                a = c.find("&&")
+                o = c.find("||")
+                pos = a if o < 0 or 0 < a < o else o
+                exprs.append(c[:pos].replace(" ", ""))
+                seps.append(c[pos:pos + 2].replace(" ", ""))
+                c = c[pos + 2:].replace(" ", "")
+            exprs.append(c)
+            seps.append(None)
+        return zip(exprs, seps)
 
-    def __create_matrix(self) -> list[list[str]]:
+    def __extract_env_var(self, constraint: list[str]) -> list[str]:
+        env = []
+        for c, _ in self._split_constraint_expr(constraint):
+            rule = self.constraint_rule.search(c)
+            reset = self.constraint_reset.search(c)
+            if rule:
+                env.append(rule["env"])
+                if rule.groupdict().get("unit"):
+                    self.env_types[rule["env"]] = rule["unit"]
+                if rule["val"][0].isalpha():
+                    self.constraint_vars.add(rule["val"])
+                # try to infer unit from constants or parameters
+                val_for_unit = rule["val"].lower().replace("()", "")
+                if val_for_unit.endswith("_ns"):
+                    self.env_types[rule["env"]] = "ns"
+                if val_for_unit.endswith("_jiffies"):
+                    self.env_types[rule["env"]] = "j"
+            if reset:
+                env.append(reset["env"])
+                # environment variables that are reset need a storage
+                self.env_stored.add(reset["env"])
+        return env
+
+    def __create_matrix(self) -> tuple[list[list[str]], dict[_ConstraintKey, list[str]]]:
         # transform the array into a dictionary
         events = self.events
         states = self.states
@@ -157,31 +291,49 @@ class Automata:
             nr_state += 1
 
         # declare the matrix....
-        matrix = [[ self.invalid_state_str for x in range(nr_event)] for y in range(nr_state)]
+        matrix = [[self.invalid_state_str for _ in range(nr_event)] for _ in range(nr_state)]
+        constraints: dict[_ConstraintKey, list[str]] = {}
 
         # and we are back! Let's fill the matrix
         cursor = self.__get_cursor_begin_events()
 
-        while self.__dot_lines[cursor].lstrip()[0] == '"':
-            if self.__dot_lines[cursor].split()[1] == "->":
-                line = self.__dot_lines[cursor].split()
-                origin_state = line[0].replace('"','').replace(',','_')
-                dest_state = line[2].replace('"','').replace(',','_')
-                possible_events = line[-2].replace('"','').replace("\\n", " ")
-                for event in possible_events.split():
+        for line in map(str.lstrip,
+                        islice(self.__dot_lines, cursor, None)):
+
+            if not line or line[0] != '"':
+                break
+
+            split_line = line.split()
+
+            if len(split_line) > 2 and split_line[1] == "->":
+                origin_state = split_line[0].replace('"', '').replace(',', '_')
+                dest_state = split_line[2].replace('"', '').replace(',', '_')
+                possible_events = "".join(split_line[split_line.index("label") + 2:-1]).replace('"', '')
+                for event in possible_events.split("\\n"):
+                    event, *constr = event.split(";")
+                    if constr:
+                        key = _EventConstraintKey(states_dict[origin_state], events_dict[event])
+                        constraints[key] = constr
+                        # those events reset also on self loops
+                        if origin_state == dest_state and "reset" in "".join(constr):
+                            self.self_loop_reset_events.add(event)
                     matrix[states_dict[origin_state]][events_dict[event]] = dest_state
-            cursor += 1
+            else:
+                state = line.split("label")[1].split('"')[1]
+                state, *constr = state.replace(" ", "").split("\\n")
+                if constr:
+                    constraints[_StateConstraintKey(states_dict[state])] = constr
 
-        return matrix
+        return matrix, constraints
 
     def __store_init_events(self) -> tuple[list[bool], list[bool]]:
         events_start = [False] * len(self.events)
         events_start_run = [False] * len(self.events)
-        for i, _ in enumerate(self.events):
+        for i in range(len(self.events)):
             curr_event_will_init = 0
             curr_event_from_init = False
             curr_event_used = 0
-            for j, _ in enumerate(self.states):
+            for j in range(len(self.states)):
                 if self.function[j][i] != self.invalid_state_str:
                     curr_event_used += 1
                 if self.function[j][i] == self.initial_state:
@@ -204,3 +356,13 @@ class Automata:
         if any(self.events_start):
             return False
         return self.events_start_run[self.events.index(event)]
+
+    def is_hybrid_automata(self) -> bool:
+        return bool(self.envs)
+
+    def is_event_constraint(self, key: _ConstraintKey) -> bool:
+        """
+        Given the key in self.constraints return true if it is an event
+        constraint, false if it is a state constraint
+        """
+        return isinstance(key, _EventConstraintKey)
diff --git a/tools/verification/rvgen/rvgen/dot2c.py b/tools/verification/rvgen/rvgen/dot2c.py
index 06a26bf15a7e..fc85ba1f649e 100644
--- a/tools/verification/rvgen/rvgen/dot2c.py
+++ b/tools/verification/rvgen/rvgen/dot2c.py
@@ -3,7 +3,7 @@
 #
 # Copyright (C) 2019-2022 Red Hat, Inc. Daniel Bristot de Oliveira <bristot@kernel.org>
 #
-# dot2c: parse an automata in dot file digraph format into a C
+# dot2c: parse an automaton in dot file digraph format into a C
 #
 # This program was written in the development of this paper:
 #  de Oliveira, D. B. and Cucinotta, T. and de Oliveira, R. S.
@@ -13,12 +13,13 @@
 # For further information, see:
 #   Documentation/trace/rv/deterministic_automata.rst
 
-from .automata import Automata
+from .automata import Automata, AutomataError
 
 class Dot2c(Automata):
     enum_suffix = ""
     enum_states_def = "states"
     enum_events_def = "events"
+    enum_envs_def = "envs"
     struct_automaton_def = "automaton"
     var_automaton_def = "aut"
 
@@ -28,17 +29,17 @@ class Dot2c(Automata):
 
     def __get_enum_states_content(self) -> list[str]:
         buff = []
-        buff.append("\t%s%s," % (self.initial_state, self.enum_suffix))
+        buff.append(f"\t{self.initial_state}{self.enum_suffix},")
         for state in self.states:
             if state != self.initial_state:
-                buff.append("\t%s%s," % (state, self.enum_suffix))
-        buff.append("\tstate_max%s," % (self.enum_suffix))
+                buff.append(f"\t{state}{self.enum_suffix},")
+        buff.append(f"\tstate_max{self.enum_suffix},")
 
         return buff
 
     def format_states_enum(self) -> list[str]:
         buff = []
-        buff.append("enum %s {" % self.enum_states_def)
+        buff.append(f"enum {self.enum_states_def} {{")
         buff += self.__get_enum_states_content()
         buff.append("};\n")
 
@@ -47,49 +48,82 @@ class Dot2c(Automata):
     def __get_enum_events_content(self) -> list[str]:
         buff = []
         for event in self.events:
-            buff.append("\t%s%s," % (event, self.enum_suffix))
+            buff.append(f"\t{event}{self.enum_suffix},")
 
-        buff.append("\tevent_max%s," % self.enum_suffix)
+        buff.append(f"\tevent_max{self.enum_suffix},")
 
         return buff
 
     def format_events_enum(self) -> list[str]:
         buff = []
-        buff.append("enum %s {" % self.enum_events_def)
+        buff.append(f"enum {self.enum_events_def} {{")
         buff += self.__get_enum_events_content()
         buff.append("};\n")
 
         return buff
 
+    def __get_non_stored_envs(self) -> list[str]:
+        return [e for e in self.envs if e not in self.env_stored]
+
+    def __get_enum_envs_content(self) -> list[str]:
+        buff = []
+        # We first place env variables that have a u64 storage.
+        # Those are limited by MAX_HA_ENV_LEN, other variables
+        # are read only and don't require a storage.
+        unstored = self.__get_non_stored_envs()
+        for env in list(self.env_stored) + unstored:
+            buff.append(f"\t{env}{self.enum_suffix},")
+
+        buff.append(f"\tenv_max{self.enum_suffix},")
+        max_stored = unstored[0] if len(unstored) else "env_max"
+        buff.append(f"\tenv_max_stored{self.enum_suffix} = {max_stored}{self.enum_suffix},")
+
+        return buff
+
+    def format_envs_enum(self) -> list[str]:
+        buff = []
+        if self.is_hybrid_automata():
+            buff.append(f"enum {self.enum_envs_def} {{")
+            buff += self.__get_enum_envs_content()
+            buff.append("};\n")
+            buff.append(f"_Static_assert(env_max_stored{self.enum_suffix} <= MAX_HA_ENV_LEN,"
+                        ' "Not enough slots");')
+            if {"ns", "us", "ms", "s"}.intersection(self.env_types.values()):
+                buff.append("#define HA_CLK_NS")
+            buff.append("")
+        return buff
+
     def get_minimun_type(self) -> str:
         min_type = "unsigned char"
 
-        if self.states.__len__() > 255:
+        if len(self.states) > 255:
             min_type = "unsigned short"
 
-        if self.states.__len__() > 65535:
+        if len(self.states) > 65535:
             min_type = "unsigned int"
 
-        if self.states.__len__() > 1000000:
-            raise Exception("Too many states: %d" % self.states.__len__())
+        if len(self.states) > 1000000:
+            raise AutomataError(f"Too many states: {len(self.states)}")
 
         return min_type
 
     def format_automaton_definition(self) -> list[str]:
         min_type = self.get_minimun_type()
         buff = []
-        buff.append("struct %s {" % self.struct_automaton_def)
-        buff.append("\tchar *state_names[state_max%s];" % (self.enum_suffix))
-        buff.append("\tchar *event_names[event_max%s];" % (self.enum_suffix))
-        buff.append("\t%s function[state_max%s][event_max%s];" % (min_type, self.enum_suffix, self.enum_suffix))
-        buff.append("\t%s initial_state;" % min_type)
-        buff.append("\tbool final_states[state_max%s];" % (self.enum_suffix))
+        buff.append(f"struct {self.struct_automaton_def} {{")
+        buff.append(f"\tchar *state_names[state_max{self.enum_suffix}];")
+        buff.append(f"\tchar *event_names[event_max{self.enum_suffix}];")
+        if self.is_hybrid_automata():
+            buff.append(f"\tchar *env_names[env_max{self.enum_suffix}];")
+        buff.append(f"\t{min_type} function[state_max{self.enum_suffix}][event_max{self.enum_suffix}];")
+        buff.append(f"\t{min_type} initial_state;")
+        buff.append(f"\tbool final_states[state_max{self.enum_suffix}];")
         buff.append("};\n")
         return buff
 
     def format_aut_init_header(self) -> list[str]:
         buff = []
-        buff.append("static const struct %s %s = {" % (self.struct_automaton_def, self.var_automaton_def))
+        buff.append(f"static const struct {self.struct_automaton_def} {self.var_automaton_def} = {{")
         return buff
 
     def __get_string_vector_per_line_content(self, entries: list[str]) -> str:
@@ -113,13 +147,24 @@ class Dot2c(Automata):
 
         return buff
 
+    def format_aut_init_envs_string(self) -> list[str]:
+        buff = []
+        if self.is_hybrid_automata():
+            buff.append("\t.env_names = {")
+            # maintain consistent order with the enum
+            ordered_envs = list(self.env_stored) + self.__get_non_stored_envs()
+            buff.append(self.__get_string_vector_per_line_content(ordered_envs))
+            buff.append("\t},")
+
+        return buff
+
     def __get_max_strlen_of_states(self) -> int:
-        max_state_name = max(self.states, key = len).__len__()
-        return max(max_state_name, self.invalid_state_str.__len__())
+        max_state_name = len(max(self.states, key=len))
+        return max(max_state_name, len(self.invalid_state_str))
 
     def get_aut_init_function(self) -> str:
-        nr_states = self.states.__len__()
-        nr_events = self.events.__len__()
+        nr_states = len(self.states)
+        nr_events = len(self.events)
         buff = []
 
         maxlen = self.__get_max_strlen_of_states() + len(self.enum_suffix)
@@ -134,10 +179,10 @@ class Dot2c(Automata):
                     next_state = self.function[x][y] + self.enum_suffix
 
                 if linetoolong:
-                    line += "\t\t\t%s" % next_state
+                    line += f"\t\t\t{next_state}"
                 else:
-                    line += "%*s" % (maxlen, next_state)
-                if y != nr_events-1:
+                    line += f"{next_state:>{maxlen}}"
+                if y != nr_events - 1:
                     line += ",\n" if linetoolong else ", "
                 else:
                     line += ",\n\t\t}," if linetoolong else " },"
@@ -180,7 +225,7 @@ class Dot2c(Automata):
 
     def format_aut_init_final_states(self) -> list[str]:
        buff = []
-       buff.append("\t.final_states = { %s }," % self.get_aut_init_final_states())
+       buff.append(f"\t.final_states = {{ {self.get_aut_init_final_states()} }},")
 
        return buff
 
@@ -196,7 +241,7 @@ class Dot2c(Automata):
 
     def format_invalid_state(self) -> list[str]:
         buff = []
-        buff.append("#define %s state_max%s\n" % (self.invalid_state_str, self.enum_suffix))
+        buff.append(f"#define {self.invalid_state_str} state_max{self.enum_suffix}\n")
 
         return buff
 
@@ -205,10 +250,12 @@ class Dot2c(Automata):
         buff += self.format_states_enum()
         buff += self.format_invalid_state()
         buff += self.format_events_enum()
+        buff += self.format_envs_enum()
         buff += self.format_automaton_definition()
         buff += self.format_aut_init_header()
         buff += self.format_aut_init_states_string()
         buff += self.format_aut_init_events_string()
+        buff += self.format_aut_init_envs_string()
         buff += self.format_aut_init_function()
         buff += self.format_aut_init_initial_state()
         buff += self.format_aut_init_final_states()
diff --git a/tools/verification/rvgen/rvgen/dot2k.py b/tools/verification/rvgen/rvgen/dot2k.py
index 6128fe238430..e6f476b903b0 100644
--- a/tools/verification/rvgen/rvgen/dot2k.py
+++ b/tools/verification/rvgen/rvgen/dot2k.py
@@ -8,8 +8,10 @@
 # For further information, see:
 #   Documentation/trace/rv/da_monitor_synthesis.rst
 
+from collections import deque
 from .dot2c import Dot2c
 from .generator import Monitor
+from .automata import _EventConstraintKey, _StateConstraintKey, AutomataError
 
 
 class dot2k(Monitor, Dot2c):
@@ -19,15 +21,22 @@ class dot2k(Monitor, Dot2c):
         self.monitor_type = MonitorType
         Monitor.__init__(self, extra_params)
         Dot2c.__init__(self, file_path, extra_params.get("model_name"))
-        self.enum_suffix = "_%s" % self.name
+        self.enum_suffix = f"_{self.name}"
+        self.enum_suffix = f"_{self.name}"
+        self.monitor_class = extra_params["monitor_class"]
 
     def fill_monitor_type(self) -> str:
-        return self.monitor_type.upper()
+        buff = [ self.monitor_type.upper() ]
+        buff += self._fill_timer_type()
+        if self.monitor_type == "per_obj":
+            buff.append("typedef /* XXX: define the target type */ *monitor_target;")
+        return "\n".join(buff)
 
     def fill_tracepoint_handlers_skel(self) -> str:
         buff = []
+        buff += self._fill_hybrid_definitions()
         for event in self.events:
-            buff.append("static void handle_%s(void *data, /* XXX: fill header */)" % event)
+            buff.append(f"static void handle_{event}(void *data, /* XXX: fill header */)")
             buff.append("{")
             handle = "handle_event"
             if self.is_start_event(event):
@@ -37,10 +46,14 @@ class dot2k(Monitor, Dot2c):
                 buff.append("\t/* XXX: validate that this event is only valid in the initial state */")
                 handle = "handle_start_run_event"
             if self.monitor_type == "per_task":
-                buff.append("\tstruct task_struct *p = /* XXX: how do I get p? */;");
-                buff.append("\tda_%s(p, %s%s);" % (handle, event, self.enum_suffix));
+                buff.append("\tstruct task_struct *p = /* XXX: how do I get p? */;")
+                buff.append(f"\tda_{handle}(p, {event}{self.enum_suffix});")
+            elif self.monitor_type == "per_obj":
+                buff.append("\tint id = /* XXX: how do I get the id? */;")
+                buff.append("\tmonitor_target t = /* XXX: how do I get t? */;")
+                buff.append(f"\tda_{handle}(id, t, {event}{self.enum_suffix});")
             else:
-                buff.append("\tda_%s(%s%s);" % (handle, event, self.enum_suffix));
+                buff.append(f"\tda_{handle}({event}{self.enum_suffix});")
             buff.append("}")
             buff.append("")
         return '\n'.join(buff)
@@ -48,25 +61,25 @@ class dot2k(Monitor, Dot2c):
     def fill_tracepoint_attach_probe(self) -> str:
         buff = []
         for event in self.events:
-            buff.append("\trv_attach_trace_probe(\"%s\", /* XXX: tracepoint */, handle_%s);" % (self.name, event))
+            buff.append(f"\trv_attach_trace_probe(\"{self.name}\", /* XXX: tracepoint */, handle_{event});")
         return '\n'.join(buff)
 
     def fill_tracepoint_detach_helper(self) -> str:
         buff = []
         for event in self.events:
-            buff.append("\trv_detach_trace_probe(\"%s\", /* XXX: tracepoint */, handle_%s);" % (self.name, event))
+            buff.append(f"\trv_detach_trace_probe(\"{self.name}\", /* XXX: tracepoint */, handle_{event});")
         return '\n'.join(buff)
 
     def fill_model_h_header(self) -> list[str]:
         buff = []
         buff.append("/* SPDX-License-Identifier: GPL-2.0 */")
         buff.append("/*")
-        buff.append(" * Automatically generated C representation of %s automaton" % (self.name))
+        buff.append(f" * Automatically generated C representation of {self.name} automaton")
         buff.append(" * For further information about this format, see kernel documentation:")
         buff.append(" *   Documentation/trace/rv/deterministic_automata.rst")
         buff.append(" */")
         buff.append("")
-        buff.append("#define MONITOR_NAME %s" % (self.name))
+        buff.append(f"#define MONITOR_NAME {self.name}")
         buff.append("")
 
         return buff
@@ -75,23 +88,27 @@ class dot2k(Monitor, Dot2c):
         #
         # Adjust the definition names
         #
-        self.enum_states_def = "states_%s" % self.name
-        self.enum_events_def = "events_%s" % self.name
-        self.struct_automaton_def = "automaton_%s" % self.name
-        self.var_automaton_def = "automaton_%s" % self.name
+        self.enum_states_def = f"states_{self.name}"
+        self.enum_events_def = f"events_{self.name}"
+        self.enum_envs_def = f"envs_{self.name}"
+        self.struct_automaton_def = f"automaton_{self.name}"
+        self.var_automaton_def = f"automaton_{self.name}"
 
         buff = self.fill_model_h_header()
         buff += self.format_model()
 
         return '\n'.join(buff)
 
+    def _is_id_monitor(self) -> bool:
+        return self.monitor_type in ("per_task", "per_obj")
+
     def fill_monitor_class_type(self) -> str:
-        if self.monitor_type == "per_task":
+        if self._is_id_monitor():
             return "DA_MON_EVENTS_ID"
         return "DA_MON_EVENTS_IMPLICIT"
 
     def fill_monitor_class(self) -> str:
-        if self.monitor_type == "per_task":
+        if self._is_id_monitor():
             return "da_monitor_id"
         return "da_monitor"
 
@@ -107,16 +124,30 @@ class dot2k(Monitor, Dot2c):
                 ("char *", "state"),
                 ("char *", "event"),
                 ]
+        tp_args_error_env = tp_args_error + [("char *", "env")]
+        tp_args_dict = {
+                "event": tp_args_event,
+                "error": tp_args_error,
+                "error_env": tp_args_error_env
+                }
         tp_args_id = ("int ", "id")
-        tp_args = tp_args_event if tp_type == "event" else tp_args_error
-        if self.monitor_type == "per_task":
+        tp_args = tp_args_dict[tp_type]
+        if self._is_id_monitor():
             tp_args.insert(0, tp_args_id)
-        tp_proto_c = ", ".join([a+b for a,b in tp_args])
-        tp_args_c = ", ".join([b for a,b in tp_args])
-        buff.append("	     TP_PROTO(%s)," % tp_proto_c)
-        buff.append("	     TP_ARGS(%s)" % tp_args_c)
+        tp_proto_c = ", ".join([a + b for a, b in tp_args])
+        tp_args_c = ", ".join([b for a, b in tp_args])
+        buff.append(f"	     TP_PROTO({tp_proto_c}),")
+        buff.append(f"	     TP_ARGS({tp_args_c})")
         return '\n'.join(buff)
 
+    def _fill_hybrid_definitions(self) -> list:
+        """Stub, not valid for deterministic automata"""
+        return []
+
+    def _fill_timer_type(self) -> list:
+        """Stub, not valid for deterministic automata"""
+        return []
+
     def fill_main_c(self) -> str:
         main_c = super().fill_main_c()
 
@@ -127,5 +158,454 @@ class dot2k(Monitor, Dot2c):
         main_c = main_c.replace("%%MIN_TYPE%%", min_type)
         main_c = main_c.replace("%%NR_EVENTS%%", str(nr_events))
         main_c = main_c.replace("%%MONITOR_TYPE%%", monitor_type)
+        main_c = main_c.replace("%%MONITOR_CLASS%%", self.monitor_class)
 
         return main_c
+
+class da2k(dot2k):
+    """Deterministic automata only"""
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        if self.is_hybrid_automata():
+            raise AutomataError("Detected hybrid automaton, use the 'ha' class")
+
+class ha2k(dot2k):
+    """Hybrid automata only"""
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        if not self.is_hybrid_automata():
+            raise AutomataError("Detected deterministic automaton, use the 'da' class")
+        self.trace_h = self._read_template_file("trace_hybrid.h")
+        self.__parse_constraints()
+
+    def fill_monitor_class_type(self) -> str:
+        if self._is_id_monitor():
+            return "HA_MON_EVENTS_ID"
+        return "HA_MON_EVENTS_IMPLICIT"
+
+    def fill_monitor_class(self) -> str:
+        """
+        Used for tracepoint classes, since they are shared we keep da
+        instead of ha (also for the ha specific tracepoints).
+        The tracepoint class is not visible to the tools.
+        """
+        return super().fill_monitor_class()
+
+    def __adjust_value(self, value: str | int, unit: str | None) -> str:
+        """Adjust the value in ns"""
+        try:
+            value = int(value)
+        except ValueError:
+            # it's a constant, a parameter or a function
+            if value.endswith("()"):
+                return value.replace("()", "(ha_mon)")
+            return value
+        match unit:
+            case "us":
+                value *= 10**3
+            case "ms":
+                value *= 10**6
+            case "s":
+                value *= 10**9
+        return str(value) + "ull"
+
+    def __parse_single_constraint(self, rule: dict, value: str) -> str:
+        return f"ha_get_env(ha_mon, {rule["env"]}{self.enum_suffix}, time_ns) {rule["op"]} {value}"
+
+    def __get_constraint_env(self, constr: str) -> str:
+        """Extract the second argument from an ha_ function"""
+        env = constr.split("(")[1].split()[1].rstrip(")").rstrip(",")
+        assert env.rstrip(f"_{self.name}") in self.envs
+        return env
+
+    def __start_to_invariant_check(self, constr: str) -> str:
+        # by default assume the timer has ns expiration
+        env = self.__get_constraint_env(constr)
+        clock_type = "ns"
+        if self.env_types.get(env.rstrip(f"_{self.name}")) == "j":
+            clock_type = "jiffy"
+
+        return f"return ha_check_invariant_{clock_type}(ha_mon, {env}, time_ns)"
+
+    def __start_to_conv(self, constr: str) -> str:
+        """
+        Undo the storage conversion done by ha_start_timer_
+        """
+        return "ha_inv_to_guard" + constr[constr.find("("):]
+
+    def __parse_timer_constraint(self, rule: dict, value: str) -> str:
+        # by default assume the timer has ns expiration
+        clock_type = "ns"
+        if self.env_types.get(rule["env"]) == "j":
+            clock_type = "jiffy"
+
+        return (f"ha_start_timer_{clock_type}(ha_mon, {rule["env"]}{self.enum_suffix},"
+                f" {value}, time_ns)")
+
+    def __format_guard_rules(self, rules: list[str]) -> list[str]:
+        """
+        Merge guard constraints as a single C return statement.
+        If the rules include a stored env, also check its validity.
+        Break lines in a best effort way that tries to keep readability.
+        """
+        if not rules:
+            return []
+
+        invalid_checks = [f"ha_monitor_env_invalid(ha_mon, {env}{self.enum_suffix}) ||"
+                          for env in self.env_stored if any(env in rule for rule in rules)]
+        if invalid_checks and len(rules) > 1:
+            rules[0] = "(" + rules[0]
+            rules[-1] = rules[-1] + ")"
+        rules = invalid_checks + rules
+
+        separator = "\n\t\t      " if sum(len(r) for r in rules) > 80 else " "
+        return ["res = " + separator.join(rules)]
+
+    def __validate_constraint(self, key: tuple[int, int] | int, constr: str,
+                              rule, reset) -> None:
+        # event constrains are tuples and allow both rules and reset
+        # state constraints are only used for expirations (e.g. clk<N)
+        if self.is_event_constraint(key):
+            if not rule and not reset:
+                raise AutomataError("Unrecognised event constraint "
+                                    f"({self.states[key[0]]}/{self.events[key[1]]}: {constr})")
+            if rule and (rule["env"] in self.env_types and
+                         rule["env"] not in self.env_stored):
+                raise AutomataError("Clocks in hybrid automata always require a storage"
+                                    f" ({rule["env"]})")
+        else:
+            if not rule:
+                raise AutomataError("Unrecognised state constraint "
+                                    f"({self.states[key]}: {constr})")
+            if rule["env"] not in self.env_stored:
+                raise AutomataError("State constraints always require a storage "
+                                    f"({rule["env"]})")
+            if rule["op"] not in ["<", "<="]:
+                raise AutomataError("State constraints must be clock expirations like"
+                                    f" clk<N ({rule.string})")
+
+    def __parse_constraints(self) -> None:
+        self.guards: dict[_EventConstraintKey, str] = {}
+        self.invariants: dict[_StateConstraintKey, str] = {}
+        for key, constraint in self.constraints.items():
+            rules = []
+            resets = []
+            for c, sep in self._split_constraint_expr(constraint):
+                rule = self.constraint_rule.search(c)
+                reset = self.constraint_reset.search(c)
+                self.__validate_constraint(key, c, rule, reset)
+                if rule:
+                    value = rule["val"]
+                    value_len = len(rule["val"])
+                    unit = None
+                    if rule.groupdict().get("unit"):
+                        value_len += len(rule["unit"])
+                        unit = rule["unit"]
+                    c = c[:-(value_len)]
+                    value = self.__adjust_value(value, unit)
+                    if self.is_event_constraint(key):
+                        c = self.__parse_single_constraint(rule, value)
+                        if sep:
+                            c += f" {sep}"
+                    else:
+                        c = self.__parse_timer_constraint(rule, value)
+                    rules.append(c)
+                if reset:
+                    c = f"ha_reset_env(ha_mon, {reset["env"]}{self.enum_suffix}, time_ns)"
+                    resets.append(c)
+            if self.is_event_constraint(key):
+                res = self.__format_guard_rules(rules) + resets
+                self.guards[key] = ";".join(res)
+            else:
+                self.invariants[key] = rules[0]
+
+    def __fill_verify_invariants_func(self) -> list[str]:
+        buff = []
+        if not self.invariants:
+            return []
+
+        buff.append(
+f"""static inline bool ha_verify_invariants(struct ha_monitor *ha_mon,
+\t\t\t\t\tenum {self.enum_states_def} curr_state, enum {self.enum_events_def} event,
+\t\t\t\t\tenum {self.enum_states_def} next_state, u64 time_ns)
+{{""")
+
+        _else = ""
+        for state, constr in sorted(self.invariants.items()):
+            check_str = self.__start_to_invariant_check(constr)
+            buff.append(f"\t{_else}if (curr_state == {self.states[state]}{self.enum_suffix})")
+            buff.append(f"\t\t{check_str};")
+            _else = "else "
+
+        buff.append("\treturn true;\n}\n")
+        return buff
+
+    def __fill_convert_inv_guard_func(self) -> list[str]:
+        buff = []
+        if not self.invariants:
+            return []
+
+        conflict_guards, conflict_invs = self.__find_inv_conflicts()
+        if not conflict_guards and not conflict_invs:
+            return []
+
+        buff.append(
+f"""static inline void ha_convert_inv_guard(struct ha_monitor *ha_mon,
+\t\t\t\t\tenum {self.enum_states_def} curr_state, enum {self.enum_events_def} event,
+\t\t\t\t\tenum {self.enum_states_def} next_state, u64 time_ns)
+{{""")
+        buff.append("\tif (curr_state == next_state)\n\t\treturn;")
+
+        _else = ""
+        for state, constr in sorted(self.invariants.items()):
+            # a state with invariant can reach us without reset
+            # multiple conflicts must have the same invariant, otherwise we cannot
+            # know how to reset the value
+            conf_i = [start for start, end in conflict_invs if end == state]
+            # we can reach a guard without reset
+            conf_g = [e for s, e in conflict_guards if s == state]
+            if not conf_i and not conf_g:
+                continue
+            buff.append(f"\t{_else}if (curr_state == {self.states[state]}{self.enum_suffix})")
+
+            buff.append(f"\t\t{self.__start_to_conv(constr)};")
+            _else = "else "
+
+        buff.append("}\n")
+        return buff
+
+    def __fill_verify_guards_func(self) -> list[str]:
+        buff = []
+        if not self.guards:
+            return []
+
+        buff.append(
+f"""static inline bool ha_verify_guards(struct ha_monitor *ha_mon,
+\t\t\t\t    enum {self.enum_states_def} curr_state, enum {self.enum_events_def} event,
+\t\t\t\t    enum {self.enum_states_def} next_state, u64 time_ns)
+{{
+\tbool res = true;
+""")
+
+        _else = ""
+        for edge, constr in sorted(self.guards.items()):
+            buff.append(f"\t{_else}if (curr_state == "
+                        f"{self.states[edge[0]]}{self.enum_suffix} && "
+                        f"event == {self.events[edge[1]]}{self.enum_suffix})")
+            if constr.count(";") > 0:
+                buff[-1] += " {"
+            buff += [f"\t\t{c};" for c in constr.split(";")]
+            if constr.count(";") > 0:
+                _else = "} else "
+            else:
+                _else = "else "
+        if _else[0] == "}":
+            buff.append("\t}")
+        buff.append("\treturn res;\n}\n")
+        return buff
+
+    def __find_inv_conflicts(self) -> tuple[set[tuple[int, _EventConstraintKey]],
+                                            set[tuple[int, _StateConstraintKey]]]:
+        """
+        Run a breadth first search from all states with an invariant.
+        Find any conflicting constraints reachable from there, this can be
+        another state with an invariant or an edge with a non-reset guard.
+        Stop when we find a reset.
+
+        Return the set of conflicting guards and invariants as tuples of
+        conflicting state and constraint key.
+        """
+        conflict_guards: set[tuple[int, _EventConstraintKey]] = set()
+        conflict_invs: set[tuple[int, _StateConstraintKey]] = set()
+        for start_idx in self.invariants:
+            queue = deque([(start_idx, 0)])  # (state_idx, distance)
+            env = self.__get_constraint_env(self.invariants[start_idx])
+
+            while queue:
+                curr_idx, distance = queue.popleft()
+
+                # Check state condition
+                if curr_idx != start_idx and curr_idx in self.invariants:
+                    conflict_invs.add((start_idx, _StateConstraintKey(curr_idx)))
+                    continue
+
+                # Check if we should stop
+                if distance > len(self.states):
+                    break
+                if curr_idx != start_idx and distance > 1:
+                    continue
+
+                for event_idx, next_state_name in enumerate(self.function[curr_idx]):
+                    if next_state_name == self.invalid_state_str:
+                        continue
+                    curr_guard = self.guards.get((curr_idx, event_idx), "")
+                    if "reset" in curr_guard and env in curr_guard:
+                        continue
+
+                    if env in curr_guard:
+                        conflict_guards.add((start_idx,
+                                             _EventConstraintKey(curr_idx, event_idx)))
+                        continue
+
+                    next_idx = self.states.index(next_state_name)
+                    queue.append((next_idx, distance + 1))
+
+        return conflict_guards, conflict_invs
+
+    def __fill_setup_invariants_func(self) -> list[str]:
+        buff = []
+        if not self.invariants:
+            return []
+
+        buff.append(
+f"""static inline void ha_setup_invariants(struct ha_monitor *ha_mon,
+\t\t\t\t       enum {self.enum_states_def} curr_state, enum {self.enum_events_def} event,
+\t\t\t\t       enum {self.enum_states_def} next_state, u64 time_ns)
+{{""")
+
+        conditions = ["next_state == curr_state"]
+        conditions += [f"event != {e}{self.enum_suffix}"
+                       for e in self.self_loop_reset_events]
+        condition_str = " && ".join(conditions)
+        buff.append(f"\tif ({condition_str})\n\t\treturn;")
+
+        _else = ""
+        for state, constr in sorted(self.invariants.items()):
+            buff.append(f"\t{_else}if (next_state == {self.states[state]}{self.enum_suffix})")
+            buff.append(f"\t\t{constr};")
+            _else = "else "
+
+        for state in self.invariants:
+            buff.append(f"\telse if (curr_state == {self.states[state]}{self.enum_suffix})")
+            buff.append("\t\tha_cancel_timer(ha_mon);")
+
+        buff.append("}\n")
+        return buff
+
+    def __fill_constr_func(self) -> list[str]:
+        buff = []
+        if not self.constraints:
+            return []
+
+        buff.append(
+"""/*
+ * These functions are used to validate state transitions.
+ *
+ * They are generated by parsing the model, there is usually no need to change them.
+ * If the monitor requires a timer, there are functions responsible to arm it when
+ * the next state has a constraint, cancel it in any other case and to check
+ * that it didn't expire before the callback run. Transitions to the same state
+ * without a reset never affect timers.
+ * Due to the different representations between invariants and guards, there is
+ * a function to convert it in case invariants or guards are reachable from
+ * another invariant without reset. Those are not present if not required in
+ * the model. This is all automatic but is worth checking because it may show
+ * errors in the model (e.g. missing resets).
+ */""")
+
+        buff += self.__fill_verify_invariants_func()
+        inv_conflicts = self.__fill_convert_inv_guard_func()
+        buff += inv_conflicts
+        buff += self.__fill_verify_guards_func()
+        buff += self.__fill_setup_invariants_func()
+
+        buff.append(
+f"""static bool ha_verify_constraint(struct ha_monitor *ha_mon,
+\t\t\t\t enum {self.enum_states_def} curr_state, enum {self.enum_events_def} event,
+\t\t\t\t enum {self.enum_states_def} next_state, u64 time_ns)
+{{""")
+
+        if self.invariants:
+            buff.append("\tif (!ha_verify_invariants(ha_mon, curr_state, "
+                        "event, next_state, time_ns))\n\t\treturn false;\n")
+        if inv_conflicts:
+            buff.append("\tha_convert_inv_guard(ha_mon, curr_state, event, "
+                        "next_state, time_ns);\n")
+
+        if self.guards:
+            buff.append("\tif (!ha_verify_guards(ha_mon, curr_state, event, "
+                        "next_state, time_ns))\n\t\treturn false;\n")
+
+        if self.invariants:
+            buff.append("\tha_setup_invariants(ha_mon, curr_state, event, next_state, time_ns);\n")
+
+        buff.append("\treturn true;\n}\n")
+        return buff
+
+    def __fill_env_getter(self, env: str) -> str:
+        if env in self.env_types:
+            match self.env_types[env]:
+                case "ns" | "us" | "ms" | "s":
+                    return "ha_get_clk_ns(ha_mon, env, time_ns);"
+                case "j":
+                    return "ha_get_clk_jiffy(ha_mon, env);"
+        return f"/* XXX: how do I read {env}? */"
+
+    def __fill_env_resetter(self, env: str) -> str:
+        if env in self.env_types:
+            match self.env_types[env]:
+                case "ns" | "us" | "ms" | "s":
+                    return "ha_reset_clk_ns(ha_mon, env, time_ns);"
+                case "j":
+                    return "ha_reset_clk_jiffy(ha_mon, env);"
+        return f"/* XXX: how do I reset {env}? */"
+
+    def __fill_hybrid_get_reset_functions(self) -> list[str]:
+        buff = []
+        if self.is_hybrid_automata():
+            for var in self.constraint_vars:
+                if var.endswith("()"):
+                    func_name = var.replace("()", "")
+                    if func_name.isupper():
+                        buff.append(f"#define {func_name}(ha_mon) "
+                                    f"/* XXX: what is {func_name}(ha_mon)? */\n")
+                    else:
+                        buff.append(f"static inline u64 {func_name}(struct ha_monitor *ha_mon)\n{{")
+                        buff.append(f"\treturn /* XXX: what is {func_name}(ha_mon)? */;")
+                        buff.append("}\n")
+                elif var.isupper():
+                    buff.append(f"#define {var} /* XXX: what is {var}? */\n")
+                else:
+                    buff.append(f"static u64 {var} = /* XXX: default value */;")
+                    buff.append(f"module_param({var}, ullong, 0644);\n")
+            buff.append("""/*
+ * These functions define how to read and reset the environment variable.
+ *
+ * Common environment variables like ns-based and jiffy-based clocks have
+ * pre-define getters and resetters you can use. The parser can infer the type
+ * of the environment variable if you supply a measure unit in the constraint.
+ * If you define your own functions, make sure to add appropriate memory
+ * barriers if required.
+ * Some environment variables don't require a storage as they read a system
+ * state (e.g. preemption count). Those variables are never reset, so we don't
+ * define a reset function on monitors only relying on this type of variables.
+ */""")
+            buff.append("static u64 ha_get_env(struct ha_monitor *ha_mon, "
+                        f"enum envs{self.enum_suffix} env, u64 time_ns)\n{{")
+            _else = ""
+            for env in self.envs:
+                buff.append(f"\t{_else}if (env == {env}{self.enum_suffix})")
+                buff.append(f"\t\treturn {self.__fill_env_getter(env)}")
+                _else = "else "
+            buff.append("\treturn ENV_INVALID_VALUE;\n}\n")
+            if len(self.env_stored):
+                buff.append("static void ha_reset_env(struct ha_monitor *ha_mon, "
+                            f"enum envs{self.enum_suffix} env, u64 time_ns)\n{{")
+                _else = ""
+                for env in self.env_stored:
+                    buff.append(f"\t{_else}if (env == {env}{self.enum_suffix})")
+                    buff.append(f"\t\t{self.__fill_env_resetter(env)}")
+                    _else = "else "
+                buff.append("}\n")
+        return buff
+
+    def _fill_hybrid_definitions(self) -> list[str]:
+        return self.__fill_hybrid_get_reset_functions() + self.__fill_constr_func()
+
+    def _fill_timer_type(self) -> list:
+        if self.invariants:
+            return [
+                    "/* XXX: If the monitor has several instances, consider HA_TIMER_WHEEL */",
+                    "#define HA_TIMER_TYPE HA_TIMER_HRTIMER"
+                    ]
+        return []
diff --git a/tools/verification/rvgen/rvgen/generator.py b/tools/verification/rvgen/rvgen/generator.py
index 3441385c1177..56f3bd8db850 100644
--- a/tools/verification/rvgen/rvgen/generator.py
+++ b/tools/verification/rvgen/rvgen/generator.py
@@ -3,7 +3,7 @@
 #
 # Copyright (C) 2019-2022 Red Hat, Inc. Daniel Bristot de Oliveira <bristot@kernel.org>
 #
-# Abtract class for generating kernel runtime verification monitors from specification file
+# Abstract class for generating kernel runtime verification monitors from specification file
 
 import platform
 import os
@@ -40,7 +40,7 @@ class RVGenerator:
         if platform.system() != "Linux":
             raise OSError("I can only run on Linux.")
 
-        kernel_path = os.path.join("/lib/modules/%s/build" % platform.release(), self.rv_dir)
+        kernel_path = os.path.join(f"/lib/modules/{platform.release()}/build", self.rv_dir)
 
         # if the current kernel is from a distro this may not be a full kernel tree
         # verify that one of the files we are going to modify is available
@@ -51,32 +51,26 @@ class RVGenerator:
         raise FileNotFoundError("Could not find the rv directory, do you have the kernel source installed?")
 
     def _read_file(self, path):
-        try:
-            fd = open(path, 'r')
-        except OSError:
-            raise Exception("Cannot open the file: %s" % path)
-
-        content = fd.read()
-
-        fd.close()
+        with open(path, 'r') as fd:
+            content = fd.read()
         return content
 
     def _read_template_file(self, file):
         try:
             path = os.path.join(self.abs_template_dir, file)
             return self._read_file(path)
-        except Exception:
+        except OSError:
             # Specific template file not found. Try the generic template file in the template/
             # directory, which is one level up
             path = os.path.join(self.abs_template_dir, "..", file)
             return self._read_file(path)
 
     def fill_parent(self):
-        return "&rv_%s" % self.parent if self.parent else "NULL"
+        return f"&rv_{self.parent}" if self.parent else "NULL"
 
     def fill_include_parent(self):
         if self.parent:
-            return "#include <monitors/%s/%s.h>\n" % (self.parent, self.parent)
+            return f"#include <monitors/{self.parent}/{self.parent}.h>\n"
         return ""
 
     def fill_tracepoint_handlers_skel(self):
@@ -122,7 +116,7 @@ class RVGenerator:
         buff = []
         buff.append("	# XXX: add dependencies if there")
         if self.parent:
-            buff.append("	depends on RV_MON_%s" % self.parent.upper())
+            buff.append(f"	depends on RV_MON_{self.parent.upper()}")
             buff.append("	default y")
         return '\n'.join(buff)
 
@@ -148,31 +142,30 @@ class RVGenerator:
         monitor_class_type = self.fill_monitor_class_type()
         if self.auto_patch:
             self._patch_file("rv_trace.h",
-                            "// Add new monitors based on CONFIG_%s here" % monitor_class_type,
-                            "#include <monitors/%s/%s_trace.h>" % (self.name, self.name))
-            return "  - Patching %s/rv_trace.h, double check the result" % self.rv_dir
+                            f"// Add new monitors based on CONFIG_{monitor_class_type} here",
+                            f"#include <monitors/{self.name}/{self.name}_trace.h>")
+            return f"  - Patching {self.rv_dir}/rv_trace.h, double check the result"
 
-        return """  - Edit %s/rv_trace.h:
-Add this line where other tracepoints are included and %s is defined:
-#include <monitors/%s/%s_trace.h>
-""" % (self.rv_dir, monitor_class_type, self.name, self.name)
+        return f"""  - Edit {self.rv_dir}/rv_trace.h:
+Add this line where other tracepoints are included and {monitor_class_type} is defined:
+#include <monitors/{self.name}/{self.name}_trace.h>
+"""
 
     def _kconfig_marker(self, container=None) -> str:
-        return "# Add new %smonitors here" % (container + " "
-                                              if container else "")
+        return f"# Add new {container + ' ' if container else ''}monitors here"
 
     def fill_kconfig_tooltip(self):
         if self.auto_patch:
             # monitors with a container should stay together in the Kconfig
             self._patch_file("Kconfig",
                              self._kconfig_marker(self.parent),
-                            "source \"kernel/trace/rv/monitors/%s/Kconfig\"" % (self.name))
-            return "  - Patching %s/Kconfig, double check the result" % self.rv_dir
+                            f"source \"kernel/trace/rv/monitors/{self.name}/Kconfig\"")
+            return f"  - Patching {self.rv_dir}/Kconfig, double check the result"
 
-        return """  - Edit %s/Kconfig:
+        return f"""  - Edit {self.rv_dir}/Kconfig:
 Add this line where other monitors are included:
-source \"kernel/trace/rv/monitors/%s/Kconfig\"
-""" % (self.rv_dir, self.name)
+source \"kernel/trace/rv/monitors/{self.name}/Kconfig\"
+"""
 
     def fill_makefile_tooltip(self):
         name = self.name
@@ -180,18 +173,18 @@ source \"kernel/trace/rv/monitors/%s/Kconfig\"
         if self.auto_patch:
             self._patch_file("Makefile",
                             "# Add new monitors here",
-                            "obj-$(CONFIG_RV_MON_%s) += monitors/%s/%s.o" % (name_up, name, name))
-            return "  - Patching %s/Makefile, double check the result" % self.rv_dir
+                            f"obj-$(CONFIG_RV_MON_{name_up}) += monitors/{name}/{name}.o")
+            return f"  - Patching {self.rv_dir}/Makefile, double check the result"
 
-        return """  - Edit %s/Makefile:
+        return f"""  - Edit {self.rv_dir}/Makefile:
 Add this line where other monitors are included:
-obj-$(CONFIG_RV_MON_%s) += monitors/%s/%s.o
-""" % (self.rv_dir, name_up, name, name)
+obj-$(CONFIG_RV_MON_{name_up}) += monitors/{name}/{name}.o
+"""
 
     def fill_monitor_tooltip(self):
         if self.auto_patch:
-            return "  - Monitor created in %s/monitors/%s" % (self.rv_dir, self. name)
-        return "  - Move %s/ to the kernel's monitor directory (%s/monitors)" % (self.name, self.rv_dir)
+            return f"  - Monitor created in {self.rv_dir}/monitors/{self.name}"
+        return f"  - Move {self.name}/ to the kernel's monitor directory ({self.rv_dir}/monitors)"
 
     def __create_directory(self):
         path = self.name
@@ -201,41 +194,27 @@ obj-$(CONFIG_RV_MON_%s) += monitors/%s/%s.o
             os.mkdir(path)
         except FileExistsError:
             return
-        except:
-            print("Fail creating the output dir: %s" % self.name)
 
     def __write_file(self, file_name, content):
-        try:
-            file = open(file_name, 'w')
-        except:
-            print("Fail writing to file: %s" % file_name)
-
-        file.write(content)
-
-        file.close()
+        with open(file_name, 'w') as file:
+            file.write(content)
 
     def _create_file(self, file_name, content):
-        path = "%s/%s" % (self.name, file_name)
+        path = f"{self.name}/{file_name}"
         if self.auto_patch:
             path = os.path.join(self.rv_dir, "monitors", path)
         self.__write_file(path, content)
 
-    def __get_main_name(self):
-        path = "%s/%s" % (self.name, "main.c")
-        if not os.path.exists(path):
-            return "main.c"
-        return "__main.c"
-
     def print_files(self):
         main_c = self.fill_main_c()
 
         self.__create_directory()
 
-        path = "%s.c" % self.name
+        path = f"{self.name}.c"
         self._create_file(path, main_c)
 
         model_h = self.fill_model_h()
-        path = "%s.h" % self.name
+        path = f"{self.name}.h"
         self._create_file(path, model_h)
 
         kconfig = self.fill_kconfig()
@@ -243,7 +222,7 @@ obj-$(CONFIG_RV_MON_%s) += monitors/%s/%s.o
 
 
 class Monitor(RVGenerator):
-    monitor_types = { "global" : 1, "per_cpu" : 2, "per_task" : 3 }
+    monitor_types = {"global": 1, "per_cpu": 2, "per_task": 3, "per_obj": 4}
 
     def __init__(self, extra_params={}):
         super().__init__(extra_params)
@@ -255,16 +234,18 @@ class Monitor(RVGenerator):
         monitor_class_type = self.fill_monitor_class_type()
         tracepoint_args_skel_event = self.fill_tracepoint_args_skel("event")
         tracepoint_args_skel_error = self.fill_tracepoint_args_skel("error")
+        tracepoint_args_skel_error_env = self.fill_tracepoint_args_skel("error_env")
         trace_h = trace_h.replace("%%MODEL_NAME%%", self.name)
         trace_h = trace_h.replace("%%MODEL_NAME_UP%%", self.name.upper())
         trace_h = trace_h.replace("%%MONITOR_CLASS%%", monitor_class)
         trace_h = trace_h.replace("%%MONITOR_CLASS_TYPE%%", monitor_class_type)
         trace_h = trace_h.replace("%%TRACEPOINT_ARGS_SKEL_EVENT%%", tracepoint_args_skel_event)
         trace_h = trace_h.replace("%%TRACEPOINT_ARGS_SKEL_ERROR%%", tracepoint_args_skel_error)
+        trace_h = trace_h.replace("%%TRACEPOINT_ARGS_SKEL_ERROR_ENV%%", tracepoint_args_skel_error_env)
         return trace_h
 
     def print_files(self):
         super().print_files()
         trace_h = self.fill_trace_h()
-        path = "%s_trace.h" % self.name
+        path = f"{self.name}_trace.h"
         self._create_file(path, trace_h)
diff --git a/tools/verification/rvgen/rvgen/ltl2ba.py b/tools/verification/rvgen/rvgen/ltl2ba.py
index f14e6760ac3d..7f538598a868 100644
--- a/tools/verification/rvgen/rvgen/ltl2ba.py
+++ b/tools/verification/rvgen/rvgen/ltl2ba.py
@@ -9,6 +9,7 @@
 
 from ply.lex import lex
 from ply.yacc import yacc
+from .automata import AutomataError
 
 # Grammar:
 # 	ltl ::= opd | ( ltl ) | ltl binop ltl | unop ltl
@@ -62,7 +63,7 @@ t_ignore_COMMENT = r'\#.*'
 t_ignore = ' \t\n'
 
 def t_error(t):
-    raise ValueError(f"Illegal character '{t.value[0]}'")
+    raise AutomataError(f"Illegal character '{t.value[0]}'")
 
 lexer = lex()
 
@@ -394,7 +395,7 @@ class Variable:
     @staticmethod
     def expand(n: ASTNode, node: GraphNode, node_set) -> set[GraphNode]:
         for f in node.old:
-            if isinstance(f, NotOp) and f.op.child is n:
+            if isinstance(f.op, NotOp) and f.op.child is n:
                 return node_set
         node.old |= {n}
         return node.expand(node_set)
@@ -487,7 +488,7 @@ def p_unop(p):
     elif p[1] == "not":
         op = NotOp(p[2])
     else:
-        raise ValueError(f"Invalid unary operator {p[1]}")
+        raise AutomataError(f"Invalid unary operator {p[1]}")
 
     p[0] = ASTNode(op)
 
@@ -507,7 +508,7 @@ def p_binop(p):
     elif p[2] == "imply":
         op = ImplyOp(p[1], p[3])
     else:
-        raise ValueError(f"Invalid binary operator {p[2]}")
+        raise AutomataError(f"Invalid binary operator {p[2]}")
 
     p[0] = ASTNode(op)
 
@@ -526,7 +527,7 @@ def parse_ltl(s: str) -> ASTNode:
             subexpr[assign[0]] = assign[1]
 
     if rule is None:
-        raise ValueError("Please define your specification in the \"RULE = <LTL spec>\" format")
+        raise AutomataError("Please define your specification in the \"RULE = <LTL spec>\" format")
 
     for node in rule:
         if not isinstance(node.op, Variable):
diff --git a/tools/verification/rvgen/rvgen/ltl2k.py b/tools/verification/rvgen/rvgen/ltl2k.py
index b075f98d50c4..81fd1f5ea5ea 100644
--- a/tools/verification/rvgen/rvgen/ltl2k.py
+++ b/tools/verification/rvgen/rvgen/ltl2k.py
@@ -4,6 +4,7 @@
 from pathlib import Path
 from . import generator
 from . import ltl2ba
+from .automata import AutomataError
 
 COLUMN_LIMIT = 100
 
@@ -43,13 +44,17 @@ def abbreviate_atoms(atoms: list[str]) -> list[str]:
         skip = ["is", "by", "or", "and"]
         return '_'.join([x[:2] for x in s.lower().split('_') if x not in skip])
 
-    abbrs = []
-    for atom in atoms:
+    def find_share_length(atom: str) -> int:
         for i in range(len(atom), -1, -1):
             if sum(a.startswith(atom[:i]) for a in atoms) > 1:
-                break
-        share = atom[:i]
-        unique = atom[i:]
+                return i
+        return 0
+
+    abbrs = []
+    for atom in atoms:
+        share_len = find_share_length(atom)
+        share = atom[:share_len]
+        unique = atom[share_len:]
         abbrs.append((shorten(share) + shorten(unique)))
     return abbrs
 
@@ -60,20 +65,23 @@ class ltl2k(generator.Monitor):
         if MonitorType != "per_task":
             raise NotImplementedError("Only per_task monitor is supported for LTL")
         super().__init__(extra_params)
-        with open(file_path) as f:
-            self.atoms, self.ba, self.ltl = ltl2ba.create_graph(f.read())
+        try:
+            with open(file_path) as f:
+                self.atoms, self.ba, self.ltl = ltl2ba.create_graph(f.read())
+        except OSError as exc:
+            raise AutomataError(exc.strerror) from exc
         self.atoms_abbr = abbreviate_atoms(self.atoms)
         self.name = extra_params.get("model_name")
         if not self.name:
             self.name = Path(file_path).stem
 
-    def _fill_states(self) -> str:
+    def _fill_states(self) -> list[str]:
         buf = [
             "enum ltl_buchi_state {",
         ]
 
         for node in self.ba:
-            buf.append("\tS%i," % node.id)
+            buf.append(f"\tS{node.id},")
         buf.append("\tRV_NUM_BA_STATES")
         buf.append("};")
         buf.append("static_assert(RV_NUM_BA_STATES <= RV_MAX_BA_STATES);")
@@ -82,7 +90,7 @@ class ltl2k(generator.Monitor):
     def _fill_atoms(self):
         buf = ["enum ltl_atom {"]
         for a in sorted(self.atoms):
-            buf.append("\tLTL_%s," % a)
+            buf.append(f"\tLTL_{a},")
         buf.append("\tLTL_NUM_ATOM")
         buf.append("};")
         buf.append("static_assert(LTL_NUM_ATOM <= RV_MAX_LTL_ATOM);")
@@ -96,7 +104,7 @@ class ltl2k(generator.Monitor):
         ]
 
         for name in self.atoms_abbr:
-            buf.append("\t\t\"%s\"," % name)
+            buf.append(f"\t\t\"{name}\",")
 
         buf.extend([
             "\t};",
@@ -113,19 +121,19 @@ class ltl2k(generator.Monitor):
                 continue
 
             if isinstance(node.op, ltl2ba.AndOp):
-                buf.append("\tbool %s = %s && %s;" % (node, node.op.left, node.op.right))
+                buf.append(f"\tbool {node} = {node.op.left} && {node.op.right};")
                 required_values |= {str(node.op.left), str(node.op.right)}
             elif isinstance(node.op, ltl2ba.OrOp):
-                buf.append("\tbool %s = %s || %s;" % (node, node.op.left, node.op.right))
+                buf.append(f"\tbool {node} = {node.op.left} || {node.op.right};")
                 required_values |= {str(node.op.left), str(node.op.right)}
             elif isinstance(node.op, ltl2ba.NotOp):
-                buf.append("\tbool %s = !%s;" % (node, node.op.child))
+                buf.append(f"\tbool {node} = !{node.op.child};")
                 required_values.add(str(node.op.child))
 
         for atom in self.atoms:
             if atom.lower() not in required_values:
                 continue
-            buf.append("\tbool %s = test_bit(LTL_%s, mon->atoms);" % (atom.lower(), atom))
+            buf.append(f"\tbool {atom.lower()} = test_bit(LTL_{atom}, mon->atoms);")
 
         buf.reverse()
 
@@ -153,7 +161,7 @@ class ltl2k(generator.Monitor):
         ])
 
         for node in self.ba:
-            buf.append("\tcase S%i:" % node.id)
+            buf.append(f"\tcase S{node.id}:")
 
             for o in sorted(node.outgoing):
                 line   = "\t\tif "
@@ -163,7 +171,7 @@ class ltl2k(generator.Monitor):
                 lines = break_long_line(line, indent)
                 buf.extend(lines)
 
-                buf.append("\t\t\t__set_bit(S%i, next);" % o.id)
+                buf.append(f"\t\t\t__set_bit(S{o.id}, next);")
             buf.append("\t\tbreak;")
         buf.extend([
             "\t}",
@@ -197,7 +205,7 @@ class ltl2k(generator.Monitor):
             lines = break_long_line(line, indent)
             buf.extend(lines)
 
-            buf.append("\t\t__set_bit(S%i, mon->states);" % node.id)
+            buf.append(f"\t\t__set_bit(S{node.id}, mon->states);")
         buf.append("}")
         return buf
 
@@ -205,23 +213,21 @@ class ltl2k(generator.Monitor):
         buff = []
         buff.append("static void handle_example_event(void *data, /* XXX: fill header */)")
         buff.append("{")
-        buff.append("\tltl_atom_update(task, LTL_%s, true/false);" % self.atoms[0])
+        buff.append(f"\tltl_atom_update(task, LTL_{self.atoms[0]}, true/false);")
         buff.append("}")
         buff.append("")
         return '\n'.join(buff)
 
     def fill_tracepoint_attach_probe(self):
-        return "\trv_attach_trace_probe(\"%s\", /* XXX: tracepoint */, handle_example_event);" \
-                % self.name
+        return f"\trv_attach_trace_probe(\"{self.name}\", /* XXX: tracepoint */, handle_example_event);"
 
     def fill_tracepoint_detach_helper(self):
-        return "\trv_detach_trace_probe(\"%s\", /* XXX: tracepoint */, handle_sample_event);" \
-                % self.name
+        return f"\trv_detach_trace_probe(\"{self.name}\", /* XXX: tracepoint */, handle_sample_event);"
 
     def fill_atoms_init(self):
         buff = []
         for a in self.atoms:
-            buff.append("\tltl_atom_set(mon, LTL_%s, true/false);" % a)
+            buff.append(f"\tltl_atom_set(mon, LTL_{a}, true/false);")
         return '\n'.join(buff)
 
     def fill_model_h(self):
diff --git a/tools/verification/rvgen/rvgen/templates/dot2k/main.c b/tools/verification/rvgen/rvgen/templates/dot2k/main.c
index a14e4f0883db..bf0999f6657a 100644
--- a/tools/verification/rvgen/rvgen/templates/dot2k/main.c
+++ b/tools/verification/rvgen/rvgen/templates/dot2k/main.c
@@ -21,7 +21,7 @@
  */
 #define RV_MON_TYPE RV_MON_%%MONITOR_TYPE%%
 #include "%%MODEL_NAME%%.h"
-#include <rv/da_monitor.h>
+#include <rv/%%MONITOR_CLASS%%_monitor.h>
 
 /*
  * This is the instrumentation part of the monitor.
diff --git a/tools/verification/rvgen/rvgen/templates/dot2k/trace_hybrid.h b/tools/verification/rvgen/rvgen/templates/dot2k/trace_hybrid.h
new file mode 100644
index 000000000000..c8290e9ba2f4
--- /dev/null
+++ b/tools/verification/rvgen/rvgen/templates/dot2k/trace_hybrid.h
@@ -0,0 +1,16 @@
+/* SPDX-License-Identifier: GPL-2.0 */
+
+/*
+ * Snippet to be included in rv_trace.h
+ */
+
+#ifdef CONFIG_RV_MON_%%MODEL_NAME_UP%%
+DEFINE_EVENT(event_%%MONITOR_CLASS%%, event_%%MODEL_NAME%%,
+%%TRACEPOINT_ARGS_SKEL_EVENT%%);
+
+DEFINE_EVENT(error_%%MONITOR_CLASS%%, error_%%MODEL_NAME%%,
+%%TRACEPOINT_ARGS_SKEL_ERROR%%);
+
+DEFINE_EVENT(error_env_%%MONITOR_CLASS%%, error_env_%%MODEL_NAME%%,
+%%TRACEPOINT_ARGS_SKEL_ERROR_ENV%%);
+#endif /* CONFIG_RV_MON_%%MODEL_NAME_UP%% */