plugin_manager — PluginManager, AIPluginSelector, PluginObservability

Overview

PluginManager is the central façade of the framework. It owns a PluginRegistry, PluginLoader, PermissionGuard, ExecutionSandbox, AIPluginSelector, and PluginObservability instance, and exposes lifecycle (startup/shutdown), execution (execute/execute_batch/execute_pipeline), AI-selection (select/select_one/auto_execute), a hook system (on/off), hot-reload control (enable_hot_reload/disable_hot_reload), and observability (dashboard/health_check_all).

PluginObservability

metrics = PluginObservability()

Lightweight in-process metrics store — no persistence, resets when the process restarts.

record_exec(name, elapsed, success, error="") -> None

Appends a record {ts, elapsed, success, error} to a per-plugin list, keeping only the last 200 entries per plugin (oldest dropped via .pop(0)). On failure, also appends to a global error list capped at 500 entries.

record_hook(event) -> None

Increments a counter for the named event.

summary() -> Dict[str, Any]

Returns total_executions, total_failures, error_rate_pct (rounded to 1 decimal, 0 if no executions), top_plugins (top 5 by execution count), hook_call_counts, and recent_errors (last 10).

plugin_report(name) -> Dict[str, Any]

Returns {"name": name, "executions": 0} if no records exist for that plugin. Otherwise returns executions, success_rate (%, 1 decimal), and avg_ms/max_ms/min_ms computed from the recorded elapsed values (converted to milliseconds).

Complexity: O(1) amortized per record_exec call (capped list); O(n) for summary()/plugin_report() in the number of stored records for the plugin(s) involved.

AIPluginSelector

selector = AIPluginSelector(registry, embed_fn=None)

Rule-based + optional keyword-semantic plugin selector. Per its docstring: "For full semantic search, inject an embed_fn (str → List[float]). Without it, falls back to keyword scoring from the registry."

select(query, plugin_type=None, max_cost=None, top_k=3) -> List[BasePlugin]

Internal workflow:

1. Candidate pool = registry.list_by_type(plugin_type) if given, else registry.list_enabled().
2. If max_cost is given, filter candidates to those whose cost_tier rank (by CostTier enum iteration order: FREE, CHEAP, MODERATE, EXPENSIVE) is <= the rank of max_cost.
3. If no candidates remain, return [] immediately.
4. Score every candidate with the same _keyword_score function used by PluginRegistry.search().
5. If embed_fn was provided, call _semantic_rerank() to blend in cosine similarity.
6. Sort descending by final score, return the top top_k.

select_one(query, **kwargs) -> Optional[BasePlugin]

Calls select(query, top_k=1, **kwargs) and returns the first result or None.

_semantic_rerank(query, scored) -> List[Tuple[float, BasePlugin]] (private)

Embeds the query once via embed_fn(query). For each candidate, builds a document string from description + use_cases + capabilities, embeds it, computes cosine similarity via the module-private _cosine() helper, and combines: final = kw_score * 0.4 + cosine_sim * 0.6. If embed_fn raises any exception, logs a warning and falls back to the original keyword-only scores rather than propagating the error.

_cosine(a, b) -> float (module-private function)

Standard cosine similarity: dot(a,b) / (||a|| * ||b|| + 1e-9). The 1e-9 epsilon guards against division by zero for a zero vector.

PluginManager

PluginManager(
    plugin_dirs: Optional[List[str]] = None,
    system_version: str = "1.0.0",
    safe_mode: bool = False,
    auto_discover: bool = True,
    default_config: Optional[PluginConfig] = None,
    embed_fn: Optional[Callable[[str], List[float]]] = None,
    hard_timeout: float = 60.0,
)

Constructor parameters

Also constructs empty structures: self._hooks (a defaultdict(list) of (priority, callback) tuples), self._hot_reload_task = None, self._hot_reload_on = False.

Lifecycle

startup() -> None

1. If auto_discover: calls loader.discover_all(). If not safe_mode, calls security.grant_all_required(plugin) for every discovered plugin.
2. For every plugin currently in the registry (including any manually registered before startup() was called), calls the private _initialize_plugin(plugin).
3. Fires the "system.startup" hook with manager=self.

shutdown() -> None

Sets self._shutting_down = True, calls disable_hot_reload(), fires "system.shutdown", then calls await plugin.cleanup() for every registered plugin (catching and logging any exception per-plugin rather than aborting the whole shutdown), setting each to PluginStatus.DISABLED on success.

Manual registration

register(plugin: BasePlugin) -> bool

Workflow (each step can short-circuit and return False):

1. Compatibility: plugin.metadata.is_compatible_with(self.system_version).
2. Dependencies: every name in plugin.metadata.dependencies must already registry.exists().
3. If safe_mode: security.check_all_required(plugin) must pass.
4. registry.register(plugin) — catches KeyError (already registered), logs, returns False.
5. _initialize_plugin(plugin); if it returns True, fires "plugin.registered" hook.

Returns True only if every step succeeds and initialization succeeds.

unregister(name: str, force: bool = False) -> bool

Returns False (with a warning) if name isn't registered. Unless force=True, checks whether any other registered plugin lists name in its own metadata.dependencies; if so, refuses and returns False. Otherwise calls plugin.cleanup(), registry.unregister(name), fires "plugin.unregistered", returns True. Any exception during cleanup/unregister is caught, logged, and results in False.

Execution

execute(name, input_data, context=None, use_retry=False) -> Any

Raises PluginError("PluginManager is shutting down.") immediately if called after shutdown() has begun. Internal workflow:

1. plugin = registry.get_or_raise(name) — raises KeyError if not found.
2. ctx = context or ExecutionContext().
3. Raises PluginError if not plugin.config.enabled.
4. If safe_mode: security.check_all_required(plugin, ctx) — note this only checks and logs, it does not itself raise; a failed check here does not currently block execution based on the source (unlike assert_permission, which does raise). Verify this against your specific build if permission enforcement at execute-time is critical for your safe-mode deployment.
5. If the plugin declares input_schema: sanitize via InputSanitizer.sanitize(..., strict=False).
6. Calls plugin.safe_execute_with_retry(...) if use_retry=True, else plugin.safe_execute(...).
7. On success: records the execution in self.metrics, fires "plugin.executed", returns the result.
8. On PluginError/PluginTimeoutError: records the failure, fires "plugin.failed", re-raises.

Raises: KeyError (plugin not found), PluginError (disabled / shutting down / wrapped internal error), PluginTimeoutError, PluginValidationError (from sanitization or plugin validate()).

execute_batch(calls, context=None, parallel=True) -> Dict[str, Any]

calls is a list of (plugin_name, input_data) tuples. In parallel=True mode (default), builds one self.execute(...) coroutine per call and runs them with asyncio.gather(*tasks, return_exceptions=True); any exception is captured into {"error": str(exception)} in the corresponding slot of the result dict rather than aborting the whole batch. In parallel=False mode, executes sequentially with a try/except around each call, same error shape.

Note: if calls contains the same plugin name twice, the returned dict — keyed by plugin name — will only retain the result of the last call for that name, since both write to the same dict key.

execute_pipeline(pipeline, initial, context=None) -> Any

Sequentially executes each plugin name in pipeline, feeding the previous plugin's return value as the next plugin's input_data (coerced to {} if the previous output wasn't a dict). Returns the final plugin's output.

AI agent helpers

select(query, plugin_type=None, max_cost=None, top_k=3) -> List[BasePlugin]

Thin passthrough to self.selector.select(...).

select_one(query, **kwargs) -> Optional[BasePlugin]

auto_execute(query, input_data=None, context=None, **selector_kwargs) -> Optional[Any]

Calls select_one(query, **selector_kwargs); if no plugin is found, logs a warning (truncating the query to 80 chars in the log message) and returns None. Otherwise executes it with input_data or {"query": query}.

tool_descriptors() -> List[Dict[str, Any]] / anthropic_tools() -> List[Dict[str, Any]]

Passthroughs to registry.to_tool_descriptors() / registry.to_anthropic_tools().

Hot reload

enable_hot_reload(interval: float = 5.0) -> None

No-op if already enabled. Otherwise starts a background asyncio.Task running _hot_reload_loop.

disable_hot_reload() -> None

Sets the flag off, cancels the task, and awaits it (swallowing asyncio.CancelledError).

_hot_reload_loop(interval) (private coroutine)

Loops while enabled and not shutting down: sleeps interval seconds, checks loader.get_changed_files(), calls loader.reload_file(path) for each, then re-initializes any plugin currently in PluginStatus.DISABLED (this is how newly-added or newly-reloaded plugins get initialized). Catches and logs any exception per-iteration rather than crashing the loop, except asyncio.CancelledError, which breaks out cleanly.

[!NOTE] Relationship to PluginHotReloader This is a separate, simpler hot-reload mechanism built directly into PluginManager — a polling loop only, with no watchdog backend. The standalone PluginHotReloader class in hot_reload.py is a different, more feature-rich watcher (with a watchdog backend, debouncing, and its own callback system) that you can optionally wire up separately; the two are not connected to each other in the source.

Hook system

on(event: str, callback: Callable, priority: int = 50) -> None

Raises TypeError if callback isn't callable. Appends (priority, callback) and re-sorts the event's callback list descending by priority after every registration (so higher-priority callbacks run first).

off(event: str, callback: Callable) -> None

Removes all (priority, cb) tuples where cb is callback (identity comparison).

_fire_hook(event, **kwargs) -> None (private)

Records the hook call in self.metrics, then calls each registered callback in priority order, awaiting it if inspect.iscoroutinefunction(cb) else calling it synchronously. Any exception from a callback is caught and logged — one misbehaving hook cannot break the manager's control flow.

Recognized events fired internally: system.startup, system.shutdown, plugin.registered, plugin.unregistered, plugin.executed, plugin.failed.

Observability

dashboard() -> Dict[str, Any]

Returns system_version, safe_mode, hot_reload (bool), total_plugins, enabled_plugins, a plugins list (each plugin's get_stats() merged with {"health": get_health()}), and metrics (self.metrics.summary()).

health_check_all() -> Dict[str, Dict[str, Any]]

{plugin.name: plugin.get_health() for plugin in registry.all()}.

Convenience accessors

get(name), list_plugins(plugin_type=None) (a simplified projection: name, version, type, enabled, status, capabilities, tags, cost_tier), grant_permission(plugin_name, permission), revoke_permission(plugin_name, permission) — thin passthroughs to security.grant/security.revoke.

_initialize_plugin(plugin) -> bool (private)

Sets status = INITIALIZING, calls await plugin.initialize(). On True: sets status = ENABLED, stamps initialized_at (via a dynamic __import__("datetime") call rather than a module-level import — functionally equivalent but stylistically unusual), returns True. On False: sets status = ERROR, error_msg = "initialize() returned False", returns False. On any exception: sets status = ERROR, error_msg = str(exc), logs, returns False (the exception itself is not re-raised — callers see it reflected only in status/error_msg).

__repr__

f"PluginManager(plugins={len(self.registry)}, system={self.system_version}, safe_mode={self.safe_mode})"

Examples

1. Basic startup and execution:

manager = PluginManager(plugin_dirs=["./plugins"])
await manager.startup()
result = await manager.execute("web_search", {"query": "RAG"}, ExecutionContext())
await manager.shutdown()

2. Batch, parallel:

results = await manager.execute_batch([
    ("web_search", {"query": "RAG papers"}),
    ("calculator", {"expression": "2+2"}),
])

3. Pipeline chaining retrieval into reranking:

final = await manager.execute_pipeline(
    pipeline=["web_search", "reranker"],
    initial={"query": "RAG"},
)

4. Error handling — plugin not found:

try:
    await manager.execute("does_not_exist", {})
except KeyError as e:
    print(e)

5. Production usage with hooks and hot reload:

manager = PluginManager(plugin_dirs=["./plugins/prod"], safe_mode=True)

async def on_failure(plugin, error):
    logging.getLogger("alerts").error("Plugin %s failed: %s", plugin.name, error)

manager.on("plugin.failed", on_failure, priority=100)
await manager.startup()
await manager.enable_hot_reload(interval=10.0)

Best practices

• Set safe_mode=True in production if you want permission checks enforced on register(); be aware of the execute()-time permission-check note above when relying on safe_mode to block unauthorized calls at execution time.
• Register hooks before calling startup() if you want to observe the "system.startup" event itself.
• Use execute_batch(parallel=True) for independent plugin calls and execute_pipeline only when each step genuinely depends on the previous step's output.

Common mistakes

• Calling manager.execute() after manager.shutdown() — this raises PluginError, not KeyError or a silent no-op.
• Assuming execute_batch preserves multiple results for a repeated plugin name — only the last call's result survives in the returned dict.
• Confusing PluginManager.enable_hot_reload() (built-in polling loop) with PluginHotReloader (the separate class in hot_reload.py) — they are independent mechanisms.

Related APIs

• PluginRegistry, PluginLoader, PermissionGuard, ExecutionSandbox — the subsystems this class composes
• PluginHotReloader — alternative, more feature-rich hot-reload
• BasePlugin — what gets executed