Skip to content

Watchers

xcore_agent.agent.watcher and xcore_agent.agent.marketplace_watcher — the CI/CD loops that poll a Hub and redeploy on version change.

Watcher (.xdeploy flow)

xcore_agent.agent.watcher

Polls XCore Hub for the latest version/tag of a project and redeploys automatically when it changes — the "CI/CD" loop: Hub publishes a new release, the agent notices on its own and rolls it out without a human running xcore-agent deploy by hand. Runs garbage collection after every successful redeploy so snapshots and cached downloads don't grow unbounded.

WatchResult dataclass

Source code in xcore_agent/agent/watcher.py 
22
23
24
25
26
@dataclass(frozen=True)
class WatchResult:
    checked_version: str
    deployed: bool
    report: DeploymentReport | None = None
checked_version: str instance-attribute
deployed: bool instance-attribute
report: DeploymentReport | None = None class-attribute instance-attribute
__init__(checked_version: str, deployed: bool, report: DeploymentReport | None = None) -> None

Watcher

Source code in xcore_agent/agent/watcher.py 
 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
 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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class Watcher:
    def __init__(
        self,
        *,
        hub: HubClient,
        credentials: DeploymentCredentials,
        workdir_root: Path,
        project_root: Path,
        trusted_signer_public_key: bytes,
        keep_snapshots: int = 3,
        supervisor: Supervisor | None = None,
        plugin_resolver: PluginResolver | None = None,
        provisioners: dict[str, Provisioner] | None = None,
        notifiers: dict[str, Notifier] | None = None,
    ) -> None:
        self._hub = hub
        self._credentials = credentials
        self._workdir_root = workdir_root
        self._project_root = project_root
        self._trusted_signer_public_key = trusted_signer_public_key
        self._keep_snapshots = keep_snapshots
        self._supervisor = supervisor
        self._plugin_resolver = plugin_resolver
        self._provisioners = provisioners
        self._notifiers = notifiers
        self._state = StateStore(project_root)

    async def check_once(self) -> WatchResult:
        session = await self._hub.authenticate(
            xdevkey=self._credentials.xdevkey, project_id=self._credentials.project_id
        )
        latest = await self._hub.get_latest_version(
            session, project_id=self._credentials.project_id
        )

        installed = self._state.read()
        if installed is not None and installed.version == latest:
            return WatchResult(checked_version=latest, deployed=False)

        runner = DeploymentRunner(
            hub=self._hub,
            credentials=self._credentials,
            version=latest,
            workdir=self._workdir_root / latest,
            project_root=self._project_root,
            trusted_signer_public_key=self._trusted_signer_public_key,
            plugin_resolver=self._plugin_resolver,
            provisioners=self._provisioners,
            notifiers=self._notifiers,
        )
        report = await runner.run()
        self._state.write(project_id=self._credentials.project_id, version=latest)
        self._collect_garbage(keep_versions=frozenset({latest}), manifest=runner.manifest)
        return WatchResult(checked_version=latest, deployed=True, report=report)

    def _collect_garbage(
        self, *, keep_versions: frozenset[str], manifest: ProjectManifest | None
    ) -> None:
        layout = Layout(
            project_root=self._project_root, extracted_root=self._workdir_root / "extracted"
        )
        gc = GarbageCollector(
            layout,
            keep_snapshots=self._keep_snapshots,
            cache_root=self._workdir_root,
            supervisor=self._supervisor,
        )
        restart_ids = [p.id for p in manifest.plugins] if manifest is not None else []
        gc.collect(keep_versions=keep_versions, restart_plugins=restart_ids)

    async def watch_forever(
        self,
        *,
        interval_seconds: float,
        on_result: Callable[[WatchResult], None] | None = None,
        on_error: Callable[[Exception], None] | None = None,
        stop_after: Callable[[], Awaitable[bool]] | None = None,
    ) -> None:
        """Loop `check_once()` every `interval_seconds`. A failed tick (Hub
        unreachable, a bad redeploy) must not kill the loop — that would
        defeat the point of an unattended CI/CD watcher — so errors go to
        `on_error` instead of propagating. `stop_after` is an injectable
        exit condition for tests; real callers just never pass it."""
        while True:
            try:
                result = await self.check_once()
            except Exception as exc:  # noqa: BLE001 — a bad tick must not stop the loop
                if on_error is not None:
                    on_error(exc)
            else:
                if on_result is not None:
                    on_result(result)

            if stop_after is not None and await stop_after():
                return
            await asyncio.sleep(interval_seconds)
__init__(*, hub: HubClient, credentials: DeploymentCredentials, workdir_root: Path, project_root: Path, trusted_signer_public_key: bytes, keep_snapshots: int = 3, supervisor: Supervisor | None = None, plugin_resolver: PluginResolver | None = None, provisioners: dict[str, Provisioner] | None = None, notifiers: dict[str, Notifier] | None = None) -> None
Source code in xcore_agent/agent/watcher.py 
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
def __init__(
    self,
    *,
    hub: HubClient,
    credentials: DeploymentCredentials,
    workdir_root: Path,
    project_root: Path,
    trusted_signer_public_key: bytes,
    keep_snapshots: int = 3,
    supervisor: Supervisor | None = None,
    plugin_resolver: PluginResolver | None = None,
    provisioners: dict[str, Provisioner] | None = None,
    notifiers: dict[str, Notifier] | None = None,
) -> None:
    self._hub = hub
    self._credentials = credentials
    self._workdir_root = workdir_root
    self._project_root = project_root
    self._trusted_signer_public_key = trusted_signer_public_key
    self._keep_snapshots = keep_snapshots
    self._supervisor = supervisor
    self._plugin_resolver = plugin_resolver
    self._provisioners = provisioners
    self._notifiers = notifiers
    self._state = StateStore(project_root)
check_once() -> WatchResult async
Source code in xcore_agent/agent/watcher.py 
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
async def check_once(self) -> WatchResult:
    session = await self._hub.authenticate(
        xdevkey=self._credentials.xdevkey, project_id=self._credentials.project_id
    )
    latest = await self._hub.get_latest_version(
        session, project_id=self._credentials.project_id
    )

    installed = self._state.read()
    if installed is not None and installed.version == latest:
        return WatchResult(checked_version=latest, deployed=False)

    runner = DeploymentRunner(
        hub=self._hub,
        credentials=self._credentials,
        version=latest,
        workdir=self._workdir_root / latest,
        project_root=self._project_root,
        trusted_signer_public_key=self._trusted_signer_public_key,
        plugin_resolver=self._plugin_resolver,
        provisioners=self._provisioners,
        notifiers=self._notifiers,
    )
    report = await runner.run()
    self._state.write(project_id=self._credentials.project_id, version=latest)
    self._collect_garbage(keep_versions=frozenset({latest}), manifest=runner.manifest)
    return WatchResult(checked_version=latest, deployed=True, report=report)
watch_forever(*, interval_seconds: float, on_result: Callable[[WatchResult], None] | None = None, on_error: Callable[[Exception], None] | None = None, stop_after: Callable[[], Awaitable[bool]] | None = None) -> None async

Loop check_once() every interval_seconds. A failed tick (Hub unreachable, a bad redeploy) must not kill the loop — that would defeat the point of an unattended CI/CD watcher — so errors go to on_error instead of propagating. stop_after is an injectable exit condition for tests; real callers just never pass it.

Source code in xcore_agent/agent/watcher.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
async def watch_forever(
    self,
    *,
    interval_seconds: float,
    on_result: Callable[[WatchResult], None] | None = None,
    on_error: Callable[[Exception], None] | None = None,
    stop_after: Callable[[], Awaitable[bool]] | None = None,
) -> None:
    """Loop `check_once()` every `interval_seconds`. A failed tick (Hub
    unreachable, a bad redeploy) must not kill the loop — that would
    defeat the point of an unattended CI/CD watcher — so errors go to
    `on_error` instead of propagating. `stop_after` is an injectable
    exit condition for tests; real callers just never pass it."""
    while True:
        try:
            result = await self.check_once()
        except Exception as exc:  # noqa: BLE001 — a bad tick must not stop the loop
            if on_error is not None:
                on_error(exc)
        else:
            if on_result is not None:
                on_result(result)

        if stop_after is not None and await stop_after():
            return
        await asyncio.sleep(interval_seconds)

MarketplaceWatcher (marketplace flow)

xcore_agent.agent.marketplace_watcher

CI/CD watch loop for the real xcore-team/marketplace flow — the counterpart to agent.watcher.Watcher for MarketplaceClient / MarketplaceDeploymentRunner instead of the invented HubClient contract.

Polls GET /{plugins|services}/{slug} for latest_version and redeploys through MarketplaceDeploymentRunner whenever it changes from what StateStore says is currently installed — same version-change-detection and GC-after-redeploy shape as Watcher, just against the Marketplace's one-slug-per-deployment model instead of a multi-plugin project bundle (see marketplace_client.py's module docstring for why that's a separate client and pipeline in the first place).

MarketplaceWatchResult dataclass

Source code in xcore_agent/agent/marketplace_watcher.py 
26
27
28
29
30
@dataclass(frozen=True)
class MarketplaceWatchResult:
    checked_version: str
    deployed: bool
    report: MarketplaceDeploymentReport | None = None
checked_version: str instance-attribute
deployed: bool instance-attribute
report: MarketplaceDeploymentReport | None = None class-attribute instance-attribute
__init__(checked_version: str, deployed: bool, report: MarketplaceDeploymentReport | None = None) -> None

MarketplaceWatcher

Source code in xcore_agent/agent/marketplace_watcher.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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
class MarketplaceWatcher:
    def __init__(
        self,
        *,
        client: MarketplaceClient,
        slug: str,
        trusted_signer_secret: bytes,
        install_plan_path: Path,
        workdir_root: Path,
        project_root: Path,
        kind: Kind = "plugin",
        host_id: str = "default",
        keep_snapshots: int = 3,
        supervisor: Supervisor | None = None,
        provisioners: dict[str, Provisioner] | None = None,
        notifiers: dict[str, Notifier] | None = None,
    ) -> None:
        self._client = client
        self._slug = slug
        self._trusted_signer_secret = trusted_signer_secret
        self._install_plan_path = install_plan_path
        self._workdir_root = workdir_root
        self._project_root = project_root
        self._kind = kind
        self._host_id = host_id
        self._keep_snapshots = keep_snapshots
        self._supervisor = supervisor
        self._provisioners = provisioners
        self._notifiers = notifiers
        # namespace=slug: see StateStore's own docstring — several
        # MarketplaceWatchers (one per slug) can share a project_root
        # without clobbering each other's recorded version.
        self._state = StateStore(project_root, namespace=slug)

    async def check_once(self) -> MarketplaceWatchResult:
        latest = await self._client.get_latest_version(slug=self._slug, kind=self._kind)

        installed = self._state.read()
        if installed is not None and installed.version == latest:
            return MarketplaceWatchResult(checked_version=latest, deployed=False)

        runner = MarketplaceDeploymentRunner(
            client=self._client,
            slug=self._slug,
            workdir=self._workdir_root / latest,
            project_root=self._project_root,
            trusted_signer_secret=self._trusted_signer_secret,
            install_plan_path=self._install_plan_path,
            version=latest,
            kind=self._kind,
            host_id=self._host_id,
            provisioners=self._provisioners,
            notifiers=self._notifiers,
        )
        report = await runner.run()
        self._state.write(project_id=self._slug, version=latest)
        self._collect_garbage(keep_versions=frozenset({latest}))
        return MarketplaceWatchResult(checked_version=latest, deployed=True, report=report)

    def _collect_garbage(self, *, keep_versions: frozenset[str]) -> None:
        layout = Layout(
            project_root=self._project_root, extracted_root=self._workdir_root / "extracted"
        )
        gc = GarbageCollector(
            layout,
            keep_snapshots=self._keep_snapshots,
            cache_root=self._workdir_root,
            supervisor=self._supervisor,
        )
        # One slug per deployment in this flow (unlike Watcher's multi-plugin
        # manifest) — restart just that plugin, and only for kind="plugin"
        # (services don't go through the Supervisor restart path).
        restart_ids = [self._slug] if self._kind == "plugin" else []
        gc.collect(keep_versions=keep_versions, restart_plugins=restart_ids)

    async def watch_forever(
        self,
        *,
        interval_seconds: float,
        on_result: Callable[[MarketplaceWatchResult], None] | None = None,
        on_error: Callable[[Exception], None] | None = None,
        stop_after: Callable[[], Awaitable[bool]] | None = None,
    ) -> None:
        """Loop `check_once()` every `interval_seconds`. A failed tick (Hub
        unreachable, a bad redeploy) must not kill the loop — errors go to
        `on_error` instead of propagating. `stop_after` is an injectable exit
        condition for tests; real callers just never pass it."""
        while True:
            try:
                result = await self.check_once()
            except Exception as exc:  # noqa: BLE001 — a bad tick must not stop the loop
                if on_error is not None:
                    on_error(exc)
            else:
                if on_result is not None:
                    on_result(result)

            if stop_after is not None and await stop_after():
                return
            await asyncio.sleep(interval_seconds)
__init__(*, client: MarketplaceClient, slug: str, trusted_signer_secret: bytes, install_plan_path: Path, workdir_root: Path, project_root: Path, kind: Kind = 'plugin', host_id: str = 'default', keep_snapshots: int = 3, supervisor: Supervisor | None = None, provisioners: dict[str, Provisioner] | None = None, notifiers: dict[str, Notifier] | None = None) -> None
Source code in xcore_agent/agent/marketplace_watcher.py 
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
def __init__(
    self,
    *,
    client: MarketplaceClient,
    slug: str,
    trusted_signer_secret: bytes,
    install_plan_path: Path,
    workdir_root: Path,
    project_root: Path,
    kind: Kind = "plugin",
    host_id: str = "default",
    keep_snapshots: int = 3,
    supervisor: Supervisor | None = None,
    provisioners: dict[str, Provisioner] | None = None,
    notifiers: dict[str, Notifier] | None = None,
) -> None:
    self._client = client
    self._slug = slug
    self._trusted_signer_secret = trusted_signer_secret
    self._install_plan_path = install_plan_path
    self._workdir_root = workdir_root
    self._project_root = project_root
    self._kind = kind
    self._host_id = host_id
    self._keep_snapshots = keep_snapshots
    self._supervisor = supervisor
    self._provisioners = provisioners
    self._notifiers = notifiers
    # namespace=slug: see StateStore's own docstring — several
    # MarketplaceWatchers (one per slug) can share a project_root
    # without clobbering each other's recorded version.
    self._state = StateStore(project_root, namespace=slug)
check_once() -> MarketplaceWatchResult async
Source code in xcore_agent/agent/marketplace_watcher.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
async def check_once(self) -> MarketplaceWatchResult:
    latest = await self._client.get_latest_version(slug=self._slug, kind=self._kind)

    installed = self._state.read()
    if installed is not None and installed.version == latest:
        return MarketplaceWatchResult(checked_version=latest, deployed=False)

    runner = MarketplaceDeploymentRunner(
        client=self._client,
        slug=self._slug,
        workdir=self._workdir_root / latest,
        project_root=self._project_root,
        trusted_signer_secret=self._trusted_signer_secret,
        install_plan_path=self._install_plan_path,
        version=latest,
        kind=self._kind,
        host_id=self._host_id,
        provisioners=self._provisioners,
        notifiers=self._notifiers,
    )
    report = await runner.run()
    self._state.write(project_id=self._slug, version=latest)
    self._collect_garbage(keep_versions=frozenset({latest}))
    return MarketplaceWatchResult(checked_version=latest, deployed=True, report=report)
watch_forever(*, interval_seconds: float, on_result: Callable[[MarketplaceWatchResult], None] | None = None, on_error: Callable[[Exception], None] | None = None, stop_after: Callable[[], Awaitable[bool]] | None = None) -> None async

Loop check_once() every interval_seconds. A failed tick (Hub unreachable, a bad redeploy) must not kill the loop — errors go to on_error instead of propagating. stop_after is an injectable exit condition for tests; real callers just never pass it.

Source code in xcore_agent/agent/marketplace_watcher.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
async def watch_forever(
    self,
    *,
    interval_seconds: float,
    on_result: Callable[[MarketplaceWatchResult], None] | None = None,
    on_error: Callable[[Exception], None] | None = None,
    stop_after: Callable[[], Awaitable[bool]] | None = None,
) -> None:
    """Loop `check_once()` every `interval_seconds`. A failed tick (Hub
    unreachable, a bad redeploy) must not kill the loop — errors go to
    `on_error` instead of propagating. `stop_after` is an injectable exit
    condition for tests; real callers just never pass it."""
    while True:
        try:
            result = await self.check_once()
        except Exception as exc:  # noqa: BLE001 — a bad tick must not stop the loop
            if on_error is not None:
                on_error(exc)
        else:
            if on_result is not None:
                on_result(result)

        if stop_after is not None and await stop_after():
            return
        await asyncio.sleep(interval_seconds)