plugin_interface

Domain-specific plugin interface for system monitoring

MonitorPlugin

Abstract base class for hardware monitoring plugins. Implementations provide platform-specific logic to gather system statistics.

Planned Implementations:

Plugin	Library	Platform
cjm-system-monitor-nvidia	nvitop / pynvml	NVIDIA GPUs
cjm-system-monitor-amd	rocm-smi	AMD GPUs
cjm-system-monitor-cpu	psutil	CPU-only systems

Integration with PluginManager:

# Register as system monitor for resource scheduling
manager.register_system_monitor("sys-mon-nvidia")

# Scheduler will call execute() to get fresh stats before each plugin execution

---

MonitorPlugin

def MonitorPlugin(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Abstract base class for hardware monitoring plugins.

CR-3 shifted MonitorPlugin from dispatcher-style execute(command=...) to a typed surface: subclasses override get_system_status() returning SystemStats and optionally list_processes() returning List[ProcessStats]. The legacy execute(command=...) dispatcher is kept as a backward-compat shim so monitors that predate CR-3 keep working until the SG-47 migration cascade.

Subclasses MUST override at least one of execute() or get_system_status() — the __init_subclass__ guard enforces this at class-definition time to prevent the recursion trap where both defaults call each other.

Implementation Guide

CR-3 supports two implementation patterns for backward compat. New monitor plugins should use the typed pattern; legacy plugins predating CR-3 continue to work via the dispatcher pattern.

Typed pattern (CR-3 onward — preferred):

from cjm_infra_plugin_system.core import ProcessStats, SystemStats
from cjm_infra_plugin_system.plugin_interface import MonitorPlugin

class NvidiaMonitorPlugin(MonitorPlugin):
    @property
    def name(self) -> str:
        return "sys-mon-nvidia"
    
    @property
    def version(self) -> str:
        return "1.0.0"
    
    def get_system_status(self) -> SystemStats:
        # Gather stats using nvitop/pynvml
        return SystemStats(
            gpu_type="NVIDIA",
            gpu_free_memory_mb=...,
            # ... other fields
        )
    
    def list_processes(self) -> list[ProcessStats]:
        # Enumerate per-process GPU usage
        return [
            ProcessStats(pid=p.pid, gpu_index=p.device.index, gpu_memory_mb=p.gpu_memory_used_mb, command=p.command)
            for p in nvitop.processes()
        ]

The inherited execute() dispatcher routes the legacy command="get_system_status"/"list_processes" strings to the typed methods automatically — no need to implement it directly.

Dispatcher pattern (REMOVE-AFTER-OVERHAUL — legacy):

class OldMonitorPlugin(MonitorPlugin):
    def execute(self, command: str = "get_system_status", **kwargs) -> dict:
        if command == "get_system_status":
            stats = SystemStats(gpu_type="NVIDIA", ...)
            return stats.to_dict()
        raise ValueError(f"Unknown command: {command}")

The inherited get_system_status() default delegates to execute() and wraps the returned dict back into SystemStats — substrate consumers calling typed methods still work. SG-47 cascade will migrate plugins from this pattern to the typed pattern.

Recursion-guard requirement: every concrete subclass must override at least one of execute() or get_system_status(). The __init_subclass__ hook on MonitorPlugin raises TypeError at class-definition time if neither is overridden, preventing the infinite-loop trap where both defaults call each other.

CR-3 tests

Verify the typed surface, the dispatcher-fallback semantics, and the __init_subclass__ recursion guard against bare subclasses.