[eric] merge eric/cloud-ui: choose where a workflow runs, and say plainly when the cloud cannot

backend/apps/outputs/publish_cloud.py

@@ -3,18 +3,10 @@ it back down. Reads the bearer directly (publish works for any signed-in account
 not just pro/free-trial), matching the cloud's requireAuthedUser gate."""
 from __future__ import annotations
 
-from typing import Optional
-
 import httpx
 
 from backend.apps.outputs.publish_common import PublishError
-from backend.apps.settings.credentials import OPENSWARM_DEFAULT_PROXY_URL
-
-
-def p_cloud_auth(settings) -> tuple[Optional[str], str]:
-    base = (getattr(settings, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL).rstrip("/")
-    token = getattr(settings, "openswarm_bearer_token", None)
-    return token, base
+from backend.apps.settings.credentials import account_auth
 
 
 def p_safe_detail(resp: httpx.Response, fallback: str) -> str:
@@ -31,7 +23,7 @@ def p_safe_detail(resp: httpx.Response, fallback: str) -> str:
 async def upload_to_cloud(
     settings, *, output_id: str, name: str, slug_hint: str, bundle: bytes, override: bool
 ) -> dict:
-    token, base = p_cloud_auth(settings)
+    token, base = account_auth(settings)
     if not token:
         raise PublishError("Sign in to your OpenSwarm account to publish apps.")
     try:
@@ -51,7 +43,7 @@ async def upload_to_cloud(
 
 
 async def unpublish_from_cloud(settings, slug: str) -> None:
-    token, base = p_cloud_auth(settings)
+    token, base = account_auth(settings)
     if not token:
         raise PublishError("Sign in to your OpenSwarm account to manage published apps.")
     try:

backend/apps/settings/credentials.py

@@ -28,6 +28,15 @@ def proxy_auth(settings: AppSettings) -> tuple[str | None, str | None]:
     return (None, None)
 
 
+def account_auth(settings: AppSettings) -> tuple[str | None, str]:
+    """(bearer, base_url) for cloud routes that identify the ACCOUNT rather than
+    route LLM traffic: publishing, hosted workflows, anything behind the cloud's
+    requireAuthedUser. Deliberately ignores connection_mode, because a signed-in
+    own-key user still has a real account. Bearer is None when signed out."""
+    base = (getattr(settings, "openswarm_proxy_url", None) or OPENSWARM_DEFAULT_PROXY_URL).rstrip("/")
+    return (getattr(settings, "openswarm_bearer_token", None) or None, base)
+
+
 def p_check_9router() -> bool:
     """Check if 9Router is running locally."""
     try:

backend/apps/workflows/cloud/client.py

@@ -0,0 +1,296 @@
+"""Talk to the cloud's hosted-workflow routes on behalf of this desktop.
+
+Two failure kinds, kept apart on purpose. CloudRefused means the server answered
+and said no, and its message is written for the user, so it is shown verbatim.
+CloudUnreachable means we never got an answer, which is NOT a no: the caller must
+render "we cannot tell" rather than inventing a denial or a usage number.
+"""
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List, Optional
+
+import httpx
+from pydantic import BaseModel, ConfigDict, Field
+from typeguard import typechecked
+
+from backend.apps.settings.credentials import account_auth
+from backend.apps.workflows.cloud.schedule import CloudSchedule
+
+# The cloud router is mounted at /api/workflows and a trailing slash 404s there, so the collection paths are the empty string, not "/".
+COLLECTION = ""
+TIMEOUT_SECONDS = 8.0
+
+
+class CloudRefused(Exception):
+    """The cloud answered and declined. `message` is user-facing prose."""
+
+    def __init__(self, message: str, status: int) -> None:
+        super().__init__(message)
+        self.message = message
+        self.status = status
+
+
+class CloudUnreachable(Exception):
+    """No answer at all: offline, timed out, 5xx, or a body we could not parse."""
+
+    def __init__(self, detail: str) -> None:
+        super().__init__(detail)
+        self.detail = detail
+
+
+class SignedOut(Exception):
+    """No bearer on this machine, or the cloud rejected the one we have."""
+
+
+class CloudLimits(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    workflows: int = 0
+    runs_per_month: int = 0
+    concurrent: int = 0
+
+
+class CloudUsage(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    workflows_enabled: int = 0
+    runs_this_month: int = 0
+
+
+class CloudCapability(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    ok: bool
+    reason: Optional[str] = None
+
+
+class HostedWorkflow(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    id: str
+    enabled: bool = False
+    next_run_at: Optional[int] = None
+
+
+class CloudPreflight(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    # None when the control plane did not name the plan. Entitlement is read off limits, never off this string.
+    plan: Optional[str] = None
+    limits: CloudLimits = Field(default_factory=CloudLimits)
+    usage: CloudUsage = Field(default_factory=CloudUsage)
+    # None when this control plane predates the capability check, which is a "we cannot tell", never an "it is fine".
+    capability: Optional[CloudCapability] = None
+    hosted: Optional[HostedWorkflow] = None
+
+
+class CloudRun(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    id: str
+    status: str
+    started_at: Optional[int] = None
+    finished_at: Optional[int] = None
+    error: Optional[str] = None
+    answer: Optional[str] = None
+    notices: List[str] = Field(default_factory=list)
+    cost_usd: Optional[float] = None
+
+
+@typechecked
+def p_message(resp: httpx.Response, fallback: str) -> str:
+    """The cloud's own words for a refusal. Hono renders an HTTPException as a bare
+    text/plain sentence, so a JSON-only reader would silently swap every written
+    reason for a generic one; both shapes are read here."""
+    raw = (resp.text or "").strip()
+    if not raw:
+        return fallback
+    try:
+        body = resp.json()
+    except (json.JSONDecodeError, ValueError):
+        body = None
+    if isinstance(body, dict):
+        for key in ("message", "error"):
+            value = body.get(key)
+            if isinstance(value, str) and value:
+                return value
+        return fallback
+    # Anything short and prose-shaped is the message itself; an HTML error page is not.
+    if len(raw) <= 500 and not raw.startswith("<"):
+        return raw
+    return fallback
+
+
+@typechecked
+async def p_call(method: str, path: str, body: Optional[Dict[str, Any]] = None) -> Any:
+    from backend.apps.settings.store import load_settings
+
+    token, base = account_auth(load_settings())
+    if not token:
+        raise SignedOut()
+    try:
+        async with httpx.AsyncClient(timeout=TIMEOUT_SECONDS) as client:
+            resp = await client.request(
+                method,
+                f"{base}/api/workflows{path}",
+                headers={"Authorization": f"Bearer {token}"},
+                json=body,
+            )
+    except httpx.HTTPError as exc:
+        raise CloudUnreachable(f"{type(exc).__name__}") from exc
+    if resp.status_code == 401:
+        raise SignedOut()
+    # A 404 on a route we expect means an older control plane; the caller decides whether that is fatal.
+    if resp.status_code >= 500:
+        raise CloudUnreachable(f"the cloud returned {resp.status_code}")
+    if resp.status_code >= 400:
+        raise CloudRefused(p_message(resp, "The cloud declined this request."), resp.status_code)
+    try:
+        return resp.json()
+    except (json.JSONDecodeError, ValueError) as exc:
+        raise CloudUnreachable("the cloud sent a response we could not read") from exc
+
+
+@typechecked
+def p_hosted(raw: Any) -> Optional[HostedWorkflow]:
+    if not isinstance(raw, dict):
+        return None
+    ident = raw.get("id")
+    if not isinstance(ident, str):
+        return None
+    nxt = raw.get("next_run_at")
+    return HostedWorkflow(
+        id=ident,
+        enabled=bool(raw.get("enabled")),
+        next_run_at=nxt if isinstance(nxt, int) else None,
+    )
+
+
+@typechecked
+def p_allowance(raw: Dict[str, Any]) -> tuple[CloudLimits, CloudUsage]:
+    """What the plan allows and what has been spent, from either shape the cloud answers in."""
+    limits = raw.get("limits") if isinstance(raw.get("limits"), dict) else {}
+    usage = raw.get("usage") if isinstance(raw.get("usage"), dict) else {}
+    return (
+        CloudLimits(
+            workflows=int(limits.get("workflows") or 0),
+            runs_per_month=int(limits.get("runsPerMonth") or 0),
+            concurrent=int(limits.get("concurrent") or 0),
+        ),
+        CloudUsage(
+            workflows_enabled=int(usage.get("enabled") or 0),
+            runs_this_month=int(usage.get("runs_this_month") or 0),
+        ),
+    )
+
+
+@typechecked
+async def preflight(definition: Dict[str, Any], hosted_id: Optional[str]) -> CloudPreflight:
+    """Plan, spend, and whether the runner could do this job, in one round trip.
+    Falls back to the plain list route when the control plane has no preflight,
+    which leaves `capability` unknown rather than pretending it passed."""
+    body: Dict[str, Any] = {"definition": definition}
+    if hosted_id:
+        body["hosted_id"] = hosted_id
+    try:
+        raw = await p_call("POST", "/preflight", body)
+    except CloudRefused as exc:
+        if exc.status != 404:
+            raise
+        return await p_preflight_from_list(hosted_id)
+    if not isinstance(raw, dict):
+        raise CloudUnreachable("the cloud sent a preflight we could not read")
+    limits, usage = p_allowance(raw)
+    capability = raw.get("capability")
+    return CloudPreflight(
+        plan=raw.get("plan") if isinstance(raw.get("plan"), str) else None,
+        limits=limits,
+        usage=usage,
+        capability=(
+            CloudCapability(ok=bool(capability.get("ok")), reason=capability.get("reason"))
+            if isinstance(capability, dict)
+            else None
+        ),
+        hosted=p_hosted(raw.get("hosted")),
+    )
+
+
+@typechecked
+async def p_preflight_from_list(hosted_id: Optional[str]) -> CloudPreflight:
+    raw = await p_call("GET", COLLECTION)
+    if not isinstance(raw, dict):
+        raise CloudUnreachable("the cloud sent a workflow list we could not read")
+    rows = raw.get("workflows") if isinstance(raw.get("workflows"), list) else []
+    match = next((r for r in rows if isinstance(r, dict) and r.get("id") == hosted_id), None)
+    limits, usage = p_allowance(raw)
+    return CloudPreflight(plan=None, limits=limits, usage=usage, capability=None, hosted=p_hosted(match))
+
+
+@typechecked
+async def put_workflow(
+    *, hosted_id: Optional[str], name: str, definition: Dict[str, Any], schedule: CloudSchedule
+) -> HostedWorkflow:
+    """Create the hosted copy, or re-push onto the existing row so an edited
+    workflow stops running last week's prose."""
+    body = {"name": name, "definition": definition, "schedule": schedule.model_dump()}
+    if hosted_id:
+        try:
+            raw = await p_call("POST", f"/{hosted_id}/update", body)
+            hosted = p_hosted(raw)
+            if hosted:
+                return hosted
+        except CloudRefused as exc:
+            # 404 is the row being gone (deleted elsewhere, or a control plane with no update route); make a fresh one.
+            if exc.status != 404:
+                raise
+    raw = await p_call("POST", COLLECTION, body)
+    hosted = p_hosted(raw)
+    if not hosted:
+        raise CloudUnreachable("the cloud accepted the workflow but did not say which one")
+    return hosted
+
+
+@typechecked
+async def set_enabled(hosted_id: str, enabled: bool) -> HostedWorkflow:
+    raw = await p_call("POST", f"/{hosted_id}/enable", {"enabled": enabled})
+    hosted = p_hosted(raw)
+    if not hosted:
+        raise CloudUnreachable("the cloud did not report the workflow back")
+    return hosted
+
+
+@typechecked
+async def delete_hosted(hosted_id: str) -> None:
+    """Stop the cloud copy. A 404 is success: it is already gone."""
+    try:
+        await p_call("POST", f"/{hosted_id}/delete", {})
+    except CloudRefused as exc:
+        if exc.status != 404:
+            raise
+
+
+@typechecked
+async def list_runs(hosted_id: str) -> List[CloudRun]:
+    raw = await p_call("GET", f"/{hosted_id}/runs")
+    rows = raw.get("runs") if isinstance(raw, dict) else None
+    if not isinstance(rows, list):
+        raise CloudUnreachable("the cloud sent a run list we could not read")
+    out: List[CloudRun] = []
+    for row in rows:
+        if not isinstance(row, dict):
+            continue
+        notices = row.get("notices")
+        out.append(
+            CloudRun(
+                id=str(row.get("id") or ""),
+                status=str(row.get("status") or "unknown"),
+                started_at=row.get("started_at") if isinstance(row.get("started_at"), int) else None,
+                finished_at=row.get("finished_at") if isinstance(row.get("finished_at"), int) else None,
+                error=row.get("error") if isinstance(row.get("error"), str) else None,
+                answer=row.get("answer") if isinstance(row.get("answer"), str) else None,
+                notices=[n for n in notices if isinstance(n, str)] if isinstance(notices, list) else [],
+                cost_usd=row.get("cost_usd") if isinstance(row.get("cost_usd"), (int, float)) else None,
+            )
+        )
+    return out

backend/apps/workflows/cloud/definition.py

@@ -0,0 +1,61 @@
+"""The copy of a workflow the cloud is allowed to hold.
+
+Two jobs. First, drop everything the runner cannot use: local session ids, run
+history, dashboard placement, and the escalation tiers (which carry the user's
+phone number and could not ring anyone from a container anyway). Second, be
+stable, so hashing it detects a real edit and not the clock ticking.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import Any, Dict
+
+from typeguard import typechecked
+
+from backend.apps.workflows.models import Workflow
+
+# Anything that changes on its own, points at something only this machine has, or is nobody else's business.
+LOCAL_ONLY_FIELDS = (
+    "deleted_at",
+    "draft_steps",
+    "next_run_at",
+    "last_run_at",
+    "last_run_status",
+    "last_run_id",
+    "created_at",
+    "updated_at",
+    "source_session_id",
+    "edit_agent_session_id",
+    "schedule_agent_session_id",
+    "last_test_session_id",
+    "dashboard_id",
+    "unsaved",
+    "auto_named",
+    "tested_signature",
+    "step_tool_usage",
+    "permissions",
+    "cloud_workflow_id",
+    "cloud_definition_signature",
+)
+
+
+@typechecked
+def cloud_definition(wf: Workflow) -> Dict[str, Any]:
+    body = wf.model_dump(mode="json")
+    for field in LOCAL_ONLY_FIELDS:
+        body.pop(field, None)
+    # The cloud owns the timer for this copy; a live schedule inside it would be a second one.
+    schedule = body.get("schedule")
+    if isinstance(schedule, dict):
+        schedule["enabled"] = False
+    body["execution_target"] = "cloud"
+    return body
+
+
+@typechecked
+def definition_signature(definition: Dict[str, Any], schedule: Dict[str, Any]) -> str:
+    """Fingerprint of exactly what we last handed the cloud, schedule included,
+    so "your edits are not up there yet" is a fact rather than a guess."""
+    payload = json.dumps({"definition": definition, "schedule": schedule}, sort_keys=True)
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

backend/apps/workflows/cloud/handover.py

@@ -0,0 +1,114 @@
+"""Moving one workflow's timer between this machine and our servers.
+
+Exactly one of the two may be armed at any moment. `execution_target` flips to
+"cloud" only once the cloud has taken the workflow, and back to "device" only
+once the cloud has let it go, so the window where both would fire does not
+exist. Every caller that stops a workflow (the toggle, Trash, purge) goes
+through take_back for the same reason.
+"""
+from __future__ import annotations
+
+import logging
+from datetime import datetime
+from typing import Optional
+
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.workflows import scheduler, storage
+from backend.apps.workflows.cloud import client as cloud
+from backend.apps.workflows.cloud.definition import cloud_definition, definition_signature
+from backend.apps.workflows.cloud.schedule import ScheduleSupported, to_cloud_schedule
+from backend.apps.workflows.cloud.status import epoch_to_datetime
+from backend.apps.workflows.models import Workflow
+
+logger = logging.getLogger(__name__)
+
+SIGN_IN_MESSAGE = "Sign in to your OpenSwarm account to run workflows in the cloud."
+UNREACHABLE_UP = (
+    "Couldn't reach the cloud, so nothing was scheduled there. "
+    "This workflow still runs on this device. Try again in a moment."
+)
+UNREACHABLE_DOWN = (
+    "Couldn't reach the cloud to stop the cloud schedule, so nothing changed. "
+    "It keeps running in the cloud until this goes through."
+)
+
+
+class TargetOutcome(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    ok: bool
+    # Present only when ok is false, and written for the user: usually the cloud's own words.
+    message: Optional[str] = None
+
+
+@typechecked
+async def hand_to_cloud(wf: Workflow, enabled: bool) -> TargetOutcome:
+    mapping = to_cloud_schedule(wf.schedule)
+    if not isinstance(mapping, ScheduleSupported):
+        return TargetOutcome(ok=False, message=mapping.reason)
+    if enabled and not scheduler.is_schedule_configured(wf.schedule):
+        return TargetOutcome(ok=False, message="Finish setting up the schedule before choosing where it runs.")
+    definition = cloud_definition(wf)
+    try:
+        hosted = await cloud.put_workflow(
+            hosted_id=wf.cloud_workflow_id,
+            name=wf.title or "Workflow",
+            definition=definition,
+            schedule=mapping.schedule,
+        )
+        if hosted.enabled != enabled:
+            hosted = await cloud.set_enabled(hosted.id, enabled)
+    except cloud.SignedOut:
+        return TargetOutcome(ok=False, message=SIGN_IN_MESSAGE)
+    except cloud.CloudRefused as exc:
+        return TargetOutcome(ok=False, message=exc.message)
+    except cloud.CloudUnreachable as exc:
+        logger.info("cloud workflow push unreachable for %s: %s", wf.id, exc.detail)
+        return TargetOutcome(ok=False, message=UNREACHABLE_UP)
+
+    wf.execution_target = "cloud"
+    wf.cloud_workflow_id = hosted.id
+    wf.cloud_definition_signature = definition_signature(definition, mapping.schedule.model_dump())
+    wf.schedule.enabled = enabled
+    wf.next_run_at = epoch_to_datetime(hosted.next_run_at) if enabled else None
+    wf.updated_at = datetime.now()
+    storage.save_workflow(wf)
+    scheduler.kick()
+    return TargetOutcome(ok=True)
+
+
+@typechecked
+async def take_back(wf: Workflow, enabled: bool) -> TargetOutcome:
+    if wf.cloud_workflow_id:
+        try:
+            await cloud.delete_hosted(wf.cloud_workflow_id)
+        except cloud.SignedOut:
+            # Signed out means the cloud copy is unreachable, not gone; arming our timer here would double-fire it.
+            return TargetOutcome(ok=False, message=SIGN_IN_MESSAGE)
+        except cloud.CloudRefused as exc:
+            return TargetOutcome(ok=False, message=exc.message)
+        except cloud.CloudUnreachable as exc:
+            logger.info("cloud workflow delete unreachable for %s: %s", wf.id, exc.detail)
+            return TargetOutcome(ok=False, message=UNREACHABLE_DOWN)
+
+    wf.execution_target = "device"
+    wf.cloud_workflow_id = None
+    wf.cloud_definition_signature = None
+    wf.schedule.enabled = enabled and scheduler.is_schedule_configured(wf.schedule)
+    wf.next_run_at = scheduler.compute_next_fire(wf) if wf.schedule.enabled else None
+    wf.updated_at = datetime.now()
+    storage.save_workflow(wf)
+    scheduler.kick()
+    return TargetOutcome(ok=True)
+
+
+@typechecked
+async def release_before_removing(wf: Workflow) -> TargetOutcome:
+    """Take the cloud copy down before a workflow disappears from this machine.
+    Trash and purge both call it: a hosted row nobody can see any more still
+    runs on its own schedule, and still costs the user money."""
+    if wf.execution_target != "cloud" and not wf.cloud_workflow_id:
+        return TargetOutcome(ok=True)
+    return await take_back(wf, False)

backend/apps/workflows/cloud/routes.py

@@ -0,0 +1,124 @@
+"""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
+
+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.workflows.cloud import client as cloud
+from backend.apps.workflows.cloud.handover import TargetOutcome, hand_to_cloud, take_back
+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():
+    yield
+
+
+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
+
+
+class CloudRunsReady(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    state: Literal["ready"] = "ready"
+    runs: List[CloudRunRow] = []
+
+
+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=[])
+    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=[])
+        return CloudRunsUnavailable(state="unknown", detail=exc.message)
+    except cloud.CloudUnreachable as exc:
+        return CloudRunsUnavailable(state="unknown", detail=exc.detail)
+    return CloudRunsReady(
+        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,
+            )
+            for r in runs
+        ]
+    )

backend/apps/workflows/cloud/schedule.py

@@ -0,0 +1,122 @@
+"""Map a local schedule onto the cloud scheduler's much smaller vocabulary.
+
+The cloud speaks two cadences: repeat every N minutes, or once a day at a UTC
+time. Everything else this app can express (set weekdays, monthly, every third
+day, stop after N runs) has no cloud equivalent, and silently rounding one of
+them off would fire a workflow on days the user never picked. So anything that
+does not map exactly is refused here, in the user's own words, and stays on
+their machine where it already works.
+"""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Literal, Optional, Union
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.workflows.models import ScheduleConfig
+from backend.apps.workflows.scheduler import host_timezone_name
+
+CADENCE_PREFIX = "Cloud runs repeat on an interval or once a day."
+
+
+class CloudIntervalSchedule(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    kind: Literal["interval"] = "interval"
+    minutes: int
+
+
+class CloudDailySchedule(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    kind: Literal["daily"] = "daily"
+    hour_utc: int
+    minute_utc: int
+
+
+CloudSchedule = Union[CloudIntervalSchedule, CloudDailySchedule]
+
+
+class ScheduleSupported(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    supported: Literal[True] = True
+    schedule: CloudSchedule
+
+
+class ScheduleUnsupported(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    supported: Literal[False] = False
+    reason: str
+
+
+ScheduleMapping = Union[ScheduleSupported, ScheduleUnsupported]
+
+
+@typechecked
+def p_zone(name: str) -> ZoneInfo:
+    if not name or name == "local":
+        name = host_timezone_name()
+    try:
+        return ZoneInfo(name)
+    except ZoneInfoNotFoundError:
+        return ZoneInfo("UTC")
+
+
+@typechecked
+def p_utc_time_of_day(sched: ScheduleConfig, ref: Optional[datetime] = None) -> CloudDailySchedule:
+    """The user's wall-clock time expressed in UTC, using today's offset."""
+    zone = p_zone(sched.timezone)
+    local = (ref or datetime.now(zone)).astimezone(zone)
+    at = local.replace(hour=sched.hour, minute=sched.minute, second=0, microsecond=0)
+    utc = at.astimezone(ZoneInfo("UTC"))
+    return CloudDailySchedule(hour_utc=utc.hour, minute_utc=utc.minute)
+
+
+@typechecked
+def to_cloud_schedule(sched: ScheduleConfig) -> ScheduleMapping:
+    if sched.max_runs is not None:
+        return ScheduleUnsupported(
+            reason=(
+                f"This schedule stops itself after {sched.max_runs} "
+                f"run{'' if sched.max_runs == 1 else 's'}, and the cloud scheduler cannot count down "
+                "to a stop. Remove the limit to run it in the cloud."
+            ),
+        )
+    if sched.ends_at is not None:
+        return ScheduleUnsupported(
+            reason=(
+                "This schedule has an end date, and the cloud scheduler cannot honour one. "
+                "Remove the end date to run it in the cloud."
+            ),
+        )
+    if sched.repeat_unit == "minute":
+        return ScheduleSupported(schedule=CloudIntervalSchedule(minutes=max(5, sched.repeat_every)))
+    if sched.repeat_unit == "hour":
+        return ScheduleSupported(schedule=CloudIntervalSchedule(minutes=max(5, sched.repeat_every * 60)))
+    if sched.repeat_unit == "day" and sched.repeat_every == 1:
+        return ScheduleSupported(schedule=p_utc_time_of_day(sched))
+    if sched.repeat_unit == "day":
+        return ScheduleUnsupported(
+            reason=(
+                f"{CADENCE_PREFIX} This one runs every {sched.repeat_every} days at a set time, "
+                "which the cloud scheduler cannot do yet, so it stays on this device."
+            ),
+        )
+    if sched.repeat_unit == "week":
+        return ScheduleUnsupported(
+            reason=(
+                f"{CADENCE_PREFIX} This one runs on the weekdays you picked, "
+                "which the cloud scheduler cannot do yet, so it stays on this device."
+            ),
+        )
+    return ScheduleUnsupported(
+        reason=(
+            f"{CADENCE_PREFIX} This one runs monthly, "
+            "which the cloud scheduler cannot do yet, so it stays on this device."
+        ),
+    )

backend/apps/workflows/cloud/status.py

@@ -0,0 +1,133 @@
+"""What the desktop is allowed to say about cloud runs for one workflow.
+
+Three answers, and they are different answers. `ready` is the cloud having told
+us something. `signed_out` is a known no-account. `unknown` is us not knowing,
+which is deliberately NOT a denial and carries no numbers at all: a failed fetch
+that renders as "0 runs left" or "not entitled" is the bug this shape prevents.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Literal, Optional, Union
+
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.workflows import storage
+from backend.apps.workflows.cloud import client as cloud
+from backend.apps.workflows.cloud.definition import cloud_definition, definition_signature
+from backend.apps.workflows.cloud.schedule import ScheduleSupported, to_cloud_schedule
+from backend.apps.workflows.models import Workflow
+
+
+class HostedState(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    id: str
+    enabled: bool
+    next_run_at: Optional[datetime] = None
+    # False when the workflow was edited after we pushed it, so the cloud is running older prose.
+    in_sync: bool
+
+
+class CloudStatusBase(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    target: Literal["device", "cloud"]
+    schedule_supported: bool
+    schedule_reason: Optional[str] = None
+
+
+class CloudStatusSignedOut(CloudStatusBase):
+    state: Literal["signed_out"] = "signed_out"
+
+
+class CloudStatusUnknown(CloudStatusBase):
+    state: Literal["unknown"] = "unknown"
+    detail: str
+
+
+class CloudStatusReady(CloudStatusBase):
+    state: Literal["ready"] = "ready"
+    plan: Optional[str] = None
+    limits: cloud.CloudLimits
+    usage: cloud.CloudUsage
+    # None when this control plane cannot tell us whether the runner could do the job.
+    capability: Optional[cloud.CloudCapability] = None
+    hosted: Optional[HostedState] = None
+
+
+CloudStatus = Union[CloudStatusReady, CloudStatusSignedOut, CloudStatusUnknown]
+
+
+@typechecked
+def epoch_to_datetime(ms: Optional[int]) -> Optional[datetime]:
+    if ms is None:
+        return None
+    return datetime.fromtimestamp(ms / 1000, tz=timezone.utc).astimezone()
+
+
+@typechecked
+def current_signature(wf: Workflow) -> Optional[str]:
+    """The fingerprint this workflow would push right now, or None when its
+    schedule has no cloud equivalent at all."""
+    mapping = to_cloud_schedule(wf.schedule)
+    if not isinstance(mapping, ScheduleSupported):
+        return None
+    return definition_signature(cloud_definition(wf), mapping.schedule.model_dump())
+
+
+@typechecked
+def p_mirror_cloud_state(wf: Workflow, hosted: Optional[cloud.HostedWorkflow]) -> None:
+    """Once the cloud holds the timer it also holds the truth about it. The local
+    scheduler stops rolling next_run_at for a cloud workflow, so every "next run"
+    in the app would sit at a time that has already passed, and an On switch read
+    off our own copy would keep saying On for a workflow paused somewhere else."""
+    if wf.execution_target != "cloud":
+        return
+    wanted = epoch_to_datetime(hosted.next_run_at) if hosted and hosted.enabled else None
+    changed = wf.next_run_at != wanted
+    wf.next_run_at = wanted
+    if hosted and wf.schedule.enabled != hosted.enabled:
+        wf.schedule.enabled = hosted.enabled
+        changed = True
+    if changed:
+        storage.save_workflow(wf)
+
+
+@typechecked
+async def compute_status(wf: Workflow) -> CloudStatus:
+    mapping = to_cloud_schedule(wf.schedule)
+    supported = isinstance(mapping, ScheduleSupported)
+    shared = {
+        "target": wf.execution_target,
+        "schedule_supported": supported,
+        "schedule_reason": None if isinstance(mapping, ScheduleSupported) else mapping.reason,
+    }
+    try:
+        pre = await cloud.preflight(cloud_definition(wf), wf.cloud_workflow_id)
+    except cloud.SignedOut:
+        return CloudStatusSignedOut(**shared)
+    except cloud.CloudUnreachable as exc:
+        return CloudStatusUnknown(detail=exc.detail, **shared)
+    except cloud.CloudRefused as exc:
+        return CloudStatusUnknown(detail=exc.message, **shared)
+
+    p_mirror_cloud_state(wf, pre.hosted)
+    hosted: Optional[HostedState] = None
+    if pre.hosted:
+        hosted = HostedState(
+            id=pre.hosted.id,
+            enabled=pre.hosted.enabled,
+            next_run_at=epoch_to_datetime(pre.hosted.next_run_at),
+            in_sync=wf.cloud_definition_signature is not None
+            and wf.cloud_definition_signature == current_signature(wf),
+        )
+    return CloudStatusReady(
+        plan=pre.plan,
+        limits=pre.limits,
+        usage=pre.usage,
+        capability=pre.capability,
+        hosted=hosted,
+        **shared,
+    )

backend/apps/workflows/models.py

@@ -93,6 +93,10 @@ class Workflow(BaseModel):
     schedule: ScheduleConfig = Field(default_factory=ScheduleConfig)
     # Where a SCHEDULED fire runs. "cloud" hands the timer to our servers outright, so this machine must never fire it nor roll next_run_at, or the same slot runs twice. Manual Run-now always stays local.
     execution_target: Literal["device", "cloud"] = "device"
+    # The cloud's own id for our hosted copy. Only backend/apps/workflows/cloud writes these two, and only after the cloud has actually accepted the workflow.
+    cloud_workflow_id: Optional[str] = None
+    # Fingerprint of what we last pushed, so "the cloud is running an older version of this" is detectable instead of silent.
+    cloud_definition_signature: Optional[str] = None
     permissions: list[PermissionTier] = Field(
         default_factory=lambda: [PermissionTier(kind="notify")]
     )

backend/apps/workflows/workflows.py

@@ -20,6 +20,7 @@ from backend.apps.workflows.models import (
     GenerateMetadataResponse,
 )
 from backend.apps.workflows import storage, scheduler, executor, audit, escalation
+from backend.apps.workflows.cloud.handover import release_before_removing
 from backend.apps.settings.models import DEFAULT_MODEL
 
 logger = logging.getLogger(__name__)
@@ -854,6 +855,10 @@ async def delete_workflow(workflow_id: str):
     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")
+    # Trashing a cloud-hosted workflow has to stop the cloud copy first, or it keeps running (and billing) with nobody able to see it.
+    released = await release_before_removing(wf)
+    if not released.ok:
+        raise HTTPException(status_code=409, detail=released.message)
     wf.deleted_at = datetime.now()
     wf.schedule.enabled = False
     wf.next_run_at = None
@@ -900,6 +905,9 @@ async def purge_workflow(workflow_id: str):
     wf = storage.get_workflow(workflow_id)
     if not wf or wf.deleted_at is None:
         raise HTTPException(status_code=404, detail="Workflow not in trash")
+    released = await release_before_removing(wf)
+    if not released.ok:
+        raise HTTPException(status_code=409, detail=released.message)
     # Before the record goes, or the ids that name the transcripts go with it and they leak forever.
     from backend.apps.workflows.owned_sessions import purge_owned_sessions
     await purge_owned_sessions(wf)

backend/config/entity_references.py

@@ -29,6 +29,7 @@ class EntityKind(str, Enum):
     WORKFLOW_RUN = "workflow_run"
     OUTPUT = "output"
     WORKSPACE = "workspace"
+    CLOUD_WORKFLOW = "cloud_workflow"
 
 
 class EntityStore(BaseModel):
@@ -61,6 +62,8 @@ ENTITY_STORES: List[EntityStore] = [
     EntityStore(kind=EntityKind.OUTPUT, module="backend.apps.outputs.workspace_io", lookup="load_output"),
     # A workspace is a folder on disk, not a record, so its only by-id lookup is the read route.
     EntityStore(kind=EntityKind.WORKSPACE, module="backend.apps.outputs.outputs", lookup="read_workspace"),
+    # The one referent that does not live on this machine. preflight asks the cloud whether it still has the row; a miss renders as "nothing is running this", never as a silent blank.
+    EntityStore(kind=EntityKind.CLOUD_WORKFLOW, module="backend.apps.workflows.cloud.client", lookup="preflight"),
 ]
 
 CROSS_ENTITY_REFERENCES: List[EntityReference] = [
@@ -96,6 +99,7 @@ CROSS_ENTITY_REFERENCES: List[EntityReference] = [
     EntityReference(module="backend.apps.skills.models", model="SkillWorkspaceSeedRequest", field="workspace_id", target=EntityKind.WORKSPACE),
     EntityReference(module="backend.apps.workflows.models", model="AskRunBody", field="run_id", target=EntityKind.WORKFLOW_RUN),
     EntityReference(module="backend.apps.workflows.models", model="MissedRun", field="workflow_id", target=EntityKind.WORKFLOW),
+    EntityReference(module="backend.apps.workflows.models", model="Workflow", field="cloud_workflow_id", target=EntityKind.CLOUD_WORKFLOW),
     EntityReference(module="backend.apps.workflows.models", model="Workflow", field="dashboard_id", target=EntityKind.DASHBOARD),
     EntityReference(module="backend.apps.workflows.models", model="Workflow", field="edit_agent_session_id", target=EntityKind.SESSION),
     EntityReference(module="backend.apps.workflows.models", model="Workflow", field="last_run_id", target=EntityKind.WORKFLOW_RUN),

backend/main.py

@@ -48,11 +48,12 @@ from backend.apps.help.bundle import help_app
 from backend.apps.agents.proxy.anthropic_proxy import anthropic_proxy
 from backend.apps.agents.core.openai_passthrough import openai_passthrough
 from backend.apps.workflows.workflows import workflows
+from backend.apps.workflows.cloud.routes import cloud_workflows
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi import WebSocket, WebSocketDisconnect
 import json
 
-main_app = MainApp([health, agents, skills, tools_lib, modes, settings, mcp_registry, skill_registry, outputs, output_versions, dashboards, swarm, service, subscription, auth, web, onboarding, voice, help_app, anthropic_proxy, workflows, openai_passthrough])
+main_app = MainApp([health, agents, skills, tools_lib, modes, settings, mcp_registry, skill_registry, outputs, output_versions, dashboards, swarm, service, subscription, auth, web, onboarding, voice, help_app, anthropic_proxy, workflows, cloud_workflows, openai_passthrough])
 app = main_app.app
 
 # Generate per-install auth token BEFORE we bind the HTTP port. By the time any request lands, the token file exists. See backend/auth.py.

backend/tests/test_cloud_workflow_definition.py

@@ -0,0 +1,70 @@
+"""What we are willing to hand the cloud, and which schedules it can honour at all.
+
+Both are pure functions, and both decide something a user reads: a schedule the cloud cannot
+express has to be refused in words rather than rounded off, and the copy we push must not carry
+anything local (session ids, a phone number) off this machine.
+"""
+from backend.apps.workflows.cloud.definition import cloud_definition, definition_signature
+from backend.apps.workflows.cloud.schedule import ScheduleSupported, ScheduleUnsupported, to_cloud_schedule
+from backend.apps.workflows.models import PermissionTier, ScheduleConfig, Workflow, WorkflowStep
+
+
+def p_sched(**overrides) -> ScheduleConfig:
+    base = dict(enabled=True, repeat_unit="day", repeat_every=1, hour=9, minute=0, timezone="UTC")
+    base.update(overrides)
+    return ScheduleConfig(**base)
+
+
+def p_wf(**overrides) -> Workflow:
+    base = dict(title="Morning digest", steps=[WorkflowStep(text="summarize the news")], schedule=p_sched())
+    base.update(overrides)
+    return Workflow(**base)
+
+
+def test_only_daily_and_interval_schedules_map_to_the_cloud():
+    assert isinstance(to_cloud_schedule(p_sched()), ScheduleSupported)
+    assert isinstance(to_cloud_schedule(p_sched(repeat_unit="minute", repeat_every=30)), ScheduleSupported)
+    assert isinstance(to_cloud_schedule(p_sched(repeat_unit="hour", repeat_every=6)), ScheduleSupported)
+    for unsupported in (
+        p_sched(repeat_unit="week", on_days=[1]),
+        p_sched(repeat_unit="month", day_of_month=1),
+        p_sched(repeat_unit="day", repeat_every=3),
+        p_sched(max_runs=5),
+    ):
+        mapping = to_cloud_schedule(unsupported)
+        assert isinstance(mapping, ScheduleUnsupported)
+        # The reason is shown to a person verbatim, so it has to read like one wrote it.
+        assert mapping.reason.endswith(".") and " " in mapping.reason
+
+
+def test_a_9am_wall_clock_becomes_the_right_utc_time():
+    mapping = to_cloud_schedule(p_sched(hour=9, minute=30, timezone="Asia/Tokyo"))
+    assert isinstance(mapping, ScheduleSupported)
+    assert mapping.schedule.model_dump() == {"kind": "daily", "hour_utc": 0, "minute_utc": 30}
+
+
+def test_the_cloud_copy_carries_no_local_secrets_and_no_live_timer():
+    wf = p_wf(
+        permissions=[PermissionTier(kind="text", after_minutes=5, phone="+15550001111")],
+        edit_agent_session_id="session-abc",
+        source_session_id="session-def",
+    )
+    body = cloud_definition(wf)
+    assert "permissions" not in body, "the escalation tiers carry a phone number and cannot ring from a container"
+    for leaked in ("edit_agent_session_id", "source_session_id", "last_run_id", "dashboard_id"):
+        assert leaked not in body
+    assert body["schedule"]["enabled"] is False
+    assert body["steps"][0]["text"] == "summarize the news"
+
+
+def test_a_signature_tracks_edits_and_ignores_the_clock():
+    wf = p_wf()
+    schedule = {"kind": "daily", "hour_utc": 9, "minute_utc": 0}
+    first = definition_signature(cloud_definition(wf), schedule)
+
+    wf.title = wf.title
+    assert definition_signature(cloud_definition(wf), schedule) == first, "a save with no edit is not a drift"
+
+    wf.steps = [WorkflowStep(id=wf.steps[0].id, text="summarize the sports news")]
+    assert definition_signature(cloud_definition(wf), schedule) != first
+    assert definition_signature(cloud_definition(wf), {"kind": "interval", "minutes": 60}) != first

backend/tests/test_cloud_workflow_target.py

@@ -0,0 +1,254 @@
+"""Handing a workflow's timer to the cloud, and taking it back.
+
+The two things worth breaking a test over: the local timer and the cloud timer must never both be
+live (that runs a workflow twice), and a cloud we could not reach must never render as a cloud that
+said no (that is a paywall built out of a dropped packet).
+"""
+import pytest
+from fastapi import HTTPException
+
+from backend.apps.workflows import storage
+from backend.apps.workflows.cloud import client as cloud
+from backend.apps.workflows.cloud.routes import TargetRequest, set_workflow_target
+from backend.apps.workflows.workflows import delete_workflow
+from backend.apps.workflows.cloud.status import compute_status
+from backend.apps.workflows.models import ScheduleConfig, Workflow, WorkflowStep
+
+pytestmark = pytest.mark.usefixtures("isolated_workflows_data")
+
+
+def p_sched(**overrides) -> ScheduleConfig:
+    base = dict(enabled=True, repeat_unit="day", repeat_every=1, hour=9, minute=0, timezone="UTC")
+    base.update(overrides)
+    return ScheduleConfig(**base)
+
+
+def p_wf(**overrides) -> Workflow:
+    base = dict(title="Morning digest", steps=[WorkflowStep(text="summarize the news")], schedule=p_sched())
+    base.update(overrides)
+    wf = Workflow(**base)
+    storage.save_workflow(wf)
+    return wf
+
+
+def p_hosted(**overrides) -> dict:
+    row = {"id": "cloud-1", "enabled": True, "next_run_at": 1893499200000}
+    row.update(overrides)
+    return row
+
+
+def p_preflight_body(**overrides) -> dict:
+    body = {
+        "plan": "pro",
+        "limits": {"workflows": 3, "runsPerMonth": 100, "concurrent": 1},
+        "usage": {"enabled": 1, "runs_this_month": 12},
+        "capability": {"ok": True, "reason": None},
+        "hosted": None,
+    }
+    body.update(overrides)
+    return body
+
+
+def p_answer(monkeypatch, handler) -> list:
+    """Replace the single network chokepoint. Every call is recorded so a test can assert we did
+    NOT talk to the cloud as well as what we said."""
+    seen: list = []
+
+    async def p_call(method: str, path: str, body=None):
+        seen.append((method, path, body))
+        return handler(method, path, body)
+
+    monkeypatch.setattr(cloud, "p_call", p_call)
+    return seen
+
+
+@pytest.mark.asyncio
+async def test_signed_out_is_a_known_answer_and_carries_no_numbers(monkeypatch):
+    wf = p_wf()
+
+    def handler(method, path, body):
+        raise cloud.SignedOut()
+
+    p_answer(monkeypatch, handler)
+    status = await compute_status(wf)
+    assert status.state == "signed_out"
+    assert not hasattr(status, "limits") and not hasattr(status, "usage")
+
+
+@pytest.mark.asyncio
+async def test_an_unreachable_cloud_is_unknown_not_denied(monkeypatch):
+    wf = p_wf()
+
+    def handler(method, path, body):
+        raise cloud.CloudUnreachable("ConnectError")
+
+    p_answer(monkeypatch, handler)
+    status = await compute_status(wf)
+    assert status.state == "unknown"
+    # The whole point: nothing here can be read as "you have no plan" or "0 runs left".
+    assert not hasattr(status, "limits")
+    assert status.schedule_supported is True
+
+
+@pytest.mark.asyncio
+async def test_a_ready_status_reports_the_plan_the_server_named(monkeypatch):
+    wf = p_wf()
+    p_answer(monkeypatch, lambda method, path, body: p_preflight_body())
+    status = await compute_status(wf)
+    assert status.state == "ready"
+    assert status.plan == "pro"
+    assert status.limits.workflows == 3
+    assert status.usage.runs_this_month == 12
+    assert status.hosted is None
+
+
+@pytest.mark.asyncio
+async def test_a_refused_flip_leaves_the_workflow_on_this_device(monkeypatch):
+    wf = p_wf()
+
+    def handler(method, path, body):
+        raise cloud.CloudRefused("Cloud workflows need a Pro plan or higher.", 402)
+
+    p_answer(monkeypatch, handler)
+    outcome = await set_workflow_target(wf.id, TargetRequest(target="cloud", enabled=True))
+    assert outcome.ok is False
+    assert outcome.message == "Cloud workflows need a Pro plan or higher."
+    assert storage.get_workflow(wf.id).execution_target == "device"
+    assert storage.get_workflow(wf.id).cloud_workflow_id is None
+
+
+@pytest.mark.asyncio
+async def test_an_unsupported_schedule_never_reaches_the_network(monkeypatch):
+    wf = p_wf(schedule=p_sched(repeat_unit="week", on_days=[1]))
+    seen = p_answer(monkeypatch, lambda method, path, body: p_preflight_body())
+    outcome = await set_workflow_target(wf.id, TargetRequest(target="cloud", enabled=True))
+    assert outcome.ok is False
+    assert seen == []
+    assert storage.get_workflow(wf.id).execution_target == "device"
+
+
+@pytest.mark.asyncio
+async def test_an_accepted_flip_records_which_copy_is_up_there(monkeypatch):
+    wf = p_wf()
+    p_answer(monkeypatch, lambda method, path, body: p_hosted())
+    outcome = await set_workflow_target(wf.id, TargetRequest(target="cloud", enabled=True))
+    assert outcome.ok is True
+
+    saved = storage.get_workflow(wf.id)
+    assert saved.execution_target == "cloud"
+    assert saved.cloud_workflow_id == "cloud-1"
+    assert saved.cloud_definition_signature is not None
+    # The cloud owns the clock now, so the app shows the cloud's next fire and not our frozen one.
+    assert saved.next_run_at is not None
+    assert saved.next_run_at.timestamp() * 1000 == p_hosted()["next_run_at"]
+
+
+@pytest.mark.asyncio
+async def test_an_unreachable_cloud_cannot_take_the_timer_back(monkeypatch):
+    wf = p_wf(execution_target="cloud", cloud_workflow_id="cloud-1")
+    storage.save_workflow(wf)
+
+    def handler(method, path, body):
+        raise cloud.CloudUnreachable("ReadTimeout")
+
+    p_answer(monkeypatch, handler)
+    outcome = await set_workflow_target(wf.id, TargetRequest(target="device", enabled=True))
+    assert outcome.ok is False
+    # Flipping the local timer on while the cloud still holds one is how a workflow runs twice.
+    assert storage.get_workflow(wf.id).execution_target == "cloud"
+    assert storage.get_workflow(wf.id).cloud_workflow_id == "cloud-1"
+
+
+@pytest.mark.asyncio
+async def test_taking_the_timer_back_clears_every_trace_of_the_cloud_copy(monkeypatch):
+    wf = p_wf(execution_target="cloud", cloud_workflow_id="cloud-1", cloud_definition_signature="abc")
+    storage.save_workflow(wf)
+    seen = p_answer(monkeypatch, lambda method, path, body: {"ok": True})
+    outcome = await set_workflow_target(wf.id, TargetRequest(target="device", enabled=True))
+    assert outcome.ok is True
+    assert seen == [("POST", "/cloud-1/delete", {})]
+
+    saved = storage.get_workflow(wf.id)
+    assert saved.execution_target == "device"
+    assert saved.cloud_workflow_id is None
+    assert saved.cloud_definition_signature is None
+    assert saved.next_run_at is not None, "the local timer has to be armed again on the way back"
+
+
+@pytest.mark.asyncio
+async def test_a_hosted_copy_that_vanished_shows_as_hosted_nothing(monkeypatch):
+    wf = p_wf(execution_target="cloud", cloud_workflow_id="cloud-gone")
+    storage.save_workflow(wf)
+    p_answer(monkeypatch, lambda method, path, body: p_preflight_body(hosted=None))
+    status = await compute_status(wf)
+    assert status.state == "ready"
+    assert status.target == "cloud" and status.hosted is None
+    assert storage.get_workflow(wf.id).next_run_at is None, "nothing is going to run it, so do not promise a time"
+
+
+@pytest.mark.asyncio
+async def test_an_edited_workflow_reads_as_out_of_sync(monkeypatch):
+    wf = p_wf()
+    p_answer(monkeypatch, lambda method, path, body: p_hosted())
+    await set_workflow_target(wf.id, TargetRequest(target="cloud", enabled=True))
+
+    p_answer(monkeypatch, lambda method, path, body: p_preflight_body(hosted=p_hosted()))
+    assert (await compute_status(storage.get_workflow(wf.id))).hosted.in_sync is True
+
+    edited = storage.get_workflow(wf.id)
+    edited.steps = [WorkflowStep(text="summarize the sports news instead")]
+    storage.save_workflow(edited)
+    assert (await compute_status(edited)).hosted.in_sync is False
+
+
+@pytest.mark.asyncio
+async def test_an_old_control_plane_leaves_capability_unknown_rather_than_ok(monkeypatch):
+    wf = p_wf()
+
+    def handler(method, path, body):
+        if path == "/preflight":
+            raise cloud.CloudRefused("Not Found", 404)
+        return {
+            "workflows": [],
+            "limits": {"workflows": 3, "runsPerMonth": 100, "concurrent": 1},
+            "usage": {"enabled": 0, "runs_this_month": 0},
+        }
+
+    seen = p_answer(monkeypatch, handler)
+    status = await compute_status(wf)
+    assert status.state == "ready"
+    assert status.capability is None
+    assert status.limits.workflows == 3
+    # The cloud router is mounted AT /api/workflows and a trailing slash 404s there, which would turn this fallback into a second refusal.
+    assert seen[-1][1] == "", "the collection path must not carry a trailing slash"
+
+
+@pytest.mark.asyncio
+async def test_trashing_a_cloud_workflow_stops_the_cloud_copy(monkeypatch):
+    wf = p_wf()
+    p_answer(monkeypatch, lambda method, path, body: p_hosted())
+    await set_workflow_target(wf.id, TargetRequest(target="cloud", enabled=True))
+
+    seen = p_answer(monkeypatch, lambda method, path, body: {"ok": True})
+    await delete_workflow(storage.get_workflow(wf.id).id)
+    assert seen == [("POST", "/cloud-1/delete", {})]
+    trashed = storage.get_workflow(wf.id)
+    assert trashed.deleted_at is not None
+    assert trashed.cloud_workflow_id is None
+
+
+@pytest.mark.asyncio
+async def test_a_workflow_the_cloud_still_holds_cannot_be_trashed_into_a_ghost(monkeypatch):
+    wf = p_wf()
+    p_answer(monkeypatch, lambda method, path, body: p_hosted())
+    await set_workflow_target(wf.id, TargetRequest(target="cloud", enabled=True))
+
+    def handler(method, path, body):
+        raise cloud.CloudUnreachable("ConnectError")
+
+    p_answer(monkeypatch, handler)
+    with pytest.raises(HTTPException) as caught:
+        await delete_workflow(wf.id)
+    assert caught.value.status_code == 409
+    # Deleting it locally would leave a hosted copy running on its own schedule, billing a user who cannot see it.
+    assert storage.get_workflow(wf.id).deleted_at is None

frontend/src/app/pages/Dashboard/canvas/DashboardOverlays.tsx

@@ -6,6 +6,7 @@ import HelpPill from '../desktop/HelpPill';
 import CardSearchPalette from '../controls/CardSearchPalette';
 import DirectionHints from '../controls/DirectionHints';
 import WorkflowRunningToast from '@/app/pages/Workflows/WorkflowRunningToast';
+import WorkflowNoticeToast from '@/app/pages/Workflows/WorkflowNoticeToast';
 import MissedRunsToast from '@/app/pages/Workflows/MissedRunsToast';
 import ProviderHealthToast from '@/app/components/overlays/ProviderHealthToast';
 import ScheduleOfferToast from '@/app/components/nudges/ScheduleOfferToast';
@@ -172,6 +173,7 @@ const DashboardOverlays: React.FC<DashboardOverlaysProps> = ({
 
       {/* Scheduled-run nudge: "your {workflow} is running now" + jump-to-canvas */}
       <WorkflowRunningToast />
+      <WorkflowNoticeToast />
 
       {/* Launch nudge when scheduled runs elapsed while the app was closed */}
       <MissedRunsToast />

frontend/src/app/pages/Workflows/WorkflowNoticeToast.tsx

@@ -0,0 +1,38 @@
+// The server's own words when it refuses something the user asked for. Today that is a delete the
+// cloud would not let go of; without it the card simply stays put and the click looks broken.
+
+import React from 'react';
+import Snackbar from '@mui/material/Snackbar';
+import Alert from '@mui/material/Alert';
+import { useClaudeTokens } from '@/shared/styles/ThemeContext';
+import { useAppDispatch, useAppSelector } from '@/shared/hooks';
+import { dismissNoticeToast } from '@/shared/state/workflowsSlice';
+
+export default function WorkflowNoticeToast() {
+  const c = useClaudeTokens();
+  const dispatch = useAppDispatch();
+  const notice = useAppSelector((s) => s.workflows.noticeToast);
+
+  return (
+    <Snackbar
+      open={Boolean(notice)}
+      autoHideDuration={9000}
+      onClose={(_, reason) => { if (reason !== 'clickaway') dispatch(dismissNoticeToast()); }}
+      anchorOrigin={{ vertical: 'bottom', horizontal: 'left' }}
+    >
+      <Alert
+        icon={false}
+        severity="warning"
+        onClose={() => dispatch(dismissNoticeToast())}
+        sx={{
+          bgcolor: c.bg.surface,
+          color: c.text.primary,
+          border: `1px solid ${c.border.medium}`,
+          maxWidth: 420,
+        }}
+      >
+        {notice}
+      </Alert>
+    </Snackbar>
+  );
+}

frontend/src/app/pages/Workflows/app/CloudRunSection.tsx

@@ -0,0 +1,160 @@
+import React from 'react';
+import type { CSSProperties } from 'react';
+import { useAppDispatch } from '@/shared/hooks';
+import { openSettingsCard } from '@/shared/state/dashboardLayoutSlice';
+import type { Workflow } from '@/shared/state/workflowsSlice';
+import { useWC } from './uiKit';
+import type { WCPalette } from './uiKit';
+import { clockOf, relativeDayLabel } from './model';
+import { cloudAvailability, usageText } from './cloudAvailability';
+import type { CloudProbe } from './cloudAvailability';
+import type { HostedState } from './cloudApi';
+import type { CloudStatusHandle } from './useCloudStatus';
+
+const CLOUD_PATH = 'M17.5 19a4.5 4.5 0 0 0 .5-8.97A6 6 0 0 0 6.2 10.5 4 4 0 0 0 6.5 19z';
+
+function hostedOf(probe: CloudProbe) {
+  if (probe.phase !== 'answered' || probe.status.state !== 'ready') return null;
+  return probe.status.hosted;
+}
+
+// The cloud's own clock, printed from the answer we just fetched; our mirrored copy only moves on a probe.
+function nextCloudRunText(hosted: HostedState): string {
+  if (!hosted.enabled) return 'Paused in the cloud';
+  if (!hosted.next_run_at) return 'No cloud run scheduled';
+  const at = new Date(hosted.next_run_at);
+  return `Next cloud run ${relativeDayLabel(at)} at ${clockOf(at)}`;
+}
+
+const Bullet: React.FC<{ wc: WCPalette; accent?: boolean; children: React.ReactNode }> = ({ wc, accent, children }) => (
+  <div style={{ marginTop: 9, display: 'flex', alignItems: 'center', gap: 8 }}>
+    <div style={{ width: 14, display: 'flex', justifyContent: 'center', flex: 'none' }}>
+      <div style={{ width: 6, height: 6, borderRadius: '50%', background: accent ? wc.accent : wc.muted }} />
+    </div>
+    <span style={{ fontSize: 12.5, color: wc.ink4 }}>{children}</span>
+  </div>
+);
+
+const Note: React.FC<{ wc: WCPalette; tone: 'quiet' | 'warn'; children: React.ReactNode }> = ({ wc, tone, children }) => (
+  <div
+    style={{
+      display: 'flex', alignItems: 'flex-start', gap: 8, marginTop: 9, padding: '8px 10px', borderRadius: 8,
+      background: tone === 'warn' ? `${wc.warn}14` : `rgba(${wc.inkRGB},0.04)`,
+      border: `1px solid ${tone === 'warn' ? `${wc.warn}40` : 'transparent'}`,
+      fontSize: 11.5, lineHeight: 1.45, color: wc.ink4,
+    }}
+  >
+    {children}
+  </div>
+);
+
+const CloudRunSection: React.FC<{ workflow: Workflow; cloud: CloudStatusHandle }> = ({ workflow, cloud }) => {
+  const WC = useWC();
+  const dispatch = useAppDispatch();
+  const availability = cloudAvailability(cloud.probe);
+  const hosted = hostedOf(cloud.probe);
+  const target = cloud.probe.phase === 'answered' ? cloud.probe.status.target : workflow.execution_target ?? 'device';
+  const onCloud = target === 'cloud';
+  const usage = usageText(cloud.probe, availability);
+  const canPickCloud = availability.kind === 'available' && !cloud.pending;
+
+  const seg = (active: boolean, enabled: boolean): CSSProperties => ({
+    flex: 1, padding: '6px 2px', borderRadius: 7, border: 'none', fontSize: 11.5, fontWeight: 600,
+    cursor: enabled ? 'pointer' : 'default',
+    background: active ? WC.paper : 'transparent',
+    color: active ? WC.ink : enabled ? WC.muted : WC.faint,
+    boxShadow: active ? WC.shadow.sm : 'none',
+  });
+
+  const link: CSSProperties = {
+    background: 'none', border: 'none', padding: 0, marginLeft: 4, cursor: 'pointer',
+    color: WC.accent, fontSize: 11.5, fontWeight: 600, textDecoration: 'underline',
+  };
+
+  return (
+    <div style={{ marginTop: 13, paddingTop: 13, borderTop: `1px solid ${WC.line}` }}>
+      <div style={{ display: 'flex', alignItems: 'center', justifyContent: 'space-between', gap: 10 }}>
+        <span style={{ display: 'flex', alignItems: 'center', gap: 8, fontSize: 13, color: WC.ink3 }}>
+          <svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke={WC.muted} strokeWidth="1.8" style={{ flex: 'none' }}><path d={CLOUD_PATH} /></svg>
+          Runs on
+        </span>
+        <div style={{ display: 'flex', width: 186, background: WC.inset, border: `1px solid ${WC.line}`, borderRadius: 9, padding: 3, gap: 2 }}>
+          <button
+            onClick={() => { if (onCloud && !cloud.pending) cloud.choose('device', workflow.schedule.enabled); }}
+            style={seg(!onCloud, !cloud.pending)}
+          >
+            This device
+          </button>
+          <button
+            onClick={() => { if (!onCloud && canPickCloud) cloud.choose('cloud', workflow.schedule.enabled); }}
+            style={seg(onCloud, canPickCloud)}
+            title={availability.kind === 'blocked' ? availability.reason : undefined}
+          >
+            Cloud
+          </button>
+        </div>
+      </div>
+
+      {cloud.pending && <Note wc={WC} tone="quiet">Talking to the cloud…</Note>}
+
+      {!cloud.pending && cloud.refusal && <Note wc={WC} tone="warn">{cloud.refusal}</Note>}
+
+      {!cloud.pending && !cloud.refusal && availability.kind === 'checking' && (
+        <Note wc={WC} tone="quiet">Checking what your account allows…</Note>
+      )}
+
+      {!cloud.pending && !cloud.refusal && availability.kind === 'unknown' && (
+        <Note wc={WC} tone="warn">
+          <span>
+            Can&apos;t reach the cloud, so we can&apos;t tell whether this can run there.
+            {onCloud
+              ? ' It stays scheduled in the cloud; nothing changed.'
+              : ' This workflow still runs on this device.'}
+            <button onClick={cloud.refresh} style={link}>Try again</button>
+          </span>
+        </Note>
+      )}
+
+      {!cloud.pending && !cloud.refusal && availability.kind === 'blocked' && (
+        <Note wc={WC} tone="quiet">
+          <span>
+            {availability.reason}
+            {availability.action === 'sign_in' && (
+              <button onClick={() => dispatch(openSettingsCard())} style={link}>Sign in</button>
+            )}
+            {availability.action === 'plans' && (
+              <button onClick={() => dispatch(openSettingsCard())} style={link}>See plans</button>
+            )}
+          </span>
+        </Note>
+      )}
+
+      {!cloud.pending && !cloud.refusal && availability.kind === 'available' && !onCloud && (
+        <Note wc={WC} tone="quiet">Cloud runs fire on our servers, so they still happen with this app closed.</Note>
+      )}
+
+      {!cloud.pending && onCloud && hosted === null && cloud.probe.phase === 'answered' && cloud.probe.status.state === 'ready' && (
+        <Note wc={WC} tone="warn">
+          <span>
+            This is set to run in the cloud, but the cloud has no copy of it, so nothing is running it.
+            <button onClick={() => cloud.choose('device', workflow.schedule.enabled)} style={link}>Move it back here</button>
+          </span>
+        </Note>
+      )}
+
+      {!cloud.pending && onCloud && hosted && !hosted.in_sync && (
+        <Note wc={WC} tone="warn">
+          <span>
+            The cloud is still running the version you sent it. Your later edits are not up there yet.
+            <button onClick={() => cloud.choose('cloud', workflow.schedule.enabled)} style={link}>Send the current version</button>
+          </span>
+        </Note>
+      )}
+
+      {onCloud && hosted && <Bullet wc={WC} accent>{nextCloudRunText(hosted)}</Bullet>}
+      {usage && <Bullet wc={WC}>{usage}</Bullet>}
+    </div>
+  );
+};
+
+export default CloudRunSection;

frontend/src/app/pages/Workflows/app/HistoryCard.tsx

@@ -3,36 +3,88 @@ import { useAppDispatch, useAppSelector } from '@/shared/hooks';
 import { fetchRuns } from '@/shared/state/workflowsSlice';
 import { openWorkflowMonitor } from '@/shared/state/dashboardLayoutSlice';
 import { useWC, FONT_SERIF, statusChip, statusDot, statusLabel } from './uiKit';
+import type { WCPalette } from './uiKit';
 import { toRunRow, whenText } from './model';
+import { useCloudRuns } from './useCloudRuns';
+import { toCloudHistoryRow } from './cloudRunRow';
+import type { CloudHistoryRow } from './cloudRunRow';
+
+interface Entry extends CloudHistoryRow {
+  where: 'device' | 'cloud';
+  open?: () => void;
+}
+
+const Row: React.FC<{ entry: Entry; wc: WCPalette; now: Date }> = ({ entry, wc, now }) => (
+  <div
+    onClick={entry.open}
+    title={entry.open ? 'Open this run' : undefined}
+    style={{ display: 'flex', alignItems: 'center', gap: 11, padding: '9px 0', borderBottom: `1px solid rgba(${wc.inkRGB},0.05)`, cursor: entry.open ? 'pointer' : 'default' }}
+  >
+    <div style={statusDot(entry.tone, wc)} />
+    <div style={{ flex: 1, minWidth: 0 }}>
+      {/* Why a run did not happen is a sentence, not a label, so let it wrap rather than ellipsing the part that answers the question. */}
+      <div title={entry.summary} style={{ fontSize: 12.5, color: wc.ink2, overflow: 'hidden', display: '-webkit-box', WebkitBoxOrient: 'vertical', WebkitLineClamp: 3 }}>{entry.summary}</div>
+      <div style={{ fontFamily: "'JetBrains Mono',monospace", fontSize: 10.5, color: wc.muted2, marginTop: 2 }}>
+        {[entry.where === 'cloud' ? 'Cloud' : null, whenText(entry.when, now), entry.durationText, entry.costText]
+          .filter(Boolean)
+          .join(' · ')}
+      </div>
+    </div>
+    <span style={statusChip(entry.tone, wc)}>{entry.label}</span>
+  </div>
+);
 
 const HistoryCard: React.FC<{ workflowId: string; title: string }> = ({ workflowId, title }) => {
   const WC = useWC();
   const dispatch = useAppDispatch();
   const runs = useAppSelector((s) => s.workflows.runs[workflowId]);
+  const workflow = useAppSelector((s) => s.workflows.items[workflowId]);
+  const onCloud = workflow?.execution_target === 'cloud';
+  const cloudRuns = useCloudRuns(workflowId, onCloud, workflow?.updated_at ?? '');
 
   useEffect(() => { dispatch(fetchRuns(workflowId)); }, [workflowId, dispatch]);
 
-  const rows = (runs || []).slice(0, 8).map((r) => toRunRow(r, title));
+  const local: Entry[] = (runs || []).map((r) => {
+    const row = toRunRow(r, title);
+    return {
+      id: row.id,
+      label: statusLabel(row.status),
+      tone: row.status,
+      summary: row.summary,
+      when: row.when,
+      durationText: row.durationText,
+      costText: '',
+      where: 'device',
+      open: () => dispatch(openWorkflowMonitor({ workflowId, runId: row.id })),
+    };
+  });
+  const remote: Entry[] =
+    cloudRuns.phase === 'answered' && cloudRuns.response.state === 'ready'
+      ? cloudRuns.response.runs.map((r) => ({ ...toCloudHistoryRow(r, title), where: 'cloud' as const }))
+      : [];
+  const rows = [...local, ...remote]
+    .sort((a, b) => (b.when?.getTime() ?? 0) - (a.when?.getTime() ?? 0))
+    .slice(0, 8);
+
+  // A history we could not load is not an empty history, and must never be drawn as one.
+  const cloudBlind =
+    onCloud && (cloudRuns.phase === 'checking' || (cloudRuns.phase === 'answered' && cloudRuns.response.state !== 'ready'));
   const now = new Date();
 
   return (
     <div style={{ background: WC.paper, border: `1px solid rgba(${WC.inkRGB},0.08)`, borderRadius: WC.radius.lg, padding: 16 }}>
       <div style={{ fontFamily: FONT_SERIF, fontSize: 16, fontWeight: 500, color: WC.ink, marginBottom: 12 }}>History</div>
-      {rows.length === 0 && <div style={{ fontSize: 12.5, color: WC.muted2 }}>No runs yet.</div>}
+      {rows.length === 0 && !cloudBlind && <div style={{ fontSize: 12.5, color: WC.muted2 }}>No runs yet.</div>}
       <div style={{ display: 'flex', flexDirection: 'column' }}>
-        {rows.map((r) => (
-          <div key={r.id} onClick={() => dispatch(openWorkflowMonitor({ workflowId, runId: r.id }))} title="Open this run" style={{ display: 'flex', alignItems: 'center', gap: 11, padding: '9px 0', borderBottom: `1px solid rgba(${WC.inkRGB},0.05)`, cursor: 'pointer' }}>
-            <div style={statusDot(r.status, WC)} />
-            <div style={{ flex: 1, minWidth: 0 }}>
-              <div style={{ fontSize: 12.5, color: WC.ink2, whiteSpace: 'nowrap', overflow: 'hidden', textOverflow: 'ellipsis' }}>{r.summary}</div>
-              <div style={{ fontFamily: "'JetBrains Mono',monospace", fontSize: 10.5, color: WC.muted2, marginTop: 2 }}>
-                {whenText(r.when, now)}{r.durationText ? ` · ${r.durationText}` : ''}
-              </div>
-            </div>
-            <span style={statusChip(r.status, WC)}>{statusLabel(r.status)}</span>
-          </div>
-        ))}
+        {rows.map((entry) => <Row key={`${entry.where}-${entry.id}`} entry={entry} wc={WC} now={now} />)}
       </div>
+      {cloudBlind && (
+        <div style={{ fontSize: 11.5, color: WC.muted2, marginTop: rows.length ? 10 : 0 }}>
+          {cloudRuns.phase === 'checking'
+            ? 'Loading cloud runs…'
+            : 'Couldn’t load this workflow’s cloud runs, so any that ran are not shown here.'}
+        </div>
+      )}
     </div>
   );
 };

frontend/src/app/pages/Workflows/app/ScheduleCard.tsx

@@ -7,6 +7,8 @@ import {
   freqOf, patchForFreq, intervalMinutes, timeInputValue, parseTimeInput, ordinal, nextRunText, type Freq,
 } from './model';
 import { useWorkflowPatch } from './useWorkflowPatch';
+import { useCloudStatus } from './useCloudStatus';
+import CloudRunSection from './CloudRunSection';
 import RepeatField from './RepeatField';
 
 const FREQS: Array<[Freq, string]> = [['daily', 'Daily'], ['weekly', 'Weekly'], ['monthly', 'Monthly'], ['interval', 'Interval']];
@@ -15,9 +17,11 @@ const DAY_LABELS: Array<[string, number]> = [['S', 0], ['M', 1], ['T', 2], ['W',
 const ScheduleCard: React.FC<{ workflow: Workflow }> = ({ workflow }) => {
   const WC = useWC();
   const patch = useWorkflowPatch();
+  const cloud = useCloudStatus(workflow);
   const sched = workflow.schedule;
   const freq = freqOf(sched);
   const enabled = sched.enabled;
+  const onCloud = workflow.execution_target === 'cloud';
 
   const patchSched = (p: Partial<ScheduleConfig>) => patch(workflow, { schedule: { ...sched, ...p } });
 
@@ -32,6 +36,11 @@ const ScheduleCard: React.FC<{ workflow: Workflow }> = ({ workflow }) => {
 
   // Turning a weekly schedule on with no days picked is "unconfigured", so the backend silently forces it back off and the switch looks dead. Seed today's weekday so the default Weekly 9am toggles on (and stays on) in one click.
   const toggleEnabled = () => {
+    // While the cloud holds the timer, this switch is the CLOUD's switch: flipping only our copy would pause nothing.
+    if (onCloud) {
+      cloud.choose('cloud', !enabled);
+      return;
+    }
     if (!enabled && sched.repeat_unit === 'week' && sched.on_days.length === 0) {
       patchSched({ enabled: true, on_days: [new Date().getDay()] });
     } else {
@@ -197,16 +206,21 @@ const ScheduleCard: React.FC<{ workflow: Workflow }> = ({ workflow }) => {
         <svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke={WC.muted} strokeWidth="1.8" style={{ flex: 'none' }}><circle cx="12" cy="12" r="9" /><path d="M12 7v5l3 2" /></svg>
         <span style={{ fontSize: 12.5, color: WC.ink4 }}>{describeSchedule(sched)}</span>
       </div>
-      <div style={{ marginTop: 7, display: 'flex', alignItems: 'center', gap: 8 }}>
-        <div style={{ width: 14, display: 'flex', justifyContent: 'center', flex: 'none' }}><div style={{ width: 6, height: 6, borderRadius: '50%', background: WC.accent }} /></div>
-        <span style={{ fontSize: 12.5, color: WC.ink4 }}>Next run {nextRunText(workflow, workflow.next_run_at ? new Date(workflow.next_run_at) : null)}</span>
-      </div>
+      {/* On cloud our copy of next_run_at is a mirror that only refreshes on a probe, so the cloud section prints the time it just fetched instead. */}
+      {!onCloud && (
+        <div style={{ marginTop: 7, display: 'flex', alignItems: 'center', gap: 8 }}>
+          <div style={{ width: 14, display: 'flex', justifyContent: 'center', flex: 'none' }}><div style={{ width: 6, height: 6, borderRadius: '50%', background: WC.accent }} /></div>
+          <span style={{ fontSize: 12.5, color: WC.ink4 }}>Next run {nextRunText(workflow, workflow.next_run_at ? new Date(workflow.next_run_at) : null)}</span>
+        </div>
+      )}
       {maxRuns != null && (
         <div style={{ marginTop: 7, display: 'flex', alignItems: 'center', gap: 8 }}>
           <div style={{ width: 14, display: 'flex', justifyContent: 'center', flex: 'none' }}><div style={{ width: 6, height: 6, borderRadius: '50%', background: WC.muted }} /></div>
           <span style={{ fontSize: 12.5, color: WC.ink4 }}>{sched.runs_count} of {maxRuns} run{maxRuns === 1 ? '' : 's'} done</span>
         </div>
       )}
+
+      <CloudRunSection workflow={workflow} cloud={cloud} />
     </div>
   );
 };

frontend/src/app/pages/Workflows/app/cloudApi.ts

@@ -0,0 +1,123 @@
+import { API_BASE, getAuthToken } from '@/shared/config';
+
+// Mirrors backend/apps/workflows/cloud/status.py. `unknown` is a first-class answer, not a
+// degraded `ready`: it carries no plan, no limits and no counts, so a failed fetch cannot be
+// rendered as "you are not entitled" or "0 runs left".
+export type CloudTarget = 'device' | 'cloud';
+
+export interface CloudLimits {
+  workflows: number;
+  runs_per_month: number;
+  concurrent: number;
+}
+
+export interface CloudUsage {
+  workflows_enabled: number;
+  runs_this_month: number;
+}
+
+export interface CloudCapability {
+  ok: boolean;
+  reason: string | null;
+}
+
+export interface HostedState {
+  id: string;
+  enabled: boolean;
+  next_run_at: string | null;
+  /** False when the workflow was edited after we pushed it, so the cloud holds older prose. */
+  in_sync: boolean;
+}
+
+interface CloudStatusShared {
+  target: CloudTarget;
+  schedule_supported: boolean;
+  schedule_reason: string | null;
+}
+
+export interface CloudStatusReady extends CloudStatusShared {
+  state: 'ready';
+  plan: string | null;
+  limits: CloudLimits;
+  usage: CloudUsage;
+  /** Null when the control plane could not tell us; create re-checks either way. */
+  capability: CloudCapability | null;
+  hosted: HostedState | null;
+}
+
+export interface CloudStatusSignedOut extends CloudStatusShared {
+  state: 'signed_out';
+}
+
+export interface CloudStatusUnknown extends CloudStatusShared {
+  state: 'unknown';
+  detail: string;
+}
+
+export type CloudStatus = CloudStatusReady | CloudStatusSignedOut | CloudStatusUnknown;
+
+export interface CloudRun {
+  id: string;
+  status: string;
+  started_at: string | null;
+  finished_at: string | null;
+  error: string | null;
+  answer: string | null;
+  notices: string[];
+  cost_usd: number | null;
+}
+
+export type CloudRunsResponse =
+  | { state: 'ready'; runs: CloudRun[] }
+  | { state: 'signed_out' | 'unknown'; detail: string | null };
+
+export interface TargetOutcome {
+  ok: boolean;
+  message: string | null;
+}
+
+const base = `${API_BASE}/cloud_workflows`;
+
+function headers(): Record<string, string> {
+  let tok = '';
+  try { tok = getAuthToken(); } catch { tok = ''; }
+  return { 'Content-Type': 'application/json', ...(tok ? { Authorization: `Bearer ${tok}` } : {}) };
+}
+
+// Null means our own backend did not answer, which the caller must render as "cannot tell" rather than as a denial.
+export async function fetchCloudStatus(workflowId: string): Promise<CloudStatus | null> {
+  try {
+    const res = await fetch(`${base}/${encodeURIComponent(workflowId)}/status`, { headers: headers() });
+    if (!res.ok) return null;
+    return (await res.json()) as CloudStatus;
+  } catch {
+    return null;
+  }
+}
+
+export async function fetchCloudRuns(workflowId: string): Promise<CloudRunsResponse> {
+  try {
+    const res = await fetch(`${base}/${encodeURIComponent(workflowId)}/runs`, { headers: headers() });
+    if (!res.ok) return { state: 'unknown', detail: null };
+    return (await res.json()) as CloudRunsResponse;
+  } catch {
+    return { state: 'unknown', detail: null };
+  }
+}
+
+export async function setCloudTarget(
+  workflowId: string,
+  body: { target: CloudTarget; enabled: boolean },
+): Promise<TargetOutcome> {
+  try {
+    const res = await fetch(`${base}/${encodeURIComponent(workflowId)}/target`, {
+      method: 'POST',
+      headers: headers(),
+      body: JSON.stringify(body),
+    });
+    if (!res.ok) return { ok: false, message: 'Something went wrong on this machine, so nothing changed.' };
+    return (await res.json()) as TargetOutcome;
+  } catch {
+    return { ok: false, message: 'Something went wrong on this machine, so nothing changed.' };
+  }
+}

frontend/src/app/pages/Workflows/app/cloudAvailability.ts

@@ -0,0 +1,68 @@
+import type { CloudStatus, CloudStatusReady } from './cloudApi';
+
+// One probe, three honest outcomes plus "still asking". Anything we have not heard back about is
+// `unknown`, never a refusal: a hiccup that renders as "not entitled" is a paywall built out of a
+// dropped packet.
+export type CloudProbe =
+  | { phase: 'checking' }
+  | { phase: 'unreachable' }
+  | { phase: 'answered'; status: CloudStatus };
+
+export type CloudAvailability =
+  | { kind: 'checking' }
+  | { kind: 'unknown'; detail: string | null }
+  | { kind: 'blocked'; reason: string; action: 'sign_in' | 'plans' | null }
+  | { kind: 'available' };
+
+const PLAN_REQUIRED = 'Cloud runs come with Pro and up. On this plan, workflows run on this device.';
+
+function blockedForAccount(status: CloudStatusReady): CloudAvailability | null {
+  if (status.limits.workflows === 0) {
+    return { kind: 'blocked', reason: PLAN_REQUIRED, action: 'plans' };
+  }
+  // A workflow already up there is holding one of the slots, so its own slot must not read as full.
+  const holdsASlot = status.hosted !== null;
+  if (!holdsASlot && status.usage.workflows_enabled >= status.limits.workflows) {
+    return {
+      kind: 'blocked',
+      reason: `${status.usage.workflows_enabled} of ${status.limits.workflows} cloud workflows used. Turn one off to move this one up.`,
+      action: null,
+    };
+  }
+  return null;
+}
+
+/** Whether the Cloud choice can be offered, and if not, the sentence that says why.
+ *  Reasons about the workflow itself come first: telling someone to upgrade for a job the runner
+ *  could never do is a sale, not an answer. */
+export function cloudAvailability(probe: CloudProbe): CloudAvailability {
+  if (probe.phase === 'checking') return { kind: 'checking' };
+  if (probe.phase === 'unreachable') return { kind: 'unknown', detail: null };
+  const status = probe.status;
+  if (!status.schedule_supported && status.schedule_reason) {
+    return { kind: 'blocked', reason: status.schedule_reason, action: null };
+  }
+  if (status.state === 'unknown') return { kind: 'unknown', detail: status.detail };
+  if (status.state === 'signed_out') {
+    return {
+      kind: 'blocked',
+      reason: 'Sign in to your OpenSwarm account to run workflows in the cloud.',
+      action: 'sign_in',
+    };
+  }
+  if (status.capability && !status.capability.ok && status.capability.reason) {
+    return { kind: 'blocked', reason: status.capability.reason, action: null };
+  }
+  return blockedForAccount(status) ?? { kind: 'available' };
+}
+
+/** The account-wide ceiling, said so plainly nobody reads it as this one workflow's count.
+ *  Null whenever we are unsure of the numbers, or cloud is not on the table for this workflow. */
+export function usageText(probe: CloudProbe, availability: CloudAvailability): string | null {
+  if (probe.phase !== 'answered' || probe.status.state !== 'ready') return null;
+  const onCloud = probe.status.target === 'cloud';
+  if (!onCloud && availability.kind !== 'available') return null;
+  const { usage, limits } = probe.status;
+  if (limits.runs_per_month === 0) return null;
+  return `Your plan: ${usage.runs_this_month} of ${limits.runs_per_month} cloud runs this month`;
+}

frontend/src/app/pages/Workflows/app/cloudRunRow.ts

@@ -0,0 +1,86 @@
+import type { CloudRun } from './cloudApi';
+import type { RunStatus } from './uiKit';
+
+// A cloud run that never started reports as "dispatch_unavailable: fly_capacity: ...", which is a
+// sentence for us, not for the person who was expecting a report at 9am. Every refusal the
+// dispatcher can produce gets a plain answer to the only question they have: did it run, and why not.
+const REFUSAL_TEXT: Record<string, string> = {
+  runner_not_configured: "Cloud runs weren't available on our side, so this didn't start. You weren't charged for it.",
+  callback_not_configured: "Cloud runs weren't available on our side, so this didn't start. You weren't charged for it.",
+  fly_unauthorized: "Cloud runs weren't available on our side, so this didn't start. You weren't charged for it.",
+  fly_rejected: "Cloud runs weren't available on our side, so this didn't start. You weren't charged for it.",
+  fly_capacity: "The cloud had no room at that moment, so this didn't start. You weren't charged for it.",
+  fly_unreachable: "We couldn't reach the machine meant to run this, so it didn't start. You weren't charged for it.",
+  workflow_definition_invalid:
+    "The cloud's copy of this workflow was unreadable, so nothing ran. Switch it back to this device and up to the cloud again to resend it.",
+  no_cloud_credential:
+    "No AI account is connected to the cloud for this workspace, so there was nothing to run this with.",
+  slot_already_run: 'This slot had already run, so it was not run a second time.',
+};
+
+const STATUS_LABEL: Record<string, string> = {
+  pending: 'Starting',
+  running: 'Running',
+  succeeded: 'Success',
+  failed: 'Failed',
+  dispatch_unavailable: "Didn't run",
+};
+
+const STATUS_TONE: Record<string, RunStatus> = {
+  pending: 'running',
+  running: 'running',
+  succeeded: 'success',
+  failed: 'failure',
+  dispatch_unavailable: 'skipped',
+};
+
+export interface CloudHistoryRow {
+  id: string;
+  label: string;
+  tone: RunStatus;
+  summary: string;
+  when: Date | null;
+  durationText: string;
+  costText: string;
+}
+
+/** Split "reason: detail" on the FIRST colon only, and only accept a reason we actually know.
+ *  An unrecognised prefix falls through to the raw text: showing the truth beats guessing at it. */
+export function explainCloudFailure(error: string | null): string {
+  if (!error) return '';
+  const at = error.indexOf(': ');
+  if (at < 0) return error;
+  const reason = error.slice(0, at);
+  const detail = error.slice(at + 2);
+  // The runner-capability detail is already the sentence we would have written.
+  if (reason === 'runner_capability') return detail;
+  return REFUSAL_TEXT[reason] ?? error;
+}
+
+function duration(run: CloudRun): string {
+  if (!run.started_at || !run.finished_at) return '';
+  const ms = new Date(run.finished_at).getTime() - new Date(run.started_at).getTime();
+  if (Number.isNaN(ms) || ms < 0) return '';
+  const s = Math.round(ms / 1000);
+  return s < 60 ? `${s}s` : `${Math.floor(s / 60)}m ${s % 60}s`;
+}
+
+function cost(run: CloudRun): string {
+  if (run.cost_usd === null || run.cost_usd === undefined) return '';
+  if (run.cost_usd === 0) return '$0.00';
+  return run.cost_usd < 0.01 ? '<$0.01' : `$${run.cost_usd.toFixed(2)}`;
+}
+
+export function toCloudHistoryRow(run: CloudRun, fallbackTitle: string): CloudHistoryRow {
+  const failed = run.status === 'failed' || run.status === 'dispatch_unavailable';
+  const explained = explainCloudFailure(run.error);
+  return {
+    id: run.id,
+    label: STATUS_LABEL[run.status] ?? run.status,
+    tone: STATUS_TONE[run.status] ?? 'skipped',
+    summary: (failed && explained) || run.answer || fallbackTitle,
+    when: run.started_at ? new Date(run.started_at) : null,
+    durationText: duration(run),
+    costText: cost(run),
+  };
+}

frontend/src/app/pages/Workflows/app/useCloudRuns.ts

@@ -0,0 +1,49 @@
+import { useEffect, useRef, useState } from 'react';
+import { fetchCloudRuns } from './cloudApi';
+import type { CloudRunsResponse } from './cloudApi';
+
+export type CloudRunsProbe =
+  | { phase: 'idle' }
+  | { phase: 'checking' }
+  | { phase: 'answered'; response: CloudRunsResponse };
+
+/** Run history for the cloud copy. Only asks when the workflow is actually up there, so a
+ *  device-only workflow never makes a network call to find out it has no cloud runs. */
+export function useCloudRuns(workflowId: string, hosted: boolean, revision: string): CloudRunsProbe {
+  const [probe, setProbe] = useState<CloudRunsProbe>({ phase: 'idle' });
+  const live = useRef(true);
+
+  useEffect(() => {
+    live.current = true;
+    return () => { live.current = false; };
+  }, []);
+
+  useEffect(() => {
+    if (!hosted) {
+      setProbe({ phase: 'idle' });
+      return;
+    }
+    setProbe((prev) => (prev.phase === 'answered' ? prev : { phase: 'checking' }));
+    fetchCloudRuns(workflowId).then((response) => {
+      if (live.current) setProbe({ phase: 'answered', response });
+    });
+  }, [workflowId, hosted, revision]);
+
+  // A cloud run reports to the cloud, not to us, so a run in flight is the one case worth asking again about. The poll stops itself the moment nothing is live.
+  const watching =
+    probe.phase === 'answered' &&
+    probe.response.state === 'ready' &&
+    probe.response.runs.some((r) => r.status === 'pending' || r.status === 'running');
+
+  useEffect(() => {
+    if (!hosted || !watching) return undefined;
+    const timer = setInterval(() => {
+      fetchCloudRuns(workflowId).then((response) => {
+        if (live.current) setProbe({ phase: 'answered', response });
+      });
+    }, 30000);
+    return () => clearInterval(timer);
+  }, [hosted, watching, workflowId]);
+
+  return probe;
+}

frontend/src/app/pages/Workflows/app/useCloudStatus.ts

@@ -0,0 +1,63 @@
+import { useCallback, useEffect, useRef, useState } from 'react';
+import { useAppDispatch } from '@/shared/hooks';
+import { fetchWorkflows } from '@/shared/state/workflowsSlice';
+import type { Workflow } from '@/shared/state/workflowsSlice';
+import { fetchCloudStatus, setCloudTarget } from './cloudApi';
+import type { CloudTarget, TargetOutcome } from './cloudApi';
+import type { CloudProbe } from './cloudAvailability';
+
+export interface CloudStatusHandle {
+  probe: CloudProbe;
+  /** True while a flip is in flight; the control must not accept a second one. */
+  pending: boolean;
+  /** Set only by a refused flip, and cleared by the next attempt. */
+  refusal: string | null;
+  choose: (target: CloudTarget, enabled: boolean) => void;
+  refresh: () => void;
+}
+
+/** The cloud's answer for one workflow, fetched off the render path so the Workflows app paints
+ *  and stays usable whether or not there is a cloud, an account, or a network. */
+export function useCloudStatus(workflow: Workflow): CloudStatusHandle {
+  const dispatch = useAppDispatch();
+  const [probe, setProbe] = useState<CloudProbe>({ phase: 'checking' });
+  const [pending, setPending] = useState(false);
+  const [refusal, setRefusal] = useState<string | null>(null);
+  const live = useRef(true);
+  const workflowId = workflow.id;
+  const dashboardId = workflow.dashboard_id;
+
+  useEffect(() => {
+    live.current = true;
+    return () => { live.current = false; };
+  }, []);
+
+  const refresh = useCallback(() => {
+    fetchCloudStatus(workflowId).then((status) => {
+      if (!live.current) return;
+      setProbe(status ? { phase: 'answered', status } : { phase: 'unreachable' });
+    });
+  }, [workflowId]);
+
+  // Only a different workflow blanks the answer. Re-asking about the SAME one keeps the last answer on screen while it happens, so an edit does not strobe the card.
+  useEffect(() => { setProbe({ phase: 'checking' }); }, [workflowId]);
+
+  // Re-ask when the workflow changes in a way the answer depends on: a different schedule can stop being expressible in the cloud, and edited steps can stop being runnable there.
+  useEffect(() => { refresh(); }, [refresh, workflow.updated_at]);
+
+  const choose = useCallback((target: CloudTarget, enabled: boolean) => {
+    if (pending) return;
+    setPending(true);
+    setRefusal(null);
+    setCloudTarget(workflowId, { target, enabled }).then((outcome: TargetOutcome) => {
+      if (!live.current) return;
+      setPending(false);
+      if (!outcome.ok) setRefusal(outcome.message);
+      // Refresh either way: a refusal usually means the reasons on screen are stale too.
+      refresh();
+      if (outcome.ok) dispatch(fetchWorkflows(dashboardId ?? undefined));
+    });
+  }, [dispatch, pending, refresh, workflowId, dashboardId]);
+
+  return { probe, pending, refusal, choose, refresh };
+}

frontend/src/shared/state/workflowsSlice.ts

@@ -74,6 +74,9 @@ export interface Workflow {
   steps: WorkflowStep[];
   actions: ActionsConfig;
   schedule: ScheduleConfig;
+  /** Where a SCHEDULED fire runs. Only the cloud_workflows routes may change it, and only once
+   *  the cloud has actually taken (or released) the workflow. Manual runs are always local. */
+  execution_target?: 'device' | 'cloud';
   permissions: PermissionTier[];
   source_session_id?: string | null;
   dashboard_id?: string | null;
@@ -200,12 +203,14 @@ interface State {
   allRuns: WorkflowRun[];
   allRunsLoading: boolean;
   runningToast: RunningToast | null;
+  /** One-off explanation for something the user asked for that the server refused. */
+  noticeToast: string | null;
   runControlPending: Record<string, WorkflowRunControlAction>;
   deleted: Workflow[];
   deletedLoading: boolean;
 }
 
-const initialState: State = { items: {}, runs: {}, openCards: {}, loaded: false, loading: false, paused: false, active: [], cloudSmsEnabled: false, allRuns: [], allRunsLoading: false, runningToast: null, runControlPending: {}, deleted: [], deletedLoading: false };
+const initialState: State = { items: {}, runs: {}, openCards: {}, loaded: false, loading: false, paused: false, active: [], cloudSmsEnabled: false, allRuns: [], allRunsLoading: false, runningToast: null, noticeToast: null, runControlPending: {}, deleted: [], deletedLoading: false };
 
 function mergeRunIntoState(state: State, r: WorkflowRun) {
   const arr = state.runs[r.workflow_id] || [];
@@ -402,8 +407,13 @@ export const discardDraft = createAsyncThunk('workflows/discardDraft', async (id
   return (await res.json()) as Workflow;
 });
 
-export const deleteWorkflow = createAsyncThunk('workflows/delete', async (id: string) => {
-  await fetch(`${API}/${id}`, { method: 'DELETE' });
+export const deleteWorkflow = createAsyncThunk('workflows/delete', async (id: string, { rejectWithValue }) => {
+  const res = await fetch(`${API}/${id}`, { method: 'DELETE' });
+  // A refused delete must not remove the card: the workflow is still there, and if it is cloud-hosted it is still running.
+  if (!res.ok) {
+    const body = await res.json().catch(() => null);
+    return rejectWithValue(typeof body?.detail === 'string' ? body.detail : "Couldn't delete this workflow. Try again in a moment.");
+  }
   return id;
 });
 
@@ -570,6 +580,9 @@ const slice = createSlice({
     dismissRunningToast(state) {
       state.runningToast = null;
     },
+    dismissNoticeToast(state) {
+      state.noticeToast = null;
+    },
   },
   extraReducers: (builder) => {
     builder
@@ -591,6 +604,9 @@ const slice = createSlice({
       .addCase(updateWorkflow.fulfilled, (state, action) => { state.items[action.payload.id] = action.payload; })
       .addCase(commitDraft.fulfilled, (state, action) => { state.items[action.payload.id] = action.payload; })
       .addCase(discardDraft.fulfilled, (state, action) => { state.items[action.payload.id] = action.payload; })
+      .addCase(deleteWorkflow.rejected, (state, action) => {
+        state.noticeToast = typeof action.payload === 'string' ? action.payload : "Couldn't delete this workflow. Try again in a moment.";
+      })
       .addCase(deleteWorkflow.fulfilled, (state, action) => {
         delete state.items[action.payload];
         delete state.runs[action.payload];
@@ -670,5 +686,6 @@ export const {
   upsertWorkflow,
   removeWorkflow,
   dismissRunningToast,
+  dismissNoticeToast,
 } = slice.actions;
 export default slice.reducer;