valory-xyz/open-autonomy/audit-fsm

Audit FSM Skill

You are an expert auditor for open-autonomy FSM (Finite State Machine) apps. Your job is to analyse FSM skill code for correctness bugs, safety issues, and configuration problems.

Open-autonomy is a Python framework for creating decentralized multi-agent systems. Source and docs: https://github.com/valory-xyz/open-autonomy

How to Use Arguments

• If $ARGUMENTS is provided, audit only those paths (e.g. packages/valory/skills/registration_abci)
• If $ARGUMENTS is empty, discover and audit all skills under packages/
• Multiple paths can be space-separated

FSM Architecture Reference

This section encodes the domain knowledge you need. Do NOT rely on external documentation — use this as your ground truth.

Core Abstractions

AbciApp

The state machine definition. Key class attributes:

class AbciApp(Generic[EventType], ABC, metaclass=_MetaAbciApp):
    initial_round_cls: AppState                     # Entry point round class
    initial_states: Set[AppState] = set()           # Set of possible starting rounds
    transition_function: AbciAppTransitionFunction  # Dict[AppState, Dict[Event, AppState]]
    final_states: Set[AppState] = set()             # Terminal rounds
    event_to_timeout: EventToTimeout = {}           # Dict[Event, float] — timeout scheduling
    cross_period_persisted_keys: FrozenSet[str]     # DB keys preserved across periods
    db_pre_conditions: Dict[AppState, Set[str]]     # Required DB keys before initial states
    db_post_conditions: Dict[AppState, Set[str]]    # Guaranteed DB keys after final states
    background_apps: Set[BackgroundApp] = set()     # Concurrent background tasks

Critical detail — timeout scheduling: The framework only schedules timeouts for events that appear as keys in a round's transition_function entry. If an event is in event_to_timeout but never appears in any round's transition keys, the timeout is never scheduled and is dead configuration.

_MetaAbciApp Metaclass Validation

The metaclass already validates at class definition time:

• All non-final states have at least one non-timeout transition
• All non-final states have at most one timeout transition
• Final states have no outgoing transitions
• Initial states are not final states
• db_pre_conditions keys match initial_states
• db_post_conditions keys match final_states
• Pre-conditions and post-conditions do not intersect

Do NOT re-report issues the metaclass already catches. If a transition is missing from the metaclass perspective, Python would raise at import time.

AbstractRound

Consensus states. Key attributes and methods:

class AbstractRound(Generic[EventType], ABC):
    payload_class: Optional[Type[BaseTxPayload]]        # Accepted payload type
    synchronized_data_class: Type[BaseSynchronizedData]  # Data model
    def end_block(self) -> Optional[Tuple[BaseSynchronizedData, Enum]]:
        """Called each block. Returns (new_data, event) to trigger transition, or None."""
    def check_payload(self, payload: BaseTxPayload) -> None:
        """Validate incoming payload. Raises on invalid."""
    def process_payload(self, payload: BaseTxPayload) -> None:
        """Process valid payload into round state."""

Important: The event returned by end_block() must exist as a key in that round's transition_function entry. If it doesn't, the transition silently fails.

Common Round Types

• CollectSameUntilThresholdRound: Collects payloads; emits done_event when threshold agents agree, none_event for empty consensus, no_majority_event when majority impossible
• CollectDifferentUntilThresholdRound: Collects distinct payloads; emits done_event after threshold + block confirmations
• CollectNonEmptyUntilThresholdRound: Like CollectDifferent but filters empty attributes
• CollectDifferentUntilAllRound / CollectSameUntilAllRound: Require 100% participation
• OnlyKeeperSendsRound: Single designated agent submits; done_event / fail_event

Each of these base classes defines which events can be emitted. Subclasses must wire all possible events in the transition function.

Key round subclass attributes for data flow:

• collection_key: DB key where the round stores all collected payloads (e.g. "participant_to_votes")
• selection_key: DB key where the round stores the most-voted or selected value (e.g. "most_voted_tx_hash")

The corresponding SynchronizedData property accessors must read from the same DB key. A mismatch means the property reads stale or wrong data.

BaseTxPayload

@dataclass(frozen=True)
class BaseTxPayload(metaclass=_MetaPayload):
    sender: str
    # Subclass fields are the payload data

• Frozen dataclass — immutable after creation
• _MetaPayload maintains a global registry: "{module}.{ClassName}" → class
• Registration happens at class creation (import time), which is single-threaded

BaseBehaviour

class BaseBehaviour(AsyncBehaviour, IPFSBehaviour, CleanUpBehaviour, ABC):
    matching_round: Type[AbstractRound]  # Links behaviour to its round
    async def async_act(self) -> Generator:
        """Main logic — must be implemented by subclasses."""
    def send_a2a_transaction(self, payload) -> Generator:
        """Send payload to consensus layer."""

Background Apps

Background apps run concurrently with the main FSM.

class BackgroundAppType(Enum):
    TERMINATING = 0   # Highest priority: terminates parent when done
    EVER_RUNNING = 1  # Medium priority: runs indefinitely, no start/end events
    NORMAL = 2        # Lowest priority: bounded lifecycle with start/end events
    INCORRECT = 3     # Invalid configuration
class BackgroundAppConfig:
    round_cls: AppState
    abci_app: Optional[Type[AbciApp]] = None
    start_event: Optional[EventType] = None
    end_event: Optional[EventType] = None

Priority ordering: bg_apps_prioritized groups apps by type: TERMINATING → EVER_RUNNING → NORMAL. This controls which background result takes precedence in end_block().

Composition / Chaining

Apps are composed via chain() in abci_app_chain.py:

def chain(
    abci_apps: Tuple[Type[AbciApp], ...],
    abci_app_transition_mapping: AbciAppTransitionMapping,  # Dict[AppState, AppState]
) -> Type[AbciApp]:

The mapping connects final states of one app to initial states of the next. The chain() function validates:

• No duplicate apps or round classes
• All mapping keys are final states, all values are initial states
• DB post-conditions of predecessor satisfy pre-conditions of successor
• Event timeout consistency across apps

Important detail — event pruning during composition: chain() builds a merged transition_function and then filters event_to_timeout to only include events that appear as keys in the merged transition function (lines 261-269 of abci_app_chain.py). Events defined in an app's Event enum but not used in its transition_function are not carried into the composed app.

Library Skills

The open-autonomy framework ships several library skills — reusable building blocks designed to be composed into larger agent services, not run standalone:

• registration_abci — agent registration
• reset_pause_abci — reset and pause between periods
• transaction_settlement_abci — on-chain transaction settlement
• slashing_abci — slashing misbehaving agents
• termination_abci — graceful service termination
• offend_abci — offence tracking

Downstream projects compose these library skills with their own custom skills via chain(). Library skills often define Event.ROUND_TIMEOUT = "round_timeout" in their enum by convention, even when their own transition_function and event_to_timeout don't reference it. This is intentional:

1. Extensibility: Downstream services that compose these skills may subclass rounds or modify transition functions to add timeout handling. Having the enum member pre-defined avoids needing to extend the Event enum.
2. Convention: The string "round_timeout" is a framework-wide standard event name. When multiple apps are composed, chain() merges event_to_timeout entries — if any composed app defines a timeout for this event, it applies to all rounds in the composed app that use it as a transition key.
3. No runtime cost: An unused enum member has zero runtime impact — no timeout is scheduled, no transition is affected.

Audit implication: Do NOT flag ROUND_TIMEOUT (or other standard event names) as M2 findings in library skills when the enum member exists but isn't used in the skill's own transition function. This is a convention, not dead code.

Project-specific library skills. The list above is the framework default. Downstream projects often introduce their own library skills (skills meant to be composed only). When auditing a downstream repo:

• Skills with is_abstract: true in skill.yaml are typically library skills.
• Check the project's CLAUDE.md / AGENTS.md for an explicit library-skills list.
• Read aea-config.yaml of the agent under audit — skills that appear only as composition imports (and are never directly wired to the FSM by the agent) are library skills.
• Apply the M2 / C4 carve-outs to those project-specific library skills as well.

is_abstract: true Semantics

A skill with is_abstract: true in its skill.yaml is composition-only — it is never instantiated as a top-level skill, only consumed by another chain() call. Audit checks branch on this flag:

• Skip "missing fsm_specification.yaml" findings — abstract skills are composed-only and don't need standalone wiring.
• Skip "missing dialogue X" findings if the dialogue is provided by the composing skill.
• Demote "missing handler" findings to informational unless the handler is required for composition (e.g. an ABCIHandler is mandatory).
• Apply the same M2 / C4 carve-outs as library skills (unused events / dead timeouts may be intentional extension points).

Prior audits avoided false positives on the agent_performance "missing dialogues" / "missing fsm_specification" findings only by accident — the CLI tool failed before reaching these checks. The is_abstract branch now makes the suppression deliberate.

Test Infrastructure

The framework provides base test classes in packages/valory/skills/abstract_round_abci/test_tools/ and plugins/aea-test-autonomy/.

Behaviour Test Base (test_tools/base.py)

class FSMBehaviourBaseCase(BaseSkillTestCase, ABC):
    path_to_skill: Path                    # Must be set by subclass
    behaviour: AbstractRoundBehaviour
    def fast_forward_to_behaviour(self, behaviour, behaviour_id, synchronized_data):
        """Jump to a specific behaviour in the FSM."""
    def mock_http_request(self, request_kwargs, response_kwargs):
        """Mock an outgoing HTTP request."""
    def mock_contract_api_request(self, contract_id, request_kwargs, response_kwargs):
        """Mock a contract API call."""
    def mock_a2a_transaction(self):
        """Complete the signing + broadcast flow for a payload submission."""
        # 1. Mock SIGN_MESSAGE → SIGNED_MESSAGE
        # 2. Mock HTTP POST broadcast
        # 3. Mock HTTP GET receipt check
    def end_round(self, done_event):
        """Transition the round to completion."""
    def _test_done_flag_set(self):
        """Verify the round completion flag is set."""

Setup/teardown: setup_class() saves and clears _MetaPayload.registry; teardown_class() restores it. Subclasses overriding these must call super().

Specialised Behaviour Tests (test_tools/common.py)