hot_reload — PluginHotReloader

Overview

A standalone background file-watcher that triggers plugin hot-reload through the existing PluginLoader, independent of PluginManager's own built-in polling loop (see the note in plugin_manager). Runs in a background thread (not an asyncio task), so it never blocks the main event loop, but it does call back into async PluginLoader methods using fresh asyncio.run() calls per reload.

Availability

plugins/__init__.py imports PluginHotReloader inside a try/except ImportError block, so it is exposed as plugins.PluginHotReloader only if the import succeeds — but note the class itself has no hard dependency; the only optional dependency is watchdog, used only within _run_watchdog / _make_watchdog_handler, both already guarded separately. PluginHotReloader is not listed in __all__.

Backends

_try_import_watchdog() attempts from watchdog.observers import Observer and the relevant event classes inside a try/except ImportError, returning None on failure. start() checks this once and picks the backend automatically — you don't choose it explicitly.

PluginHotReloader

PluginHotReloader(
    loader: Any,                          # a PluginLoader instance
    manager: Optional[Any] = None,         # a PluginManager instance (optional)
    directories: Optional[List[str]] = None,
    poll_interval: float = 1.0,
    debounce_sec: float = 0.5,
)

Constructor behavior

If directories is not given, falls back to [str(d) for d in loader.plugin_dirs]. Converts to Path objects. Initializes a threading.Event (_stop_event), a threading.Lock, an empty thread handle, debounce tracking dict, and empty reload/error callback lists. Public attributes backend ("unknown" until start()), is_running (False), reload_count (0) are set up front.

Callback registration

on_reload(fn: Callable) -> Callable

Registers fn(plugin_names: list[str], file_path: str) -> None, called after each successful reload. Returns fn unchanged, so it can be used as a decorator: @reloader.on_reload.

on_error(fn: Callable) -> Callable

Registers fn(exc: Exception, file_path: str) -> None, called when a reload raises. Also usable as a decorator.

_trigger_reload(file_path: Path) -> None (private)

Internal workflow:

1. Debounce check: if this file was reloaded less than debounce_sec seconds ago, return immediately (thread-safe via self._lock).
2. Calls asyncio.run(self._loader.reload_file(file_path)) — this creates and tears down a new event loop for every single reload, since asyncio.run() cannot be called from within an already-running loop and this method runs on a plain background thread with no loop of its own.
3. If plugins were reloaded and a manager was supplied, calls asyncio.run(self._manager.initialize_plugin(name)) for each — note: this calls a method named initialize_plugin on the manager, but PluginManager's actual method (per plugin_manager.py) is the private _initialize_plugin. Calling the public name as written here would raise AttributeError unless the manager object you pass in defines a public initialize_plugin method yourself. This is a direct observation from the source, not a guess — if you wire a real PluginManager in here, verify this call path before relying on it in production.
4. Increments reload_count, logs, and fires every registered on_reload callback (each wrapped in its own try/except so one bad callback doesn't stop the others).
5. On any exception from steps 2–3: logs an error and fires every registered on_error callback (also individually guarded).

Watchdog backend

_run_watchdog(self) -> None (private)

Builds a watchdog.events.FileSystemEventHandler subclass on the fly (via the module-level _make_watchdog_handler function) whose on_modified/on_created both call _trigger_reload for any non-directory path ending in .py. Schedules a non-recursive watch (recursive=False) on each directory that exists, starts the Observer, then blocks on self._stop_event.wait() until stop() is called, after which it stops and joins the observer.

Polling backend

_run_polling(self) -> None (private)

Loops with self._stop_event.wait(self._poll_interval) as the sleep/exit mechanism (returns True and exits the loop as soon as stop() sets the event, so shutdown latency is bounded by one poll interval, not a fixed sleep). Each iteration calls loader.get_changed_files() and triggers a reload for each. Catches and logs any exception per-iteration.

Public API

start() -> PluginHotReloader

No-op with a warning if already running. Otherwise clears the stop event, picks a backend (see table above), spawns a daemon thread running the chosen backend method, sets is_running = True, logs, and returns self (enabling reloader = PluginHotReloader(...).start()).

stop(timeout: float = 3.0) -> None

No-op if not running. Sets the stop event and joins the background thread with the given timeout. Logs the final reload_count.

Note: because the thread is a daemon thread, the process can still exit even if stop() is never called — but any reload in progress at process exit could be interrupted mid-way since daemon threads are killed abruptly on interpreter shutdown.

reload_all() -> List[str]

Synchronous, immediate reload of every .py file (excluding those starting with _) in every watched directory that currently exists, via asyncio.run(loader.reload_file(fp)) per file. Aggregates and returns all reloaded plugin names. Does not go through the debounce mechanism (that's only in _trigger_reload). Any per-file exception is caught, logged, and skipped rather than aborting the whole sweep.

status() -> dict

Returns is_running, backend, directories (as strings), reload_count, poll_interval.

__repr__

f"PluginHotReloader(backend={self.backend!r}, running={self.is_running}, reloads={self.reload_count})"

Context manager support

def __enter__(self) -> "PluginHotReloader":
    return self.start()

def __exit__(self, *_: Any) -> None:
    self.stop()

Enables with PluginHotReloader(loader) as reloader: ... — starts on entry, stops on exit regardless of how the block ends.

Thread safety

The class is explicitly designed to run its watcher on a background thread while the rest of your application (including PluginLoader/PluginRegistry) runs on the main thread/event loop. Because _trigger_reload calls asyncio.run() fresh each time from the background thread, and PluginLoader/PluginRegistry were not documented as thread-safe against concurrent access (see notes in loader and registry), using PluginHotReloader concurrently with active plugin execution on the main event loop carries a real risk of races on shared registry/loader state — the source does not add any cross-thread locking around the registry itself to prevent this.

Side effects

Spawns a background daemon thread; performs file I/O (globbing, stat/watch); dynamically re-imports .py files (arbitrary code execution) via the loader; mutates sys.modules and the target registry (through the loader); structured logging.

Examples

1. Basic usage:

from zerda.plugins import PluginLoader, PluginRegistry
from zerda.plugins.hot_reload import PluginHotReloader

registry = PluginRegistry()
loader = PluginLoader(registry, plugin_dirs=["./plugins/custom"])

reloader = PluginHotReloader(loader=loader, directories=["./plugins/custom"], poll_interval=1.0)
reloader.start()
# ... application runs ...
reloader.stop()

2. Callback-based observation:

@reloader.on_reload
def log_reload(plugin_names, file_path):
    print(f"Reloaded: {plugin_names} from {file_path}")

@reloader.on_error
def log_error(exc, file_path):
    print(f"Reload error in {file_path}: {exc}")

3. Context manager:

with PluginHotReloader(loader) as reloader:
    run_app()
# stop() called automatically

4. Manual, immediate reload of everything:

names = reloader.reload_all()
print(f"Reloaded {len(names)} plugins")

5. Checking backend and status:

reloader.start()
print(reloader.status())
# {"is_running": True, "backend": "watchdog" | "polling", "directories": [...], ...}

Best practices

• Install watchdog in any environment where reload latency matters — the polling fallback only detects changes once per poll_interval.
• Always pair start() with stop() (or use the context manager) to ensure the background thread joins cleanly.
• Given the thread-safety caveat above, prefer enabling this only in development, or ensure your application doesn't execute plugins concurrently with reload windows in production.

Common mistakes

• Passing a manager and expecting initialize_plugin(name) to work against a stock PluginManager — as noted above, verify this attribute exists on the object you pass, since PluginManager itself exposes the equivalent logic as the private _initialize_plugin.
• Assuming reload_all() respects the debounce window — it doesn't; it always reloads every matching file immediately.

Related APIs

• PluginLoader.reload_file / get_changed_files — what this class calls
• PluginManager.enable_hot_reload — the separate, simpler built-in polling alternative