Events API Reference¶

This page provides detailed technical information about the Events system in Meta Agents Research Environments. For a higher-level overview, see Foundations.

Event Structure ¶

All events inherit from the AbstractEvent class, which contains shared fields between completed and future events:

@dataclass(order=True)
class AbstractEvent(ABC):
    """
    Abstract event class, that contains shared field between completed and future events.
    """
    # The action that will be executed when the event happens, which is an App API call
    action: Action
    # The type of the event, either AGENT, ENV or USER.
    event_type: EventType = field(default=EventType.ENV)
    # The time at which the event will happen, this can get overridden in various places for e.g. in case of conditional triggers.
    event_time: float = field(default_factory=lambda: global_time.time())
    # The relative time wrt the simulation start time
    # WARNING when the event is going to be added to the queue, this information is going to be used to set event_time
    event_relative_time: float | None = field(default=None)
    # The unique id of the event, this is used to identify the event in the logs.
    # This is created automatically and does NOT need to be handled by the user
    event_id: str = field(default=None)

Note

Events have an event_type which describes the origin of the event. By default, each app_tool call made by the agent will be of type EventType.AGENT, each event definition in the scenario script will be of type EventType.ENV. You can also decide to manually add the type EventType.USER to events when defining a scenario to simulate that the user is the source of the event. This field is only used for logging purposes and does not have any impact on the execution of the scenario.

Event Types by Origin ¶

EventType.AGENT: Events initiated by agent tool calls
EventType.ENV: Events defined in scenario scripts
EventType.USER: Events simulating user interactions

Actions ¶

Events execute Actions, which define what actually happens:

@dataclass
class Action:
    function: Callable               # Function to call
    args: dict[str, Any]            # Function arguments
    app: App | None = None          # App instance
    action_id: str = None           # Unique identifier

Event Management Overview ¶

The environment manages events through two main data structures:

Event Queue: Stores future events waiting to be processed
Event Log: Contains the history of completed events

# Schedule an event
env.schedule(event)

# Access event history
completed_events = env.event_log.list_view()

Event Graph Nodes ¶

Event graphs consist of three types of nodes:

Event Nodes ¶

Standard events as described above

ConditionCheckEvent Nodes ¶

Check functions that return boolean values

ValidationEvent Nodes ¶

Validation functions with checklists of expected agent actions

Advanced Event Properties ¶

ConditionCheckEvent and ValidationEvent nodes support additional properties:

timeout_tick: Maximum time to wait for condition/validation
every_tick: Frequency of condition/validation checks

Example ValidationEvent ¶

def validation_function(env: AbstractEvent, event: AbstractEvent) -> bool:
    aui = event.get_app(AgentUserInterface.__name__)
    return all([
        event.action.app == aui,
        event.function_name() == "send_message_to_user",
    ])

validation = ValidationEvent(
    check_list=[validation_function],
    schedule_every_ticks=1,
    timeout=20,
)

Event Dependencies ¶

Events can be chained with dependencies and delays:

e1 = Event(...)
e2 = Event(...)

e1.depends_on(None, delay_seconds=30)  # Start after 30 seconds
e2.depends_on(e1, delay_seconds=5)     # Start 5 seconds after e1

Here e1 is delayed by 30 seconds after the start of the simulation and e2 will happen 5 seconds after e1 is executed.

API Methods ¶

Event Creation ¶

from are.simulation.core.events import Event
from are.simulation.core.action import Action

# Create a basic event
action = Action(function=my_function, args={"param": "value"})
event = Event(action=action, event_time=simulation_time + 30)

# Schedule the event
environment.schedule(event)

Event Scheduling ¶

# Schedule with relative time
event.event_relative_time = 60  # 60 seconds from simulation start
environment.schedule(event)

# Schedule with dependencies
event1 = Event(...)
event2 = Event(...)
event2.depends_on(event1, delay_seconds=10)

Event Querying ¶

# Get all completed events
completed = environment.event_log.list_view()

# Filter events by type
agent_events = [e for e in completed if e.event_type == EventType.AGENT]

Validation Events ¶

Validation events are special events that check whether certain conditions are met during scenario execution:

def both_responses_validator(env: AbstractEnvironment) -> bool:
    """Check if agent gave both responses."""
    aui_app = env.get_app("AgentUserInterface")
    agent_messages = aui_app.get_all_messages_from_agent()
    has_correct = any(msg.content == "I am a robot" for msg in agent_messages)
    has_incorrect = any(
        msg.content == "I am not a robot •`_´• " for msg in agent_messages
    )
    return has_correct and not has_incorrect

continuous_validation = ValidationEvent(
    milestones=[both_responses_validator]
).with_id("continuous_validation")
continuous_validation.depends_on(user_request, delay_seconds=1)
continuous_validation.schedule(every_ticks=1, timeout=15)
self.events.append(continuous_validation)

Condition Check Events ¶

Condition check events monitor the environment state and trigger other events when conditions are met:

from are.simulation.core.events import ConditionCheckEvent

def check_file_exists(env, event):
    """Check if a specific file exists"""
    fs_app = env.get_app("FileSystemApp")
    return fs_app.file_exists("/important/document.txt")

condition = ConditionCheckEvent(
    condition_function=check_file_exists,
    schedule_every_ticks=5,  # Check every 5 ticks
    timeout=600,             # Timeout after 10 minutes
)

# Trigger another event when condition is met
followup_event = Event(...)
followup_event.depends_on(condition)

For practical examples of event usage, see the tutorials and scenario development guides.

Oracle Events ¶

Oracle Events are specialized events that represent actions the agent should perform during scenario execution. They serve as templates or blueprints for generating actual executable events at runtime. Oracle Events are particularly useful for defining expected agent behaviors in scenarios and for creating validation trajectories.

Key Characteristics ¶

Agent-focused: Oracle Events default to EventType.AGENT, indicating they represent actions the agent should take
Dynamic creation: They use a make_event callable to generate the actual event when needed
Timing flexibility: Support event time comparators for flexible scheduling
Action descriptions: Store metadata about the action to be performed

Creating Oracle Events ¶

Oracle Events can be created in several ways:

From Existing Events

from are.simulation.core.events import Event, OracleEvent
from are.simulation.core.action import Action

# Create a regular event
action = Action(function=email_app.send_email, args={"to": "user@example.com"})
regular_event = Event(action=action)

# Convert to Oracle Event
oracle_event = OracleEvent.from_event(regular_event)

# Or use the convenience method
oracle_event = regular_event.oracle()

Direct Creation

def make_email_event(env):
    email_app = env.get_app("EmailClientApp")
    return Event.from_function(
        email_app.send_email,
        recipients=["client@company.com"],
        subject="Project Update"
    )

oracle_event = OracleEvent(
    make_event=make_email_event,
    event_type=EventType.AGENT,
    event_relative_time=30  # Execute 30 seconds after dependencies
)

Generating Executable Events ¶

Oracle Events generate actual executable events through the make() method:

# Generate the executable event
executable_event = oracle_event.make(environment)

# The generated event inherits timing and dependencies
assert executable_event.event_time == oracle_event.event_time
assert executable_event.dependencies == oracle_event.dependencies

Event Time Comparators ¶

Oracle Events support time-based filtering through event time comparators:

from are.simulation.types import EventTimeComparator

oracle_event = OracleEvent(
    make_event=make_email_event,
    event_time_comparator=EventTimeComparator.LESS_THAN
)

# This oracle event can be used to find events that occur before a certain time

Action Descriptions ¶

Oracle Events automatically capture action metadata when created from existing events:

# The action description includes app name, function name, and arguments
oracle_event = regular_event.oracle()

action_desc = oracle_event.action_desc
print(f"App: {action_desc.app}")
print(f"Function: {action_desc.function}")
print(f"Args: {action_desc.args}")

Completed Oracle Events

When Oracle Events are executed, they become CompletedOracleEvent instances that preserve timing information:

from are.simulation.types import CompletedOracleEvent

# After execution, create a completed oracle event
completed_oracle = CompletedOracleEvent.from_completed_event_and_oracle_event(
    completed_event, original_oracle_event
)

# Access timing metadata
print(f"Original scheduled time: {completed_oracle.absolute_event_time}")
print(f"Time comparator: {completed_oracle.event_time_comparator}")

Integration with Event System ¶

Oracle Events integrate seamlessly with the standard event system:

They inherit from AbstractEvent and support all standard event operations
They can have dependencies and successors like regular events
They participate in event scheduling and timing calculations
They can be serialized and deserialized for persistence

Event Registration ¶

The @event_registered decorator and EventRegisterer class are used for automatic event registration when app methods are called:

Event Registerer

class are.simulation.types.EventRegisterer[source]

Bases: object

Class that handles all the logic for registering events.

classmethod is_active()[source]: Checks whether we should fire events or not.

classmethod is_capture_mode()[source]: Capture mode makes sure a function call is not executed but only a “fictitious” CompletedEvent is returned. This is useful for debugging and testing, as well as easily creating CompletedEvent instances for validation

classmethod set_active(state)[source]: Sets whether we should fire events or not.

classmethod set_capture_mode(state)[source]: Sets whether capture mode is active or not.

classmethod event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT)[source]

This decorator is used to wrap API calls, so that we can register the event and add the appropriate CompletedEvent instance. The CompletedEvent instance is only added to the Event Log if the App is already registered in the environment.

This decorator is also used to capture fictitious CompletedEvent instances when capture mode is active. Capture mode allows to easily simulate and create CompletedEvent instances without actually executing the API call. This is useful for debugging and testing, as well as defining validation trajectories.

classmethod disable()[source]: Context manager to disable event firing, and just return the result of the function call as if decorator is not applied.

classmethod capture_mode()[source]: Context manager to replace a function call decorated with event_registered, by just a fictitious CompletedEvent instance.

Event Registration Decorator

The @event_registered decorator is used to automatically register events when app methods are called:

are.simulation.types.event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT)¶

This decorator wraps app methods to automatically create and log events when they are executed.

It supports different operation types and event types, and can be used in capture mode for testing and validation.

Example usage:

from are.simulation.types import event_registered, OperationType, EventType
from are.simulation.apps.app import App

class MyApp(App):
    @event_registered(operation_type=OperationType.WRITE, event_type=EventType.AGENT)
    def send_message(self, message: str):
        # This method call will automatically create and log an event
        return f"Sent: {message}"

Core Event Classes ¶

All event classes are defined in the are.simulation.types module:

class are.simulation.types.EnvironmentState(value)[source]

Bases: Enum

The state of the environment. - SETUP: the environment is being setup, this is the initial state of the environment - RUNNING: the environment event loop is running and events are being registered and logged - PAUSED: the environment is paused, and no events are being registered or logged, but can be restarted - STOPPED: the environment is completely stopped

SETUP = 'SETUP'

RUNNING = 'RUNNING'

STOPPED = 'STOPPED'

PAUSED = 'PAUSED'

FAILED = 'FAILED'

class are.simulation.types.EventTimeComparator(value)[source]

Bases: Enum

Comparator for event time filtering. - LESS_THAN: Less than comparison - GREATER_THAN: Greater than comparison - EQUAL: Equal comparison

LESS_THAN = 'LESS_THAN'

GREATER_THAN = 'GREATER_THAN'

EQUAL = 'EQUAL'

class are.simulation.types.AbstractEnvironment[source]

Bases: ABC

state: EnvironmentState | None

start()[source]

pause()[source]

resume()[source]

get_state()[source]

Return type:: dict[str, Any]

get_app(app_name)[source]

get_event_log_size()[source]

Return type:: int

get_event_queue_length()[source]

Return type:: int

final_validation_checks()[source]

Return type:: None

class are.simulation.types.HintType(value)[source]

Bases: Enum

Type of the hint, depends on the linked event - TASK_HINT: hints initiated by the send_message_to_agent

TASK_HINT = 'TASK_HINT'

ENVIRONMENT_HINT = 'ENVIRONMENT_HINT'

class are.simulation.types.Hint(*, hint_type, content, associated_event_id)[source]

Bases: object

Hint associated with an event - hint_type: Type of the hint, depends on the linked event - content: Content of the hint - associated_event_id: The id of the event that this hint is associated with

hint_type: HintType

content: str

associated_event_id: str

class are.simulation.types.EnvironmentType(value)[source]

Bases: Enum

The type of the environment.

UNKNOWN = 'UNKNOWN'

CLI = 'CLI'

GUI = 'GUI'

class are.simulation.types.EventType(value)[source]

Bases: Enum

Type of the events, depends on who initiated them. - AGENT: events initiated by the agent - ENV: events initiated by the environment, or the scenario designer. - USER: events initiated by the user or user proxy (unused for now). - CONDITION: events that check a condition and trigger other events. - VALIDATION: events that validate the state of the environment. - STOP: events that stop the simulation.

AGENT = 'AGENT'

ENV = 'ENV'

CONDITION = 'CONDITION'

VALIDATION = 'VALIDATION'

USER = 'USER'

STOP = 'STOP'

class are.simulation.types.Action(function, args=<factory>, resolved_args=<factory>, app=None, action_id=None, operation_type=OperationType.READ, tool_metadata=None)[source]

Bases: object

Action associated with an event, this is a function that will be called when the event is executed.

function: Function to be called when the action is executed, it can be a class method, or a regular function
args: Dict mapping the argument names to values to call the function with at execution time
app: The actual App instance to use for the action execution
If not specified at creation time, it can be deducted from the function instance e.g. if function=email_app.add_emails then we can deduct that app=email_app
action_id: The unique id of the action, this is used to identify the event in the logs.
This is created automatically and does NOT need to be handled by the user
tool_metadata: Optional metadata from the AppTool that this action is associated with

function: Callable

args: dict[str, Any]

resolved_args: dict[str, Any]

app: App | None = None

action_id: str = None

operation_type: OperationType | None = 'read'

tool_metadata: AppTool | None = None

execute_on_app(app)[source]

execute()[source]

property function_name

property class_name

property app_name

to_dict()[source]

classmethod from_dict(d)[source]

class are.simulation.types.ActionDescription(app, function, args)[source]

Bases: object

app: str

function: str

args: list[dict[str, Any]]

class are.simulation.types.ConditionCheckAction(function, action_id=None)[source]

Bases: object

function: Callable[[AbstractEnvironment], bool]

action_id: str = None

property function_name

property class_name

to_dict()[source]

classmethod from_dict(d)[source]

class are.simulation.types.EventMetadata(return_value=None, exception=None, exception_stack_trace=None, completed=True)[source]

Bases: object

Metadata for a completed event, which includes details such as the return value, exception, etc. This is particularly useful for completed events from App API calls, where: - return_value: the return value of the API call. - exception: any exception that occurred during the API call. - exception_stack_trace: the stack trace of the exception, if any. - completed: a flag indicating whether the execution was completed.

return_value: Optional[Any] = None

exception: str | None = None

exception_stack_trace: str | None = None

completed: bool = True

to_dict()[source]

classmethod from_dict(d)[source]

Return type:: EventMetadata

class are.simulation.types.AbstractEvent(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>)[source]

Bases: ABC

Abstract event class, that contains shared field between completed and future events.

event_type: the type of the event, either AGENT, ENV or USER.
action: the action that will be executed when the event happens, either directly a function or an Action obj.
event_time: the time at which the event will happen, this can get overridden in various placed for e.g. in case of conditional triggers.
event_relative_time: the relative time wrt the simulation start time
WARNING when the event is going to be added to the queue, this information is going to be used to set event_time
event_id: the unique id of the event, this is used to identify the event in the logs.

event_type: EventType = 'ENV'

event_time: float | None = None

event_relative_time: float | None = None

event_id: str = None

successors: list[AbstractEvent]

dependencies: list[AbstractEvent]

to_dict()[source]

depends_on(events=None, delay_seconds=0)[source]

This function is used to add dependencies to the event. If e1 depends on e2 and e3, then e1 will only be executed after e2 and e3 are executed. If a delay is specified, then the event will be executed after the delay after the dependencies are executed.

followed_by(events, delay_seconds=0.0)[source]

This function is used to add successors to the event. If e1 is followed by e2 and e3, then e2 and e3 will only be executed after e1 is executed. If a delay is specified, then the event will be executed after the delay after the dependencies are executed.

is_ready()[source]

This function is used to check if the event is ready to be scheduled i.e. put into the event_queue. An event is ready to be executed if all its dependencies are executed. When an event has its event_time set, it means it is ready to be scheduled.

Return type:: bool

compute_absolute_time(start_time=0)[source]

Here we compute the absolute time of an event based on its relative time as well as the time of its dependencies.

delayed(delay)[source]

with_id(id)[source]

with_type(event_type)[source]

at_absolute_time(time)[source]

reset_dependencies()[source]

copy()[source]

class are.simulation.types.Event(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, action=None)[source]

Bases: AbstractEvent

Represents an event that will happen in the future. This is what we create often when populating a scenario, and what gets added to the event queue.

action: Action = None

execute(fire_individual_events=False)[source]

Executes the action corresponding to the events and returns the completed event with its metadata. Here by default we make sure we only have ONE event, and that whatever happens inside the action is not registered as individual events. This is to guarantee 2 things: 1/ Events are transactional, either the whole Event (and associated action) happened or not 2/ Not duplicating what is registered in the event log (otherwise we will register BOTH the current event, and every underlying one)

Return type:: CompletedEvent

copy()[source]

classmethod from_function(function, event_type=None, **kwargs)[source]

app_class_name()[source]

Return type:: str | None

app_name()[source]

Return type:: str | None

function_name()[source]

Return type:: str | None

oracle()[source]

to_dict()[source]

classmethod from_dict(d)[source]

class are.simulation.types.StopEvent(*args, **kwargs)[source]

Bases: AbstractEvent

execute(fire_individual_events=False)[source]

Return type:: CompletedEvent

successors: list[AbstractEvent]

dependencies: list[AbstractEvent]

class are.simulation.types.CompletedEvent(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, action=None, metadata=None, _tool_name=None)[source]

Bases: AbstractEvent

Represents an event that already happened, and thus we have some additional metadata on it.

action: Action | ConditionCheckAction = None

metadata: EventMetadata = None

app_class_name()[source]

Return type:: str | None

app_name()[source]

Return type:: str | None

function_name()[source]

Return type:: str | None

replay()[source]

failed()[source]

Return type:: bool

copy()[source]

to_future_event()[source]

to_dict()[source]

classmethod from_dict(d)[source]

property tool_name: Tool name used for validation

get_args()[source]

Return type:: dict[str, Any]

class are.simulation.types.ConditionCheckEvent(event_type=EventType.CONDITION, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, action=None, schedule_every_ticks=1, timeout=None, _internal_check_count=0)[source]

Bases: AbstractEvent

event_type: EventType = 'CONDITION'

action: ConditionCheckAction = None

schedule_every_ticks: int = 1

timeout: int | None = None

with_id(id)[source]

is_timeout()[source]

Return type:: bool

classmethod from_condition(condition, every_tick=1, timeout=None)[source]

copy()[source]

check(env)[source]

Return type:: tuple[bool, CompletedEvent]

depends_on(events=None, delay_seconds=0, schedule_every_ticks=None, timeout=None)[source]

get_next_check_event(time_increment_in_seconds)[source]

class are.simulation.types.OracleEvent(event_type=EventType.AGENT, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, make_event=None, event_time_comparator=None, action_desc=None)[source]

Bases: AbstractEvent

make_event: Callable[[AbstractEnvironment], AbstractEvent] = None

event_type: EventType = 'AGENT'

event_time_comparator: EventTimeComparator | None = None

action_desc: ActionDescription | None = None

make(env)[source]

classmethod from_event(event)[source]

to_dict()[source]

classmethod action_desc_from_event(event)[source]

class are.simulation.types.CompletedOracleEvent(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, action=None, metadata=None, _tool_name=None, absolute_event_time=None, event_time_comparator=None)[source]

Bases: CompletedEvent

A completed oracle event with timing information from the original oracle event.

absolute_event_time: float | None = None

event_time_comparator: EventTimeComparator | None = None

classmethod from_completed_event(completed_event, absolute_event_time=None, event_time_comparator=None)[source]

classmethod from_completed_event_and_oracle_event(completed_event, oracle_event)[source]

Create a completed oracle event from a completed event and an oracle event. The absolute event time is taken from the oracle event, and the event time comparator is taken from the oracle event if it exists.

class are.simulation.types.ValidationResult(success, achieved_milestones=<factory>)[source]

Bases: object

Represents the result of a validation event. - success: whether the validation was successful or not. - message: a message describing the result of the validation. - failed_milestones: the list of milestones that failed during the validation. - triggered_minefields: the list of minefields that were triggered during the validation.

success: bool

achieved_milestones: list[Callable[[AbstractEnvironment], bool]] | list[Callable[[AbstractEnvironment, AbstractEvent], bool]]

to_dict()[source]

class are.simulation.types.AgentActionValidator(milestones=<factory>, minefields=<factory>, timeout=None, _start_tick=0, _internal_check_count=0, achieved_milestones=<factory>)[source]

Bases: object

milestones: list[Callable[[AbstractEnvironment, AbstractEvent], bool]]

minefields: list[Callable[[AbstractEnvironment, AbstractEvent], bool]]

timeout: int | None = None

achieved_milestones: list[Callable[[AbstractEnvironment, AbstractEvent], bool]]

is_timeout()[source]

Return type:: bool

update_tick_count(current_env_tick)[source]

validate(env, event)[source]

Return type:: ValidationResult

class are.simulation.types.AgentValidationEvent(event_type=EventType.VALIDATION, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, validators=<factory>, milestones=<factory>, minefields=<factory>, timeout=None)[source]

Bases: AbstractEvent

event_type: EventType = 'VALIDATION'

validators: list[AgentActionValidator]

milestones: list[Callable[[AbstractEnvironment, AbstractEvent], bool]]

minefields: list[Callable[[AbstractEnvironment, AbstractEvent], bool]]

timeout: int | None = None

get_validator()[source]

depends_on(events=None, delay_seconds=0, schedule_every_ticks=None, timeout=None)[source]

schedule(every_ticks, timeout=None)[source]

exception are.simulation.types.ValidationException[source]: Bases: Exception

class are.simulation.types.ValidationEvent(event_type=EventType.VALIDATION, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, milestones=<factory>, minefields=<factory>, schedule_every_ticks=1, timeout=1, _internal_check_count=0, achieved_milestones=<factory>)[source]

Bases: AbstractEvent

event_type: EventType = 'VALIDATION'

milestones: list[Callable[[AbstractEnvironment], bool]]

minefields: list[Callable[[AbstractEnvironment], bool]]

schedule_every_ticks: int = 1

timeout: int | None = 1

achieved_milestones: list[Callable[[AbstractEnvironment], bool]]

is_timeout()[source]

Return type:: bool

validate(env)[source]

Return type:: tuple[ValidationResult, CompletedEvent]

get_next_event(time_increment_in_seconds=1)[source]

Return type:: ValidationEvent | None

copy()[source]

depends_on(events=None, delay_seconds=0, schedule_every_ticks=None, timeout=None)[source]

schedule(every_ticks, timeout=None)[source]

class are.simulation.types.EventLog(past_events=<factory>)[source]

Bases: object

Event log, contains all the events that happened so far in the environment.

past_events: PriorityQueue[CompletedEvent]

put(event)[source]

list_view()[source]

Return type:: list[CompletedEvent]

to_dict()[source]

static from_list_view(events)[source]

class are.simulation.types.EventQueue(future_events=<factory>, already_scheduled=<factory>)[source]

Bases: object

Event queue, contains all the events that will happen in the future.

future_events: PriorityQueue[AbstractEvent]

already_scheduled: set[str]

put(events)[source]

pop_events_to_process(timestamp)[source]

peek()[source]

list_view()[source]

Return type:: list[Event]

to_dict()[source]

from_list_view(events)[source]

class are.simulation.types.EventRegisterer[source]

Bases: object

Class that handles all the logic for registering events.

classmethod is_active()[source]: Checks whether we should fire events or not.

classmethod is_capture_mode()[source]: Capture mode makes sure a function call is not executed but only a “fictitious” CompletedEvent is returned. This is useful for debugging and testing, as well as easily creating CompletedEvent instances for validation

classmethod set_active(state)[source]: Sets whether we should fire events or not.

classmethod set_capture_mode(state)[source]: Sets whether capture mode is active or not.

classmethod event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT)[source]

classmethod disable()[source]: Context manager to disable event firing, and just return the result of the function call as if decorator is not applied.

classmethod capture_mode()[source]: Context manager to replace a function call decorated with event_registered, by just a fictitious CompletedEvent instance.

are.simulation.types.event_registered(operation_type=OperationType.READ, event_type=EventType.AGENT)

are.simulation.types.disable_events(): Context manager to disable event firing, and just return the result of the function call as if decorator is not applied.

class are.simulation.types.CapabilityTag(value)[source]

Bases: Enum

An enumeration.

Planning = 'Planning'

Memory = 'Memory'

Collaboration = 'Collaboration'

Exploration = 'Exploration'

UnitTest = 'UnitTest'

PromptInjection = 'PromptInjection'

Universe = 'Universe'

Safety = 'Safety'

Adaptability = 'Adaptability'

Ambiguity = 'Ambiguity'

Execution = 'Execution'

Search = 'Search'

Time = 'Time'

Security = 'Security'

classmethod gaia2_capabilities()[source]

class are.simulation.types.ScenarioGUIConfig(show_timestamps=False)[source]

Bases: object

show_timestamps: bool = False

class are.simulation.types.ToolAugmentationConfig(tool_failure_probability=0.1, apply_tool_name_augmentation=True, apply_tool_description_augmentation=True)[source]

Bases: object

tool_failure_probability: float = 0.1

apply_tool_name_augmentation: bool = True

apply_tool_description_augmentation: bool = True

class are.simulation.types.SimulatedGenerationTimeConfig(mode='measured', seconds=1.0)[source]

Bases: object

Configuration for simulating the LLM’s generation time in are.simulation.

Modes:

fixed: The simulated generation time is set to a fixed to seconds.
measured: The simulated generation time is measured from the first successful generation.

mode: Literal['fixed', 'measured'] = 'measured'

seconds: float | None = 1.0

class are.simulation.types.PlaceholderMetadata(parent_tool_name, parent_turn_idx, parent_event_id, placeholder_turn_idx, placeholder_event_id)[source]

Bases: object

Metadata for a placeholder.

parent_tool_name: str

parent_turn_idx: int

parent_event_id: str

placeholder_turn_idx: int

placeholder_event_id: str

class are.simulation.types.ExecutionMetadata(has_placeholder_conflicts, placeholders)[source]

Bases: object

Data structure for execution data.

has_placeholder_conflicts: bool

placeholders: list[PlaceholderMetadata]

Abstract Event ¶

class are.simulation.types.AbstractEvent(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>)[source]¶

Bases: ABC

Abstract event class, that contains shared field between completed and future events.

event_type: the type of the event, either AGENT, ENV or USER.
action: the action that will be executed when the event happens, either directly a function or an Action obj.
event_time: the time at which the event will happen, this can get overridden in various placed for e.g. in case of conditional triggers.
event_relative_time: the relative time wrt the simulation start time
WARNING when the event is going to be added to the queue, this information is going to be used to set event_time
event_id: the unique id of the event, this is used to identify the event in the logs.

event_type: EventType = 'ENV'¶

event_time: float | None = None¶

event_relative_time: float | None = None¶

event_id: str = None¶

successors: list[AbstractEvent]¶

dependencies: list[AbstractEvent]¶

to_dict()[source]¶

depends_on(events=None, delay_seconds=0)[source]¶

followed_by(events, delay_seconds=0.0)[source]¶

is_ready()[source]¶

Return type:: bool

compute_absolute_time(start_time=0)[source]¶

Here we compute the absolute time of an event based on its relative time as well as the time of its dependencies.

delayed(delay)[source]¶

with_id(id)[source]¶

with_type(event_type)[source]¶

at_absolute_time(time)[source]¶

reset_dependencies()[source]¶

copy()[source]¶

__init__(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>)¶

Event ¶

class are.simulation.types.AbstractEvent(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>)[source]

Bases: ABC

Abstract event class, that contains shared field between completed and future events.

event_type: the type of the event, either AGENT, ENV or USER.
action: the action that will be executed when the event happens, either directly a function or an Action obj.
event_time: the time at which the event will happen, this can get overridden in various placed for e.g. in case of conditional triggers.
event_relative_time: the relative time wrt the simulation start time
WARNING when the event is going to be added to the queue, this information is going to be used to set event_time
event_id: the unique id of the event, this is used to identify the event in the logs.

event_type: EventType = 'ENV'

event_time: float | None = None

event_relative_time: float | None = None

event_id: str = None

successors: list[AbstractEvent]

dependencies: list[AbstractEvent]

to_dict()[source]

depends_on(events=None, delay_seconds=0)[source]

followed_by(events, delay_seconds=0.0)[source]

is_ready()[source]

Return type:: bool

compute_absolute_time(start_time=0)[source]

Here we compute the absolute time of an event based on its relative time as well as the time of its dependencies.

delayed(delay)[source]

with_id(id)[source]

with_type(event_type)[source]

at_absolute_time(time)[source]

reset_dependencies()[source]

copy()[source]

__init__(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>)

Bases: AbstractEvent

Represents an event that will happen in the future. This is what we create often when populating a scenario, and what gets added to the event queue.

action: Action = None

execute(fire_individual_events=False)[source]

Return type:: CompletedEvent

copy()[source]

classmethod from_function(function, event_type=None, **kwargs)[source]

app_class_name()[source]

Return type:: str | None

app_name()[source]

Return type:: str | None

function_name()[source]

Return type:: str | None

oracle()[source]

to_dict()[source]

classmethod from_dict(d)[source]

__init__(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, action=None)

successors: list[AbstractEvent]

dependencies: list[AbstractEvent]

Bases: AbstractEvent

event_type: EventType = 'VALIDATION'

milestones: list[Callable[[AbstractEnvironment], bool]]

minefields: list[Callable[[AbstractEnvironment], bool]]

schedule_every_ticks: int = 1

timeout: int | None = 1

achieved_milestones: list[Callable[[AbstractEnvironment], bool]]

is_timeout()[source]

Return type:: bool

validate(env)[source]

Return type:: tuple[ValidationResult, CompletedEvent]

get_next_event(time_increment_in_seconds=1)[source]

Return type:: ValidationEvent | None

copy()[source]

depends_on(events=None, delay_seconds=0, schedule_every_ticks=None, timeout=None)[source]

schedule(every_ticks, timeout=None)[source]

__init__(event_type=EventType.VALIDATION, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, milestones=<factory>, minefields=<factory>, schedule_every_ticks=1, timeout=1, _internal_check_count=0, achieved_milestones=<factory>)

successors: list[AbstractEvent]

dependencies: list[AbstractEvent]

Bases: AbstractEvent

event_type: EventType = 'CONDITION'

action: ConditionCheckAction = None

schedule_every_ticks: int = 1

timeout: int | None = None

with_id(id)[source]

is_timeout()[source]

Return type:: bool

classmethod from_condition(condition, every_tick=1, timeout=None)[source]

copy()[source]

check(env)[source]

Return type:: tuple[bool, CompletedEvent]

depends_on(events=None, delay_seconds=0, schedule_every_ticks=None, timeout=None)[source]

get_next_check_event(time_increment_in_seconds)[source]

__init__(event_type=EventType.CONDITION, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, action=None, schedule_every_ticks=1, timeout=None, _internal_check_count=0)

successors: list[AbstractEvent]

dependencies: list[AbstractEvent]

Bases: AbstractEvent

Represents an event that already happened, and thus we have some additional metadata on it.

action: Action | ConditionCheckAction = None

metadata: EventMetadata = None

app_class_name()[source]

Return type:: str | None

app_name()[source]

Return type:: str | None

function_name()[source]

Return type:: str | None

replay()[source]

failed()[source]

Return type:: bool

copy()[source]

to_future_event()[source]

to_dict()[source]

classmethod from_dict(d)[source]

property tool_name: Tool name used for validation

get_args()[source]

Return type:: dict[str, Any]

__init__(event_type=EventType.ENV, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, action=None, metadata=None, _tool_name=None)

successors: list[AbstractEvent]

dependencies: list[AbstractEvent]

Bases: AbstractEvent

make_event: Callable[[AbstractEnvironment], AbstractEvent] = None

event_type: EventType = 'AGENT'

event_time_comparator: EventTimeComparator | None = None

action_desc: ActionDescription | None = None

make(env)[source]

classmethod from_event(event)[source]

to_dict()[source]

classmethod action_desc_from_event(event)[source]

__init__(event_type=EventType.AGENT, event_time=None, event_relative_time=None, event_id=None, successors=<factory>, dependencies=<factory>, make_event=None, event_time_comparator=None, action_desc=None)

successors: list[AbstractEvent]

dependencies: list[AbstractEvent]

Action Class ¶

class are.simulation.types.Action(function, args=<factory>, resolved_args=<factory>, app=None, action_id=None, operation_type=OperationType.READ, tool_metadata=None)[source]

Bases: object

Action associated with an event, this is a function that will be called when the event is executed.

function: Function to be called when the action is executed, it can be a class method, or a regular function
args: Dict mapping the argument names to values to call the function with at execution time
app: The actual App instance to use for the action execution
If not specified at creation time, it can be deducted from the function instance e.g. if function=email_app.add_emails then we can deduct that app=email_app
action_id: The unique id of the action, this is used to identify the event in the logs.
This is created automatically and does NOT need to be handled by the user
tool_metadata: Optional metadata from the AppTool that this action is associated with

function: Callable

args: dict[str, Any]

resolved_args: dict[str, Any]

app: App | None = None

action_id: str = None

operation_type: OperationType | None = 'read'

tool_metadata: AppTool | None = None

execute_on_app(app)[source]

execute()[source]

property function_name

property class_name

property app_name

to_dict()[source]

classmethod from_dict(d)[source]

__init__(function, args=<factory>, resolved_args=<factory>, app=None, action_id=None, operation_type=OperationType.READ, tool_metadata=None)

Event Types and Enums ¶

class are.simulation.types.EventType(value)[source]

Bases: Enum

AGENT = 'AGENT'

ENV = 'ENV'

CONDITION = 'CONDITION'

VALIDATION = 'VALIDATION'

USER = 'USER'

STOP = 'STOP'

Event Management ¶

Event Queue

class are.simulation.types.EventQueue(future_events=<factory>, already_scheduled=<factory>)[source]

Bases: object

Event queue, contains all the events that will happen in the future.

future_events: PriorityQueue[AbstractEvent]

already_scheduled: set[str]

put(events)[source]

pop_events_to_process(timestamp)[source]

peek()[source]

list_view()[source]

Return type:: list[Event]

to_dict()[source]

from_list_view(events)[source]

Event Log

class are.simulation.types.EventLog(past_events=<factory>)[source]

Bases: object

Event log, contains all the events that happened so far in the environment.

past_events: PriorityQueue[CompletedEvent]

put(event)[source]

list_view()[source]

Return type:: list[CompletedEvent]

to_dict()[source]

static from_list_view(events)[source]

Additional Types and Classes ¶

class are.simulation.types.AbstractEnvironment[source]

Bases: ABC

state: EnvironmentState | None

event_log: EventLog

event_queue: EventQueue

start()[source]

pause()[source]

resume()[source]

get_state()[source]

Return type:: dict[str, Any]

get_app(app_name)[source]

get_event_log_size()[source]

Return type:: int

get_event_queue_length()[source]

Return type:: int

final_validation_checks()[source]

Return type:: None

class are.simulation.types.ActionDescription(app, function, args)[source]

Bases: object

app: str

function: str

args: list[dict[str, Any]]

Bases: object

milestones: list[Callable[[AbstractEnvironment, AbstractEvent], bool]]

minefields: list[Callable[[AbstractEnvironment, AbstractEvent], bool]]

timeout: int | None = None

achieved_milestones: list[Callable[[AbstractEnvironment, AbstractEvent], bool]]

is_timeout()[source]

Return type:: bool

update_tick_count(current_env_tick)[source]

validate(env, event)[source]

Return type:: ValidationResult

class are.simulation.types.CapabilityTag(value)[source]

Bases: Enum

An enumeration.

Planning = 'Planning'

Memory = 'Memory'

Collaboration = 'Collaboration'

Exploration = 'Exploration'

UnitTest = 'UnitTest'

PromptInjection = 'PromptInjection'

Universe = 'Universe'

Safety = 'Safety'

Adaptability = 'Adaptability'

Ambiguity = 'Ambiguity'

Execution = 'Execution'

Search = 'Search'

Time = 'Time'

Security = 'Security'

classmethod gaia2_capabilities()[source]

class are.simulation.types.ConditionCheckAction(function, action_id=None)[source]

Bases: object

function: Callable[[AbstractEnvironment], bool]

action_id: str = None

property function_name

property class_name

to_dict()[source]

classmethod from_dict(d)[source]

class are.simulation.types.EventMetadata(return_value=None, exception=None, exception_stack_trace=None, completed=True)[source]

Bases: object

return_value: Optional[Any] = None

exception: str | None = None

exception_stack_trace: str | None = None

completed: bool = True

to_dict()[source]

classmethod from_dict(d)[source]

Return type:: EventMetadata

class are.simulation.types.EventTimeComparator(value)[source]

Bases: Enum

Comparator for event time filtering. - LESS_THAN: Less than comparison - GREATER_THAN: Greater than comparison - EQUAL: Equal comparison

LESS_THAN = 'LESS_THAN'

GREATER_THAN = 'GREATER_THAN'

EQUAL = 'EQUAL'