[eric] merge eric/runner-parity: files come back, apps build, skills load, and MCP refuses honestly

backend/apps/workflows/cloud/client.py

@@ -20,6 +20,8 @@ from backend.apps.workflows.cloud.schedule import CloudSchedule, wire
 # 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
+# Files are up to 20MB each, so they get their own budget rather than the chatty-call one.
+DOWNLOAD_TIMEOUT_SECONDS = 120.0
 
 
 class CloudRefused(Exception):
@@ -88,6 +90,18 @@ class CloudPreflight(BaseModel):
     hosted: Optional[HostedWorkflow] = None
 
 
+class CloudRunFile(BaseModel):
+    """One file a cloud run produced. `refusal` set means it exists nowhere and says why."""
+
+    model_config = ConfigDict(validate_assignment=True)
+
+    id: str
+    path: str
+    size_bytes: int = 0
+    sha256: Optional[str] = None
+    refusal: Optional[str] = None
+
+
 class CloudRun(BaseModel):
     model_config = ConfigDict(validate_assignment=True)
 
@@ -99,6 +113,7 @@ class CloudRun(BaseModel):
     answer: Optional[str] = None
     notices: List[str] = Field(default_factory=list)
     cost_usd: Optional[float] = None
+    files: List[CloudRunFile] = Field(default_factory=list)
 
 
 @typechecked
@@ -232,12 +247,16 @@ async def p_preflight_from_list(hosted_id: Optional[str]) -> CloudPreflight:
 @typechecked
 async def put_workflow(
     *, hosted_id: Optional[str], name: str, definition: Dict[str, Any],
-    schedule: CloudSchedule, runs_before: int = 0,
+    schedule: CloudSchedule, runs_before: int = 0, context: Optional[Dict[str, Any]] = None,
 ) -> HostedWorkflow:
     """Create the hosted copy, or re-push onto the existing row so an edited workflow stops running
     last week's prose. runs_before rides only on the create: an edit that resent it would hand a
-    nearly-spent run cap its whole budget back."""
+    nearly-spent run cap its whole budget back. `context` carries the user's skills and the names of
+    the apps a cloud run cannot reach; an older control plane ignores the extra keys, which costs a
+    run its skills but never its run."""
     body: Dict[str, Any] = {"name": name, "definition": definition, "schedule": wire(schedule)}
+    if context:
+        body.update(context)
     if hosted_id:
         try:
             raw = await p_call("POST", f"/{hosted_id}/update", body)
@@ -274,6 +293,26 @@ async def delete_hosted(hosted_id: str) -> None:
             raise
 
 
+@typechecked
+def p_files(raw: Any) -> List[CloudRunFile]:
+    """A control plane with no file support answers without the key, which is an empty list, not an error."""
+    if not isinstance(raw, list):
+        return []
+    out: List[CloudRunFile] = []
+    for row in raw:
+        if not isinstance(row, dict) or not isinstance(row.get("id"), str) or not isinstance(row.get("path"), str):
+            continue
+        size = row.get("size_bytes")
+        out.append(CloudRunFile(
+            id=row["id"],
+            path=row["path"],
+            size_bytes=size if isinstance(size, int) else 0,
+            sha256=row.get("sha256") if isinstance(row.get("sha256"), str) else None,
+            refusal=row.get("refusal") if isinstance(row.get("refusal"), str) else None,
+        ))
+    return out
+
+
 @typechecked
 async def list_runs(hosted_id: str) -> List[CloudRun]:
     raw = await p_call("GET", f"/{hosted_id}/runs")
@@ -294,5 +333,29 @@ async def list_runs(hosted_id: str) -> List[CloudRun]:
             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,
+            files=p_files(row.get("files")),
         ))
     return out
+
+
+@typechecked
+async def download_run_file(hosted_id: str, run_id: str, file_id: str) -> bytes:
+    """The file's bytes. Its own call rather than p_call because this answer is not JSON."""
+    from backend.apps.settings.store import load_settings
+
+    token, base = account_auth(load_settings())
+    if not token:
+        raise SignedOut()
+    url = f"{base}/api/workflows/{hosted_id}/runs/{run_id}/files/{file_id}"
+    try:
+        async with httpx.AsyncClient(timeout=DOWNLOAD_TIMEOUT_SECONDS) as client:
+            resp = await client.get(url, headers={"Authorization": f"Bearer {token}"})
+    except httpx.HTTPError as exc:
+        raise CloudUnreachable(f"{type(exc).__name__}") from exc
+    if resp.status_code == 401:
+        raise SignedOut()
+    if resp.status_code >= 500:
+        raise CloudUnreachable(f"the cloud returned {resp.status_code}")
+    if resp.status_code >= 400:
+        raise CloudRefused(p_message(resp, "That file could not be downloaded."), resp.status_code)
+    return resp.content

backend/apps/workflows/cloud/definition.py

@@ -9,7 +9,7 @@ from __future__ import annotations
 
 import hashlib
 import json
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 
 from typeguard import typechecked
 
@@ -54,8 +54,13 @@ def cloud_definition(wf: Workflow) -> Dict[str, Any]:
 
 
 @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)
+def definition_signature(
+    definition: Dict[str, Any], schedule: Dict[str, Any], context: Optional[Dict[str, Any]] = None
+) -> str:
+    """Fingerprint of exactly what we last handed the cloud, schedule and skills included,
+    so "your edits are not up there yet" is a fact rather than a guess. Editing a skill a
+    workflow leans on is an edit to that workflow's behaviour, so it belongs in here."""
+    payload = json.dumps(
+        {"definition": definition, "schedule": schedule, "context": context or {}}, sort_keys=True
+    )
     return hashlib.sha256(payload.encode("utf-8")).hexdigest()

backend/apps/workflows/cloud/handover.py

@@ -18,6 +18,7 @@ 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.portable_context import portable_context
 from backend.apps.workflows.cloud.schedule import ScheduleSupported, to_cloud_schedule, wire
 from backend.apps.workflows.cloud.status import epoch_to_datetime
 from backend.apps.workflows.models import Workflow
@@ -51,6 +52,7 @@ async def hand_to_cloud(wf: Workflow, enabled: bool) -> TargetOutcome:
     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)
+    context = portable_context().as_body()
     try:
         hosted = await cloud.put_workflow(
             hosted_id=wf.cloud_workflow_id,
@@ -58,6 +60,7 @@ async def hand_to_cloud(wf: Workflow, enabled: bool) -> TargetOutcome:
             definition=definition,
             schedule=mapping.schedule,
             runs_before=wf.schedule.runs_count,
+            context=context,
         )
         if hosted.enabled != enabled:
             hosted = await cloud.set_enabled(hosted.id, enabled)
@@ -71,7 +74,7 @@ async def hand_to_cloud(wf: Workflow, enabled: bool) -> TargetOutcome:
 
     wf.execution_target = "cloud"
     wf.cloud_workflow_id = hosted.id
-    wf.cloud_definition_signature = definition_signature(definition, wire(mapping.schedule))
+    wf.cloud_definition_signature = definition_signature(definition, wire(mapping.schedule), context)
     wf.schedule.enabled = enabled
     wf.next_run_at = epoch_to_datetime(hosted.next_run_at) if enabled else None
     wf.updated_at = datetime.now()

backend/apps/workflows/cloud/portable_context.py

@@ -0,0 +1,157 @@
+"""What travels with a workflow besides the workflow, and the sharp line about what does not.
+
+Two things go up:
+
+* **Skills.** The user's own writing. A container without them does not get a weaker Skill tool,
+  it gets none at all (the backend gates the whole tool on one non-built-in skill existing), and
+  the agent then answers from general knowledge in exactly the same confident voice. So the
+  skills travel and the cloud run knows what the user's house ratio actually is.
+* **The NAMES of connected apps.** Names only, so the agent can say "I cannot reach your Notion
+  from a cloud run" instead of inventing what is in it.
+
+One thing must never go up: the credentials behind those apps. A ToolDefinition's `credentials`
+and `oauth_tokens` hold Slack session cookies, Notion and GitHub access tokens that do not expire,
+and Google refresh tokens. The destination is an ephemeral machine executing the user's own agent
+prose with Bash in it. Sending them would put a permanent, full-scope key to someone's email and
+documents inside a box designed to be thrown away, so this module reads `tool.name` and nothing
+else, and the runner's own RunSpec has no field that could hold a secret even if someone tried.
+"""
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any, Dict, List
+
+from pydantic import BaseModel, ConfigDict, Field
+from typeguard import typechecked
+
+logger = logging.getLogger(__name__)
+
+# Matches openswarm-runner/runner/run_spec.py. The runner refuses past these anyway; refusing here
+# too means an oversized push is a legible error on the user's own machine, not a dead cloud run.
+MAX_SKILLS = 60
+MAX_SKILL_FILE_CHARS = 200_000
+MAX_TOTAL_SKILL_CHARS = 1_000_000
+# A skill is prose and small scripts. Anything else in the folder is somebody's stray download.
+PORTABLE_SUFFIXES = (".md", ".txt", ".py", ".sh", ".json", ".yaml", ".yml", ".csv", ".ts", ".js")
+
+
+class PortableSkillFile(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    path: str
+    text: str
+
+
+class PortableSkill(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    id: str
+    files: List[PortableSkillFile]
+
+
+class PortableContext(BaseModel):
+    model_config = ConfigDict(validate_assignment=True)
+
+    skills: List[PortableSkill] = Field(default_factory=list)
+    # Names of apps connected here whose sign-in details stay here.
+    unavailable_mcp_servers: List[str] = Field(default_factory=list)
+
+    @typechecked
+    def as_body(self) -> Dict[str, Any]:
+        return {
+            "skills": [skill.model_dump(mode="json") for skill in self.skills],
+            "unavailable_mcp_servers": self.unavailable_mcp_servers,
+        }
+
+
+@typechecked
+def p_read_text(path: str) -> str:
+    """The file's text, or empty when it is binary, unreadable, or too big to carry."""
+    try:
+        if os.path.getsize(path) > MAX_SKILL_FILE_CHARS:
+            return ""
+        with open(path, "r", encoding="utf-8") as handle:
+            return handle.read()
+    except (OSError, UnicodeDecodeError):
+        return ""
+
+
+@typechecked
+def p_skill_files(skill: Any) -> List[PortableSkillFile]:
+    """One skill flattened to SKILL.md plus whatever else is in its folder.
+
+    A legacy flat `<id>.md` skill goes up as a folder with a SKILL.md too, so the wire has one
+    shape and the container has one layout.
+    """
+    files = [PortableSkillFile(path="SKILL.md", text=skill.content)]
+    folder = getattr(skill, "dir_path", "") or ""
+    if not folder or not os.path.isdir(folder):
+        return files
+    for directory, subdirs, filenames in os.walk(folder):
+        subdirs[:] = sorted(name for name in subdirs if not name.startswith("."))
+        for filename in sorted(filenames):
+            if filename == "SKILL.md" or not filename.lower().endswith(PORTABLE_SUFFIXES):
+                continue
+            absolute = os.path.join(directory, filename)
+            if os.path.islink(absolute):
+                continue
+            text = p_read_text(absolute)
+            if text:
+                files.append(PortableSkillFile(
+                    path=os.path.relpath(absolute, folder).replace(os.sep, "/"),
+                    text=text,
+                ))
+    return files
+
+
+@typechecked
+def portable_skills() -> List[PortableSkill]:
+    """Every skill the user wrote or installed. Built-ins are skipped: the container seeds its own."""
+    from backend.apps.skills.skills import safe_slug, sync_skills
+
+    out: List[PortableSkill] = []
+    budget = MAX_TOTAL_SKILL_CHARS
+    for skill in sync_skills():
+        if skill.built_in or len(out) >= MAX_SKILLS:
+            continue
+        # The id becomes a directory name in the container, so it has to survive being one.
+        slug = safe_slug(skill.id)
+        if not slug:
+            continue
+        files = p_skill_files(skill)
+        cost = sum(len(f.text) for f in files)
+        if cost > budget:
+            logger.info("skill %s not sent to the cloud: the run spec's skill budget is spent", skill.id)
+            continue
+        budget -= cost
+        out.append(PortableSkill(id=slug, files=files))
+    return out
+
+
+@typechecked
+def unavailable_mcp_servers() -> List[str]:
+    """Connected apps, by name. Reads `tool.name` and deliberately nothing else."""
+    from backend.apps.tools_lib.tools_lib import load_all_tools
+
+    names: List[str] = []
+    for tool in load_all_tools():
+        if tool.mcp_config and tool.enabled and tool.auth_status in ("configured", "connected"):
+            names.append(tool.name)
+    return sorted(set(names))
+
+
+@typechecked
+def portable_context() -> PortableContext:
+    """Everything a cloud run should know that the workflow itself does not carry."""
+    try:
+        skills = portable_skills()
+    except Exception:
+        logger.exception("could not gather skills for the cloud; the run will have none")
+        skills = []
+    try:
+        servers = unavailable_mcp_servers()
+    except Exception:
+        logger.exception("could not gather connected app names for the cloud")
+        servers = []
+    return PortableContext(skills=skills, unavailable_mcp_servers=servers)

backend/apps/workflows/cloud/routes.py

@@ -7,6 +7,7 @@ cloud re-decides that at create and again at dispatch.
 """
 from __future__ import annotations
 
+import asyncio
 from contextlib import asynccontextmanager
 from datetime import datetime
 from typing import List, Literal, Optional, Union
@@ -18,6 +19,7 @@ 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.run_files import LocalRunFile, described, downloads_root, fetch_missing
 from backend.apps.workflows.cloud.status import CloudStatus, compute_status, epoch_to_datetime
 from backend.apps.workflows.models import Workflow
 from backend.config.Apps import SubApp
@@ -50,6 +52,7 @@ class CloudRunRow(BaseModel):
     answer: Optional[str] = None
     notices: List[str] = []
     cost_usd: Optional[float] = None
+    files: List[LocalRunFile] = []
 
 
 class CloudRunsReady(BaseModel):
@@ -57,6 +60,9 @@ class CloudRunsReady(BaseModel):
 
     state: Literal["ready"] = "ready"
     runs: List[CloudRunRow] = []
+    # Where this machine puts a cloud run's files, so the UI can say it out loud even when a run
+    # has none yet. A folder the user is told about is a folder they can find later.
+    files_folder: str = ""
 
 
 class CloudRunsUnavailable(BaseModel):
@@ -95,7 +101,7 @@ async def set_workflow_target(workflow_id: str, body: TargetRequest) -> TargetOu
 async def workflow_cloud_runs(workflow_id: str) -> CloudRunsResponse:
     wf = p_workflow(workflow_id)
     if not wf.cloud_workflow_id:
-        return CloudRunsReady(runs=[])
+        return CloudRunsReady(runs=[], files_folder=downloads_root())
     try:
         runs = await cloud.list_runs(wf.cloud_workflow_id)
     except cloud.SignedOut:
@@ -103,11 +109,17 @@ async def workflow_cloud_runs(workflow_id: str) -> CloudRunsResponse:
     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 CloudRunsReady(runs=[], files_folder=downloads_root())
         return CloudRunsUnavailable(state="unknown", detail=exc.message)
     except cloud.CloudUnreachable as exc:
         return CloudRunsUnavailable(state="unknown", detail=exc.detail)
+
+    # Answered now, files fetched behind it. A 20MB attachment must not hold up the history the
+    # user asked for, and a run's files appear a moment later without them doing anything.
+    asyncio.create_task(fetch_missing(wf.cloud_workflow_id, wf, runs))
+
     return CloudRunsReady(
+        files_folder=downloads_root(),
         runs=[
             CloudRunRow(
                 id=r.id,
@@ -118,7 +130,8 @@ async def workflow_cloud_runs(workflow_id: str) -> CloudRunsResponse:
                 answer=r.answer,
                 notices=r.notices,
                 cost_usd=r.cost_usd,
+                files=described(wf, r),
             )
             for r in runs
-        ]
+        ],
     )

backend/apps/workflows/cloud/run_files.py

@@ -0,0 +1,143 @@
+"""Bring a cloud run's files down to the machine the user actually sits at.
+
+A cloud run finishes while the laptop is shut, so the files live in the cloud until
+something fetches them. That something is this module, and it puts them in Downloads
+rather than anywhere clever: the whole point of a deliverable is that the user can
+find it without being told where to look, and "it is in Downloads" is the one answer
+nobody needs explained.
+
+Every file is fetched at most once. The local path is derived, never stored, so a
+user who moves or deletes a file just gets it back on the next look rather than
+staring at a broken link.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+import re
+from typing import List, Optional
+
+from pydantic import BaseModel, ConfigDict
+from typeguard import typechecked
+
+from backend.apps.workflows.cloud import client as cloud
+from backend.apps.workflows.models import Workflow
+
+logger = logging.getLogger(__name__)
+
+# Bounded so one History open cannot spend minutes pulling a whole month of runs.
+MAX_CONCURRENT_DOWNLOADS = 3
+
+
+class LocalRunFile(BaseModel):
+    """A cloud file plus where it is (or would be) on this machine."""
+
+    model_config = ConfigDict(validate_assignment=True)
+
+    id: str
+    path: str
+    size_bytes: int
+    refusal: Optional[str] = None
+    # Set once the bytes are on disk. None means "not here yet", never "not coming".
+    local_path: Optional[str] = None
+
+
+@typechecked
+def safe_component(raw: str) -> str:
+    """One path segment a filesystem will take, with the user's own name still legible."""
+    cleaned = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "-", raw).strip(" .")
+    return cleaned[:60] or "untitled"
+
+
+@typechecked
+def downloads_root() -> str:
+    home = os.path.expanduser("~")
+    downloads = os.path.join(home, "Downloads")
+    return os.path.join(downloads if os.path.isdir(downloads) else home, "OpenSwarm")
+
+
+@typechecked
+def run_folder(workflow: Workflow, run: cloud.CloudRun) -> str:
+    """Where this run's files go: one folder per run, named so a list of them reads as a history."""
+    stamp = run.finished_at or run.started_at
+    when = "unknown-date"
+    if stamp:
+        from datetime import datetime, timezone
+
+        when = datetime.fromtimestamp(stamp / 1000, tz=timezone.utc).astimezone().strftime("%Y-%m-%d %H%M")
+    return os.path.join(
+        downloads_root(),
+        safe_component(workflow.title or "Workflow"),
+        safe_component(f"{when} {run.id[:8]}"),
+    )
+
+
+@typechecked
+def local_path_for(folder: str, file: cloud.CloudRunFile) -> str:
+    """The file's home, with every segment of the run-relative path made filesystem-safe."""
+    parts = [safe_component(part) for part in file.path.split("/") if part not in ("", ".", "..")]
+    return os.path.join(folder, *(parts or ["file"]))
+
+
+@typechecked
+def p_already_here(path: str, size_bytes: int) -> bool:
+    try:
+        return os.path.getsize(path) == size_bytes
+    except OSError:
+        return False
+
+
+@typechecked
+def described(workflow: Workflow, run: cloud.CloudRun) -> List[LocalRunFile]:
+    """This run's files and where they are right now. Reads disk, never fetches."""
+    folder = run_folder(workflow, run)
+    out: List[LocalRunFile] = []
+    for file in run.files:
+        local = local_path_for(folder, file)
+        out.append(LocalRunFile(
+            id=file.id,
+            path=file.path,
+            size_bytes=file.size_bytes,
+            refusal=file.refusal,
+            local_path=local if file.refusal is None and p_already_here(local, file.size_bytes) else None,
+        ))
+    return out
+
+
+@typechecked
+async def p_fetch_one(hosted_id: str, run: cloud.CloudRun, file: cloud.CloudRunFile, target: str) -> None:
+    payload = await cloud.download_run_file(hosted_id, run.id, file.id)
+    os.makedirs(os.path.dirname(target), exist_ok=True)
+    temporary = f"{target}.part"
+    with open(temporary, "wb") as handle:
+        handle.write(payload)
+    os.replace(temporary, target)
+    logger.info("cloud run %s: saved %s to %s", run.id, file.path, target)
+
+
+@typechecked
+async def fetch_missing(hosted_id: str, workflow: Workflow, runs: List[cloud.CloudRun]) -> None:
+    """Pull down anything not already here. Never raises: a run's answer must render
+    even when its attachments cannot be fetched, and the next look tries again."""
+    semaphore = asyncio.Semaphore(MAX_CONCURRENT_DOWNLOADS)
+
+    async def p_guarded(run: cloud.CloudRun, file: cloud.CloudRunFile, target: str) -> None:
+        async with semaphore:
+            try:
+                await p_fetch_one(hosted_id, run, file, target)
+            except (cloud.SignedOut, cloud.CloudRefused, cloud.CloudUnreachable, OSError) as exc:
+                logger.info("cloud run %s: %s not fetched (%s)", run.id, file.path, exc)
+
+    pending = []
+    for run in runs:
+        folder = run_folder(workflow, run)
+        for file in run.files:
+            if file.refusal is not None:
+                continue
+            target = local_path_for(folder, file)
+            if p_already_here(target, file.size_bytes):
+                continue
+            pending.append(p_guarded(run, file, target))
+    if pending:
+        await asyncio.gather(*pending)

backend/apps/workflows/cloud/status.py

@@ -16,6 +16,7 @@ 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.portable_context import portable_context
 from backend.apps.workflows.cloud.schedule import ScheduleSupported, to_cloud_schedule, wire
 from backend.apps.workflows.models import Workflow
 
@@ -74,7 +75,10 @@ def current_signature(wf: Workflow) -> Optional[str]:
     mapping = to_cloud_schedule(wf.schedule)
     if not isinstance(mapping, ScheduleSupported):
         return None
-    return definition_signature(cloud_definition(wf), wire(mapping.schedule))
+    # wire() is the richer serializer (weekly, zones, ends_at); portable_context is what the runner needs to reproduce the user's setup.
+    return definition_signature(
+        cloud_definition(wf), wire(mapping.schedule), portable_context().as_body()
+    )
 
 
 @typechecked