"""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 datetime import UTC, datetime 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, TaskCancellationResponse, TaskListItem, TaskListResponse, TaskPlannerDecisionItem, TaskPlannerDecisionListResponse, TaskStatusResponse, TaskSubmissionRequest, TaskSubmissionResponse, ) from fastapi import APIRouter, HTTPException, Query, Request, Response, 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", "cancelled" ] | 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.post( "/tasks/{task_id}/cancel", response_model=TaskCancellationResponse, responses={ status.HTTP_202_ACCEPTED: {"model": TaskCancellationResponse}, }, ) def cancel_task( task_id: str, request: Request, response: Response ) -> TaskCancellationResponse: _authorize(request, TASKS_SUBMIT_SCOPE) result = scheduler.store.request_task_cancellation( task_id, requested_at=datetime.now(UTC) ) if result == "not_found": raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"task {task_id!r} not found", ) if result == "already_terminal": raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=f"task {task_id!r} has already reached a terminal state", ) task = scheduler.store.get_task(task_id) assert task is not None if result == "requested" and task.status != "cancelled": response.status_code = status.HTTP_202_ACCEPTED return TaskCancellationResponse(task_id=task_id, status=task.status) @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, )