Skip to content

Install driver

xcore_agent.agent.install_driver — filesystem install/snapshot/rollback, plus the Supervisor and Provisioner protocols.

xcore_agent.agent.install_driver

Executes individual install.yaml steps against a target host.

Filesystem operations (installing/configuring plugins, writing env files) are implemented for real against a local directory layout. Process supervision (start/stop/restart/healthcheck) is intentionally left as a Supervisor protocol: which init system or orchestrator runs the actual processes (systemd, Docker, Dockploy, k8s...) is the client's infrastructure choice, not something xcore-agent should hardcode. NullSupervisor is provided for tests and dry runs.

logger = logging.getLogger(__name__) module-attribute

Provisioner = Callable[[ProvisionStep], None] module-attribute

Notifier = Callable[[NotifyStep], None] module-attribute

Layout dataclass

On-disk layout for one deployed project.

Mirrors xcore's own convention of keeping secrets host-side, never inside the artifact: <project_root>/plugins/<plugin>.env.

Source code in xcore_agent/agent/install_driver.py 
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
@dataclass
class Layout:
    """On-disk layout for one deployed project.

    Mirrors xcore's own convention of keeping secrets host-side, never inside
    the artifact: `<project_root>/plugins/<plugin>.env`.
    """

    project_root: Path
    extracted_root: Path
    # Which directory plugins live under, both inside the extracted artifact
    # and on the target host — "plugins" unless the source project's own
    # integration.yaml overrides it (e.g. "app" — see
    # packer.builder._read_plugins_dirname). Defaults to "plugins" so every
    # caller that never sets it (tests, `gc`, a manifest-less marketplace
    # deploy) is unaffected; DeploymentRunner sets it from
    # `manifest.plugins_dirname` once the manifest is parsed and verified —
    # see pipeline.py's `_verify_manifest`.
    plugins_dirname: str = "plugins"

    @property
    def plugins_dir(self) -> Path:
        return self.project_root / self.plugins_dirname

    @property
    def snapshots_dir(self) -> Path:
        return self.project_root / ".snapshots"

    def plugin_dir(self, plugin_id: str) -> Path:
        return self.plugins_dir / plugin_id

    def plugin_env_file(self, plugin_id: str) -> Path:
        return self.plugins_dir / f"{plugin_id}.env"

    @property
    def extensions_dir(self) -> Path:
        return self.project_root / "extensions"

    def extension_dir(self, extension_id: str) -> Path:
        return self.extensions_dir / extension_id
project_root: Path instance-attribute
extracted_root: Path instance-attribute
plugins_dirname: str = 'plugins' class-attribute instance-attribute
plugins_dir: Path property
snapshots_dir: Path property
extensions_dir: Path property
__init__(project_root: Path, extracted_root: Path, plugins_dirname: str = 'plugins') -> None
plugin_dir(plugin_id: str) -> Path
Source code in xcore_agent/agent/install_driver.py 
81
82
def plugin_dir(self, plugin_id: str) -> Path:
    return self.plugins_dir / plugin_id
plugin_env_file(plugin_id: str) -> Path
Source code in xcore_agent/agent/install_driver.py 
84
85
def plugin_env_file(self, plugin_id: str) -> Path:
    return self.plugins_dir / f"{plugin_id}.env"
extension_dir(extension_id: str) -> Path
Source code in xcore_agent/agent/install_driver.py 
91
92
def extension_dir(self, extension_id: str) -> Path:
    return self.extensions_dir / extension_id

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: ...

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

SnapshotRecord dataclass

Source code in xcore_agent/agent/install_driver.py 
118
119
120
121
122
123
124
125
126
127
128
129
@dataclass
class SnapshotRecord:
    step_id: str
    plugin_id: str
    # None means the plugin directory did not exist before this step ran —
    # i.e. this was a fresh install, so "rolling back" means deleting it,
    # not restoring a prior copy.
    saved_path: Path | None
    # "plugin" (default, matches every snapshot taken before extensions
    # existed) or "extension" — picks plugin_dir() vs extension_dir() when
    # rolling back, see rollback() below.
    kind: str = "plugin"
step_id: str instance-attribute
plugin_id: str instance-attribute
saved_path: Path | None instance-attribute
kind: str = 'plugin' class-attribute instance-attribute
__init__(step_id: str, plugin_id: str, saved_path: Path | None, kind: str = 'plugin') -> None

InstallDriver

Executes filesystem-touching install.yaml steps for real, with snapshot/rollback support for any step marked snapshot: true.

Source code in xcore_agent/agent/install_driver.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
class InstallDriver:
    """Executes filesystem-touching install.yaml steps for real, with
    snapshot/rollback support for any step marked `snapshot: true`."""

    def __init__(
        self,
        layout: Layout,
        supervisor: Supervisor | None = None,
        *,
        provisioners: dict[str, Provisioner] | None = None,
        notifiers: dict[str, Notifier] | None = None,
        manifest: ProjectManifest | None = None,
        plugin_secret_key: bytes | None = None,
    ) -> None:
        self.layout = layout
        self._supervisor = supervisor or NullSupervisor()
        self._provisioners = provisioners or {}
        self._notifiers = notifiers or {}
        # Set by DeploymentRunner once manifest.json has been parsed and
        # verified — used to validate required environment variables in
        # write_env(). None in tests/callers that don't need that check.
        self.manifest = manifest
        # The target host's own `plugins.secret_key` (integration.yaml) —
        # host-local, never embedded in the artifact, passed in the same
        # spirit as `.env` values. None (default) skips plugin.sig signing
        # entirely: a caller with no strict_trusted host to satisfy pays
        # nothing for this. See plugin_signing.py for what this enables.
        self._plugin_secret_key = plugin_secret_key
        self._snapshots: list[SnapshotRecord] = []

    def snapshot_before(self, step_id: str, plugin_id: str, *, kind: str = "plugin") -> None:
        target = (
            self.layout.plugin_dir(plugin_id)
            if kind == "plugin"
            else self.layout.extension_dir(plugin_id)
        )
        if not target.exists():
            self._snapshots.append(
                SnapshotRecord(step_id=step_id, plugin_id=plugin_id, saved_path=None, kind=kind)
            )
            return
        self.layout.snapshots_dir.mkdir(parents=True, exist_ok=True)
        saved = self.layout.snapshots_dir / f"{step_id}-{plugin_id}-{int(time.time() * 1000)}"
        shutil.copytree(target, saved)
        self._snapshots.append(
            SnapshotRecord(step_id=step_id, plugin_id=plugin_id, saved_path=saved, kind=kind)
        )

    def rollback(self, *, to_step_id: str | None = None) -> None:
        """Restore snapshots in reverse order, optionally stopping once
        `to_step_id` is reached (exclusive) instead of rolling back everything.
        A step that had no prior state (fresh install) is undone by deleting
        what it created, not by restoring a copy that never existed."""
        for record in reversed(self._snapshots):
            if to_step_id is not None and record.step_id == to_step_id:
                break
            target = (
                self.layout.plugin_dir(record.plugin_id)
                if record.kind == "plugin"
                else self.layout.extension_dir(record.plugin_id)
            )
            if target.exists():
                shutil.rmtree(target)
            if record.saved_path is not None and record.saved_path.exists():
                shutil.copytree(record.saved_path, target)

    def provision(self, step: ProvisionStep) -> None:
        provisioner = self._provisioners.get(step.plugin)
        if provisioner is None:
            raise InstallError(
                f"no provisioner registered for plugin {step.plugin!r} — pass one via "
                f"InstallDriver(provisioners={{{step.plugin!r}: ...}}); provisioning a "
                "backing service is infra-specific, so there is no generic default"
            )
        provisioner(step)

    def notify(self, step: NotifyStep) -> None:
        """Best-effort, unlike `provision`: notifying is a side channel, not
        part of what makes a deployment succeed or fail (same reasoning as
        MarketplaceClient.report_deployment — see its docstring) — a missing
        or failing notifier is logged and swallowed, never raised."""
        notifier = self._notifiers.get(step.event)
        if notifier is None:
            logger.debug("no notifier registered for event %r — skipping", step.event)
            return
        try:
            notifier(step)
        except Exception as exc:
            logger.warning("notifier for event %r failed: %s", step.event, exc)

    def install_plugin(self, step: InstallPluginStep) -> None:
        source = self.layout.extracted_root / self.layout.plugins_dirname / step.plugin
        if not source.is_dir():
            raise InstallError(f"plugin {step.plugin!r} not found in extracted artifact")
        target = self.layout.plugin_dir(step.plugin)
        if step.snapshot:
            self.snapshot_before(step.id, step.plugin)
        if target.exists():
            shutil.rmtree(target)
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copytree(source, target)

        # A no-op unless `target/plugin.yaml` declares `execution_mode:
        # trusted` AND a plugin_secret_key was configured — a project with
        # no trusted plugins, or a deployment onto a host that doesn't run
        # strict_trusted, pays nothing for this. See plugin_signing.py.
        if self._plugin_secret_key is not None:
            sign_installed_plugin(target, self._plugin_secret_key)

    def install_extension(self, step: InstallExtensionStep) -> None:
        source = self.layout.extracted_root / "extensions" / step.extension
        if not source.is_dir():
            raise InstallError(f"extension {step.extension!r} not found in extracted artifact")
        target = self.layout.extension_dir(step.extension)
        if step.snapshot:
            self.snapshot_before(step.id, step.extension, kind="extension")
        if target.exists():
            shutil.rmtree(target)
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copytree(source, target)

    def configure_plugin(self, step: ConfigurePluginStep) -> None:
        # Actual plugin configuration is applied by xcore's own runtime
        # (kernel.runtime.loader) once the host loads the plugin — this step
        # only needs to exist so install.yaml can express ordering/dependencies.
        return None

    def write_env(self, step: WriteEnvStep) -> None:
        target = self.layout.plugin_env_file(step.plugin)
        if not target.exists():
            # Never overwrite secrets already configured on the host — only
            # seed the file if it's missing. Seeding prefers a matching
            # value already exported in this process's own OS environment
            # (the operator's shell, a systemd unit's Environment=, ...)
            # over the template's own placeholder — still host-local, never
            # embedded in the artifact, just read from a place xcore-agent
            # already has access to instead of requiring a manual SSH edit
            # afterwards. A key absent from the OS environment falls back
            # to whatever the template already has (usually empty).
            template = self.layout.extracted_root / step.from_
            if not template.is_file():
                raise InstallError(f"env template {step.from_!r} not found in artifact")
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_text(_seed_env_from_os_environ(template.read_text(), os.environ))
            target.chmod(0o600)

        self._check_required_env(step.plugin, target)

    def _check_required_env(self, plugin_id: str, env_file: Path) -> None:
        if self.manifest is None:
            return
        try:
            plugin = self.manifest.plugin(plugin_id)
        except KeyError:
            return
        if plugin.environment is None or not plugin.environment.required:
            return

        values = _parse_env_file(env_file)
        missing = [key for key in plugin.environment.required if not values.get(key)]
        if missing:
            raise InstallError(
                f"plugin {plugin_id!r} is missing required environment variable(s) "
                f"{', '.join(missing)} in {env_file} — set them on the host before deploying"
            )

    def start(self, step: StartStep) -> None:
        self._supervisor.start(step.plugin)

    def stop(self, step: StopStep) -> None:
        self._supervisor.stop(step.plugin)

    def restart(self, step: RestartStep) -> None:
        self._supervisor.restart(step.plugin)

    def healthcheck(self, step: HealthcheckStep) -> None:
        try:
            self._supervisor.healthcheck(
                step.plugin, timeout_seconds=step.timeout_seconds, retries=step.retries
            )
        except Exception as exc:
            raise HealthcheckError(str(exc)) from exc
layout = layout instance-attribute
manifest = manifest instance-attribute
__init__(layout: Layout, supervisor: Supervisor | None = None, *, provisioners: dict[str, Provisioner] | None = None, notifiers: dict[str, Notifier] | None = None, manifest: ProjectManifest | None = None, plugin_secret_key: bytes | None = None) -> None
Source code in xcore_agent/agent/install_driver.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def __init__(
    self,
    layout: Layout,
    supervisor: Supervisor | None = None,
    *,
    provisioners: dict[str, Provisioner] | None = None,
    notifiers: dict[str, Notifier] | None = None,
    manifest: ProjectManifest | None = None,
    plugin_secret_key: bytes | None = None,
) -> None:
    self.layout = layout
    self._supervisor = supervisor or NullSupervisor()
    self._provisioners = provisioners or {}
    self._notifiers = notifiers or {}
    # Set by DeploymentRunner once manifest.json has been parsed and
    # verified — used to validate required environment variables in
    # write_env(). None in tests/callers that don't need that check.
    self.manifest = manifest
    # The target host's own `plugins.secret_key` (integration.yaml) —
    # host-local, never embedded in the artifact, passed in the same
    # spirit as `.env` values. None (default) skips plugin.sig signing
    # entirely: a caller with no strict_trusted host to satisfy pays
    # nothing for this. See plugin_signing.py for what this enables.
    self._plugin_secret_key = plugin_secret_key
    self._snapshots: list[SnapshotRecord] = []
snapshot_before(step_id: str, plugin_id: str, *, kind: str = 'plugin') -> None
Source code in xcore_agent/agent/install_driver.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def snapshot_before(self, step_id: str, plugin_id: str, *, kind: str = "plugin") -> None:
    target = (
        self.layout.plugin_dir(plugin_id)
        if kind == "plugin"
        else self.layout.extension_dir(plugin_id)
    )
    if not target.exists():
        self._snapshots.append(
            SnapshotRecord(step_id=step_id, plugin_id=plugin_id, saved_path=None, kind=kind)
        )
        return
    self.layout.snapshots_dir.mkdir(parents=True, exist_ok=True)
    saved = self.layout.snapshots_dir / f"{step_id}-{plugin_id}-{int(time.time() * 1000)}"
    shutil.copytree(target, saved)
    self._snapshots.append(
        SnapshotRecord(step_id=step_id, plugin_id=plugin_id, saved_path=saved, kind=kind)
    )
rollback(*, to_step_id: str | None = None) -> None

Restore snapshots in reverse order, optionally stopping once to_step_id is reached (exclusive) instead of rolling back everything. A step that had no prior state (fresh install) is undone by deleting what it created, not by restoring a copy that never existed.

Source code in xcore_agent/agent/install_driver.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def rollback(self, *, to_step_id: str | None = None) -> None:
    """Restore snapshots in reverse order, optionally stopping once
    `to_step_id` is reached (exclusive) instead of rolling back everything.
    A step that had no prior state (fresh install) is undone by deleting
    what it created, not by restoring a copy that never existed."""
    for record in reversed(self._snapshots):
        if to_step_id is not None and record.step_id == to_step_id:
            break
        target = (
            self.layout.plugin_dir(record.plugin_id)
            if record.kind == "plugin"
            else self.layout.extension_dir(record.plugin_id)
        )
        if target.exists():
            shutil.rmtree(target)
        if record.saved_path is not None and record.saved_path.exists():
            shutil.copytree(record.saved_path, target)
provision(step: ProvisionStep) -> None
Source code in xcore_agent/agent/install_driver.py 
198
199
200
201
202
203
204
205
206
def provision(self, step: ProvisionStep) -> None:
    provisioner = self._provisioners.get(step.plugin)
    if provisioner is None:
        raise InstallError(
            f"no provisioner registered for plugin {step.plugin!r} — pass one via "
            f"InstallDriver(provisioners={{{step.plugin!r}: ...}}); provisioning a "
            "backing service is infra-specific, so there is no generic default"
        )
    provisioner(step)
notify(step: NotifyStep) -> None

Best-effort, unlike provision: notifying is a side channel, not part of what makes a deployment succeed or fail (same reasoning as MarketplaceClient.report_deployment — see its docstring) — a missing or failing notifier is logged and swallowed, never raised.

Source code in xcore_agent/agent/install_driver.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
def notify(self, step: NotifyStep) -> None:
    """Best-effort, unlike `provision`: notifying is a side channel, not
    part of what makes a deployment succeed or fail (same reasoning as
    MarketplaceClient.report_deployment — see its docstring) — a missing
    or failing notifier is logged and swallowed, never raised."""
    notifier = self._notifiers.get(step.event)
    if notifier is None:
        logger.debug("no notifier registered for event %r — skipping", step.event)
        return
    try:
        notifier(step)
    except Exception as exc:
        logger.warning("notifier for event %r failed: %s", step.event, exc)
install_plugin(step: InstallPluginStep) -> None
Source code in xcore_agent/agent/install_driver.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def install_plugin(self, step: InstallPluginStep) -> None:
    source = self.layout.extracted_root / self.layout.plugins_dirname / step.plugin
    if not source.is_dir():
        raise InstallError(f"plugin {step.plugin!r} not found in extracted artifact")
    target = self.layout.plugin_dir(step.plugin)
    if step.snapshot:
        self.snapshot_before(step.id, step.plugin)
    if target.exists():
        shutil.rmtree(target)
    target.parent.mkdir(parents=True, exist_ok=True)
    shutil.copytree(source, target)

    # A no-op unless `target/plugin.yaml` declares `execution_mode:
    # trusted` AND a plugin_secret_key was configured — a project with
    # no trusted plugins, or a deployment onto a host that doesn't run
    # strict_trusted, pays nothing for this. See plugin_signing.py.
    if self._plugin_secret_key is not None:
        sign_installed_plugin(target, self._plugin_secret_key)
install_extension(step: InstallExtensionStep) -> None
Source code in xcore_agent/agent/install_driver.py 
241
242
243
244
245
246
247
248
249
250
251
def install_extension(self, step: InstallExtensionStep) -> None:
    source = self.layout.extracted_root / "extensions" / step.extension
    if not source.is_dir():
        raise InstallError(f"extension {step.extension!r} not found in extracted artifact")
    target = self.layout.extension_dir(step.extension)
    if step.snapshot:
        self.snapshot_before(step.id, step.extension, kind="extension")
    if target.exists():
        shutil.rmtree(target)
    target.parent.mkdir(parents=True, exist_ok=True)
    shutil.copytree(source, target)
configure_plugin(step: ConfigurePluginStep) -> None
Source code in xcore_agent/agent/install_driver.py 
253
254
255
256
257
def configure_plugin(self, step: ConfigurePluginStep) -> None:
    # Actual plugin configuration is applied by xcore's own runtime
    # (kernel.runtime.loader) once the host loads the plugin — this step
    # only needs to exist so install.yaml can express ordering/dependencies.
    return None
write_env(step: WriteEnvStep) -> None
Source code in xcore_agent/agent/install_driver.py 
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def write_env(self, step: WriteEnvStep) -> None:
    target = self.layout.plugin_env_file(step.plugin)
    if not target.exists():
        # Never overwrite secrets already configured on the host — only
        # seed the file if it's missing. Seeding prefers a matching
        # value already exported in this process's own OS environment
        # (the operator's shell, a systemd unit's Environment=, ...)
        # over the template's own placeholder — still host-local, never
        # embedded in the artifact, just read from a place xcore-agent
        # already has access to instead of requiring a manual SSH edit
        # afterwards. A key absent from the OS environment falls back
        # to whatever the template already has (usually empty).
        template = self.layout.extracted_root / step.from_
        if not template.is_file():
            raise InstallError(f"env template {step.from_!r} not found in artifact")
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(_seed_env_from_os_environ(template.read_text(), os.environ))
        target.chmod(0o600)

    self._check_required_env(step.plugin, target)
start(step: StartStep) -> None
Source code in xcore_agent/agent/install_driver.py 
298
299
def start(self, step: StartStep) -> None:
    self._supervisor.start(step.plugin)
stop(step: StopStep) -> None
Source code in xcore_agent/agent/install_driver.py 
301
302
def stop(self, step: StopStep) -> None:
    self._supervisor.stop(step.plugin)
restart(step: RestartStep) -> None
Source code in xcore_agent/agent/install_driver.py 
304
305
def restart(self, step: RestartStep) -> None:
    self._supervisor.restart(step.plugin)
healthcheck(step: HealthcheckStep) -> None
Source code in xcore_agent/agent/install_driver.py 
307
308
309
310
311
312
313
def healthcheck(self, step: HealthcheckStep) -> None:
    try:
        self._supervisor.healthcheck(
            step.plugin, timeout_seconds=step.timeout_seconds, retries=step.retries
        )
    except Exception as exc:
        raise HealthcheckError(str(exc)) from exc