agentic-mobile-control/packages/cloud-platform/cloud/sdk/api.py

"""Versioned REST API for external integrators.

Capability: ``platform-sdk``.

Mirrors ``api/console.py``'s ``create_console_router`` shape: a factory that
returns a ``fastapi.APIRouter`` mounted under a versioned ``/v1`` prefix.
Every route flows through an ``AuthProvider`` hook (no-op default) so real
authentication can be added later without changing route signatures.
"""

from __future__ import annotations

from typing import TYPE_CHECKING, Literal

from cloud.auth import (
    PLUGINS_ADMIN_SCOPE,
    PLUGINS_READ_SCOPE,
    POOL_READ_SCOPE,
    TASKS_READ_SCOPE,
    TASKS_SUBMIT_SCOPE,
    AuthProvider,
    NullAuthProvider,
    Principal,
)
from cloud.sdk.models import (
    DeviceResponse,
    ErrorResponse,
    HostResponse,
    PluginRegistrationRequest,
    PluginResponse,
    TaskAttemptResponse,
    TaskListItem,
    TaskListResponse,
    TaskStatusResponse,
    TaskSubmissionRequest,
    TaskSubmissionResponse,
)
from fastapi import APIRouter, HTTPException, Query, Request, status

if TYPE_CHECKING:
    from cloud.plugins import PluginRegistry
    from cloud.pool import DevicePool
    from cloud.scheduler import TaskScheduler


def create_cloud_router(
    *,
    pool: "DevicePool",
    scheduler: "TaskScheduler",
    plugin_registry: "PluginRegistry",
    auth_provider: AuthProvider | None = None,
    version_prefix: str = "/v1",
) -> APIRouter:
    """Build the ``/v1`` APIRouter exposing the platform SDK surface."""
    auth = auth_provider or NullAuthProvider()
    router = APIRouter(prefix=version_prefix, tags=["cloud-platform"])

    def _authorize(request: Request, required_scope: str) -> Principal:
        principal = auth.authenticate(request)
        if principal is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="unauthorized",
                headers={"WWW-Authenticate": "Bearer"},
            )
        if not principal.has_scope(required_scope):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"missing required scope: {required_scope}",
            )
        return principal

    @router.post(
        "/tasks",
        response_model=TaskSubmissionResponse,
        status_code=status.HTTP_201_CREATED,
    )
    def submit_task(
        payload: TaskSubmissionRequest,
        request: Request,
    ) -> TaskSubmissionResponse:
        _authorize(request, TASKS_SUBMIT_SCOPE)
        task_constraints = _build_constraints(payload.constraints)
        try:
            task_id = scheduler.submit(
                goal=payload.goal,
                workflow_definition_id=payload.workflow_definition_id,
                constraints=task_constraints,
            )
        except (ValueError, RuntimeError) as exc:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=str(exc),
            ) from exc
        return TaskSubmissionResponse(task_id=task_id)

    @router.get("/tasks/{task_id}", response_model=TaskStatusResponse)
    def get_task_status(task_id: str, request: Request) -> TaskStatusResponse:
        _authorize(request, TASKS_READ_SCOPE)
        task = scheduler.store.get_task(task_id)
        if task is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"task {task_id!r} not found",
            )
        return TaskStatusResponse(
            id=task.id,
            status=task.status,
            goal=task.goal,
            workflow_definition_id=task.workflow_definition_id,
            assigned_device_id=task.assigned_device_id,
            assigned_host_id=task.assigned_host_id,
            attempt_count=task.attempt_count,
            lease_expires_at=task.lease_expires_at,
            failure_reason=task.failure_reason,
        )

    @router.get("/tasks", response_model=TaskListResponse)
    def list_tasks(
        request: Request,
        status_filter: Literal[
            "queued", "assigned", "dispatched", "done", "failed"
        ]
        | None = Query(default=None, alias="status"),
        limit: int = Query(default=50, ge=1, le=100),
        offset: int = Query(default=0, ge=0),
    ) -> TaskListResponse:
        _authorize(request, TASKS_READ_SCOPE)
        tasks = scheduler.store.list_tasks(
            status=status_filter,
            limit=limit,
            offset=offset,
        )
        total = scheduler.store.count_tasks(status=status_filter)
        return TaskListResponse(
            items=[
                TaskListItem(
                    id=task.id,
                    status=task.status,
                    goal=task.goal,
                    workflow_definition_id=task.workflow_definition_id,
                    assigned_device_id=task.assigned_device_id,
                    assigned_host_id=task.assigned_host_id,
                    attempt_count=task.attempt_count,
                    failure_reason=task.failure_reason,
                    created_at=task.created_at,
                )
                for task in tasks
            ],
            total=total,
            limit=limit,
            offset=offset,
        )

    @router.get(
        "/tasks/{task_id}/attempts",
        response_model=list[TaskAttemptResponse],
    )
    def list_task_attempts(
        task_id: str,
        request: Request,
    ) -> list[TaskAttemptResponse]:
        _authorize(request, TASKS_READ_SCOPE)
        task = scheduler.store.get_task(task_id)
        if task is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail=f"task {task_id!r} not found",
            )
        attempts = scheduler.store.list_task_attempts(task_id)
        return [
            TaskAttemptResponse(
                task_id=attempt.task_id,
                attempt=attempt.attempt,
                lease_id=attempt.lease_id,
                host_id=attempt.host_id,
                device_id=attempt.device_id,
                status=attempt.status,
                lease_expires_at=attempt.lease_expires_at,
                created_at=attempt.created_at,
                completed_at=attempt.completed_at,
                failure_reason=attempt.failure_reason,
                terminal_result=attempt.terminal_result,
            )
            for attempt in attempts
        ]

    @router.get("/devices", response_model=list[DeviceResponse])
    def list_devices(request: Request) -> list[DeviceResponse]:
        _authorize(request, POOL_READ_SCOPE)
        return [
            DeviceResponse(
                device_id=d.device_id,
                host_id=d.host_id,
                driver_type=d.driver_type,
                status=d.status,
                capability_tags=list(d.capability_tags),
            )
            for d in pool.list_devices()
        ]

    @router.get("/hosts", response_model=list[HostResponse])
    def list_hosts(request: Request) -> list[HostResponse]:
        _authorize(request, POOL_READ_SCOPE)
        return [
            HostResponse(
                host_id=h.host_id,
                address=h.address,
                last_seen_at=h.last_seen_at.isoformat() if h.last_seen_at else "",
            )
            for h in pool.list_hosts()
        ]

    @router.get("/plugins", response_model=list[PluginResponse])
    def list_plugins(request: Request) -> list[PluginResponse]:
        _authorize(request, PLUGINS_READ_SCOPE)
        return [
            PluginResponse(
                name=manifest.name,
                version=manifest.version,
                entry_point_kind=manifest.entry_point_kind,
                target=manifest.target,
                wired=wired,
            )
            for manifest, wired in plugin_registry.list()
        ]

    @router.post(
        "/plugins",
        response_model=PluginResponse,
        status_code=status.HTTP_201_CREATED,
        responses={
            status.HTTP_400_BAD_REQUEST: {"model": ErrorResponse},
            status.HTTP_409_CONFLICT: {"model": ErrorResponse},
        },
    )
    def register_plugin(
        payload: PluginRegistrationRequest,
        request: Request,
    ) -> PluginResponse:
        _authorize(request, PLUGINS_ADMIN_SCOPE)
        from cloud.plugins import (
            DriverRegistryUnavailableError,
            DuplicatePluginError,
            PluginManifest,
            PluginValidationError,
        )

        manifest = PluginManifest(
            name=payload.name,
            version=payload.version,
            entry_point_kind=payload.entry_point_kind,
            target=payload.target,
        )
        try:
            plugin_registry.register(manifest)
        except DuplicatePluginError as exc:
            raise HTTPException(
                status_code=status.HTTP_409_CONFLICT,
                detail=str(exc),
            ) from exc
        except DriverRegistryUnavailableError as exc:
            raise HTTPException(
                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                detail=str(exc),
            ) from exc
        except (PluginValidationError, ValueError) as exc:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=str(exc),
            ) from exc
        stored = plugin_registry.store.get_plugin(manifest.name)
        wired = stored[1] if stored is not None else False
        return PluginResponse(
            name=manifest.name,
            version=manifest.version,
            entry_point_kind=manifest.entry_point_kind,
            target=manifest.target,
            wired=wired,
        )

    return router


def _build_constraints(model):
    from cloud.scheduler import TaskConstraints

    return TaskConstraints(
        driver_type=model.driver_type,
        capability_tags=list(model.capability_tags),
    )