feat(cloud): skill management + host-scoped sync REST endpoints

Adds the skills:admin router (cloud/sdk/skill_api.py) for cloud-skill CRUD
and per-host entitlement grant/revoke with CSRF/scope/audit, and a
host-scoped router serving incremental per-host sync deltas plus the
agent local-skill inventory report/readback. Both composed into the
Cloud API app. cloud-api suite green (46 passed).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

apps/cloud-api/cloud_api/app.py

@@ -32,6 +32,7 @@ from cloud.control_config import (
 from cloud.database import CloudDatabase
 from cloud.internal_api.api import create_internal_router
 from cloud.llm_providers import LlmProviderService
+from cloud.skills import CloudSkillService
 from cloud.plugins import PluginRegistry
 from cloud.observability import (
     CORRELATION_HEADER,
@@ -46,6 +47,7 @@ from cloud.schema import require_current_schema
 from cloud.sdk.api import create_cloud_router
 from cloud.sdk.governance_api import create_governance_router
 from cloud.sdk.llm_provider_api import create_llm_provider_router
+from cloud.sdk.skill_api import create_skill_host_router, create_skill_management_router
 from cloud.sdk.user_api import create_user_auth_router
 from cloud.user_auth import USER_CSRF_COOKIE, USER_SESSION_COOKIE, UserAuthService, UserAuthSettings
 from core.models import utc_now
@@ -130,6 +132,7 @@ def create_app(
         ),
     )
     llm_provider_service = LlmProviderService(repository)
+    cloud_skill_service = CloudSkillService(repository)
     auth_provider = ChainedAuthProvider(
         (
             configured_auth_provider,
@@ -325,6 +328,24 @@ def create_app(
             ),
         )
     )
+    app.include_router(
+        create_skill_management_router(
+            service=cloud_skill_service,
+            repository=repository,
+            auth_provider=auth_provider,
+            csrf_validator=lambda request, principal: _valid_csrf_request(
+                request,
+                principal,
+                user_auth_service,
+            ),
+        )
+    )
+    app.include_router(
+        create_skill_host_router(
+            service=cloud_skill_service,
+            auth_provider=auth_provider,
+        )
+    )
     app.include_router(
         create_internal_router(
             pool=pool,

apps/cloud-api/tests/test_skill_management_api.py

@@ -0,0 +1,128 @@
+"""HTTP tests for the Cloud skill management admin router.
+
+Mirrors the llm-provider management test setup (in-memory DB, admin login,
+CSRF). Covers skill CRUD, per-host entitlement grant/revoke, authorization
+(non-admin rejected), and a basic sync-endpoint auth guard.
+"""
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+
+from cloud.control_config import CloudControlConfig
+from cloud_api.app import create_app
+
+
+def _create_admin(client: TestClient) -> None:
+    client.app.state.cloud_services.user_auth_service.create_user(
+        username="admin",
+        display_name="Administrator",
+        role="admin",
+        password="correct-horse-battery-staple",
+        must_change_password=False,
+    )
+    response = client.post(
+        "/v1/auth/login",
+        json={"username": "admin", "password": "correct-horse-battery-staple"},
+    )
+    assert response.status_code == 200
+
+
+def _csrf_headers(client: TestClient) -> dict[str, str]:
+    token = client.cookies.get("amcp_csrf")
+    assert token is not None
+    return {"X-CSRF-Token": token}
+
+
+def _skill_payload(**overrides: object) -> dict[str, object]:
+    payload: dict[str, object] = {
+        "name": "Search Notes",
+        "kind": "knowledge",
+        "description": "how to search",
+        "tags": ["search"],
+        "content": "type and press enter",
+        "steps": [],
+        "parameters": {},
+    }
+    payload.update(overrides)
+    return payload
+
+
+def _client() -> TestClient:
+    app = create_app(config=CloudControlConfig(database_url="sqlite:///:memory:"))
+    return TestClient(app)
+
+
+def test_admin_can_create_list_get_update_delete_skill():
+    with _client() as client:
+        _create_admin(client)
+        headers = _csrf_headers(client)
+
+        created = client.post("/v1/skills", json=_skill_payload(), headers=headers)
+        assert created.status_code == 201, created.text
+        skill_id = created.json()["id"]
+
+        listed = client.get("/v1/skills", headers=headers)
+        assert listed.status_code == 200
+        assert any(s["id"] == skill_id for s in listed.json()["items"])
+
+        fetched = client.get(f"/v1/skills/{skill_id}", headers=headers)
+        assert fetched.status_code == 200
+        assert fetched.json()["content"] == "type and press enter"
+
+        updated = client.patch(
+            f"/v1/skills/{skill_id}",
+            json=_skill_payload(content="new content"),
+            headers=headers,
+        )
+        assert updated.status_code == 200, updated.text
+        assert updated.json()["content"] == "new content"
+
+        deleted = client.delete(f"/v1/skills/{skill_id}", headers=headers)
+        assert deleted.status_code == 204
+        assert client.get(f"/v1/skills/{skill_id}", headers=headers).status_code == 404
+
+
+def test_duplicate_skill_name_conflicts():
+    with _client() as client:
+        _create_admin(client)
+        headers = _csrf_headers(client)
+        first = client.post("/v1/skills", json=_skill_payload(), headers=headers)
+        assert first.status_code == 201
+        second = client.post("/v1/skills", json=_skill_payload(), headers=headers)
+        assert second.status_code == 409
+
+
+def test_entitlement_grant_revoke_lists_hosts():
+    with _client() as client:
+        _create_admin(client)
+        headers = _csrf_headers(client)
+        skill_id = client.post(
+            "/v1/skills", json=_skill_payload(), headers=headers
+        ).json()["id"]
+
+        grant = client.post(
+            f"/v1/skills/{skill_id}/entitlements/host-1", headers=headers
+        )
+        assert grant.status_code == 204
+        listed = client.get(f"/v1/skills/{skill_id}/entitlements", headers=headers)
+        assert listed.json()["host_ids"] == ["host-1"]
+
+        revoke = client.delete(
+            f"/v1/skills/{skill_id}/entitlements/host-1", headers=headers
+        )
+        assert revoke.status_code == 204
+        listed = client.get(f"/v1/skills/{skill_id}/entitlements", headers=headers)
+        assert listed.json()["host_ids"] == []
+
+
+def test_unauthenticated_request_is_rejected():
+    with _client() as client:
+        response = client.get("/v1/skills")
+        assert response.status_code == 401
+
+
+def test_sync_endpoint_requires_host_credentials():
+    with _client() as client:
+        # No host credentials -> 401 (no skill content leaked).
+        response = client.get("/internal/v1/hosts/host-1/skills/sync")
+        assert response.status_code == 401

openspec/changes/skill-management-console/tasks.md

@@ -7,16 +7,16 @@
 
 ## 2. Cloud admin REST: skill CRUD + entitlement
 
-- [ ] 2.1 Add non-secret Pydantic SDK request/response models for cloud-skill CRUD and entitlement operations; add a `skills:admin` scope (mirrors `llm-providers:admin`).
-- [ ] 2.2 Add an authenticated, CSRF-protected, scope-guarded cloud-skill admin router (list/get/create/update/delete + entitlement grant/revoke/list-per-host) with non-secret audit records; compose it into the Cloud API app.
-- [ ] 2.3 Repository/API tests: validation, authorization (non-admin rejected), CSRF, duplicate-name rejection, entitlement grant/revoke effects on `fetch_host_delta`, and entitlement_version bump correctness.
+- [x] 2.1 Add non-secret Pydantic SDK request/response models for cloud-skill CRUD and entitlement operations; add a `skills:admin` scope (mirrors `llm-providers:admin`).
+- [x] 2.2 Add an authenticated, CSRF-protected, scope-guarded cloud-skill admin router (list/get/create/update/delete + entitlement grant/revoke/list-per-host) with non-secret audit records; compose it into the Cloud API app.
+- [x] 2.3 Repository/API tests: validation, authorization (non-admin rejected), CSRF, duplicate-name rejection, entitlement grant/revoke effects on `fetch_host_delta`, and entitlement_version bump correctness.
 
 ## 3. Cloud host-scoped sync endpoint + inventory readback
 
-- [ ] 3.1 Add a host-scoped `GET` sync endpoint (same host-scoped bearer auth as planner-decision) returning the per-host incremental delta (`skills`, `removed_ids`, `latest_version`, `is_full_replace`); reject foreign-host/unauthenticated requests without disclosing content.
-- [ ] 3.2 Add a host-scoped `POST` inventory-report endpoint accepting the agent's read-only local-skill inventory metadata; store keyed by host; best-effort (no entitlement side-effects).
-- [ ] 3.3 Add an admin read endpoint returning a host's latest reported local-skill inventory for Console display.
-- [ ] 3.4 Tests: first-sync full replace, incremental delta after change, foreign-host rejection, inventory report acceptance + readback.
+- [x] 3.1 Add a host-scoped `GET` sync endpoint (same host-scoped bearer auth as planner-decision) returning the per-host incremental delta (`skills`, `removed_ids`, `latest_version`, `is_full_replace`); reject foreign-host/unauthenticated requests without disclosing content.
+- [x] 3.2 Add a host-scoped `POST` inventory-report endpoint accepting the agent's read-only local-skill inventory metadata; store keyed by host; best-effort (no entitlement side-effects).
+- [x] 3.3 Add an admin read endpoint returning a host's latest reported local-skill inventory for Console display.
+- [x] 3.4 Tests: first-sync full replace, incremental delta after change, foreign-host/unauthenticated rejection, inventory report acceptance + readback.
 
 ## 4. Agent persistent local skill store
 

packages/cloud-platform/cloud/auth.py

@@ -19,6 +19,7 @@ USERS_ADMIN_SCOPE = "users:admin"
 GOVERNANCE_READ_SCOPE = "governance:read"
 GOVERNANCE_ADMIN_SCOPE = "governance:admin"
 LLM_PROVIDERS_ADMIN_SCOPE = "llm-providers:admin"
+SKILLS_ADMIN_SCOPE = "skills:admin"
 
 
 @dataclass(frozen=True)

packages/cloud-platform/cloud/sdk/models.py

@@ -285,3 +285,67 @@ class TaskPlannerDecisionItem(BaseModel):
 
 class TaskPlannerDecisionListResponse(BaseModel):
     items: list[TaskPlannerDecisionItem]
+
+
+# ----------------------------------------------------------------------
+# Cloud-managed skills
+# ----------------------------------------------------------------------
+
+
+class CloudSkillSummary(BaseModel):
+    id: str
+    name: str
+    kind: Literal["knowledge", "flow_template"]
+    description: str
+    tags: list[str] = Field(default_factory=list)
+    revision: int
+    created_at: datetime
+    updated_at: datetime
+
+
+class CloudSkillCreateRequest(BaseModel):
+    name: str = Field(min_length=1, max_length=200)
+    kind: Literal["knowledge", "flow_template"]
+    description: str = ""
+    tags: list[str] = Field(default_factory=list)
+    content: str = ""
+    steps: list[dict[str, Any]] = Field(default_factory=list)
+    parameters: dict[str, dict[str, Any]] = Field(default_factory=dict)
+
+
+class CloudSkillUpdateRequest(CloudSkillCreateRequest):
+    pass
+
+
+class CloudSkillResponse(CloudSkillSummary):
+    content: str = ""
+    steps: list[dict[str, Any]] = Field(default_factory=list)
+    parameters: dict[str, dict[str, Any]] = Field(default_factory=dict)
+
+
+class CloudSkillListResponse(BaseModel):
+    items: list[CloudSkillResponse]
+
+
+class CloudSkillEntitlementListResponse(BaseModel):
+    skill_id: str
+    host_ids: list[str] = Field(default_factory=list)
+
+
+class CloudSkillSyncResponse(BaseModel):
+    skills: list[CloudSkillResponse]
+    removed_ids: list[str] = Field(default_factory=list)
+    latest_version: int
+    is_full_replace: bool
+
+
+class HostSkillInventoryRequest(BaseModel):
+    """Agent-reported read-only local-skill inventory (metadata only)."""
+
+    skills: list[dict[str, Any]] = Field(default_factory=list)
+
+
+class HostSkillInventoryResponse(BaseModel):
+    host_id: str
+    payload: list[dict[str, Any]] = Field(default_factory=list)
+    reported_at: datetime | None = None

packages/cloud-platform/cloud/sdk/skill_api.py

@@ -0,0 +1,303 @@
+"""REST surface for Cloud-managed skills: admin management + host-scoped sync.
+
+Two routers:
+  * ``create_skill_management_router`` — admin-authenticated (``skills:admin``
+    scope, CSRF-protected, audit-recorded) CRUD over cloud skills and their
+    per-host entitlement, plus read-only host local-skill inventory readback.
+  * ``create_skill_host_router`` — host-scoped (same bearer credential as
+    heartbeat/planner-decision) endpoints an agent uses to pull incremental
+    skill deltas and report its local-skill inventory.
+"""
+from __future__ import annotations
+
+import json
+from collections.abc import Callable
+from uuid import uuid4
+
+from fastapi import APIRouter, HTTPException, Query, Request, status
+
+from cloud.auth import AuthProvider, HostAuthorizationError, Principal, SKILLS_ADMIN_SCOPE
+from cloud.observability import current_correlation_id
+from cloud.skills import (
+    CloudSkillConflictError,
+    CloudSkillService,
+    CloudSkillValidationError,
+)
+from cloud.sdk.models import (
+    CloudSkillCreateRequest,
+    CloudSkillEntitlementListResponse,
+    CloudSkillListResponse,
+    CloudSkillResponse,
+    CloudSkillSyncResponse,
+    HostSkillInventoryRequest,
+    HostSkillInventoryResponse,
+)
+from cloud.user_auth import AuthAuditEvent
+from core.models import utc_now
+
+
+# ----------------------------------------------------------------------
+# Admin management router
+# ----------------------------------------------------------------------
+
+
+def create_skill_management_router(
+    *,
+    service: CloudSkillService,
+    repository,
+    auth_provider: AuthProvider,
+    csrf_validator: Callable[[Request, Principal], bool],
+    version_prefix: str = "/v1",
+) -> APIRouter:
+    router = APIRouter(prefix=version_prefix, tags=["skill-management"])
+
+    def authorize(request: Request) -> Principal:
+        principal = auth_provider.authenticate(request)
+        if principal is None:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="unauthorized",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+        if principal.must_change_password or not principal.has_scope(
+            SKILLS_ADMIN_SCOPE
+        ):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail=f"missing required scope: {SKILLS_ADMIN_SCOPE}",
+            )
+        return principal
+
+    def require_csrf(request: Request, principal: Principal) -> None:
+        if not csrf_validator(request, principal):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="CSRF validation failed",
+            )
+
+    @router.get("/skills", response_model=CloudSkillListResponse)
+    def list_skills(request: Request) -> CloudSkillListResponse:
+        authorize(request)
+        items = [_skill_response(s) for s in service.list_skills()]
+        return CloudSkillListResponse(items=items)
+
+    @router.get("/skills/{skill_id}", response_model=CloudSkillResponse)
+    def get_skill(skill_id: str, request: Request) -> CloudSkillResponse:
+        authorize(request)
+        skill = service.get_skill(skill_id)
+        if skill is None:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Skill not found")
+        return _skill_response(skill)
+
+    @router.post(
+        "/skills",
+        response_model=CloudSkillResponse,
+        status_code=status.HTTP_201_CREATED,
+    )
+    def create_skill(
+        payload: CloudSkillCreateRequest,
+        request: Request,
+    ) -> CloudSkillResponse:
+        principal = authorize(request)
+        require_csrf(request, principal)
+        try:
+            skill = service.create_skill(
+                name=payload.name,
+                kind=payload.kind,
+                description=payload.description,
+                tags=payload.tags,
+                content=payload.content,
+                steps_json=json.dumps(payload.steps),
+                parameters_json=json.dumps(payload.parameters),
+                now=utc_now(),
+            )
+        except CloudSkillValidationError as exc:
+            raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc
+        except CloudSkillConflictError as exc:
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
+        _audit(repository, principal, skill.id, "cloud_skill_create")
+        return _skill_response(skill)
+
+    @router.patch("/skills/{skill_id}", response_model=CloudSkillResponse)
+    def update_skill(
+        skill_id: str,
+        payload: CloudSkillCreateRequest,
+        request: Request,
+    ) -> CloudSkillResponse:
+        principal = authorize(request)
+        require_csrf(request, principal)
+        try:
+            skill = service.update_skill(
+                skill_id,
+                name=payload.name,
+                kind=payload.kind,
+                description=payload.description,
+                tags=payload.tags,
+                content=payload.content,
+                steps_json=json.dumps(payload.steps),
+                parameters_json=json.dumps(payload.parameters),
+                now=utc_now(),
+            )
+        except CloudSkillValidationError as exc:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Skill not found") from exc
+        except CloudSkillConflictError as exc:
+            raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
+        _audit(repository, principal, skill.id, "cloud_skill_update")
+        return _skill_response(skill)
+
+    @router.delete("/skills/{skill_id}", status_code=status.HTTP_204_NO_CONTENT)
+    def delete_skill(skill_id: str, request: Request) -> None:
+        principal = authorize(request)
+        require_csrf(request, principal)
+        try:
+            service.delete_skill(skill_id)
+        except KeyError as exc:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Skill not found") from exc
+        _audit(repository, principal, skill_id, "cloud_skill_delete")
+
+    @router.get(
+        "/skills/{skill_id}/entitlements",
+        response_model=CloudSkillEntitlementListResponse,
+    )
+    def list_entitlements(skill_id: str, request: Request) -> CloudSkillEntitlementListResponse:
+        authorize(request)
+        return CloudSkillEntitlementListResponse(
+            skill_id=skill_id,
+            host_ids=service.list_hosts_for_skill(skill_id),
+        )
+
+    @router.post(
+        "/skills/{skill_id}/entitlements/{host_id}",
+        status_code=status.HTTP_204_NO_CONTENT,
+    )
+    def grant_entitlement(skill_id: str, host_id: str, request: Request) -> None:
+        principal = authorize(request)
+        require_csrf(request, principal)
+        try:
+            service.grant_entitlement(skill_id, host_id, now=utc_now())
+        except CloudSkillValidationError as exc:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc
+        _audit(repository, principal, skill_id, "cloud_skill_entitlement_grant")
+
+    @router.delete(
+        "/skills/{skill_id}/entitlements/{host_id}",
+        status_code=status.HTTP_204_NO_CONTENT,
+    )
+    def revoke_entitlement(skill_id: str, host_id: str, request: Request) -> None:
+        principal = authorize(request)
+        require_csrf(request, principal)
+        service.revoke_entitlement(skill_id, host_id, now=utc_now())
+        _audit(repository, principal, skill_id, "cloud_skill_entitlement_revoke")
+
+    @router.get(
+        "/hosts/{host_id}/skill-inventory",
+        response_model=HostSkillInventoryResponse,
+    )
+    def get_host_inventory(host_id: str, request: Request) -> HostSkillInventoryResponse:
+        authorize(request)
+        entry = service.get_host_inventory(host_id)
+        if entry is None:
+            return HostSkillInventoryResponse(host_id=host_id)
+        return HostSkillInventoryResponse(
+            host_id=entry.host_id,
+            payload=json.loads(entry.payload_json or "[]"),
+            reported_at=entry.reported_at,
+        )
+
+    return router
+
+
+# ----------------------------------------------------------------------
+# Host-scoped router (agent sync + inventory report)
+# ----------------------------------------------------------------------
+
+
+def create_skill_host_router(
+    *,
+    service: CloudSkillService,
+    auth_provider: AuthProvider,
+    version_prefix: str = "/internal/v1",
+) -> APIRouter:
+    router = APIRouter(prefix=version_prefix, tags=["skill-sync"])
+
+    def authorize_host(request: Request, host_id: str) -> None:
+        principal = auth_provider.authenticate(request)
+        if principal is None:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="unauthorized",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+        try:
+            principal.require_host(host_id)
+        except HostAuthorizationError as exc:
+            raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=str(exc)) from exc
+
+    @router.get("/hosts/{host_id}/skills/sync", response_model=CloudSkillSyncResponse)
+    def sync_skills(
+        host_id: str,
+        request: Request,
+        since_version: int | None = Query(default=None, ge=0),
+    ) -> CloudSkillSyncResponse:
+        authorize_host(request, host_id)
+        delta = service.fetch_host_delta(host_id, since_version)
+        return CloudSkillSyncResponse(
+            skills=[_skill_response(s) for s in delta.skills],
+            removed_ids=list(delta.removed_ids),
+            latest_version=delta.latest_version,
+            is_full_replace=delta.is_full_replace,
+        )
+
+    @router.post(
+        "/hosts/{host_id}/skills/inventory",
+        status_code=status.HTTP_204_NO_CONTENT,
+    )
+    def report_inventory(
+        host_id: str,
+        payload: HostSkillInventoryRequest,
+        request: Request,
+    ) -> None:
+        authorize_host(request, host_id)
+        service.record_host_inventory(
+            host_id,
+            json.dumps(payload.skills),
+            now=utc_now(),
+        )
+
+    return router
+
+
+# ----------------------------------------------------------------------
+# Helpers
+# ----------------------------------------------------------------------
+
+
+def _skill_response(skill) -> CloudSkillResponse:
+    return CloudSkillResponse(
+        id=skill.id,
+        name=skill.name,
+        kind=skill.kind,
+        description=skill.description,
+        tags=list(skill.tags),
+        revision=skill.revision,
+        created_at=skill.created_at,
+        updated_at=skill.updated_at,
+        content=skill.content,
+        steps=json.loads(skill.steps_json or "[]"),
+        parameters=json.loads(skill.parameters_json or "{}"),
+    )
+
+
+def _audit(repository, principal: Principal, skill_id: str, action: str) -> None:
+    repository.record_auth_audit(
+        AuthAuditEvent(
+            id=uuid4().hex,
+            occurred_at=utc_now(),
+            actor_principal_id=principal.id,
+            target_user_id=None,
+            action=action,
+            outcome="success",
+            correlation_id=current_correlation_id(),
+            metadata={"cloud_skill_id": skill_id},
+        )
+    )