Lifecycle Management

ErisPulse provides a unified hook/lifecycle system for monitoring the operational status of system components and enabling extensions such as auditing, statistics, and custom logic.

The system supports three trigger methods:

await lifecycle.emit("event", data) — A concise version, passing arbitrary data (directed delivery when to="Owner")
lifecycle.emit_sync("event", data) — Synchronous version (for non-async contexts)
await lifecycle.submit_event("event", ...) — Backward compatible, automatically constructs standard event format

Event Handling Mechanism

Registering Handlers

from ErisPulse import sdk

# Decorator pattern
@sdk.lifecycle.on("module.load")
async def on_module_load(data):
    print(f"Module loaded: {data}")

# Programmatic registration
sdk.lifecycle.register("module.load", on_module_load, priority=10)

# Unregister
sdk.lifecycle.unregister("module.load", on_module_load)

# Batch unregister by owner (automatically called by framework during module/unloader adapter)
removed = sdk.lifecycle.unregister_by_owner("MyModule")
print(f"Cleaned up {removed} lifecycle hooks")

Priority

Handlers support the priority parameter, with higher values executed first (consistent with module loader):

@sdk.lifecycle.on("adapter.event.receive", priority=10)  # Executed first
async def first_handler(data):
    pass

@sdk.lifecycle.on("adapter.event.receive", priority=0)  # Executed later
async def second_handler(data):
    pass

Dot-Structure Events

When a specific event is triggered, its parent events are also triggered:

Triggering module.load also triggers module
Triggering adapter.event.receive also triggers adapter.event and adapter

Wildcards

@sdk.lifecycle.on("*")
async def on_anything(data):
    print(f"Received event: {data}")

Directed Propagation (emit to=)

Note

This feature requires ErisPulse 2.8.0+.

When emit() specifies the to parameter, it enters directed propagation: events are only distributed to handlers registered with that owner (module hooks registered in on_load automatically belong to the module), other modules and wildcard * handlers do not receive them.

# Emitter: events only delivered to hooks registered by Chat module
await sdk.lifecycle.emit("message_received", {"text": "hi"}, to="Chat")

# Subscriber (within Chat module): register same event name, owner is automatically recorded during registration
@sdk.lifecycle.on("message_received")
async def on_message_received(data): ...

@sdk.lifecycle.on("message")   # Dot-structure parent prefixes also apply (filtered by owner)
async def on_any(data): ...

If the target owner has no registered hooks → the event is not consumed (can be probed beforehand using has_handlers())
When data is a dict, _trace_id is automatically included (without overriding existing values)
emit_sync / submit_event also support the to= parameter
Three-layer model for inter-module communication (RPC / directed / broadcast) is detailed in Module Communication

One-Time Registration (once)

Since 2.7.0, handlers registered with lifecycle.once() are automatically unregistered after being triggered once, suitable for "first ready" type hooks:

@sdk.lifecycle.once("core.init.complete")
async def on_first_ready(data):
    print("First ready, will not trigger again")

Same priority semantics as on() (higher priority value means earlier execution)
Automatically unregistered, no need for manual unregister
Supported for both synchronous and asynchronous handlers

Listener Query (has_handlers)

In hot-path short-circuit scenarios, has_handlers() can be used to check if listeners exist beforehand, avoiding unnecessary event traversal and task scheduling:

if sdk.lifecycle.has_handlers("message.sending"):
    await sdk.lifecycle.emit("message.sending", send_ctx)

Covers exact event name, wildcard *, and parent event matching
Returns False if there are no listeners, allowing safe skipping of emit

Hook Breakpoints Overview

A typical sequence of lifecycle events for a message from platform entry into the framework until processing completion:

sequenceDiagram
    participant P as Platform
    participant A as Adapter
    participant F as Framework Core
    participant M as Module Processor

    P->>A: Native event arrives
    A->>F: adapter.event.receive (earliest)
    F->>F: event.pre_process (before handler execution)
    F->>M: Dispatch to processor (commands/messages/notifications, etc.)
    M->>M: command.matched / command.executed
    M->>F: event.reply()
    F->>F: message.sending (before sending)
    F->>A: SendDSL send
    A->>P: Send to platform
    A->>F: message.sent (after sending)
    F->>F: adapter.event.dispatched (after dispatch)

The framework provides the following built-in hook breakpoints, which users can listen to with @sdk.lifecycle.on() to implement custom logic.

Core Initialization

Hook Name	Trigger Timing	Data
`core.init.start`	SDK initialization starts	`{}`
`core.init.stage`	Each initialization stage starts (emitted in background)	`{"stage": str}`, values: `discovery` / `adapter_register` / `adapter_start` / `module_register` / `module_init` / `adapter_start_deferred` / `router_start`
`core.init.complete`	SDK initialization completes	`{"duration": float, "success": bool, "stages": {stage: float}, "adapters": {"enabled": [str], "disabled": [str]}, "modules": {"enabled": [str], "disabled": [str]}, "error": str (only on failure)}`
`core.uninit.complete`	SDK deinitialization completes	`{"duration": float, "success": bool, "adapters_closed": int, "modules_unloaded": int, "module_properties_cleared": int, "module_properties_to_clear": [str], "error": str (only on failure)}`

Example: Displaying Startup Progress

@sdk.lifecycle.on("core.init.stage")
def show_stage(data):
    print(f"[Startup] Entering stage: {data['stage']}")

Configuration Changes

Hook Name	Trigger Timing	Data
`config.set`	A configuration item is modified	`{"key": str, "old_value": Any, "new_value": Any}`
`config.updated`	Entire config tree is updated after external edit to config.toml	`{"old_config": dict, "new_config": dict, "config_file": str}`

Example: Configuration Audit

@sdk.lifecycle.on("config.set")
def audit_config(data):
    print(f"[Audit] {data['key']}: {data['old_value']} -> {data['new_value']}")

Module Lifecycle

Hook Name	Trigger Timing	Data
`module.register`	Module class is registered to manager	`{"module_name": str, "success": bool}`
`module.load`	Module loading completes (successful instantiation)	`{"module_name": str, "success": bool}`
`module.init`	Module initialization completes (including lazy loading)	`{"module_name": str, "success": bool}`
`module.unload`	Module is unloaded	`{"module_name": str, "success": bool}`
`module.reload`	Module hot reload completes (including cascading reload of dependencies)	`{"module_name": str, "success": bool, "full": bool}`; when full reload (`reload_all`) occurs, `module_name` is `"All"`, payload additionally contains `"results": dict[str, bool]`

Adapter Lifecycle

Hook Name	Trigger Timing	Data
`adapter.load`	Adapter registration completes	`{"platform": str, "success": bool}`
`adapter.start`	Adapter starts	`{"platforms": [str]}`
`adapter.status.change`	Adapter status changes	`{"platform": str, "status": str, "retry_count": int, "error": str (only on failure)}`; complete status values: `starting` / `started` / `start_failed` / `stopping` / `stopped` / `stop_failed` / `skipped-dependency` / `disabled`
`adapter.stop`	Adapter stops	`{"platforms": [str]}`
`adapter.stopped`	Adapter stop completes	`{"platforms": [str]}`
`adapter.bot.online`	Bot goes online	`{"platform": str, "bot_id": str, "info": dict, "status": str}`
`adapter.bot.offline`	Bot goes offline	`{"platform": str, "bot_id": str, "status": str}`

Event Reception and Processing

Hook Name	Trigger Timing	Data
`adapter.event.receive`	External platform event is received (earliest)	`{"platform": str, "event_type": str, "raw_event_type": str}`
`adapter.event.blocked`	Middleware rejects event (returns `False`, event is discarded and not passed to any handler)	`{"middleware": str, "platform": str, "event_type": str, "detail_type": str, "event": dict, "_trace_id": str}`
`adapter.event.dispatched`	Event dispatch completes	`{"platform": str, "event_type": str, "raw_event_type": str, "onebot_handlers_count": int}`
`event.pre_process`	Event handler begins execution	`{"event_type": str, "platform": str, "detail_type": str}`

Example: Event Counting

event_counter = {}

@sdk.lifecycle.on("adapter.event.receive")
def count_events(data):
    platform = data["platform"]
    event_counter[platform] = event_counter.get(platform, 0) + 1

@sdk.lifecycle.on("adapter.event.dispatched")
def log_unhandled(data):
    if data["onebot_handlers_count"] == 0:
        print(f"[Unhandled] {data['platform']}/{data['event_type']}")

Message Sending

Hook Name	Trigger Timing	Data
`message.sending`	Message is about to be sent	`{"platform": str, "method": str, "detail_type": str, "target_id": str, "bot_id": str}`
`message.sent`	Message sending completes	`{"platform": str, "method": str, "detail_type": str, "target_id": str, "bot_id": str}`

Example: Message Sending Audit

@sdk.lifecycle.on("message.sending")
def log_sending(data):
    print(f"[Sending] -> {data['platform']}/{data['detail_type']}/{data['target_id']} via {data['method']}")

Command System

Hook Name	Trigger Timing	Data
`command.matched`	Command is matched and about to execute	`{"command": str, "args": list[str], "platform": str, "user_id": str}`
`command.executed`	Command execution completes	`{"command": str, "args": list[str], "platform": str, "user_id": str, "success": bool, "error": str (only on failure)}`

Example: Command Counting

@sdk.lifecycle.on("command.matched")
def count_commands(data):
    print(f"[Command] /{data['command']} from {data['user_id']}@{data['platform']}")

HTTP Routing

Hook Name	Trigger Timing	Data
`server.request`	HTTP request is received	`{"method": str, "path": str, "client_ip": str}`
`server.response`	HTTP response is sent	`{"method": str, "path": str, "status_code": int, "client_ip": str}`

Example: Request Logging

@sdk.lifecycle.on("server.response")
def log_http(data):
    print(f"[HTTP] {data['method']} {data['path']} -> {data['status_code']}")

WebSocket

Hook Name	Trigger Timing	Data
`server.start`	Route server starts	`{"base_url": str, "host": str, "port": int, "success": bool, "error": str (only on failure)}`
`server.stop`	Route server stops	`{}`
`server.websocket.connect`	WebSocket connection established	`{"path": str, "module_name": str, "client_ip": str}`
`server.websocket.disconnect`	WebSocket connection disconnected	`{"path": str, "module_name": str, "reason": str, "error": str (only on abnormal disconnection)}`

Example: WebSocket Connection Monitoring

@sdk.lifecycle.on("server.websocket.connect")
def on_ws_connect(data):
    print(f"[WS] Connection: {data['path']} from {data['client_ip']}")

@sdk.lifecycle.on("server.websocket.disconnect")
def on_ws_disconnect(data):
    print(f"[WS] Disconnection: {data['path']} ({data['reason']})")

Storage Connection Status

Establishment, failure, and recovery of storage backend connection pools (all emitted in background, do not block storage operations):

Hook Name	Trigger Timing	Data
`storage.ready`	Storage backend connection pool is ready (first successful pool creation per event loop)	`{"backend": str}`
`storage.unreachable`	Connection retries exhausted, entering cooldown period (operations fail quickly during cooldown)	`{"backend": str, "error": str, "cooldown": float}`
`storage.recovered`	Cooldown ends, reconnection successful, storage becomes available again	`{"backend": str}`

Example: Storage Failure Alert

@sdk.lifecycle.on("storage.unreachable")
def alert_storage_down(data):
    print(f"[Alert] Storage backend {data['backend']} unreachable: {data['error']}, automatic reconnection in {data['cooldown']}s")

@sdk.lifecycle.on("storage.recovered")
def notify_storage_back(data):
    print(f"[Recovery] Storage backend {data['backend']} is available again")

HTTP Client

Events related to sdk.client requests and connections (all emitted in background):

Hook Name	Trigger Timing	Data
`client.request.success`	HTTP request succeeds	`{"method": str, "url": str, "status": int, "elapsed": float}`
`client.request.failed`	HTTP request fails after exhausting retries	`{"method": str, "url": str, "error": str, "attempts": int, "elapsed": float}`
`client.ws.connect`	WebSocket connection established	`{"url": str}`

Internationalization

Hook Name	Trigger Timing	Data
`i18n.language.changed`	Framework language switch (via `i18n.set_language`)	`{"language": str, "previous": str}`

Standard Event Definitions

STANDARD_EVENTS = {
    "core": ["init.start", "init.stage", "init.complete", "uninit.complete"],
    "module": ["load", "init", "unload", "register", "reload"],
    "adapter": [
        "load", "start", "status.change", "stop", "stopped",
        "event.receive", "event.dispatched",
        "bot.online", "bot.offline",
    ],
    "server": [
        "start", "stop",
        "request", "response",
        "websocket.connect", "websocket.disconnect",
    ],
    "event": ["pre_process"],
    "message": ["sending", "sent"],
    "command": ["matched", "executed"],
    "config": ["set", "updated"],
    "storage": ["ready", "unreachable", "recovered"],
    "client": ["request.success", "request.failed", "ws.connect"],
    "i18n": ["language.changed"],
}

Complete API Reference

Registration and Unregistration

Method	Description
`@lifecycle.on(event, *, priority=0)`	Decorator registration of handler
`lifecycle.register(event, handler, *, priority=0)`	Programmatic registration
`lifecycle.unregister(event, handler=None)`	Unregister (if handler=None, unregister all handlers for that event)

Triggering

Method	Description
`await lifecycle.emit(event, data=None, *, to=None)`	Asynchronous trigger, handlers execute in parallel (non-blocking, returns when all complete), returns non-None values in priority order for chained data replacement; `to` specifies owner for directed delivery
`lifecycle.fire(event, data=None, *, to=None)`	Fire in background (fire and forget): handlers execute in background tasks in parallel, no wait, no return value; zero overhead if no listeners; suitable for high-frequency hot paths and pure observation events; for shutdown sequences and ordered consumption (e.g., `config.set`) use `emit`
`lifecycle.emit_sync(event, data=None, *, to=None)`	Synchronous trigger, asynchronous handlers scheduled via create_task
`await lifecycle.submit_event(event_type, *, source, msg, data, to=None, background=False)`	Backward compatible, automatically constructs standard event format; if `background=True`, uses `fire` background firing

Utilities

Method	Description
`lifecycle.start_timer(timer_id)`	Start timing
`lifecycle.get_duration(timer_id)`	Get elapsed duration (seconds)
`lifecycle.stop_timer(timer_id)`	Stop timing and return elapsed duration
`lifecycle.list_hooks()`	List all registered hooks and handler counts
`lifecycle.clear()`	Clear all handlers and timers

Module Usage Examples

from ErisPulse.Core.Bases import BaseModule
from ErisPulse import sdk

class Main(BaseModule):
    async def on_load(self, event):
        # Implement simple message statistics
        self.msg_count = 0
        
        @sdk.lifecycle.on("adapter.event.receive")
        async def count(data):
            if data["event_type"] == "message":
                self.msg_count += 1
        
        # Monitor all commands
        @sdk.lifecycle.on("command.matched")
        async def log_cmd(data):
            sdk.logger.info(f"Command executed: /{data['command']} by {data['user_id']}")
        
        # Configuration change audit
        @sdk.lifecycle.on("config.set")
        def audit(data):
            sdk.logger.info(f"Configuration change: {data['key']} = {data['new_value']}")

Background Task Ownership and Automatic Cancellation

Note

This feature requires ErisPulse 2.8.0+.

Background tasks created by modules that are not canceled in on_unload will hold a reference to self, preventing the module instance from being recycled (old instances remain after hot reload). The framework provides the following fallback mechanism:

self.spawn(coro) (recommended within modules): Tasks are automatically assigned to the module name, and the framework cancels unfinished tasks and logs warnings after on_unload when the module is unloaded
spawn_background(coro) (on ErisPulse.runtime): Automatically captures the current owner_scope context; cancel_owner_tasks(owner) cancels tasks by assignment, cancel_all_background_tasks() is provided for sdk.uninit() fallback
Adapters: When closing, background tasks under the platform name are also canceled as a fallback

async def on_load(self, event):
    # Recommended: Use self.spawn() for background tasks, framework automatically cancels as fallback after unload
    self.spawn(self._poll())

async def on_unload(self, event):
    # For fine-grained control, still recommend manually canceling and waiting for completion
    if self._poll_task:
        self._poll_task.cancel()
        await asyncio.gather(self._poll_task, return_exceptions=True)

async def _poll(self):
    while True:
        await asyncio.sleep(60)
        ...

Important

The framework's fallback is forced cancellation (cancel_owner_tasks), which occurs after on_unload returns. Therefore, tasks requiring graceful termination (flush buffers, persist state, close connections) must be manually canceled and awaited in on_unload—do not rely on the fallback to preserve termination logic. The framework only guarantees that "tasks holding self are not left behind," not that they terminate "gracefully." Tasks requiring await results should be awaited directly, not thrown into background tasks.

Notes

Handlers can be synchronous or asynchronous: The system automatically identifies and correctly calls them
Data passing: In emit() mode, handlers returning non-None values modify the data passed to subsequent handlers
Event naming convention: It is recommended to use dot-structure naming for events, which facilitates listening to parent events
Error isolation: An exception in a single handler does not affect the execution of other handlers
Synchronous trigger limitation: In emit_sync(), asynchronous handlers are scheduled in a fire-and-forget manner, and return values cannot be returned
Lifecycle cleanup: When sdk.uninit() is called, all registered handlers and timers are cleared
Load priority: If you need to listen to events during the framework initialization phase, it is recommended to set a high priority and disable lazy loading

Module Development Guide - Learn about module lifecycle methods
Best Practices - Lifecycle event usage recommendations

Lifecycle Management

Event Handling Mechanism

Registering Handlers

Priority

Dot-Structure Events

Wildcards

Directed Propagation (emit to=)

One-Time Registration (once)

Listener Query (has_handlers)

Hook Breakpoints Overview

Core Initialization

Configuration Changes

Module Lifecycle

Adapter Lifecycle

Event Reception and Processing

Message Sending

Command System

HTTP Routing

WebSocket

Storage Connection Status

HTTP Client

Internationalization

Standard Event Definitions

Complete API Reference

Registration and Unregistration

Triggering

Utilities

Module Usage Examples

Background Task Ownership and Automatic Cancellation

Notes

Related Documentation