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

import json
from typing import TYPE_CHECKING, Callable, 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.governance import TaskSubmissionPolicyError, enforce_user_submission_policy
from cloud.sdk.models import (
    DeviceResponse,
    ErrorResponse,
    HostResponse,
    PluginRegistrationRequest,
    PluginResponse,
    TaskAttemptResponse,
    TaskListItem,
    TaskListResponse,
    TaskPlannerDecisionItem,
    TaskPlannerDecisionListResponse,
    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,
    csrf_validator: Callable[[Request, Principal], bool] | 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 principal.must_change_password:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="password must be changed before accessing this resource",
            )
        if not principal.has_scope(required_scope):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"missing required scope: {required_scope}",
            )
        if (
            principal.session_id is not None
            and request.method in {"POST", "PUT", "PATCH", "DELETE"}
            and (csrf_validator is None or not csrf_validator(request, principal))
        ):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="CSRF validation failed",
            )
        return principal

    @router.post(
        "/tasks",
        response_model=TaskSubmissionResponse,
        status_code=status.HTTP_201_CREATED,
    )
    def submit_task(
        payload: TaskSubmissionRequest,
        request: Request,
    ) -> TaskSubmissionResponse:
        principal = _authorize(request, TASKS_SUBMIT_SCOPE)
        try:
            task_constraints = _build_constraints(payload.constraints)
            _validate_task_target(pool, task_constraints)
            _enforce_user_policy(principal, scheduler.store, task_constraints)
        except TaskSubmissionPolicyError as exc:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=str(exc),
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=str(exc),
            ) from exc
        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,
            target_host_id=task.constraints.target_host_id,
            target_device_id=task.constraints.target_device_id,
            progress_step_index=task.progress_step_index,
            progress_step_status=task.progress_step_status,
            progress_summary=task.progress_summary,
            progress_updated_at=task.progress_updated_at,
        )

    @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,
                    target_host_id=task.constraints.target_host_id,
                    target_device_id=task.constraints.target_device_id,
                    created_at=task.created_at,
                    progress_step_index=task.progress_step_index,
                    progress_step_status=task.progress_step_status,
                    progress_summary=task.progress_summary,
                    progress_updated_at=task.progress_updated_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(
        "/tasks/{task_id}/planner-decisions",
        response_model=TaskPlannerDecisionListResponse,
    )
    def list_task_planner_decisions(
        task_id: str,
        request: Request,
        attempt: int = Query(ge=0),
    ) -> TaskPlannerDecisionListResponse:
        _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",
            )
        records = scheduler.store.list_planner_decisions(
            task_id=task_id,
            attempt=attempt,
        )
        items: list[TaskPlannerDecisionItem] = []
        for rec in records:
            try:
                parsed_args = (
                    json.loads(rec.arguments_json) if rec.arguments_json else {}
                )
            except Exception:
                parsed_args = {}
            items.append(
                TaskPlannerDecisionItem(
                    step_index=rec.step_index,
                    attempt=rec.attempt,
                    system_prompt=rec.system_prompt,
                    user_prompt=rec.user_prompt,
                    tool_name=rec.tool_name,
                    arguments=parsed_args,
                    created_at=rec.created_at,
                )
            )
        return TaskPlannerDecisionListResponse(items=items)

    @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 "",
                planner_transport=h.planner_transport,
            )
            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),
        target_host_id=model.target_host_id,
        target_device_id=model.target_device_id,
    )


def _validate_task_target(pool, constraints) -> None:
    if constraints.target_device_id and not constraints.target_host_id:
        raise ValueError("target_device_id requires target_host_id")
    if constraints.target_host_id is None:
        return
    if not any(
        host.host_id == constraints.target_host_id for host in pool.list_hosts()
    ):
        raise ValueError(f"target host {constraints.target_host_id!r} is not known")
    if constraints.target_device_id is not None and not any(
        device.host_id == constraints.target_host_id
        and device.device_id == constraints.target_device_id
        for device in pool.list_devices()
    ):
        raise ValueError(
            f"target device {constraints.target_device_id!r} is not owned by "
            f"host {constraints.target_host_id!r}"
        )


def _enforce_user_policy(principal, store, constraints) -> None:
    if not principal.id.startswith("user:"):
        return
    policy = store.get_user_submission_policy(principal.id.removeprefix("user:"))
    enforce_user_submission_policy(
        policy,
        target_host_id=constraints.target_host_id,
        target_device_id=constraints.target_device_id,
    )