diff --git a/apps/device-host-agent/host_agent/app.py b/apps/device-host-agent/host_agent/app.py index 46bd0b1..8dd8053 100644 --- a/apps/device-host-agent/host_agent/app.py +++ b/apps/device-host-agent/host_agent/app.py @@ -24,6 +24,7 @@ from host_agent.local_account import LocalAccountStore from host_agent.policy_cache import HostPolicyCacheStore from host_agent.processor import AssignmentProcessingResult, AssignmentProcessor from host_agent.retention import prune_task_history +from host_agent.skill_sync import HostAgentSkillSync from host_agent.status import AgentStatusTracker from host_agent.web.app import create_console_app from host_agent.web.auth import SessionManager @@ -42,6 +43,7 @@ class HostAgentApplication: console_enrollment_client: HostAgentEnrollmentClient | None = None dependency_supervisor: DependencySupervisor | None = None instance_lock: InstanceLock | None = None + skill_sync: HostAgentSkillSync | None = None def run(self) -> None: asyncio.run(self.run_async()) @@ -60,6 +62,8 @@ class HostAgentApplication: ) heartbeat_stop = asyncio.Event() heartbeat_task = asyncio.create_task(self.heartbeat.run(heartbeat_stop)) + if self.skill_sync is not None: + self.skill_sync.start() console_task = ( asyncio.create_task(self.console_server.serve()) if self.console_server is not None @@ -110,6 +114,8 @@ class HostAgentApplication: finally: if self.console_enrollment_client is not None: self.console_enrollment_client.close() + if self.skill_sync is not None: + self.skill_sync.stop() await self.client.aclose() if self.instance_lock is not None: self.instance_lock.release() @@ -263,6 +269,9 @@ def create_application( dependency_supervisor = DependencySupervisor.from_host_agent_config( resolved_config ) + skill_sync: HostAgentSkillSync | None = None + if resolved_config.host_id and resolved_config.token: + skill_sync = HostAgentSkillSync(resolved_config) return HostAgentApplication( client=client, heartbeat=heartbeat, @@ -271,6 +280,7 @@ def create_application( console_enrollment_client=console_enrollment_client, dependency_supervisor=dependency_supervisor, instance_lock=instance_lock, + skill_sync=skill_sync, ) except BaseException: instance_lock.release() diff --git a/apps/device-host-agent/host_agent/config.py b/apps/device-host-agent/host_agent/config.py index 302989d..d3e7049 100644 --- a/apps/device-host-agent/host_agent/config.py +++ b/apps/device-host-agent/host_agent/config.py @@ -47,6 +47,7 @@ class HostAgentConfig: task_artifact_dir: Path = Path("host_agent_data/history") task_retention_max_count: int = 50 task_retention_max_age_days: int = 7 + skill_sync_interval_seconds: float = 300.0 def load_host_agent_config( @@ -156,6 +157,9 @@ def load_host_agent_config( task_retention_max_age_days=_positive_int( values, "HOST_AGENT_TASK_RETENTION_MAX_AGE_DAYS", 7 ), + skill_sync_interval_seconds=_positive_float( + values, "HOST_AGENT_SKILL_SYNC_INTERVAL_SECONDS", 300.0 + ), ) if config.max_retry_backoff_seconds < config.retry_backoff_seconds: raise HostAgentConfigurationError( diff --git a/apps/device-host-agent/host_agent/skill_sync.py b/apps/device-host-agent/host_agent/skill_sync.py new file mode 100644 index 0000000..eff573d --- /dev/null +++ b/apps/device-host-agent/host_agent/skill_sync.py @@ -0,0 +1,104 @@ +"""Host-agent wiring for skill sync against the Cloud Control Plane. + +Constructs the synced catalog store, the local skill store, the Cloud API +sync client, and the :class:`SkillSyncRunner`, then drives them on the host +agent's lifecycle: the runner pulls incremental per-host skill deltas into the +synced catalog (forking local overrides on revocation, design D9), and a +best-effort inventory of the agent's local skills is reported to the Cloud +(design D7). Lives in ``host_agent`` (not ``runtime``) for the same boundary +reasons as :mod:`host_agent.cloud_planner_client`. +""" +from __future__ import annotations + +import logging +import threading + +from host_agent.config import HostAgentConfig + +log = logging.getLogger(__name__) + + +class HostAgentSkillSync: + """Owns the skill stores, sync runner, and inventory reporting thread.""" + + def __init__( + self, + config: HostAgentConfig, + *, + poll_interval: float | None = None, + http_client=None, + ) -> None: + from api.skill_sync import CloudApiSkillClient, SkillSyncRunner + from storage.local_skills import LocalSkillStore + from storage.skill_catalog import SkillCatalogStore + + state_dir = config.identity_path.parent + self.config = config + self.synced_store = SkillCatalogStore(db_path=state_dir / "skills.sqlite3") + self.local_store = LocalSkillStore(db_path=state_dir / "local_skills.sqlite3") + self.client = CloudApiSkillClient( + config.control_plane_url, + host_token=config.token, + client=http_client, + ) + self.runner = SkillSyncRunner( + store=self.synced_store, + client=self.client, + subscriptions=[config.host_id], + poll_interval=poll_interval or config.skill_sync_interval_seconds, + local_store=self.local_store, + ) + self._inventory_stop = threading.Event() + self._inventory_thread: threading.Thread | None = None + + def start(self) -> None: + """Start the sync poll loop + periodic inventory reporting.""" + self.runner.start_background() + self._inventory_stop.clear() + self._inventory_thread = threading.Thread( + target=self._report_inventory_forever, daemon=True + ) + self._inventory_thread.start() + log.info("skill sync started for host %s", self.config.host_id) + + def stop(self) -> None: + """Stop the sync loop + inventory thread and close the HTTP client.""" + self.runner.stop_background() + self._inventory_stop.set() + if self._inventory_thread is not None: + self._inventory_thread.join(timeout=5.0) + self._inventory_thread = None + self.client.close() + + def report_inventory_once(self) -> None: + """Report the current local-skill inventory to the Cloud (best-effort).""" + try: + inventory = [ + { + "id": meta.id, + "name": meta.name, + "kind": meta.kind, + "origin": "local", + } + for meta in self.local_store.list_local() + ] + for override in self.local_store.list_overrides(): + inventory.append( + { + "id": override.metadata.id, + "name": override.metadata.name, + "kind": override.metadata.kind, + "origin": "cloud", + "locally_overridden": True, + } + ) + self.client.report_inventory(self.config.host_id, inventory) + except Exception: # best-effort: never impair local operation + log.debug("skill inventory report failed", exc_info=True) + + def _report_inventory_forever(self) -> None: + # Report once at startup, then on the sync cadence. + self.report_inventory_once() + interval = self.config.skill_sync_interval_seconds + while not self._inventory_stop.wait(interval): + self.report_inventory_once() diff --git a/apps/device-host-agent/tests/test_skill_sync_wiring.py b/apps/device-host-agent/tests/test_skill_sync_wiring.py new file mode 100644 index 0000000..79d19f3 --- /dev/null +++ b/apps/device-host-agent/tests/test_skill_sync_wiring.py @@ -0,0 +1,93 @@ +"""Tests for the host-agent skill sync wiring (ยง7.2/7.4).""" +from __future__ import annotations + +import httpx +import pytest + +from host_agent.config import HostAgentConfig +from host_agent.skill_sync import HostAgentSkillSync + + +def _config(tmp_path) -> HostAgentConfig: + return HostAgentConfig( + control_plane_url="https://cloud.example", + host_id="host-1", + token="host-token", + identity_path=tmp_path / "identity.json", + skill_sync_interval_seconds=0.01, + ) + + +def test_skill_sync_pulls_delta_and_reports_inventory(tmp_path): + seen_inventory = {"host": None, "body": None} + + def handler(request: httpx.Request) -> httpx.Response: + path = request.url.path + if path.endswith("/skills/sync"): + return httpx.Response( + 200, + json={ + "skills": [ + { + "id": "c1", + "name": "Cloud Skill", + "kind": "knowledge", + "description": "", + "tags": [], + "revision": 1, + "created_at": "2026-01-01T00:00:00+00:00", + "updated_at": "2026-01-01T00:00:00+00:00", + "content": "body", + "steps": [], + "parameters": {}, + } + ], + "removed_ids": [], + "latest_version": 1, + "is_full_replace": True, + }, + ) + if path.endswith("/skills/inventory"): + seen_inventory["host"] = request.url.path.split("/")[4] + seen_inventory["body"] = request.read() + return httpx.Response(204) + return httpx.Response(404) + + sync = HostAgentSkillSync( + _config(tmp_path), + http_client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + # One manual tick applies the cloud skill to the synced store. + outcomes = sync.runner.tick() + assert outcomes["host-1"].success is True + visible = sync.synced_store.list_skills({"host-1"}) + assert [m.name for m in visible] == ["Cloud Skill"] + + # Inventory report of an authored local skill is best-effort and payload-shaped. + from skills_learning.models import KnowledgeSkill, SkillMetadata + + sync.local_store.create_local( + KnowledgeSkill(metadata=SkillMetadata(name="Local Note", kind="knowledge"), content="x") + ) + sync.report_inventory_once() + assert seen_inventory["host"] == "host-1" + assert b"Local Note" in seen_inventory["body"] + + # start/stop lifecycle does not raise. + sync.start() + sync.stop() + + +def test_skill_sync_inventory_failure_is_isolated(tmp_path): + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/skills/inventory"): + return httpx.Response(500) + return httpx.Response(200, json={"skills": [], "removed_ids": [], "latest_version": 0, "is_full_replace": True}) + + sync = HostAgentSkillSync( + _config(tmp_path), + http_client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + # A failed inventory report must not raise. + sync.report_inventory_once() + sync.client.close() diff --git a/openspec/changes/skill-management-console/tasks.md b/openspec/changes/skill-management-console/tasks.md index f7b89c6..05ecb4d 100644 --- a/openspec/changes/skill-management-console/tasks.md +++ b/openspec/changes/skill-management-console/tasks.md @@ -39,10 +39,10 @@ ## 7. Sync repoint + runner wiring + inventory report - [x] 7.1 Repoint `api/skill_sync.py`'s concrete client at the Cloud API per-host sync endpoint (host-scoped bearer; `since_version` incremental; full-replace on first/stale); keep the `SubscriptionClient` Protocol / `SyncDelta` shape. The agent's "subscription_id" becomes its host identifier. -- [ ] 7.2 Construct and start `SkillSyncRunner` in the host-agent app bootstrap behind the existing cloud-transport configuration; configurable poll interval; failures non-fatal (preserve cache + record error). +- [x] 7.2 Construct and start `SkillSyncRunner` in the host-agent app bootstrap behind the existing cloud-transport configuration; configurable poll interval; failures non-fatal (preserve cache + record error). - [x] 7.3 On a sync `removed_ids` entry that has a local override, trigger the fork (D9) via `LocalSkillStore.fork_override_to_local`. -- [ ] 7.4 Add a periodic best-effort local-skill inventory report from the agent to the Cloud inventory endpoint (metadata only). -- [ ] 7.5 Tests: incremental apply, full-replace, fork-on-revocation end-to-end, runner lifecycle, inventory report payload + failure-isolation. +- [x] 7.4 Add a periodic best-effort local-skill inventory report from the agent to the Cloud inventory endpoint (metadata only). +- [x] 7.5 Tests: incremental apply, full-replace, fork-on-revocation end-to-end, runner lifecycle, inventory report payload + failure-isolation. ## 8. Cloud Console Skills view