Startup Process and Manual Control

ErisPulse's await sdk.run() / await sdk.init() encapsulates the entire startup chain into a single line of code. However, when you need to fully customize the startup process (for example, partial loading, dynamic registration, hot plugging, injecting custom loading strategies), you need to understand what happens inside this chain and how to manually drive each step.

This article breaks down the startup chain into independent components, explains the responsibilities of each, the order of calls, and provides an example of how to manually perform a complete startup.

This article assumes you have already run through the first bot and understand the two modes of sdk.run(keep_running=True/False). This article focuses on the internal breakdown of the init() chain and lower-level entry points such as init() / init_task() / init_sync().

Overview of SDK Top-Level Entry Points

In addition to the two keep_running modes of run(), the SDK also provides several lower-level initialization entry points, which differ in terms of asynchronicity, return value, and whether exceptions are wrapped:

Common Misconception: await sdk.init() is not equivalent to await sdk.run(keep_running=False). There are two differences: ① init() returns bool (returns False on failure), while run() returns None; ② init() only performs initialization and does not automatically unload, whereas run() automatically uninit() when the event loop ends. Therefore, when manually pairing unloading or customizing the lifecycle, use init() + uninit().

Chain Startup Overview

sdk.init() (specifically, its internal Initializer.init()) initiates the framework in the following sequence:

flowchart TD
    A[0. Prepare Environment<br/>Configuration Loading / Exception Handling] --> B
    B[1. Parallel Discovery and Loading<br/>AdapterLoader.load / ModuleLoader.load<br/>Internal call to Finder.find_all] --> C
    C[2. Register Adapters<br/>AdapterLoader.register_to_manager] --> D
    D[3. Start Adapters<br/>adapter.startup] --> E
    E[4. Register Modules<br/>ModuleLoader.register_to_manager] --> F
    F[5. Initialize Modules<br/>ModuleLoader.initialize_modules<br/>Instantiate and mount to sdk] --> G
    G[6. Start Router Server<br/>router.start]

Corresponding core components:

Important: Finder and Loader are two layers. The Loader internally already holds a Finder (AdapterLoader has its own AdapterFinder, ModuleLoader has its own ModuleFinder). In most scenarios, you only need to use Loader. You would only use Finder separately when you need to "list without importing".

Detailed Explanation of Each Step

1. Discovery Layer: Finder

The Finder is responsible only for "finding which packages provide adapters/modules", without importing or instantiating them.

from ErisPulse.finders import AdapterFinder, ModuleFinder

adapter_finder = AdapterFinder()
module_finder = ModuleFinder()

# Find all installed adapter/module entry-points
adapter_entries = adapter_finder.find_all()    # list[EntryPoint]
module_entries = module_finder.find_all()      # list[EntryPoint]

# Find a single entry-point by name
entry = module_finder.find_by_name("MyModule")  # EntryPoint | None

Each EntryPoint can be loaded via .load() to obtain the corresponding class, but typically you do not need to do this manually—the Loader will handle it.

2. Loading Layer: Loader

The Loader performs "importing + reading metadata + determining enabled/disabled" on top of the Finder.

from ErisPulse.loaders import AdapterLoader, ModuleLoader
from ErisPulse import sdk

adapter_loader = AdapterLoader()
module_loader = ModuleLoader()

# load() internally: calls finder.find_all() → processes each entry-point → returns a triple
adapter_objs, enabled_adapters, disabled_adapters = await adapter_loader.load(sdk.adapter)
module_objs, enabled_modules, disabled_modules = await module_loader.load(sdk.module)

The triple returned by load():

Diagnostic Information on Loading Failures

When a module/adapter throws an exception during loading or initialization, the framework skips that component and continues loading others, while outputting a user code frame summary. This allows you to locate the error position at the default INFO level, without manually enabling DEBUG:

[ERROR] [ModuleLoader] Failed to load module MyModule from entry-point, skipped: 'NoneType' object has no attribute 'platform'
  → MyModule/Core.py:42 in on_load
      adapter = sdk.platform
  → AttributeError: 'NoneType' object has no attribute 'platform'
  → Hint: Increase log level to DEBUG to view full stack trace; check implementation code of module MyModule

Diagnostic information is generated through the ErisPulse.runtime.diagnostics module, which automatically filters out internal framework frames and retains only your code frames. If you need to reuse this in custom loading logic:

from ErisPulse.runtime import log_diagnostic

try:
    risky_init()
except Exception as e:
    log_diagnostic(e)  # Automatically extracts user code frames and writes to ERROR log

This module also provides two low-level functions: extract_user_frame() (returns structured frame information) and format_diagnostic_block() (returns multi-line text).

3. Registration Layer: register_to_manager

Registers the objects produced by the Loader into the manager, so that sdk.adapter / sdk.module can recognize them.

# Register adapters (returns bool indicating success)
await adapter_loader.register_to_manager(enabled_adapters, adapter_objs, sdk.adapter)

# Register modules
await module_loader.register_to_manager(enabled_modules, module_objs, sdk.module)

After registration, adapters are registered into the adapter manager, and modules are registered into the module manager, but they are not yet started/instantiated.

4. Starting Adapters

# Start all registered adapters
await sdk.adapter.startup()
# Or specify a platform
await sdk.adapter.startup("yunhu")
await sdk.adapter.startup(["yunhu", "telegram"])

Registration ≠ Starting. register_to_manager only registers; startup calls the adapter's start() method to establish a connection with the platform.

5. Initializing Modules

Modules have an additional step compared to adapters—they need to be instantiated and attached to sdk (so you can call sdk.MyModule.xxx). This step also handles module dependencies and topological sorting.

success = await module_loader.initialize_modules(
    enabled_modules, module_objs, sdk.module, sdk
)

After successful instantiation, the module appears at sdk.<ModuleName>.

6. Starting the Router Server

await sdk.router.start(
    host="0.0.0.0",
    port=8000,
    ssl_certfile=None,
    ssl_keyfile=None,
)

The router server is responsible for receiving webhook/WebSocket callbacks from adapters. Without starting it, server-mode adapters cannot receive messages.

Complete Manual Startup Example

The following code is equivalent to the core process of await sdk.init(), but exposes every step, allowing you to insert custom logic at any point:

import asyncio
from ErisPulse import sdk
from ErisPulse.loaders import AdapterLoader, ModuleLoader

async def manual_startup():
    # 0. Prepare environment (load configuration, register global exception handling)
    #    _prepare_environment is a pre-step inside init(); for manual flow, it must be called first,
    #    otherwise the Loader won't read the configuration and will incorrectly disable all adapters/modules.
    if not await sdk._prepare_environment():
        print("Environment preparation failed")
        return False

    # 1. Create loaders (each internally holds a Finder)
    adapter_loader = AdapterLoader()
    module_loader = ModuleLoader()

    # 2. Parallel discovery and loading (same as internal gather in init())
    (adapter_objs, enabled_adapters, disabled_adapters), \
    (module_objs, enabled_modules, disabled_modules) = await asyncio.gather(
        adapter_loader.load(sdk.adapter),
        module_loader.load(sdk.module),
    )

    # 3. Register adapters
    await adapter_loader.register_to_manager(
        enabled_adapters, adapter_objs, sdk.adapter
    )

    # 4. Start adapters
    if enabled_adapters:
        await sdk.adapter.startup()

    # 5. Register modules
    await module_loader.register_to_manager(
        enabled_modules, module_objs, sdk.module
    )

    # 6. Initialize modules (instantiation + attach to sdk)
    if enabled_modules:
        await module_loader.initialize_modules(
            enabled_modules, module_objs, sdk.module, sdk
        )

    # 7. Start the routing server
    await sdk.router.start(host="0.0.0.0", port=8000)

    print("Manual startup completed")
    return True

async def main():
    ok = await manual_startup()
    if ok:
        # Block to keep running (manual flow won't automatically block)
        await asyncio.Event().wait()

if __name__ == "__main__":
    asyncio.run(main())

When Should You Use Manual Startup?

In most cases, manual startup is not necessary as await sdk.run() already handles all the above steps. Manual startup is only valuable in these scenarios:

• Partial loading: Load only specific adapters/modules, skipping others
• Dynamic registration: Register new adapters/modules at runtime based on conditions
• Custom order: Need to change the default loading order (e.g., start a module before starting an adapter)
• Injection strategies: Inject custom strict-mode managers, loading strategies, etc., into the Loader
• Debugging/diagnosis: Manually drive execution at a specific step when failure occurs to locate the issue

Runtime Fine-grained Control

Even after using sdk.run() to start the SDK, you can still control individual subsystems at runtime without having to restart the entire SDK:

Hot Restart of Adapters

# Hot restart a specific adapter (to fix a connection, without affecting other platforms)
await sdk.adapter.shutdown("yunhu")
await sdk.adapter.startup("yunhu")

# Bring up a new platform during runtime
await sdk.adapter.startup("telegram")

# Temporarily take a platform offline
await sdk.adapter.shutdown("telegram")

adapter.startup() requires the adapter to be registered with the manager. Registration occurs internally within init()/run(), so this is a fine-grained control that operates after startup.

Router Server

# Temporarily take the webhook server offline
await sdk.router.stop()

# Restart it (for example, after changing the port)
await sdk.router.start(host="0.0.0.0", port=9000)

On-demand Module Loading

# Manually load a module (which may be lazily loaded)
await sdk.load_module("MyModule")

Graceful Shutdown

Starting from version 2.7.0, sdk.shutdown() provides programmatic graceful shutdown: it sets a shutdown event, allowing the main loop that is hanging with await sdk.run(keep_running=True) to return, which in turn triggers uninit() to complete resource cleanup.

# Call from any coroutine to trigger graceful exit (run() will return and uninit is automatically triggered)
sdk.shutdown()

Typical use cases:

async def shutdown_after_idle():
    await asyncio.sleep(3600)
    sdk.shutdown()  # Gracefully exit after being idle for 1 hour

Signal Handling: run() internally registers SIGTERM / SIGHUP handlers, converting system signals into graceful shutdown—when stopping services via container orchestration (e.g., Docker docker stop) or systemd, the process will complete uninit() cleanup instead of being forcefully killed.

• On Windows, loop.add_signal_handler is not supported, so signal handlers are automatically skipped (graceful shutdown can still be triggered via sdk.shutdown() or Ctrl+C)
• Repeatedly calling sdk.shutdown() is safe (subsequent calls after the event is set are no-ops)

Uninstall Process

The reverse operation of startup is await sdk.uninit(), which performs cleanup in the reverse order:

1. Shut down all adapters (adapter.shutdown())
2. Unload all modules
3. Clean up all event handlers
4. Clean up module properties on managers and the SDK

In scenarios where the process is manually started, remember to call uninit() before exiting to ensure a graceful shutdown:

try:
    await asyncio.Event().wait()   # Keep running
finally:
    await sdk.uninit()

Restart

The SDK provides two restart methods, both of which do not require you to manually uninstall first—the framework handles it automatically:

# Hot restart: Reload within the same process (most common)
await sdk.restart()

# Hard restart: Exit the process, let an external supervisor restart (see "Supervisor Guide" below)
await sdk.hard_restart()

Two points to note:

1. Both methods execute the restart in the background, immediately returning True to indicate that the "restart task has been scheduled", not that the restart is completed. The actual restart runs in the background to avoid interrupting the current event chain.
2. The principle of hard_restart() is: after unloading and saving the configuration, exit the process with exit code 42 (HARD_RESTART_EXIT_CODE) — it does not start a new process itself. It must be restarted by an external supervisor that detects exit code 42. If you run python main.py directly without any supervisor, the process exits with code 42 and does not automatically restart (the framework will issue a warning).

When to Use Hard Restart?

Hard restart is not just a "more thorough restart"—it is more suitable and even more efficient than hot restart in the following scenarios:

• Binary library (C extension) side effects: Hot restart occurs within the same process and cannot release C extensions, open file descriptors, threads, or other process-level resources. Hard restart starts a new process, completely clearing these side effects.
• Resource leak troubleshooting: When you suspect memory or handle leaks, hard restart provides a clean environment.
• Frequent restarts sensitive to performance: Hard restart avoids the overhead of unloading and reloading within the same process, making it more efficient than hot restart.

The "Framework Restart" function in the Dashboard management panel internally calls hard_restart().

Exit Code 42 Contract

Hard restart is a cross-process collaboration: SDK is responsible for exiting (code 42), and the supervisor is responsible for restarting.

sdk.is_supervised() can check if the current process was started by a supervisor (by checking the environment variable ERISPULSE_SUPERVISED). The CLI run command automatically injects this flag when starting a subprocess; external supervisors like systemd or Docker do not inject it, so is_supervised() returns False, and the framework will issue a "no supervisor detected" warning after hard restart.

Supervisor Guide

Choose a supervisor that suits your needs to make hard restart work effectively:

1. CLI run command (Development/Simple Deployment, Recommended)

epsdk run main.py includes a built-in supervision loop: it detects the child process exit code, restarts immediately if it is 42; for other abnormal exit codes, it automatically retries with exponential backoff; Ctrl+C first gracefully terminates the child process (exit code 0 is considered normal, so it will not be restarted).

epsdk run main.py

2. systemd (Linux Server)

RestartForceExitStatus=42 makes exit code 42 trigger a restart (by default on-failure only applies to non-zero codes):

[Service]
ExecStart=/usr/bin/python3 /opt/mybot/main.py
Restart=on-failure
RestartForceExitStatus=42
RestartSec=2
User=mybot

3. Docker / docker-compose

The container's PID 1 is the application process, and it exits with code 42, causing the container to exit—use the restart policy to make it automatically restart:

services:
  bot:
    build: .
    restart: unless-stopped   # Restart for any exit (including 42)

4. PM2 (Node Ecosystem Operations)

pm2 start main.py --name mybot --interpreter python3
# 42 is treated as an exit code, and PM2 restarts by default; set restart_delay to debounce
pm2 set mybot.restart_delay 2000

5. supervisord

[program:mybot]
command=python3 /opt/mybot/main.py
autorestart=true
exitcodes=0,2,42    # 42 is also considered "normal exit, restart needed"

6. Pure Python Custom Supervisor

import subprocess, sys, time

while True:
    p = subprocess.Popen([sys.executable, "main.py"])
    code = p.wait()
    if code == 42:          # Hard restart request
        time.sleep(0.5)
        continue
    if code == 0:           # Normal exit
        break
    time.sleep(3)           # Abnormal exit, retry with backoff

Behavior without a supervisor: Running directly with python main.py, calling hard_restart() causes the process to exit with code 42 and not restart. In this case, you should integrate one of the supervisors above.

Related Documentation

• Create Your First Bot - Introduction to the two basic modes of keep_running
• Lifecycle Management - Listen to startup events such as core.init.start / core.init.complete
• Lazy Loading System - Module lazy loading mechanism and load_module