openswarm/backend/apps/workflows/cloud/routes.py

"""HTTP surface for running a workflow on our servers instead of this machine.

Its own SubApp so the already-large workflows.py does not grow. Prefix:
/api/cloud_workflows. The local half of the handover (which timer is armed)
lives in cloud/handover.py; nothing here is the authority on entitlement, the
cloud re-decides that at create and again at dispatch.
"""
from __future__ import annotations

import asyncio
from contextlib import asynccontextmanager
from datetime import datetime
from typing import List, Literal, Optional, Union

from fastapi import HTTPException
from pydantic import BaseModel, ConfigDict
from typeguard import typechecked

from backend.apps.workflows import storage
from backend.apps.nine_router.lent_credential_refresh import lent_credential_loop
from backend.apps.workflows.cloud import client as cloud
from backend.apps.workflows.cloud.handover import TargetOutcome, hand_to_cloud, take_back
from backend.apps.workflows.cloud.run_files import LocalRunFile, described, downloads_root, fetch_missing
from backend.apps.workflows.cloud.status import CloudStatus, compute_status, epoch_to_datetime
from backend.apps.workflows.models import Workflow
from backend.config.Apps import SubApp


@asynccontextmanager
async def cloud_workflows_lifespan():
    # Lending a credential upward strips this device's ability to renew it, so something has to
    # ask the cloud for a fresh one before the old one dies. Without this, turning on a cloud
    # workflow quietly stops local agents a few hours later.
    task = asyncio.create_task(lent_credential_loop())
    try:
        yield
    finally:
        task.cancel()


cloud_workflows = SubApp("cloud_workflows", cloud_workflows_lifespan)


class TargetRequest(BaseModel):
    model_config = ConfigDict(validate_assignment=True)

    # The whole desired state, not a delta: where the schedule runs and whether it runs at all.
    target: Literal["device", "cloud"]
    enabled: bool


class CloudRunRow(BaseModel):
    model_config = ConfigDict(validate_assignment=True)

    id: str
    status: str
    started_at: Optional[datetime] = None
    finished_at: Optional[datetime] = None
    error: Optional[str] = None
    answer: Optional[str] = None
    notices: List[str] = []
    cost_usd: Optional[float] = None
    files: List[LocalRunFile] = []


class CloudRunsReady(BaseModel):
    model_config = ConfigDict(validate_assignment=True)

    state: Literal["ready"] = "ready"
    runs: List[CloudRunRow] = []
    # Where this machine puts a cloud run's files, so the UI can say it out loud even when a run
    # has none yet. A folder the user is told about is a folder they can find later.
    files_folder: str = ""


class CloudRunsUnavailable(BaseModel):
    model_config = ConfigDict(validate_assignment=True)

    # Same rule as the status route: not knowing is its own answer, and it shows no runs rather than "no runs".
    state: Literal["signed_out", "unknown"]
    detail: Optional[str] = None


CloudRunsResponse = Union[CloudRunsReady, CloudRunsUnavailable]


@typechecked
def p_workflow(workflow_id: str) -> Workflow:
    wf = storage.get_workflow(workflow_id)
    if not wf or wf.deleted_at is not None:
        raise HTTPException(status_code=404, detail="Workflow not found")
    return wf


@cloud_workflows.router.get("/{workflow_id}/status")
async def workflow_cloud_status(workflow_id: str) -> CloudStatus:
    return await compute_status(p_workflow(workflow_id))


@cloud_workflows.router.post("/{workflow_id}/target")
async def set_workflow_target(workflow_id: str, body: TargetRequest) -> TargetOutcome:
    wf = p_workflow(workflow_id)
    if body.target == "cloud":
        return await hand_to_cloud(wf, body.enabled)
    return await take_back(wf, body.enabled)


@cloud_workflows.router.get("/{workflow_id}/runs")
async def workflow_cloud_runs(workflow_id: str) -> CloudRunsResponse:
    wf = p_workflow(workflow_id)
    if not wf.cloud_workflow_id:
        return CloudRunsReady(runs=[], files_folder=downloads_root())
    try:
        runs = await cloud.list_runs(wf.cloud_workflow_id)
    except cloud.SignedOut:
        return CloudRunsUnavailable(state="signed_out")
    except cloud.CloudRefused as exc:
        # A 404 here is the hosted copy being gone, which is an empty history, not a broken one.
        if exc.status == 404:
            return CloudRunsReady(runs=[], files_folder=downloads_root())
        return CloudRunsUnavailable(state="unknown", detail=exc.message)
    except cloud.CloudUnreachable as exc:
        return CloudRunsUnavailable(state="unknown", detail=exc.detail)

    # Answered now, files fetched behind it. A 20MB attachment must not hold up the history the
    # user asked for, and a run's files appear a moment later without them doing anything.
    asyncio.create_task(fetch_missing(wf.cloud_workflow_id, wf, runs))

    return CloudRunsReady(
        files_folder=downloads_root(),
        runs=[
            CloudRunRow(
                id=r.id,
                status=r.status,
                started_at=epoch_to_datetime(r.started_at),
                finished_at=epoch_to_datetime(r.finished_at),
                error=r.error,
                answer=r.answer,
                notices=r.notices,
                cost_usd=r.cost_usd,
                files=described(wf, r),
            )
            for r in runs
        ],
    )