Skip to content

Supervisors

The three Supervisor implementations that start/stop/restart/healthcheck plugins, plus the shared protocol.

Protocol & NullSupervisor

xcore_agent.agent.install_driver.Supervisor

Bases: Protocol

Source code in xcore_agent/agent/install_driver.py 
95
96
97
98
99
class Supervisor(Protocol):
    def start(self, plugin_id: str | None) -> None: ...
    def stop(self, plugin_id: str | None) -> None: ...
    def restart(self, plugin_id: str | None) -> None: ...
    def healthcheck(self, plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None: ...

start(plugin_id: str | None) -> None

Source code in xcore_agent/agent/install_driver.py 
96
def start(self, plugin_id: str | None) -> None: ...

stop(plugin_id: str | None) -> None

Source code in xcore_agent/agent/install_driver.py 
97
def stop(self, plugin_id: str | None) -> None: ...

restart(plugin_id: str | None) -> None

Source code in xcore_agent/agent/install_driver.py 
98
def restart(self, plugin_id: str | None) -> None: ...

healthcheck(plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None

Source code in xcore_agent/agent/install_driver.py 
99
def healthcheck(self, plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None: ...

xcore_agent.agent.install_driver.NullSupervisor

No-op supervisor for dry runs and tests.

Source code in xcore_agent/agent/install_driver.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
class NullSupervisor:
    """No-op supervisor for dry runs and tests."""

    def start(self, plugin_id: str | None) -> None:
        return None

    def stop(self, plugin_id: str | None) -> None:
        return None

    def restart(self, plugin_id: str | None) -> None:
        return None

    def healthcheck(self, plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None:
        return None

start(plugin_id: str | None) -> None

Source code in xcore_agent/agent/install_driver.py 
105
106
def start(self, plugin_id: str | None) -> None:
    return None

stop(plugin_id: str | None) -> None

Source code in xcore_agent/agent/install_driver.py 
108
109
def stop(self, plugin_id: str | None) -> None:
    return None

restart(plugin_id: str | None) -> None

Source code in xcore_agent/agent/install_driver.py 
111
112
def restart(self, plugin_id: str | None) -> None:
    return None

healthcheck(plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None

Source code in xcore_agent/agent/install_driver.py 
114
115
def healthcheck(self, plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None:
    return None

SystemdSupervisor

Backed by systemctl (optionally systemctl --user).

xcore_agent.agent.systemd_supervisor

A Supervisor (see install_driver.py) backed by systemctl — the concrete default for a project deployed directly on a VPS, as opposed to inside Docker/k8s where the client's own orchestrator plays this role (see README's "what's real vs. stubbed" table).

Expects one systemd unit per plugin, named <unit_prefix><plugin_id>.service (default prefix xcore-plugin-), plus one project_unit used for steps that don't name a specific plugin (a project-wide start/stop/restart). Provisioning those units (writing the .service files, daemon-reload) is a deployment/ops concern outside this class's scope — it only ever calls start / stop / restart / is-active on units that already exist.

SystemdCommandError

Bases: Exception

Raised when a systemctl invocation itself fails (bad unit, permission, systemd not running, ...) — distinct from a healthcheck simply reporting the unit as not active.

Source code in xcore_agent/agent/systemd_supervisor.py 
21
22
23
24
class SystemdCommandError(Exception):
    """Raised when a `systemctl` invocation itself fails (bad unit, permission,
    systemd not running, ...) — distinct from a healthcheck simply reporting
    the unit as not active."""

SystemdSupervisor dataclass

Source code in xcore_agent/agent/systemd_supervisor.py 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@dataclass
class SystemdSupervisor:
    unit_prefix: str = "xcore-plugin-"
    project_unit: str = "xcore-project.service"
    user_scope: bool = True
    healthcheck_poll_interval_seconds: float = 1.0

    def _unit(self, plugin_id: str | None) -> str:
        return f"{self.unit_prefix}{plugin_id}.service" if plugin_id else self.project_unit

    def _run(self, *args: str, timeout: float | None = None) -> subprocess.CompletedProcess:
        cmd = ["systemctl", *(["--user"] if self.user_scope else []), *args]
        try:
            return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
        except subprocess.TimeoutExpired as exc:
            raise SystemdCommandError(f"{' '.join(cmd)} timed out after {timeout}s") from exc
        except FileNotFoundError as exc:
            raise SystemdCommandError("systemctl not found on this host") from exc

    def _run_checked(self, *args: str) -> None:
        result = self._run(*args)
        if result.returncode != 0:
            raise SystemdCommandError(
                f"systemctl {' '.join(args)} failed (exit {result.returncode}): "
                f"{result.stderr.strip() or result.stdout.strip()}"
            )

    def start(self, plugin_id: str | None) -> None:
        self._run_checked("start", self._unit(plugin_id))

    def stop(self, plugin_id: str | None) -> None:
        self._run_checked("stop", self._unit(plugin_id))

    def restart(self, plugin_id: str | None) -> None:
        self._run_checked("restart", self._unit(plugin_id))

    def is_active(self, plugin_id: str | None) -> bool:
        result = self._run("is-active", self._unit(plugin_id))
        return result.stdout.strip() == "active"

    def healthcheck(self, plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None:
        unit = self._unit(plugin_id)
        last_status = "unknown"
        for attempt in range(retries + 1):
            result = self._run("is-active", unit, timeout=timeout_seconds)
            last_status = result.stdout.strip() or "unknown"
            if last_status == "active":
                return
            if attempt < retries:
                time.sleep(self.healthcheck_poll_interval_seconds)
        raise HealthcheckError(
            f"{unit} did not become active after {retries + 1} attempt(s) "
            f"(last status: {last_status!r})"
        )
unit_prefix: str = 'xcore-plugin-' class-attribute instance-attribute
project_unit: str = 'xcore-project.service' class-attribute instance-attribute
user_scope: bool = True class-attribute instance-attribute
healthcheck_poll_interval_seconds: float = 1.0 class-attribute instance-attribute
__init__(unit_prefix: str = 'xcore-plugin-', project_unit: str = 'xcore-project.service', user_scope: bool = True, healthcheck_poll_interval_seconds: float = 1.0) -> None
start(plugin_id: str | None) -> None
Source code in xcore_agent/agent/systemd_supervisor.py 
54
55
def start(self, plugin_id: str | None) -> None:
    self._run_checked("start", self._unit(plugin_id))
stop(plugin_id: str | None) -> None
Source code in xcore_agent/agent/systemd_supervisor.py 
57
58
def stop(self, plugin_id: str | None) -> None:
    self._run_checked("stop", self._unit(plugin_id))
restart(plugin_id: str | None) -> None
Source code in xcore_agent/agent/systemd_supervisor.py 
60
61
def restart(self, plugin_id: str | None) -> None:
    self._run_checked("restart", self._unit(plugin_id))
is_active(plugin_id: str | None) -> bool
Source code in xcore_agent/agent/systemd_supervisor.py 
63
64
65
def is_active(self, plugin_id: str | None) -> bool:
    result = self._run("is-active", self._unit(plugin_id))
    return result.stdout.strip() == "active"
healthcheck(plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None
Source code in xcore_agent/agent/systemd_supervisor.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def healthcheck(self, plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None:
    unit = self._unit(plugin_id)
    last_status = "unknown"
    for attempt in range(retries + 1):
        result = self._run("is-active", unit, timeout=timeout_seconds)
        last_status = result.stdout.strip() or "unknown"
        if last_status == "active":
            return
        if attempt < retries:
            time.sleep(self.healthcheck_poll_interval_seconds)
    raise HealthcheckError(
        f"{unit} did not become active after {retries + 1} attempt(s) "
        f"(last status: {last_status!r})"
    )

DockerSupervisor

Backed by the docker CLI.

xcore_agent.agent.docker_supervisor

A Supervisor (see install_driver.py) backed by the docker CLI — an alternative to SystemdSupervisor for projects deployed as containers instead of directly on the host.

Expects one container per plugin, named <container_prefix><plugin_id> (default prefix xcore-plugin-), plus one project_container used for steps that don't name a specific plugin. Creating/updating those containers (image, env, volumes, docker run vs docker compose, ...) is a deployment/ops concern outside this class's scope — it only ever calls start / stop / restart / inspect on containers that already exist.

DockerCommandError

Bases: Exception

Raised when a docker invocation itself fails (unknown container, daemon not running, permission denied, ...) — distinct from a healthcheck simply reporting the container as not running.

Source code in xcore_agent/agent/docker_supervisor.py 
20
21
22
23
class DockerCommandError(Exception):
    """Raised when a `docker` invocation itself fails (unknown container,
    daemon not running, permission denied, ...) — distinct from a
    healthcheck simply reporting the container as not running."""

DockerSupervisor dataclass

Source code in xcore_agent/agent/docker_supervisor.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@dataclass
class DockerSupervisor:
    container_prefix: str = "xcore-plugin-"
    project_container: str = "xcore-project"
    healthcheck_poll_interval_seconds: float = 1.0

    def _container(self, plugin_id: str | None) -> str:
        return f"{self.container_prefix}{plugin_id}" if plugin_id else self.project_container

    def _run(self, *args: str, timeout: float | None = None) -> subprocess.CompletedProcess:
        cmd = ["docker", *args]
        try:
            return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
        except subprocess.TimeoutExpired as exc:
            raise DockerCommandError(f"{' '.join(cmd)} timed out after {timeout}s") from exc
        except FileNotFoundError as exc:
            raise DockerCommandError("docker not found on this host") from exc

    def _run_checked(self, *args: str) -> None:
        result = self._run(*args)
        if result.returncode != 0:
            raise DockerCommandError(
                f"docker {' '.join(args)} failed (exit {result.returncode}): "
                f"{result.stderr.strip() or result.stdout.strip()}"
            )

    def start(self, plugin_id: str | None) -> None:
        self._run_checked("start", self._container(plugin_id))

    def stop(self, plugin_id: str | None) -> None:
        self._run_checked("stop", self._container(plugin_id))

    def restart(self, plugin_id: str | None) -> None:
        self._run_checked("restart", self._container(plugin_id))

    def is_running(self, plugin_id: str | None) -> bool:
        result = self._run("inspect", "--format", "{{.State.Running}}", self._container(plugin_id))
        return result.returncode == 0 and result.stdout.strip() == "true"

    def healthcheck(self, plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None:
        container = self._container(plugin_id)
        last_status = "unknown"
        for attempt in range(retries + 1):
            result = self._run(
                "inspect", "--format", "{{.State.Status}}", container, timeout=timeout_seconds
            )
            last_status = result.stdout.strip() or "unknown"
            if result.returncode == 0 and last_status == "running":
                return
            if attempt < retries:
                time.sleep(self.healthcheck_poll_interval_seconds)
        raise HealthcheckError(
            f"container {container!r} did not become running after {retries + 1} "
            f"attempt(s) (last status: {last_status!r})"
        )
container_prefix: str = 'xcore-plugin-' class-attribute instance-attribute
project_container: str = 'xcore-project' class-attribute instance-attribute
healthcheck_poll_interval_seconds: float = 1.0 class-attribute instance-attribute
__init__(container_prefix: str = 'xcore-plugin-', project_container: str = 'xcore-project', healthcheck_poll_interval_seconds: float = 1.0) -> None
start(plugin_id: str | None) -> None
Source code in xcore_agent/agent/docker_supervisor.py 
52
53
def start(self, plugin_id: str | None) -> None:
    self._run_checked("start", self._container(plugin_id))
stop(plugin_id: str | None) -> None
Source code in xcore_agent/agent/docker_supervisor.py 
55
56
def stop(self, plugin_id: str | None) -> None:
    self._run_checked("stop", self._container(plugin_id))
restart(plugin_id: str | None) -> None
Source code in xcore_agent/agent/docker_supervisor.py 
58
59
def restart(self, plugin_id: str | None) -> None:
    self._run_checked("restart", self._container(plugin_id))
is_running(plugin_id: str | None) -> bool
Source code in xcore_agent/agent/docker_supervisor.py 
61
62
63
def is_running(self, plugin_id: str | None) -> bool:
    result = self._run("inspect", "--format", "{{.State.Running}}", self._container(plugin_id))
    return result.returncode == 0 and result.stdout.strip() == "true"
healthcheck(plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None
Source code in xcore_agent/agent/docker_supervisor.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
def healthcheck(self, plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None:
    container = self._container(plugin_id)
    last_status = "unknown"
    for attempt in range(retries + 1):
        result = self._run(
            "inspect", "--format", "{{.State.Status}}", container, timeout=timeout_seconds
        )
        last_status = result.stdout.strip() or "unknown"
        if result.returncode == 0 and last_status == "running":
            return
        if attempt < retries:
            time.sleep(self.healthcheck_poll_interval_seconds)
    raise HealthcheckError(
        f"container {container!r} did not become running after {retries + 1} "
        f"attempt(s) (last status: {last_status!r})"
    )

KubernetesSupervisor

Backed by the kubectl CLI — scale/rollout-restart/rollout-status, one Deployment per plugin (xcore-plugin-<id> by default).

xcore_agent.agent.kubernetes_supervisor

A Supervisor (see install_driver.py) backed by the kubectl CLI — an alternative to SystemdSupervisor/DockerSupervisor for projects deployed onto a Kubernetes cluster. Same shell-out-to-CLI shape as DockerSupervisor (no kubernetes Python client dependency), because the agent otherwise has no way to assume kubeconfig/cluster access is even present.

Expects one Deployment per plugin, named <deployment_prefix><plugin_id> (default prefix xcore-plugin-) in namespace, plus one project_deployment for steps that don't name a specific plugin. Creating those Deployments (image, env, resources, ...) is a deployment/ops concern outside this class's scope — it only ever scales, restarts, and checks the rollout status of Deployments that already exist.

Kubernetes has no direct "start/stop a container" verb the way docker start/docker stop do; the equivalent for a Deployment is scaling replicas to 1 or 0, and "restart" is kubectl rollout restart, whose completion is observed via kubectl rollout status — which doubles as the healthcheck.

KubectlCommandError

Bases: Exception

Raised when a kubectl invocation itself fails (unknown deployment, cluster unreachable, permission denied, ...) — distinct from a healthcheck simply reporting the rollout as not yet complete.

Source code in xcore_agent/agent/kubernetes_supervisor.py 
27
28
29
30
class KubectlCommandError(Exception):
    """Raised when a `kubectl` invocation itself fails (unknown deployment,
    cluster unreachable, permission denied, ...) — distinct from a
    healthcheck simply reporting the rollout as not yet complete."""

KubernetesSupervisor dataclass

Source code in xcore_agent/agent/kubernetes_supervisor.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@dataclass
class KubernetesSupervisor:
    namespace: str = "default"
    deployment_prefix: str = "xcore-plugin-"
    project_deployment: str = "xcore-project"
    kubeconfig: str | None = None
    context: str | None = None
    healthcheck_poll_interval_seconds: float = 1.0

    def _deployment(self, plugin_id: str | None) -> str:
        return f"{self.deployment_prefix}{plugin_id}" if plugin_id else self.project_deployment

    def _base_args(self) -> list[str]:
        args = ["--namespace", self.namespace]
        if self.kubeconfig:
            args += ["--kubeconfig", self.kubeconfig]
        if self.context:
            args += ["--context", self.context]
        return args

    def _run(self, *args: str, timeout: float | None = None) -> subprocess.CompletedProcess:
        cmd = ["kubectl", *self._base_args(), *args]
        try:
            return subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
        except subprocess.TimeoutExpired as exc:
            raise KubectlCommandError(f"{' '.join(cmd)} timed out after {timeout}s") from exc
        except FileNotFoundError as exc:
            raise KubectlCommandError("kubectl not found on this host") from exc

    def _run_checked(self, *args: str) -> None:
        result = self._run(*args)
        if result.returncode != 0:
            raise KubectlCommandError(
                f"kubectl {' '.join(args)} failed (exit {result.returncode}): "
                f"{result.stderr.strip() or result.stdout.strip()}"
            )

    def start(self, plugin_id: str | None) -> None:
        self._run_checked("scale", f"deployment/{self._deployment(plugin_id)}", "--replicas=1")

    def stop(self, plugin_id: str | None) -> None:
        self._run_checked("scale", f"deployment/{self._deployment(plugin_id)}", "--replicas=0")

    def restart(self, plugin_id: str | None) -> None:
        self._run_checked("rollout", "restart", f"deployment/{self._deployment(plugin_id)}")

    def is_running(self, plugin_id: str | None) -> bool:
        result = self._run(
            "get",
            f"deployment/{self._deployment(plugin_id)}",
            "-o",
            "jsonpath={.status.readyReplicas}",
        )
        if result.returncode != 0:
            return False
        ready = result.stdout.strip()
        return ready.isdigit() and int(ready) > 0

    def healthcheck(self, plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None:
        deployment = self._deployment(plugin_id)
        last_error = ""
        for attempt in range(retries + 1):
            result = self._run(
                "rollout",
                "status",
                f"deployment/{deployment}",
                f"--timeout={timeout_seconds}s",
                timeout=timeout_seconds + 5,
            )
            if result.returncode == 0:
                return
            last_error = result.stderr.strip() or result.stdout.strip() or "unknown error"
            if attempt < retries:
                time.sleep(self.healthcheck_poll_interval_seconds)
        raise HealthcheckError(
            f"deployment {deployment!r} did not become ready after {retries + 1} "
            f"attempt(s) (last error: {last_error!r})"
        )
namespace: str = 'default' class-attribute instance-attribute
deployment_prefix: str = 'xcore-plugin-' class-attribute instance-attribute
project_deployment: str = 'xcore-project' class-attribute instance-attribute
kubeconfig: str | None = None class-attribute instance-attribute
context: str | None = None class-attribute instance-attribute
healthcheck_poll_interval_seconds: float = 1.0 class-attribute instance-attribute
__init__(namespace: str = 'default', deployment_prefix: str = 'xcore-plugin-', project_deployment: str = 'xcore-project', kubeconfig: str | None = None, context: str | None = None, healthcheck_poll_interval_seconds: float = 1.0) -> None
start(plugin_id: str | None) -> None
Source code in xcore_agent/agent/kubernetes_supervisor.py 
70
71
def start(self, plugin_id: str | None) -> None:
    self._run_checked("scale", f"deployment/{self._deployment(plugin_id)}", "--replicas=1")
stop(plugin_id: str | None) -> None
Source code in xcore_agent/agent/kubernetes_supervisor.py 
73
74
def stop(self, plugin_id: str | None) -> None:
    self._run_checked("scale", f"deployment/{self._deployment(plugin_id)}", "--replicas=0")
restart(plugin_id: str | None) -> None
Source code in xcore_agent/agent/kubernetes_supervisor.py 
76
77
def restart(self, plugin_id: str | None) -> None:
    self._run_checked("rollout", "restart", f"deployment/{self._deployment(plugin_id)}")
is_running(plugin_id: str | None) -> bool
Source code in xcore_agent/agent/kubernetes_supervisor.py 
79
80
81
82
83
84
85
86
87
88
89
def is_running(self, plugin_id: str | None) -> bool:
    result = self._run(
        "get",
        f"deployment/{self._deployment(plugin_id)}",
        "-o",
        "jsonpath={.status.readyReplicas}",
    )
    if result.returncode != 0:
        return False
    ready = result.stdout.strip()
    return ready.isdigit() and int(ready) > 0
healthcheck(plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None
Source code in xcore_agent/agent/kubernetes_supervisor.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def healthcheck(self, plugin_id: str | None, *, timeout_seconds: int, retries: int) -> None:
    deployment = self._deployment(plugin_id)
    last_error = ""
    for attempt in range(retries + 1):
        result = self._run(
            "rollout",
            "status",
            f"deployment/{deployment}",
            f"--timeout={timeout_seconds}s",
            timeout=timeout_seconds + 5,
        )
        if result.returncode == 0:
            return
        last_error = result.stderr.strip() or result.stdout.strip() or "unknown error"
        if attempt < retries:
            time.sleep(self.healthcheck_poll_interval_seconds)
    raise HealthcheckError(
        f"deployment {deployment!r} did not become ready after {retries + 1} "
        f"attempt(s) (last error: {last_error!r})"
    )