unaiverse.hsm.action

What this module does 🔴

Defines the Action abstraction used by the hybrid state machine to bind callables to transitions, manage timing/timeouts, wildcards, and an ordered list of pending interactions per requester.

action

█████ █████ ██████ █████ █████ █████ █████ ██████████ ███████████ █████████ ██████████ ░░███ ░░███ ░░██████ ░░███ ░░███ ░░███ ░░███ ░░███░░░░░█░░███░░░░░███ ███░░░░░███░░███░░░░░█ ░███ ░███ ░███░███ ░███ ██████ ░███ ░███ ░███ ░███ █ ░ ░███ ░███ ░███ ░░░ ░███ █ ░ ░███ ░███ ░███░░███░███ ░░░░░███ ░███ ░███ ░███ ░██████ ░██████████ ░░█████████ ░██████
░███ ░███ ░███ ░░██████ ███████ ░███ ░░███ ███ ░███░░█ ░███░░░░░███ ░░░░░░░░███ ░███░░█
░███ ░███ ░███ ░░█████ ███░░███ ░███ ░░░█████░ ░███ ░ █ ░███ ░███ ███ ░███ ░███ ░ █ ░░████████ █████ ░░█████░░████████ █████ ░░███ ██████████ █████ █████░░█████████ ██████████ ░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░ ░░░░░ ░░░ ░░░░░░░░░░ ░░░░░ ░░░░░ ░░░░░░░░░ ░░░░░░░░░░ A Collectionless AI Project (https://collectionless.ai) Registration/Login: https://unaiverse.io Code Repositories: https://github.com/collectionlessai/ Main Developers: Stefano Melacci (Project Leader), Christian Di Maio, Tommaso Guidi

Action

Action(name: str, args: dict, actionable: object, idx: int = -1, ready: bool = True, msg: str | None = None, avoid_changing_ready: bool = False, teleport: bool = False, high_priority: bool = False, max_duration: float | str = 0.0, retry_timeout: float | str = 0, delay: float | str = 0)

A single executable action inside a Hybrid State Machine (HSM) state.

An Action wraps a named method on an actionable object together with a fixed set of keyword arguments and a collection of time-based constraints (maximum duration, retry timeout, delay). The HSM schedules and dispatches actions; calling an instance directly (via __call__) runs the underlying method and returns a status code that tells the HSM whether to retry, move on, or mark the step as done.

Actions are either inner-ready (the HSM can execute them autonomously) or outer-ready (they require an external Interaction request to be triggered). The two flags (inner / outer) are set on construction and can be changed at runtime with set_as_ready and set_as_not_ready. A high-priority flag (high_priority) is a hint to the policy to prefer this action over others in the same state.

Wildcard placeholders (strings matching <name> patterns) may appear in args, msg, and the time parameters. When set_state_machine registers the parent HybridStateMachine, its wildcard table is applied to expand all placeholders to concrete values. The expansion can be re-applied later via apply_wildcards.

Attributes:

Name	Type	Description
name		Name of the underlying callable on actionable.
args		Keyword arguments passed to the callable, after wildcard expansion.
actionable		The object that owns the method named name.
interactions		The ActionInteractionList holding pending external requests.
id		Optional unique integer ID (-1 when not needed).
msg		Optional human-readable description logged when the action runs.
inner		True when the action can be executed autonomously (inner readiness).
outer		True when the action accepts external interaction requests.
state_machine		The parent HybridStateMachine, set by set_state_machine.
teleport		When True the action is hidden from the drawn state-machine diagram.
high_priority		When True the policy prefers this action over lower-priority peers.
param_list		Ordered list of parameter names accepted by the underlying callable.
param_to_default_value		Mapping from parameter name to its default value.
wildcards		Mapping from placeholder strings to their concrete replacement values.
args_with_wildcards		Backup of args before wildcard expansion (restored on each apply_wildcards call).
msg_with_wildcards		Backup of msg before wildcard expansion.
system_interaction		The synthetic Interaction representing an autonomous (non-requester-driven) invocation. Rebuilt whenever wildcards are applied.

Examples:

The framework constructs Action objects automatically when loading HSM behaviour definitions. Direct construction is rarely needed outside tests:

>>> class MyAgent:
...     async def greet(self, name: str = "world") -> bool:
...         print(f"Hello, {name}!")
...         return True
>>>
>>> agent = MyAgent()
>>> action = Action(name="greet", args={"name": "Alice"}, actionable=agent)
>>> action.get_name()
'greet'
>>> action.is_ready()
True

Initialize an Action wrapping a named method on actionable.

Resolves the callable from actionable via getattr, inspects its signature to populate param_list and param_to_default_value, and extracts any time-based parameters (max_duration, retry_timeout, delay) from args when the corresponding wildcard keys are present. If any parameter in the callable's signature belongs to Custom.INTERACTION_INJECT_NAMES and avoid_changing_ready is False, the action is forced to outer-only mode (inner=False, outer=True) so it waits for an external Interaction.

HTML entities in msg are unescaped via html.unescape before storage.

A lightweight system_interaction is built at the very end of initialization so that the action can be dispatched autonomously without an external requester.

Parameters:

Name	Type	Description	Default
name	str	Name of the method to look up on actionable.	required
args	dict	Keyword arguments to pass to the method. Special time-related keys (see Custom.SECONDS_ARG_NAMES, Custom.TIMEOUT_ARG_NAMES, Custom.DELAY_ARG_NAMES) are extracted and removed from args.	required
actionable	object	Object on which the method named name is resolved.	required
idx	int	Optional unique integer ID for this action. Defaults to -1.	-1
ready	bool	Initial inner-readiness flag. Defaults to True.	True
msg	str | None	Optional human-readable description logged when the action runs. HTML entities are unescaped automatically. Defaults to None.	None
avoid_changing_ready	bool	When True, suppresses the automatic override that forces outer-only mode for interaction-injecting parameters. Defaults to False.	False
teleport	bool	When True, the action is hidden from the state-machine diagram. Defaults to False.	False
high_priority	bool	When True, the policy prefers this action over others. Defaults to False.	False
max_duration	float | str	Maximum wall-clock seconds the action may run before the HSM considers it expired. Values <= 0 mean no limit. May be a wildcard string (e.g. "<max_time>"). Defaults to 0.	0.0
retry_timeout	float | str	Maximum seconds the HSM keeps re-attempting this action before giving up. Values <= 0 mean no timeout. May be a wildcard string. Defaults to 0.	0
delay	float | str	Seconds that must elapse after entering the parent state before this action is considered for execution. Values <= 0 mean no delay. May be a wildcard string. Defaults to 0.	0

Raises:

Type	Description
ValueError	If name is not a valid attribute on actionable.
ValueError	If any key in args is a reserved framework name or is not in the callable's parameter list (checked by check_provided_args with exception=True).

Source code in unaiverse/hsm/action.py

def __init__(self, name: str, args: dict, actionable: object,
             idx: int = -1,
             ready: bool = True,
             msg: str | None = None,
             avoid_changing_ready: bool = False,
             teleport: bool = False,
             high_priority: bool = False,
             max_duration: float | str = 0.,
             retry_timeout: float | str = 0,
             delay: float | str = 0):
    """Initialize an ``Action`` wrapping a named method on ``actionable``.

    Resolves the callable from ``actionable`` via ``getattr``, inspects its signature
    to populate ``param_list`` and ``param_to_default_value``, and extracts any
    time-based parameters (``max_duration``, ``retry_timeout``, ``delay``) from
    ``args`` when the corresponding wildcard keys are present. If any parameter in the
    callable's signature belongs to ``Custom.INTERACTION_INJECT_NAMES`` and
    ``avoid_changing_ready`` is ``False``, the action is forced to *outer-only* mode
    (``inner=False``, ``outer=True``) so it waits for an external ``Interaction``.

    HTML entities in ``msg`` are unescaped via ``html.unescape`` before storage.

    A lightweight ``system_interaction`` is built at the very end of initialization
    so that the action can be dispatched autonomously without an external requester.

    Args:
        name: Name of the method to look up on ``actionable``.
        args: Keyword arguments to pass to the method. Special time-related keys
            (see ``Custom.SECONDS_ARG_NAMES``, ``Custom.TIMEOUT_ARG_NAMES``,
            ``Custom.DELAY_ARG_NAMES``) are extracted and removed from ``args``.
        actionable: Object on which the method named ``name`` is resolved.
        idx: Optional unique integer ID for this action. Defaults to -1.
        ready: Initial inner-readiness flag. Defaults to True.
        msg: Optional human-readable description logged when the action runs.
            HTML entities are unescaped automatically. Defaults to None.
        avoid_changing_ready: When True, suppresses the automatic override that
            forces outer-only mode for interaction-injecting parameters.
            Defaults to False.
        teleport: When True, the action is hidden from the state-machine diagram.
            Defaults to False.
        high_priority: When True, the policy prefers this action over others.
            Defaults to False.
        max_duration: Maximum wall-clock seconds the action may run before the HSM
            considers it expired. Values <= 0 mean no limit. May be a wildcard
            string (e.g. ``"<max_time>"``). Defaults to 0.
        retry_timeout: Maximum seconds the HSM keeps re-attempting this action
            before giving up. Values <= 0 mean no timeout. May be a wildcard
            string. Defaults to 0.
        delay: Seconds that must elapse after entering the parent state before
            this action is considered for execution. Values <= 0 mean no delay.
            May be a wildcard string. Defaults to 0.

    Raises:
        ValueError: If ``name`` is not a valid attribute on ``actionable``.
        ValueError: If any key in ``args`` is a reserved framework name or is not
            in the callable's parameter list (checked by ``check_provided_args``
            with ``exception=True``).
    """
    # Basic properties
    self.name = name  # Name of the action (name of the corresponding method)
    self.args = args.copy()  # Dictionary of arguments to pass to the action (shallow copy, we will remove some)
    self.actionable = actionable  # Object on which the method whose name is self.name is searched
    self.interactions = ActionInteractionList()  # List of interactions to make this action ready to be executed
    self.id = idx  # Unique ID of the action (-1 if not needed)
    self.msg = msg  # Human-readable message associated to this instance of action
    self.inner = ready
    self.outer = True
    self.state_machine = None
    self.teleport = teleport
    self.high_priority = high_priority
    self.__mark = None

    # Fix UNICODE chars
    if self.msg is not None:
        self.msg = html.unescape(self.msg)

    # Reference elements
    self.__fcn = self.__action_name_to_callable(name)  # The real method to be called
    self.__sig = inspect.signature(self.__fcn)  # Signature of the method for argument inspection

    # Parameter names and default values
    self.param_list = []  # Full list of the parameters that the action supports
    self.param_to_default_value = {}  # From parameter to its default value, if any
    self.__get_action_params()  # This will fill the two attributes above

    # Time-based metrics
    self.__action_max_duration = max_duration  # A total time <= 0 means "no total time at all"
    self.__action_max_duration_with_wildcard = max_duration if isinstance(max_duration, str) else None
    self.__guess_total_time(self.args)  # This will "guess" the value of self.__action_max_duration from args dict

    # Time-based metrics
    self.__action_retry_timeout = retry_timeout  # A retry_timeout <= 0 means "no total time at all"
    self.__action_retry_timeout_with_wildcard = retry_timeout if isinstance(retry_timeout, str) else None
    self.__guess_timeout(self.args)  # This will "guess" the value of self.__action_retry_timeout from the args dict

    # Time-based metrics
    self.__delay = delay
    self.__delay_with_wildcard = delay if isinstance(delay, str) else None
    self.__guess_delay(self.args)  # This will "guess" the value of self.__delay from the args dict

    # Checking arguments (also removing 'retry_timeout', 'delay', etc...)
    self.check_provided_args(self.args, exception=True, remove_special_arguments=True)

    # Argument values replaced by wildcards (commonly assumed to be in the format <value>)
    self.wildcards = {}  # Value-to-value (es: <playlist> to this:and:this)
    self.args_with_wildcards = copy.deepcopy(self.args)  # Backup of the originally provided arguments
    self.msg_with_wildcards = self.msg

    # Fixing (forcing NOT-ready on some actions)
    if not avoid_changing_ready:
        for p in self.param_list:
            if p in Custom.INTERACTION_INJECT_NAMES:
                self.inner = False
                self.outer = True
                break

    # Whether for this Action we've already register_lazy a system_interaction,
    # and whether `copy_sys` was requested for the current round.
    self._lazy_registered = False
    self._must_lazy_register = False
    self._copy_sys_on_register = False

    # Build the initial system_interaction (lightweight dummy or interaction-shaped). MUST stay
    # at the very end of __init__.
    self.__build_system_interaction()

name instance-attribute

name = name

args instance-attribute

args = copy()

actionable instance-attribute

actionable = actionable

interactions instance-attribute

interactions = ActionInteractionList()

id instance-attribute

id = idx

msg instance-attribute

msg = msg

inner instance-attribute

inner = ready

outer instance-attribute

outer = True

state_machine instance-attribute

state_machine = None

teleport instance-attribute

teleport = teleport

high_priority instance-attribute

high_priority = high_priority

param_list instance-attribute

param_list = []

param_to_default_value instance-attribute

param_to_default_value = {}

wildcards instance-attribute

wildcards = {}

args_with_wildcards instance-attribute

args_with_wildcards = deepcopy(args)

msg_with_wildcards instance-attribute

msg_with_wildcards = msg

ready property

ready: bool

Return the inner readiness flag of the action.

This property exposes the inner attribute, which is True when the action can be dispatched autonomously by the HSM without an external Interaction. Use is_ready for a full readiness check that also accounts for pending interactions and the configured delay.

Returns:

Type	Description
bool	True if the action is inner-ready, False otherwise.

is_high_priority property

is_high_priority: bool

Return whether the action is marked as high priority.

High-priority actions are preferred by the policy over other actions in the same HSM state when multiple actions are simultaneously ready. The flag is set at construction time or via set_high_priority.

Returns:

Type	Description
bool	True if the action is high priority, False otherwise.

set_high_priority

set_high_priority() -> None

Mark the action as high priority.

Sets high_priority to True, signalling to the HSM policy that this action should be preferred over non-high-priority actions in the same state when multiple are ready at the same time. See also is_high_priority.

Source code in unaiverse/hsm/action.py

def set_high_priority(self) -> None:
    """Mark the action as high priority.

    Sets ``high_priority`` to ``True``, signalling to the HSM policy that this action
    should be preferred over non-high-priority actions in the same state when
    multiple are ready at the same time. See also ``is_high_priority``.
    """
    self.high_priority = True

set_state_machine

set_state_machine(hsm: object) -> None

Register the parent state machine that owns this action.

Stores a reference to hsm in state_machine and immediately calls set_wildcards with the wildcards from hsm.get_wildcards(), so that all placeholder values in args, msg, and the time parameters are resolved to their concrete values.

Parameters:

Name	Type	Description	Default
hsm	object	The HybridStateMachine instance that owns this action.	required

Source code in unaiverse/hsm/action.py

def set_state_machine(self, hsm: object) -> None:
    """Register the parent state machine that owns this action.

    Stores a reference to ``hsm`` in ``state_machine`` and immediately calls
    ``set_wildcards`` with the wildcards from ``hsm.get_wildcards()``, so that all
    placeholder values in ``args``, ``msg``, and the time parameters are resolved to
    their concrete values.

    Args:
        hsm: The ``HybridStateMachine`` instance that owns this action.
    """
    self.state_machine = hsm
    self.set_wildcards(hsm.get_wildcards())

get_name

get_name() -> str

Return the name of the underlying callable.

Returns:

Type	Description
str	The string name of the method on actionable that this action invokes.

Source code in unaiverse/hsm/action.py

def get_name(self) -> str:
    """Return the name of the underlying callable.

    Returns:
        The string name of the method on ``actionable`` that this action invokes.
    """
    return self.name

set_mark

set_mark(mark: object) -> None

Store an arbitrary marker object on the action.

The mark is a lightweight tag that the HSM or external code can attach to an action for bookkeeping purposes (for example, to associate a state transition token). It has no effect on action scheduling or dispatch. Use get_mark to retrieve the stored value and clear_mark to reset it to None.

Parameters:

Name	Type	Description	Default
mark	object	Any object to store as the marker. None is a valid value.	required

Source code in unaiverse/hsm/action.py

def set_mark(self, mark: object) -> None:
    """Store an arbitrary marker object on the action.

    The mark is a lightweight tag that the HSM or external code can attach to an
    action for bookkeeping purposes (for example, to associate a state transition
    token). It has no effect on action scheduling or dispatch. Use ``get_mark`` to
    retrieve the stored value and ``clear_mark`` to reset it to ``None``.

    Args:
        mark: Any object to store as the marker. ``None`` is a valid value.
    """
    self.__mark = mark

get_mark

get_mark() -> object

Return the marker object previously stored by set_mark.

Returns None when no mark has been set or after clear_mark has been called.

Returns:

Type	Description
object	The stored marker object, or None if no mark is present.

Source code in unaiverse/hsm/action.py

def get_mark(self) -> object:
    """Return the marker object previously stored by ``set_mark``.

    Returns ``None`` when no mark has been set or after ``clear_mark`` has been
    called.

    Returns:
        The stored marker object, or ``None`` if no mark is present.
    """
    return self.__mark

clear_mark

clear_mark() -> None

Clear the marker object stored on the action.

Resets the internal mark to None. Equivalent to calling set_mark(None). This is called automatically at the end of each action invocation to prevent stale marks from leaking across execution rounds.

Source code in unaiverse/hsm/action.py

def clear_mark(self) -> None:
    """Clear the marker object stored on the action.

    Resets the internal mark to ``None``. Equivalent to calling
    ``set_mark(None)``. This is called automatically at the end of each action
    invocation to prevent stale marks from leaking across execution rounds.
    """
    self.__mark = None

to_code_str

to_code_str() -> str

Return a compact code-style string representation of the action for logging.

The format is aai:<name>|<args>| followed by one indented line per pending interaction (using each interaction's own to_code_str representation), or the suffix no-int when no interactions are present.

Returns:

Type	Description
str	A single- or multi-line string summarising the action and its interactions.

Source code in unaiverse/hsm/action.py

def to_code_str(self) -> str:
    """Return a compact code-style string representation of the action for logging.

    The format is ``aai:<name>|<args>|`` followed by one indented line per pending
    interaction (using each interaction's own ``to_code_str`` representation), or the
    suffix ``no-int`` when no interactions are present.

    Returns:
        A single- or multi-line string summarising the action and its interactions.
    """
    s = f"aai:{self.name}|{self.args}|"
    if len(self.interactions) > 0:
        return s + "\n" + "\n".join(f"   {inter.to_code_str(True)}" for inter in self.interactions)
    else:
        return s + "no-int"

set_as_ready

set_as_ready() -> None

Set the action to inner-ready mode, enabling autonomous execution.

Sets inner to True and outer to False. In this mode the HSM can dispatch the action without waiting for an external Interaction. Accepting external requests is disabled (outer=False) to prevent double-dispatching for actions that are already unconditionally available. See also set_as_not_ready.

Source code in unaiverse/hsm/action.py

def set_as_ready(self) -> None:
    """Set the action to inner-ready mode, enabling autonomous execution.

    Sets ``inner`` to ``True`` and ``outer`` to ``False``. In this mode the HSM can
    dispatch the action without waiting for an external ``Interaction``. Accepting
    external requests is disabled (``outer=False``) to prevent double-dispatching for
    actions that are already unconditionally available. See also ``set_as_not_ready``.
    """
    self.outer = False  # For the moment, I prefer to avoid "solid" actions to also take requests from the outside
    self.inner = True

set_as_not_ready

set_as_not_ready() -> None

Set the action to outer-only mode, requiring an external interaction to run.

Sets inner to False and outer to True. In this mode the action will not be dispatched autonomously; it must be explicitly requested by an external agent via an Interaction. See also set_as_ready.

Source code in unaiverse/hsm/action.py

def set_as_not_ready(self) -> None:
    """Set the action to outer-only mode, requiring an external interaction to run.

    Sets ``inner`` to ``False`` and ``outer`` to ``True``. In this mode the action
    will not be dispatched autonomously; it must be explicitly requested by an
    external agent via an ``Interaction``. See also ``set_as_ready``.
    """
    self.inner = False
    self.outer = True

set_msg

set_msg(msg: str | None) -> None

Set the human-readable message associated with this action.

HTML entities in msg are automatically unescaped via html.unescape before storage. Both msg and msg_with_wildcards are updated together so that subsequent wildcard expansions (apply_wildcards) start from the new value. Passing None clears any previously set message.

Parameters:

Name	Type	Description	Default
msg	str | None	The new message string, or None to clear the current message.	required

Source code in unaiverse/hsm/action.py

def set_msg(self, msg: str | None) -> None:
    """Set the human-readable message associated with this action.

    HTML entities in ``msg`` are automatically unescaped via ``html.unescape`` before
    storage. Both ``msg`` and ``msg_with_wildcards`` are updated together so that
    subsequent wildcard expansions (``apply_wildcards``) start from the new value.
    Passing ``None`` clears any previously set message.

    Args:
        msg: The new message string, or ``None`` to clear the current message.
    """

    if msg is not None:
        self.msg = html.unescape(msg)
        self.msg_with_wildcards = self.msg
    else:
        self.msg = None
        self.msg_with_wildcards = None

is_ready

is_ready(consider_interactions: bool = True, delay_starting_time: float = -1.0) -> bool

Return True if the action is eligible for execution right now.

An action is considered ready when all of the following conditions hold:

1. The configured delay has elapsed (or no delay is set). The delay is measured from delay_starting_time using time.perf_counter().

2. At least one of the following sub-conditions is true:

3. inner is True (the action can run autonomously), or

4. consider_interactions is True, outer is True, and at least one doable interaction exists in interactions.

Parameters:

Name	Type	Description	Default
consider_interactions	bool	When True, pending external interactions count towards readiness. Defaults to True.	True
delay_starting_time	float	The time.perf_counter() timestamp from which the configured delay is measured. A value <= 0 disables the delay check. Defaults to -1.	-1.0

Returns:

Type	Description
bool	True if the action is eligible for execution, False otherwise.

Source code in unaiverse/hsm/action.py

def is_ready(self, consider_interactions: bool = True, delay_starting_time: float = -1.) -> bool:
    """Return True if the action is eligible for execution right now.

    An action is considered ready when all of the following conditions hold:

    1. The configured delay has elapsed (or no delay is set). The delay is measured
       from ``delay_starting_time`` using ``time.perf_counter()``.
    2. At least one of the following sub-conditions is true:

       - ``inner`` is ``True`` (the action can run autonomously), or
       - ``consider_interactions`` is ``True``, ``outer`` is ``True``, and at least
         one doable interaction exists in ``interactions``.

    Args:
        consider_interactions: When ``True``, pending external interactions count
            towards readiness. Defaults to True.
        delay_starting_time: The ``time.perf_counter()`` timestamp from which the
            configured delay is measured. A value <= 0 disables the delay check.
            Defaults to -1.

    Returns:
        True if the action is eligible for execution, False otherwise.
    """
    is_delayed = (delay_starting_time > 0 and self.__delay > 0 and
                  (time.perf_counter() - delay_starting_time) <= self.__delay)
    return not is_delayed and (self.inner or (consider_interactions and self.outer and len(
        self.interactions.get_interactions(doable_only=True)) > 0))

is_teleport

is_teleport() -> bool

Return whether this action is a teleport action.

Teleport actions are hidden from the state-machine diagram rendered by the framework. They are typically used for instantaneous, bookkeeping-only transitions that do not represent meaningful user-visible behaviour.

Returns:

Type	Description
bool	True if the action is a teleport action, False otherwise.

Source code in unaiverse/hsm/action.py

def is_teleport(self) -> bool:
    """Return whether this action is a teleport action.

    Teleport actions are hidden from the state-machine diagram rendered by the
    framework. They are typically used for instantaneous, bookkeeping-only
    transitions that do not represent meaningful user-visible behaviour.

    Returns:
        True if the action is a teleport action, False otherwise.
    """
    return self.teleport

allows_outer_interactions

allows_outer_interactions() -> bool

Return whether the action accepts externally triggered interactions.

When True (outer=True), external agents may push Interaction requests into the action's interactions list, and the action becomes eligible for dispatch when at least one doable interaction is present. When False, calls to add_interaction are silently rejected.

Returns:

Type	Description
bool	True if the action accepts outer interactions, False otherwise.

Source code in unaiverse/hsm/action.py

def allows_outer_interactions(self) -> bool:
    """Return whether the action accepts externally triggered interactions.

    When ``True`` (``outer=True``), external agents may push ``Interaction`` requests
    into the action's ``interactions`` list, and the action becomes eligible for
    dispatch when at least one doable interaction is present. When ``False``, calls
    to ``add_interaction`` are silently rejected.

    Returns:
        True if the action accepts outer interactions, False otherwise.
    """
    return self.outer

allows_inner_interactions

allows_inner_interactions() -> bool

Return whether the action can be dispatched autonomously by the HSM.

When True (inner=True), the HSM may execute the action without any external Interaction request. This corresponds to the ready property.

Returns:

Type	Description
bool	True if the action can be triggered internally, False otherwise.

Source code in unaiverse/hsm/action.py

def allows_inner_interactions(self) -> bool:
    """Return whether the action can be dispatched autonomously by the HSM.

    When ``True`` (``inner=True``), the HSM may execute the action without any
    external ``Interaction`` request. This corresponds to the ``ready`` property.

    Returns:
        True if the action can be triggered internally, False otherwise.
    """
    return self.inner

set_timeout

set_timeout(retry_timeout: float) -> None

Set the retry timeout to a custom value.

Overrides the timeout configured at construction time or via the args dict. The timeout controls how long the HSM keeps re-attempting this action before giving up. A value <= 0 disables the timeout. See also set_default_timeout.

Parameters:

Name	Type	Description	Default
retry_timeout	float	Maximum seconds to retry the action. Values <= 0 disable the timeout entirely.	required

Source code in unaiverse/hsm/action.py

def set_timeout(self, retry_timeout: float) -> None:
    """Set the retry timeout to a custom value.

    Overrides the timeout configured at construction time or via the ``args`` dict.
    The timeout controls how long the HSM keeps re-attempting this action before
    giving up. A value <= 0 disables the timeout. See also ``set_default_timeout``.

    Args:
        retry_timeout: Maximum seconds to retry the action. Values <= 0 disable
            the timeout entirely.
    """
    self.__action_retry_timeout = retry_timeout

set_default_timeout

set_default_timeout() -> None

Reset the retry timeout to the framework-wide default value.

Sets the internal timeout to Custom.DEFAULT_TIMEOUT, overriding any value that was previously set at construction time, via args, or by a call to set_timeout. See also set_timeout.

Source code in unaiverse/hsm/action.py

def set_default_timeout(self) -> None:
    """Reset the retry timeout to the framework-wide default value.

    Sets the internal timeout to ``Custom.DEFAULT_TIMEOUT``, overriding any value
    that was previously set at construction time, via ``args``, or by a call to
    ``set_timeout``. See also ``set_timeout``.
    """
    self.__action_retry_timeout = Custom.DEFAULT_TIMEOUT

is_pedantic

is_pedantic() -> bool

Return whether a retry timeout is configured for this action.

An action is pedantic when retry_timeout is greater than zero, meaning the HSM will keep re-attempting it rather than skipping it immediately on failure. Non-pedantic actions (retry_timeout <= 0) are skipped after the first failed execution attempt.

Returns:

Type	Description
bool	True if a positive retry timeout has been configured, False otherwise.

Source code in unaiverse/hsm/action.py

def is_pedantic(self) -> bool:
    """Return whether a retry timeout is configured for this action.

    An action is *pedantic* when ``retry_timeout`` is greater than zero, meaning the
    HSM will keep re-attempting it rather than skipping it immediately on failure.
    Non-pedantic actions (``retry_timeout <= 0``) are skipped after the first failed
    execution attempt.

    Returns:
        True if a positive retry timeout has been configured, False otherwise.
    """
    return self.__action_retry_timeout > 0

get_total_time

get_total_time() -> float | str

Return the maximum wall-clock duration configured for this action.

Values greater than zero cap the total time the action may run. A value of 0 (or any value <= 0) means there is no upper time limit. When the parameter was specified as a wildcard placeholder string (e.g. "<max_time>"), the raw wildcard string is returned until apply_wildcards has resolved it.

Returns:

Type	Description
float | str	The maximum duration in seconds as a float, or a wildcard placeholder
float | str	str if the value has not yet been resolved.

Source code in unaiverse/hsm/action.py

def get_total_time(self) -> float | str:
    """Return the maximum wall-clock duration configured for this action.

    Values greater than zero cap the total time the action may run. A value of ``0``
    (or any value <= 0) means there is no upper time limit. When the parameter was
    specified as a wildcard placeholder string (e.g. ``"<max_time>"``), the raw
    wildcard string is returned until ``apply_wildcards`` has resolved it.

    Returns:
        The maximum duration in seconds as a ``float``, or a wildcard placeholder
        ``str`` if the value has not yet been resolved.
    """
    return self.__action_max_duration

get_timeout

get_timeout() -> float | str

Return the retry timeout configured for this action.

The retry timeout is the maximum number of seconds the HSM keeps re-attempting this action before giving up. A value of 0 (or any value <= 0) disables the timeout. When the parameter was specified as a wildcard placeholder string, the raw string is returned until apply_wildcards has resolved it. See also is_pedantic and set_timeout.

Returns:

Type	Description
float | str	The retry timeout in seconds as a float, or a wildcard placeholder
float | str	str if the value has not yet been resolved.

Source code in unaiverse/hsm/action.py

def get_timeout(self) -> float | str:
    """Return the retry timeout configured for this action.

    The retry timeout is the maximum number of seconds the HSM keeps re-attempting
    this action before giving up. A value of ``0`` (or any value <= 0) disables the
    timeout. When the parameter was specified as a wildcard placeholder string, the
    raw string is returned until ``apply_wildcards`` has resolved it. See also
    ``is_pedantic`` and ``set_timeout``.

    Returns:
        The retry timeout in seconds as a ``float``, or a wildcard placeholder
        ``str`` if the value has not yet been resolved.
    """
    return self.__action_retry_timeout

get_delay

get_delay() -> float | str

Return the entry delay configured for this action.

The delay is the number of seconds that must elapse after the parent HSM state is entered before this action becomes eligible for execution. A value of 0 (or any value <= 0) means the action is immediately eligible. When the parameter was specified as a wildcard placeholder string, the raw string is returned until apply_wildcards has resolved it.

Returns:

Type	Description
float | str	The delay in seconds as a float, or a wildcard placeholder str if
float | str	the value has not yet been resolved.

Source code in unaiverse/hsm/action.py

def get_delay(self) -> float | str:
    """Return the entry delay configured for this action.

    The delay is the number of seconds that must elapse after the parent HSM state is
    entered before this action becomes eligible for execution. A value of ``0`` (or
    any value <= 0) means the action is immediately eligible. When the parameter was
    specified as a wildcard placeholder string, the raw string is returned until
    ``apply_wildcards`` has resolved it.

    Returns:
        The delay in seconds as a ``float``, or a wildcard placeholder ``str`` if
        the value has not yet been resolved.
    """
    return self.__delay

to_list

to_list() -> list

Return a flat list snapshot of the action's properties for serialization or comparison.

The returned list has the fixed layout: [name, args, inner, outer, total_time, retry_timeout, delay, msg]. Time fields (total_time, retry_timeout, delay) are normalized to 0. when their value is a non-positive number, so comparisons against the default "no limit" state are straightforward. Wildcard placeholder strings are kept as-is. See also to_dict for a named-key representation.

Returns:

Type	Description
list	A list of eight elements containing the action's name, args dict, inner and
list	outer readiness flags, time constraints, and message.

Source code in unaiverse/hsm/action.py

def to_list(self) -> list:
    """Return a flat list snapshot of the action's properties for serialization or comparison.

    The returned list has the fixed layout:
    ``[name, args, inner, outer, total_time, retry_timeout, delay, msg]``.
    Time fields (``total_time``, ``retry_timeout``, ``delay``) are normalized to
    ``0.`` when their value is a non-positive number, so comparisons against the
    default "no limit" state are straightforward. Wildcard placeholder strings are
    kept as-is. See also ``to_dict`` for a named-key representation.

    Returns:
        A list of eight elements containing the action's name, args dict, inner and
        outer readiness flags, time constraints, and message.
    """
    total_time = 0.
    retry_timeout = 0.
    delay = 0.
    if isinstance(self.__action_max_duration, str) or self.__action_max_duration > 0:
        total_time = self.__action_max_duration
    if isinstance(self.__action_retry_timeout, str) or self.__action_retry_timeout > 0.:
        retry_timeout = self.__action_retry_timeout
    if isinstance(self.__delay, str) or self.__delay > 0.:
        delay = self.__delay
    return [self.name, self.args, self.inner, self.outer, total_time, retry_timeout, delay, self.msg]

to_dict

to_dict() -> dict

Return a named-key dictionary snapshot of the action's properties for serialization.

The returned dictionary contains the keys "action", "action_kwargs", "msg", "ready", "high_priority", and the canonical time-constraint key names from Custom.SECONDS_ARG_NAMES[0], Custom.TIMEOUT_ARG_NAMES[0], and Custom.DELAY_ARG_NAMES[0]. Non-ASCII characters in msg are encoded as XML character references so the result is safe to embed in JSON or HTML. See also to_list for a positional-list representation.

Returns:

Type	Description
dict	A dictionary mapping property names to their current values.

Source code in unaiverse/hsm/action.py

def to_dict(self) -> dict:
    """Return a named-key dictionary snapshot of the action's properties for serialization.

    The returned dictionary contains the keys ``"action"``, ``"action_kwargs"``,
    ``"msg"``, ``"ready"``, ``"high_priority"``, and the canonical time-constraint
    key names from ``Custom.SECONDS_ARG_NAMES[0]``, ``Custom.TIMEOUT_ARG_NAMES[0]``,
    and ``Custom.DELAY_ARG_NAMES[0]``. Non-ASCII characters in ``msg`` are encoded
    as XML character references so the result is safe to embed in JSON or HTML. See
    also ``to_list`` for a positional-list representation.

    Returns:
        A dictionary mapping property names to their current values.
    """
    return {
        "action": self.name,
        "action_kwargs": self.args,
        "msg": self.msg.encode("ascii",
                               "xmlcharrefreplace").decode("ascii") if self.msg is not None else None,
        "ready": self.inner,
        "high_priority": self.is_high_priority,
        Custom.SECONDS_ARG_NAMES[0]: self.__action_max_duration,
        Custom.TIMEOUT_ARG_NAMES[0]: self.__action_retry_timeout,
        Custom.DELAY_ARG_NAMES[0]: self.__delay
    }

same_as

same_as(name: str, args: dict | None) -> bool

Return whether this action matches the given name and argument subset.

Two actions are considered the same when all of the following hold:

1. Their names are equal.
2. Every key in args is a recognized parameter of this action (validated by check_provided_args).
3. For every key-value pair in args that is not in Custom.RESERVED_IN_ACTION_KWARGS, the value matches the corresponding value in self.args (when the key is present there).

Keys not present in args are assumed to match, so same_as accepts a subset of arguments. Reserved framework keys (time-based, sentinel, etc.) are excluded from value comparison.

Parameters:

Name	Type	Description	Default
name	str	The action name to compare against.	required
args	dict | None	A (possibly partial) argument dictionary to compare. None is treated as an empty dictionary.	required

Returns:

Type	Description
bool	True if this action is considered identical to the described action, False
bool	otherwise.

Source code in unaiverse/hsm/action.py

def same_as(self, name: str, args: dict | None) -> bool:
    """Return whether this action matches the given name and argument subset.

    Two actions are considered the same when all of the following hold:

    1. Their names are equal.
    2. Every key in ``args`` is a recognized parameter of this action (validated by
       ``check_provided_args``).
    3. For every key-value pair in ``args`` that is not in
       ``Custom.RESERVED_IN_ACTION_KWARGS``, the value matches the corresponding
       value in ``self.args`` (when the key is present there).

    Keys not present in ``args`` are assumed to match, so ``same_as`` accepts a
    *subset* of arguments. Reserved framework keys (time-based, sentinel, etc.) are
    excluded from value comparison.

    Args:
        name: The action name to compare against.
        args: A (possibly partial) argument dictionary to compare. ``None`` is
            treated as an empty dictionary.

    Returns:
        True if this action is considered identical to the described action, False
        otherwise.
    """
    if args is None:
        args = {}

    # The current action is the same of another action called with some arguments "args" if:
    # 1) it has the same name of the other action
    # 2) the name of the arguments in "args" are known and valid
    # 3) the values of the arguments in "args" matches the ones of the current action, being them default or not
    # the values of those arguments that are not in "args" are assumed to the equivalent to the ones in the current
    # action, so:
    # - if the current action is act(a=3, b=4), then it is the same_as(name='act', args={'a': 3})
    # - if the current action is act(a=3, b=4), then it is the same_as(name='act', args={'a': 3, 'b': 4, 'c': 5})
    args_to_exclude = Custom.RESERVED_IN_ACTION_KWARGS
    return (name == self.name and
            self.check_provided_args(args) and
            all(k in args_to_exclude or k not in self.args or self.args[k] == v for k, v in args.items()))

check_provided_args

check_provided_args(args: dict, exception: bool = False, remove_special_arguments: bool = False) -> bool

Validate that every key in args is an accepted parameter for this action.

Each key is checked against the following categories in order:

• HSM transit meta names (Custom.HSM_TRANSIT_META_NAMES): silently removed when remove_special_arguments is True; otherwise rejected.
• Wire sentinels / injection names (Custom.WIRE_SENTINEL_NAMES, Custom.INTERACTION_INJECT_NAMES): always rejected.
• Interaction field names (Custom.INTERACTION_FIELD_NAMES): always silently accepted (skipped).
• Unknown names (not in self.param_list): rejected.

Parameters:

Name	Type	Description	Default
args	dict	The keyword-argument dictionary to validate. The dictionary may be mutated in-place when remove_special_arguments is True.	required
exception	bool	When True, a ValueError is raised on the first invalid key instead of returning False. Defaults to False.	False
remove_special_arguments	bool	When True, HSM-transit meta keys are deleted from args in-place rather than triggering a rejection. Defaults to False.	False

Returns:

Type	Description
bool	True if every key is valid (or has been silently removed), False on the
bool	first rejected key when exception is False.

Raises:

Type	Description
ValueError	If an invalid key is found and exception is True.

Source code in unaiverse/hsm/action.py

def check_provided_args(self, args: dict, exception: bool = False,
                        remove_special_arguments: bool = False) -> bool:
    """Validate that every key in ``args`` is an accepted parameter for this action.

    Each key is checked against the following categories in order:

    - **HSM transit meta names** (``Custom.HSM_TRANSIT_META_NAMES``): silently
      removed when ``remove_special_arguments`` is ``True``; otherwise rejected.
    - **Wire sentinels / injection names** (``Custom.WIRE_SENTINEL_NAMES``,
      ``Custom.INTERACTION_INJECT_NAMES``): always rejected.
    - **Interaction field names** (``Custom.INTERACTION_FIELD_NAMES``): always
      silently accepted (skipped).
    - **Unknown names** (not in ``self.param_list``): rejected.

    Args:
        args: The keyword-argument dictionary to validate. The dictionary may be
            mutated in-place when ``remove_special_arguments`` is ``True``.
        exception: When ``True``, a ``ValueError`` is raised on the first invalid
            key instead of returning ``False``. Defaults to False.
        remove_special_arguments: When ``True``, HSM-transit meta keys are deleted
            from ``args`` in-place rather than triggering a rejection. Defaults to
            False.

    Returns:
        True if every key is valid (or has been silently removed), False on the
        first rejected key when ``exception`` is ``False``.

    Raises:
        ValueError: If an invalid key is found and ``exception`` is ``True``.
    """
    if args is not None:
        for k in list(args.keys()):
            if k in Custom.HSM_TRANSIT_META_NAMES:
                if remove_special_arguments:
                    del args[k]
                elif exception:
                    raise ValueError(f"Parameter {k} is not a valid param for action"
                                     f" {self.name} (it is a private HSM transit meta name)")
                else:
                    return False
            elif k in Custom.WIRE_SENTINEL_NAMES or k in Custom.INTERACTION_INJECT_NAMES:
                if exception:
                    raise ValueError(f"Parameter {k} is not a valid param for action"
                                     f" {self.name} (it is reserved)")
                else:
                    return False
            elif k in Custom.INTERACTION_FIELD_NAMES:
                continue  # Skipping
            elif k not in self.param_list:
                if exception:
                    raise ValueError(f"Parameter {k} is not a valid param for action"
                                     f" {self.name} (it is not part of the action signature)")
                else:
                    return False
    return True

set_wildcards

set_wildcards(wildcards: dict[str, str | float | int] | None) -> None

Store a new wildcard table for later expansion.

The wildcard table maps placeholder strings (typically in the <name> format) to their concrete replacement values. Passing None clears the current table (equivalent to an empty dict). The replacements are not applied immediately; call apply_wildcards afterward to expand all placeholders in args, msg, and the time parameters. set_state_machine calls this method automatically when the action is registered with a parent HybridStateMachine.

Parameters:

Name	Type	Description	Default
wildcards	dict[str, str | float | int] | None	A mapping from placeholder strings to concrete replacement values (str, float, or int). None is treated as an empty mapping.	required

Source code in unaiverse/hsm/action.py

def set_wildcards(self, wildcards: dict[str, str | float | int] | None) -> None:
    """Store a new wildcard table for later expansion.

    The wildcard table maps placeholder strings (typically in the ``<name>`` format)
    to their concrete replacement values. Passing ``None`` clears the current table
    (equivalent to an empty dict). The replacements are not applied immediately;
    call ``apply_wildcards`` afterward to expand all placeholders in ``args``,
    ``msg``, and the time parameters. ``set_state_machine`` calls this method
    automatically when the action is registered with a parent
    ``HybridStateMachine``.

    Args:
        wildcards: A mapping from placeholder strings to concrete replacement
            values (``str``, ``float``, or ``int``). ``None`` is treated as an
            empty mapping.
    """
    self.wildcards = wildcards if wildcards is not None else {}

add_interaction

add_interaction(interaction: Interaction) -> bool

Add a pending external interaction request to this action.

Registers interaction as a pending request by attaching a back-reference to this action (via interaction.set_action_ref) and appending it to the internal ActionInteractionList. If the action does not accept outer interactions (outer=False), the call is a no-op and False is returned. Once at least one doable interaction is present, the action becomes ready for execution (see is_ready). See also clear_interactions and clear_interaction.

Parameters:

Name	Type	Description	Default
interaction	Interaction	The Interaction object representing the external request.	required

Returns:

Type	Description
bool	True if the interaction was added, False if the action does not accept outer
bool	interactions.

Source code in unaiverse/hsm/action.py

def add_interaction(self, interaction: Interaction) -> bool:
    """Add a pending external interaction request to this action.

    Registers ``interaction`` as a pending request by attaching a back-reference to
    this action (via ``interaction.set_action_ref``) and appending it to the internal
    ``ActionInteractionList``. If the action does not accept outer interactions
    (``outer=False``), the call is a no-op and ``False`` is returned. Once at least
    one doable interaction is present, the action becomes ready for execution (see
    ``is_ready``). See also ``clear_interactions`` and ``clear_interaction``.

    Args:
        interaction: The ``Interaction`` object representing the external request.

    Returns:
        True if the interaction was added, False if the action does not accept outer
        interactions.
    """

    # Let's augment interaction object
    if not self.allows_outer_interactions():
        return False
    interaction.set_action_ref(self)
    self.interactions.add(interaction)
    return True

clear_interactions

clear_interactions(requester: str | None = None) -> None

Remove pending interaction requests from this action.

When requester is None, all pending interactions are discarded and the internal ActionInteractionList is replaced with a fresh empty one. When requester is provided, only the interactions belonging to that requester are removed, leaving interactions from other requesters untouched. See also clear_interaction for removing a single specific interaction.

Parameters:

Name	Type	Description	Default
requester	str | None	The peer ID whose interactions should be removed, or None to remove all interactions regardless of requester. Defaults to None.	None

Source code in unaiverse/hsm/action.py

def clear_interactions(self, requester: str | None = None) -> None:
    """Remove pending interaction requests from this action.

    When ``requester`` is ``None``, all pending interactions are discarded and the
    internal ``ActionInteractionList`` is replaced with a fresh empty one. When
    ``requester`` is provided, only the interactions belonging to that requester are
    removed, leaving interactions from other requesters untouched. See also
    ``clear_interaction`` for removing a single specific interaction.

    Args:
        requester: The peer ID whose interactions should be removed, or ``None``
            to remove all interactions regardless of requester. Defaults to None.
    """
    if requester is None:
        self.interactions = ActionInteractionList()
    else:
        if self.interactions.is_requester_known(requester):
            requests = self.interactions.get_interactions(requester)
            for req in requests:
                self.interactions.remove(req)

clear_interaction

clear_interaction(requester: str, req_id: int) -> None

Remove a single specific interaction identified by its requester and insertion-order ID.

Looks up the interaction at position req_id in the per-requester sub-list and removes it from the internal ActionInteractionList. If no such interaction exists the call is a no-op. See also clear_interactions for bulk removal.

Parameters:

Name	Type	Description	Default
requester	str	The peer ID of the requester who owns the interaction.	required
req_id	int	The per-requester insertion-order index of the interaction to remove.	required

Source code in unaiverse/hsm/action.py

def clear_interaction(self, requester: str, req_id: int) -> None:
    """Remove a single specific interaction identified by its requester and insertion-order ID.

    Looks up the interaction at position ``req_id`` in the per-requester sub-list and
    removes it from the internal ``ActionInteractionList``. If no such interaction
    exists the call is a no-op. See also ``clear_interactions`` for bulk removal.

    Args:
        requester: The peer ID of the requester who owns the interaction.
        req_id: The per-requester insertion-order index of the interaction to remove.
    """
    req = self.interactions.get_interaction(req_id, requester)
    if req is not None:
        self.interactions.remove(req)

get_list_of_interactions

get_list_of_interactions() -> ActionInteractionList

Return the ActionInteractionList holding all pending interaction requests.

The returned object provides fine-grained access to pending requests: filter by requester, retrieve by insertion order, check doability, or iterate over all stored Interaction objects. The list is live - mutations affect the action's internal state directly. See add_interaction and clear_interactions for the standard way to add or remove entries.

Returns:

Type	Description
ActionInteractionList	The ActionInteractionList holding all currently pending interactions.

Source code in unaiverse/hsm/action.py

def get_list_of_interactions(self) -> 'ActionInteractionList':
    """Return the ``ActionInteractionList`` holding all pending interaction requests.

    The returned object provides fine-grained access to pending requests: filter by
    requester, retrieve by insertion order, check doability, or iterate over all
    stored ``Interaction`` objects. The list is live - mutations affect the action's
    internal state directly. See ``add_interaction`` and ``clear_interactions`` for
    the standard way to add or remove entries.

    Returns:
        The ``ActionInteractionList`` holding all currently pending interactions.
    """
    return self.interactions

get_actual_params

get_actual_params(additional_args: dict | None) -> dict | None

Resolve the full set of keyword arguments to pass to the underlying callable.

Builds the final argument dictionary by consulting, in priority order:

1. self.args - the action's own stored arguments (highest priority).
2. additional_args - extra arguments supplied at call time (e.g. from an incoming Interaction's action_kwargs).
3. self.param_to_default_value - default values from the method signature.

Only parameters listed in self.param_list are included. If a required parameter has no value in any of the three sources, an error is logged and None is returned.

Parameters:

Name	Type	Description	Default
additional_args	dict | None	A supplementary keyword-argument dictionary merged after self.args but before signature defaults. May be None.	required

Returns:

Type	Description
dict | None	A dictionary mapping every parameter name to its resolved value, or
dict | None	None if a required parameter could not be resolved.

Source code in unaiverse/hsm/action.py

def get_actual_params(self, additional_args: dict | None) -> dict | None:
    """Resolve the full set of keyword arguments to pass to the underlying callable.

    Builds the final argument dictionary by consulting, in priority order:

    1. ``self.args`` - the action's own stored arguments (highest priority).
    2. ``additional_args`` - extra arguments supplied at call time (e.g. from an
       incoming ``Interaction``'s ``action_kwargs``).
    3. ``self.param_to_default_value`` - default values from the method signature.

    Only parameters listed in ``self.param_list`` are included. If a required
    parameter has no value in any of the three sources, an error is logged and
    ``None`` is returned.

    Args:
        additional_args: A supplementary keyword-argument dictionary merged after
            ``self.args`` but before signature defaults. May be ``None``.

    Returns:
        A dictionary mapping every parameter name to its resolved value, or
        ``None`` if a required parameter could not be resolved.
    """
    actual_params = {}
    params = self.param_list
    defaults = self.param_to_default_value
    for param_name in params:
        if param_name in self.args:
            actual_params[param_name] = self.args[param_name]
        elif additional_args is not None and param_name in additional_args:
            actual_params[param_name] = additional_args[param_name]
        elif param_name in defaults:
            actual_params[param_name] = defaults[param_name]
        else:
            log.error(f"Getting actual params for {self.name}; missing param: {param_name}")
            return None
    return actual_params

apply_wildcards

apply_wildcards() -> None

Expand all wildcard placeholders using the current wildcards table.

Re-reads the pre-expansion backups (args_with_wildcards and msg_with_wildcards) and rebuilds args and msg with every placeholder string replaced by its concrete value from wildcards. Wildcard substitution also covers max_duration, retry_timeout, and delay when those were originally provided as placeholder strings (e.g. "<max_time>"). String wildcards support partial replacement (substring match); non-string wildcards are replaced only when the entire value equals the placeholder.

After expansion, __build_system_interaction is called so the system_interaction reflects the post-expansion args (important when args contains interaction-shaping keys such as streams or num_steps). This method is called automatically by set_state_machine and should be called again whenever the wildcard table is updated.

Note

The backup fields (args_with_wildcards, msg_with_wildcards) are never mutated by this method, so apply_wildcards can be called multiple times safely to re-apply a changed wildcard table.

Source code in unaiverse/hsm/action.py

def apply_wildcards(self) -> None:
    """Expand all wildcard placeholders using the current ``wildcards`` table.

    Re-reads the pre-expansion backups (``args_with_wildcards`` and
    ``msg_with_wildcards``) and rebuilds ``args`` and ``msg`` with every placeholder
    string replaced by its concrete value from ``wildcards``. Wildcard substitution
    also covers ``max_duration``, ``retry_timeout``, and ``delay`` when those were
    originally provided as placeholder strings (e.g. ``"<max_time>"``). String
    wildcards support partial replacement (substring match); non-string wildcards are
    replaced only when the entire value equals the placeholder.

    After expansion, ``__build_system_interaction`` is called so the
    ``system_interaction`` reflects the post-expansion ``args`` (important when
    ``args`` contains interaction-shaping keys such as ``streams`` or ``num_steps``).
    This method is called automatically by ``set_state_machine`` and should be called
    again whenever the wildcard table is updated.

    Note:
        The backup fields (``args_with_wildcards``, ``msg_with_wildcards``) are
        never mutated by this method, so ``apply_wildcards`` can be called multiple
        times safely to re-apply a changed wildcard table.
    """

    # Setting up the original wildcard-based arguments and messages
    if self.args_with_wildcards is None:
        self.args_with_wildcards = copy.deepcopy(self.args)  # Backup before applying wildcards (1st time only)
    else:
        self.args = copy.deepcopy(self.args_with_wildcards)  # Restore a backup before applying wildcards
    if self.msg_with_wildcards is not None:
        self.msg = self.msg_with_wildcards

    # Applying wildcard-suggested replacements to arguments
    self.__apply_wildcards_to_args(self.args)

    # Applying wildcard-suggested replacements to message and other stuff
    for wildcard_from, wildcard_to in self.wildcards.items():

        # ... to message
        if self.msg is not None:
            self.msg = self.msg.replace(wildcard_from, str(wildcard_to))

        # ... to the rest
        if (self.__action_max_duration_with_wildcard is not None and
                wildcard_from in self.__action_max_duration_with_wildcard):
            self.__action_max_duration = (
                float(self.__action_max_duration_with_wildcard.replace(wildcard_from, str(wildcard_to))))
        if (self.__action_retry_timeout_with_wildcard is not None and
                wildcard_from in self.__action_retry_timeout_with_wildcard):
            self.__action_retry_timeout = (
                float(self.__action_retry_timeout_with_wildcard.replace(wildcard_from, str(wildcard_to))))
        if self.__delay_with_wildcard is not None and wildcard_from in self.__delay_with_wildcard:
            self.__delay = float(self.__delay_with_wildcard.replace(wildcard_from, str(wildcard_to)))

    # Now that wildcards have been resolved in self.args, rebuild the system_interaction so any
    # streams / num_steps / etc. reflect the concrete (post-wildcard) values.
    self.__build_system_interaction()

ActionInteractionList

ActionInteractionList(max_per_requester: int = -1)

An ordered, dual-indexed container for the Interaction objects pending on an Action.

ActionInteractionList maintains two parallel indices over the same set of Interaction objects:

• Global insertion order (by_insertion_order): a flat list that records every interaction in the order it was added. Used to iterate over all pending requests, retrieve the oldest or most-recent entry, and enforce the wall-clock timeout via remove_due_to_timeout.
• Per-requester insertion order (by_requester_and_by_insertion_order): a dictionary keyed by requester peer ID, each value being the sub-list of that requester's interactions in insertion order. Used to retrieve, move, or remove a specific requester's requests efficiently.

Both indices are kept in sync on every add and remove operation by adjusting the by_insertion_order_id and by_requester_insertion_order_id attributes stored directly on each Interaction object.

An optional per-requester cap (max_per_requester) evicts the oldest entry for a requester whenever the cap would be exceeded on add.

Attributes:

Name	Type	Description
by_insertion_order		Flat list of all interactions in global insertion order.
by_requester_and_by_insertion_order		Mapping from requester peer ID to the sub-list of that requester's interactions.
max_per_requester		Maximum interactions stored per requester, or -1 for no limit.
by_insertion_order_entering_time		Parallel list of time.perf_counter() timestamps recording when each entry in by_insertion_order was added.

Examples:

>>> from unittest.mock import MagicMock
>>> il = ActionInteractionList(max_per_requester=2)
>>> inter = MagicMock()
>>> inter.requester = "agent_1"
>>> inter.uuid = "abc"
>>> inter.running = False
>>> inter.is_valid.return_value = True
>>> il.add(inter)
>>> len(il)
1
>>> il.is_requester_known("agent_1")
True

Initialize an empty interaction list with an optional per-requester cap.

Both the global insertion-order list and the per-requester index start empty. The entering-time list is also initialized empty and grows in lockstep with the global insertion-order list.

Parameters:

Name	Type	Description	Default
max_per_requester	int	Maximum number of interactions to retain per requester. When a new interaction would exceed this limit, the oldest interaction from that requester is evicted before the new one is added. -1 disables the cap. Defaults to -1.	-1

Source code in unaiverse/hsm/action.py

def __init__(self, max_per_requester: int = -1):
    """Initialize an empty interaction list with an optional per-requester cap.

    Both the global insertion-order list and the per-requester index start empty.
    The entering-time list is also initialized empty and grows in lockstep with the
    global insertion-order list.

    Args:
        max_per_requester: Maximum number of interactions to retain per requester.
            When a new interaction would exceed this limit, the oldest interaction
            from that requester is evicted before the new one is added. ``-1``
            disables the cap. Defaults to -1.
    """
    self.by_insertion_order = []
    self.by_requester_and_by_insertion_order = {}
    self.max_per_requester = max_per_requester
    self.by_insertion_order_entering_time = []

by_insertion_order instance-attribute

by_insertion_order = []

by_requester_and_by_insertion_order instance-attribute

by_requester_and_by_insertion_order = {}

max_per_requester instance-attribute

max_per_requester = max_per_requester

by_insertion_order_entering_time instance-attribute

by_insertion_order_entering_time = []

add

add(interaction: Interaction) -> None

Append an interaction to the list, updating both internal indices.

Before appending, the list checks whether an interaction with the same requester and UUID already exists:

• If a duplicate is found and its running flag is set, the new interaction is silently dropped (the HSM is already handling that UUID).
• If a duplicate is found but is not running, the existing slot is refreshed in-place with the new Interaction object, preserving its position in both indices.

When no duplicate exists, the interaction is appended to both the global list and the per-requester sub-list. If max_per_requester is positive and adding the new entry would exceed the cap, the oldest entry for that requester is evicted first via remove. The current time.perf_counter() timestamp is recorded in by_insertion_order_entering_time for use by remove_due_to_timeout.

Parameters:

Name	Type	Description	Default
interaction	Interaction	The Interaction object to add.	required

Source code in unaiverse/hsm/action.py

def add(self, interaction: Interaction) -> None:
    """Append an interaction to the list, updating both internal indices.

    Before appending, the list checks whether an interaction with the same requester
    and UUID already exists:

    - If a duplicate is found and its ``running`` flag is set, the new interaction
      is silently dropped (the HSM is already handling that UUID).
    - If a duplicate is found but is not running, the existing slot is refreshed
      in-place with the new ``Interaction`` object, preserving its position in both
      indices.

    When no duplicate exists, the interaction is appended to both the global list and
    the per-requester sub-list. If ``max_per_requester`` is positive and adding the
    new entry would exceed the cap, the oldest entry for that requester is evicted
    first via ``remove``. The current ``time.perf_counter()`` timestamp is recorded
    in ``by_insertion_order_entering_time`` for use by ``remove_due_to_timeout``.

    Args:
        interaction: The ``Interaction`` object to add.
    """

    # Searching for already existing interactions with this UUID, FROM THE SAME REQUESTER
    # If already there - do not accumulate multiple requests with same UUID (useful also for system interactions)
    existing_request_same_uuid = self.get_interaction_by_uuid(interaction.requester, interaction.uuid)
    if existing_request_same_uuid:

        # We skip this request if we are already taking care (running) of another one with the same UUID
        if existing_request_same_uuid.running:  # This means the HSM is taking care of it, don't mess up things!
            log.error(f"Tried to add an interaction with the UUID of an already existing one that was in "
                      f"'running' state, skipping (requester: {interaction.requester}")
            return

        # Since we know the requester is the same, we just "refresh" the existing interaction object,
        # without altering its position in the list
        self.by_insertion_order[existing_request_same_uuid.by_insertion_order_id] = interaction
        interaction.by_insertion_order_id = existing_request_same_uuid.by_insertion_order_id

        self.by_requester_and_by_insertion_order[
            interaction.requester][existing_request_same_uuid.by_requester_insertion_order_id] = (
            interaction)
        interaction.by_requester_insertion_order_id = existing_request_same_uuid.by_requester_insertion_order_id
        return  # We stop here in this case, no need to do any other things

    # Updating by-requester index
    if interaction.requester not in self.by_requester_and_by_insertion_order:
        self.by_requester_and_by_insertion_order[interaction.requester] = []

    if 0 < self.max_per_requester <= len(self.by_requester_and_by_insertion_order[interaction.requester]):
        self.remove(self.get_oldest_interaction(interaction.requester))
    by_requester_insertion_order_id = len(self.by_requester_and_by_insertion_order[interaction.requester])
    self.by_requester_and_by_insertion_order[interaction.requester].append(interaction)

    # Updating direct global index
    insertion_order_id = len(self.by_insertion_order)
    self.by_insertion_order.append(interaction)

    # Updating reverse indices
    interaction.by_insertion_order_id = insertion_order_id
    interaction.by_requester_insertion_order_id = by_requester_insertion_order_id

    # Saving joining time
    self.by_insertion_order_entering_time.append(time.perf_counter())

remove

remove(interaction: Interaction) -> None

Remove an interaction from the list and reindex all subsequent entries.

After deletion, every interaction that followed the removed entry in the global list has its by_insertion_order_id decremented by one. The same reindexing is applied to the per-requester sub-list. If the requester has no remaining interactions, its key is deleted from by_requester_and_by_insertion_order. The removed Interaction's index attributes are set to -1 to mark it as invalidated. The call is a no-op when interaction.is_valid() returns False.

Parameters:

Name	Type	Description	Default
interaction	Interaction	The Interaction object to remove. Must currently be stored in this list; invalid interactions are silently ignored.	required

Source code in unaiverse/hsm/action.py

def remove(self, interaction: Interaction) -> None:
    """Remove an interaction from the list and reindex all subsequent entries.

    After deletion, every interaction that followed the removed entry in the global
    list has its ``by_insertion_order_id`` decremented by one. The same reindexing
    is applied to the per-requester sub-list. If the requester has no remaining
    interactions, its key is deleted from ``by_requester_and_by_insertion_order``.
    The removed ``Interaction``'s index attributes are set to ``-1`` to mark it as
    invalidated. The call is a no-op when ``interaction.is_valid()`` returns
    ``False``.

    Args:
        interaction: The ``Interaction`` object to remove. Must currently be stored
            in this list; invalid interactions are silently ignored.
    """
    if interaction.is_valid():
        if (interaction.by_insertion_order_id < len(self.by_insertion_order) and
                self.by_insertion_order[interaction.by_insertion_order_id] == interaction):

            for i in range(interaction.by_insertion_order_id + 1, len(self.by_insertion_order)):
                self.by_insertion_order[i].by_insertion_order_id -= 1
            del self.by_insertion_order[interaction.by_insertion_order_id]
            del self.by_insertion_order_entering_time[interaction.by_insertion_order_id]

            d = self.by_requester_and_by_insertion_order[interaction.requester]
            for i in range(interaction.by_requester_insertion_order_id + 1, len(d)):
                d[i].by_requester_insertion_order_id -= 1
            del d[interaction.by_requester_insertion_order_id]
            if len(d) == 0:
                del self.by_requester_and_by_insertion_order[interaction.requester]
            interaction.by_insertion_order_id = -1
            interaction.by_requester_insertion_order_id = -1

remove_due_to_timeout

remove_due_to_timeout(timeout_secs: float) -> None

Remove all interactions that have been waiting longer than timeout_secs.

Compares each interaction's entering time (recorded in by_insertion_order_entering_time at add time) against the current time.perf_counter() value. Any interaction whose age (elapsed seconds since entering) is greater than or equal to timeout_secs is collected and then removed via remove. The two-phase collect-then-remove approach avoids index corruption during iteration.

Parameters:

Name	Type	Description	Default
timeout_secs	float	Maximum age in seconds. Interactions that have waited at least this long are evicted.	required

Source code in unaiverse/hsm/action.py

def remove_due_to_timeout(self, timeout_secs: float) -> None:
    """Remove all interactions that have been waiting longer than ``timeout_secs``.

    Compares each interaction's entering time (recorded in
    ``by_insertion_order_entering_time`` at ``add`` time) against the current
    ``time.perf_counter()`` value. Any interaction whose age (elapsed seconds since
    entering) is greater than or equal to ``timeout_secs`` is collected and then
    removed via ``remove``. The two-phase collect-then-remove approach avoids
    index corruption during iteration.

    Args:
        timeout_secs: Maximum age in seconds. Interactions that have waited at
            least this long are evicted.
    """
    to_remove = []
    for i, req in enumerate(self.by_insertion_order):
        if (time.perf_counter() - self.by_insertion_order_entering_time[i]) >= timeout_secs:
            to_remove.append(req)
    for req in to_remove:
        self.remove(req)

remove_completed

remove_completed() -> None

Removes all interactions that have been marked as completed.

Source code in unaiverse/hsm/action.py

def remove_completed(self) -> None:
    """Removes all interactions that have been marked as completed."""
    to_remove = []
    for i, req in enumerate(self.by_insertion_order):
        if req.completed:
            to_remove.append(req)
    for req in to_remove:
        self.remove(req)

move_interaction_to_back

move_interaction_to_back(interaction: Interaction) -> None

Moves an interaction to the back of the insertion-order list, preserving its original entering time.

Parameters:

Name	Type	Description	Default
interaction	Interaction	The Interaction object to reposition.	required

Source code in unaiverse/hsm/action.py

def move_interaction_to_back(self, interaction: Interaction) -> None:
    """Moves an interaction to the back of the insertion-order list, preserving its original entering time.

    Args:
        interaction: The Interaction object to reposition.
    """
    if len(self.by_insertion_order) > 1 and interaction.is_valid():
        try:
            entering_time = self.by_insertion_order_entering_time[interaction.by_insertion_order_id]
            self.remove(interaction)
            self.add(interaction)
            self.by_insertion_order_entering_time[interaction.by_insertion_order_id] = entering_time
        except Exception as e:
            raise e

move_requester_to_back

move_requester_to_back(requester: str) -> None

Moves all interactions belonging to a requester to the back of the list, preserving their entering times.

Parameters:

Name	Type	Description	Default
requester	str	The peer ID whose interactions should be moved to the back.	required

Source code in unaiverse/hsm/action.py

def move_requester_to_back(self, requester: str) -> None:
    """Moves all interactions belonging to a requester to the back of the list, preserving their entering times.

    Args:
        requester: The peer ID whose interactions should be moved to the back.
    """
    requests = self.get_interactions(requester)
    if requests is not None and len(requests) > 0:
        requests_copy = []
        entering_times = []
        for req in requests:
            if req.is_valid():
                requests_copy.append(req)
                entering_times.append(self.by_insertion_order_entering_time[req.by_insertion_order_id])
                self.remove(req)
        for i, req in enumerate(requests_copy):
            self.add(req)
            self.by_insertion_order_entering_time[req.by_insertion_order_id] = entering_times[i]

get_interaction

get_interaction(req_order_id: int, requester: str | None = None) -> Interaction | None

Retrieves an interaction by its insertion-order index, optionally scoped to a specific requester.

Parameters:

Name	Type	Description	Default
req_order_id	int	The insertion-order index of the interaction to retrieve.	required
requester	str | None	If provided, scopes the lookup to the sub-list of this requester.	None

Returns:

Type	Description
Interaction | None	The matching Interaction, or None if not found.

Source code in unaiverse/hsm/action.py

def get_interaction(self, req_order_id: int, requester: str | None = None) -> Interaction | None:
    """Retrieves an interaction by its insertion-order index, optionally scoped to a specific requester.

    Args:
        req_order_id: The insertion-order index of the interaction to retrieve.
        requester: If provided, scopes the lookup to the sub-list of this requester.

    Returns:
        The matching Interaction, or None if not found.
    """
    if req_order_id < 0 and req_order_id != -1:
        return None
    if requester is None:
        return self.by_insertion_order[req_order_id] if req_order_id < len(self.by_insertion_order) else None
    else:
        if requester not in self.by_requester_and_by_insertion_order:
            return None
        return self.by_requester_and_by_insertion_order[requester][req_order_id] \
            if req_order_id < len(self.by_requester_and_by_insertion_order[requester]) else None

get_oldest_interaction

get_oldest_interaction(requester: str | None = None, ignore_completed: bool = False) -> Interaction | None

Returns the oldest (first-added) interaction, optionally scoped to a specific requester.

Parameters:

Name	Type	Description	Default
requester	str | None	If provided, scopes the lookup to the sub-list of this requester.	None
ignore_completed	bool	If True, completed interactions will be ignored/skipped.	False

Returns:

Type	Description
Interaction | None	The oldest Interaction, or None if the list is empty.

Source code in unaiverse/hsm/action.py

def get_oldest_interaction(self, requester: str | None = None, ignore_completed: bool = False) \
        -> Interaction | None:
    """Returns the oldest (first-added) interaction, optionally scoped to a specific requester.

    Args:
        requester: If provided, scopes the lookup to the sub-list of this requester.
        ignore_completed: If True, completed interactions will be ignored/skipped.

    Returns:
        The oldest Interaction, or None if the list is empty.
    """
    if not ignore_completed:
        return self.get_interaction(0, requester)
    else:
        req = self.get_interaction(0, requester)
        i = 1
        while req.completed:
            if i < len(self):
                req = self.get_interaction(i, requester)
                i += 1
            else:
                return None
        return req

get_most_recent_interaction

get_most_recent_interaction(requester: str | None = None, ignore_completed: bool = False) -> Interaction | None

Returns the most recently added interaction, optionally scoped to a specific requester.

Parameters:

Name	Type	Description	Default
requester	str | None	If provided, scopes the lookup to the sub-list of this requester.	None
ignore_completed	bool	If True, completed interactions will be ignored/skipped.	False

Returns:

Type	Description
Interaction | None	The most recent Interaction, or None if the list is empty.

Source code in unaiverse/hsm/action.py

def get_most_recent_interaction(self, requester: str | None = None, ignore_completed: bool = False) \
        -> Interaction | None:
    """Returns the most recently added interaction, optionally scoped to a specific requester.

    Args:
        requester: If provided, scopes the lookup to the sub-list of this requester.
        ignore_completed: If True, completed interactions will be ignored/skipped.

    Returns:
        The most recent Interaction, or None if the list is empty.
    """
    if not ignore_completed:
        return self.get_interaction(-1, requester)
    else:
        req = self.get_interaction(-1, requester)
        i = len(self) - 2
        while req.completed:
            if i >= 0:
                req = self.get_interaction(i, requester)
                i -= 1
            else:
                return None
        return req

get_interaction_by_uuid

get_interaction_by_uuid(requester: str, uuid: str | None) -> Interaction | None

Finds the first interaction for a given requester that matches the specified UUID.

Parameters:

Name	Type	Description	Default
requester	str	The peer ID of the requester.	required
uuid	str | None	The UUID to search for (None is a valid UUID).	required

Returns:

Type	Description
Interaction | None	The matching Interaction, or None if not found.

Source code in unaiverse/hsm/action.py

def get_interaction_by_uuid(self, requester: str, uuid: str | None) -> Interaction | None:
    """Finds the first interaction for a given requester that matches the specified UUID.

    Args:
        requester: The peer ID of the requester.
        uuid: The UUID to search for (None is a valid UUID).

    Returns:
        The matching Interaction, or None if not found.
    """
    requests = self.get_interactions(requester)
    if requests is None or len(requests) == 0:
        return None

    for req in requests:
        if req.uuid == uuid:
            return req

keep_only_the_most_recent_interaction

keep_only_the_most_recent_interaction(ignore_completed: bool = False) -> None

Discards all interactions except the most recently added one, preserving its entering time.

Parameters:

Name	Type	Description	Default
ignore_completed	bool	If True, completed interactions will be ignored/skipped.	False

Returns:

Type	Description
None	None

Source code in unaiverse/hsm/action.py

def keep_only_the_most_recent_interaction(self, ignore_completed: bool = False) -> None:
    """Discards all interactions except the most recently added one, preserving its entering time.

    Args:
        ignore_completed: If True, completed interactions will be ignored/skipped.

    Returns:
        None
    """
    req = self.get_most_recent_interaction(ignore_completed=ignore_completed)
    if req is not None:
        entering_time = self.by_insertion_order_entering_time[req.by_insertion_order_id]
        self.clear()
        self.add(req)
        self.by_insertion_order_entering_time[req.by_insertion_order_id] = entering_time
    else:
        self.clear()

get_interactions

get_interactions(requester: str | None = None, to_str: bool = False, doable_only: bool = False) -> list[Interaction] | str

Returns interactions, optionally filtered by requester or "do-ability", or as a JSON string.

Parameters:

Name	Type	Description	Default
requester	str | None	If provided, returns only interactions from this requester.	None
to_str	bool	If True, returns a JSON-encoded string instead of a list.	False
doable_only	bool	If True, filters to only interactions that pass check_if_doable().	False

Returns:

Type	Description
list[Interaction] | str	A list of Interaction objects, or a JSON-encoded string if to_str is True.

Source code in unaiverse/hsm/action.py

def get_interactions(self, requester: str | None = None, to_str: bool = False,
                     doable_only: bool = False) -> list[Interaction] | str:
    """Returns interactions, optionally filtered by requester or "do-ability", or as a JSON string.

    Args:
        requester: If provided, returns only interactions from this requester.
        to_str: If True, returns a JSON-encoded string instead of a list.
        doable_only: If True, filters to only interactions that pass ``check_if_doable()``.

    Returns:
        A list of Interaction objects, or a JSON-encoded string if ``to_str`` is True.
    """
    if requester is None:
        reqs = self.by_insertion_order
        if doable_only:
            reqs = [req for req in reqs if req.check_if_doable()]  # This excludes completed too
        if not to_str:
            return reqs
        else:
            return json.dumps([req.to_str() for req in reqs])
    else:
        if requester in self.by_requester_and_by_insertion_order:
            reqs = self.by_requester_and_by_insertion_order[requester]
            if doable_only:
                reqs = [req for req in reqs if req.check_if_doable()]
            if not to_str:
                return reqs
            else:
                return json.dumps([req.to_str() for req in reqs])
        else:
            if not to_str:
                return []
            else:
                return json.dumps([])

clear

clear() -> None

Removes all interactions and resets all internal indices.

Source code in unaiverse/hsm/action.py

def clear(self) -> None:
    """Removes all interactions and resets all internal indices."""
    self.by_insertion_order.clear()
    self.by_requester_and_by_insertion_order.clear()
    self.by_insertion_order_entering_time.clear()

is_requester_known

is_requester_known(requester: str) -> bool

Checks whether any interactions from the given requester are currently stored.

Parameters:

Name	Type	Description	Default
requester	str	The peer ID to look up.	required

Returns:

Type	Description
bool	True if the requester has at least one interaction in the list, False otherwise.

Source code in unaiverse/hsm/action.py

def is_requester_known(self, requester: str) -> bool:
    """Checks whether any interactions from the given requester are currently stored.

    Args:
        requester: The peer ID to look up.

    Returns:
        True if the requester has at least one interaction in the list, False otherwise.
    """
    return requester in self.by_requester_and_by_insertion_order