[eric] backend: a finished cloud run's files land in Downloads, where a user would actually look

backend/apps/workflows/cloud/client.py

@@ -20,6 +20,8 @@ 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
+# 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):
@@ -85,6 +87,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)
 
@@ -96,6 +110,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
@@ -229,11 +244,22 @@ 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
+    *,
+    hosted_id: Optional[str],
+    name: str,
+    definition: Dict[str, Any],
+    schedule: CloudSchedule,
+    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."""
+    workflow stops running last week's prose.
+
+    `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 = {"name": name, "definition": definition, "schedule": schedule.model_dump()}
+    if context:
+        body.update(context)
     if hosted_id:
         try:
             raw = await p_call("POST", f"/{hosted_id}/update", body)
@@ -270,6 +296,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")
@@ -291,6 +337,30 @@ 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/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/tests/test_cloud_run_files.py

@@ -0,0 +1,152 @@
+"""Where a cloud run's files land on the machine the user sits at, and what they are told.
+
+A deliverable the user cannot find has not been delivered, so the two things worth pinning
+are the folder (Downloads, because that is where people look) and the honesty of the list:
+a file that was refused must still appear, with its reason, and must never be fetched.
+"""
+
+import os
+
+import pytest
+
+from backend.apps.workflows.cloud.client import CloudRun, CloudRunFile
+from backend.apps.workflows.cloud.run_files import (
+    described,
+    downloads_root,
+    fetch_missing,
+    local_path_for,
+    safe_component,
+    run_folder,
+)
+from backend.apps.workflows.models import Workflow
+
+
+def p_run(**overrides) -> CloudRun:
+    body = {
+        "id": "run-abcdef123456",
+        "status": "succeeded",
+        "finished_at": 1_785_000_000_000,
+        "files": [],
+    }
+    body.update(overrides)
+    return CloudRun.model_validate(body)
+
+
+def test_files_land_in_downloads_because_that_is_where_people_look() -> None:
+    root = downloads_root()
+    assert root.endswith(os.path.join("OpenSwarm"))
+    assert os.path.expanduser("~") in root
+
+
+def test_a_run_gets_its_own_folder_named_so_a_list_of_them_reads_as_history() -> None:
+    workflow = Workflow(title="Weekly numbers")
+    folder = run_folder(workflow, p_run())
+    assert "Weekly numbers" in folder
+    # The run id's head disambiguates two runs in the same minute without being a wall of hex.
+    assert "run-abcd" in folder
+
+
+@pytest.mark.parametrize(
+    "raw,expected",
+    [
+        ("Weekly numbers", "Weekly numbers"),
+        ("bad/name", "bad-name"),
+        ("../escape", "-escape"),
+        ("with:colon", "with-colon"),
+        ("", "untitled"),
+        ("...", "untitled"),
+    ],
+)
+def test_a_title_is_made_safe_without_becoming_unrecognisable(raw: str, expected: str) -> None:
+    assert safe_component(raw) == expected
+
+
+def test_a_nested_run_path_stays_nested_locally() -> None:
+    target = local_path_for("/tmp/folder", CloudRunFile(id="f1", path="notes/method.txt", size_bytes=5))
+    assert target == os.path.join("/tmp/folder", "notes", "method.txt")
+
+
+def test_a_path_that_tries_to_climb_out_cannot(tmp_path) -> None:
+    target = local_path_for(str(tmp_path), CloudRunFile(id="f1", path="../../etc/passwd", size_bytes=1))
+    assert os.path.abspath(target).startswith(str(tmp_path))
+
+
+def test_a_file_not_yet_downloaded_reports_no_local_path_rather_than_a_broken_one(tmp_path, monkeypatch) -> None:
+    monkeypatch.setattr("backend.apps.workflows.cloud.run_files.downloads_root", lambda: str(tmp_path))
+    run = p_run(files=[{"id": "f1", "path": "summary.md", "size_bytes": 42}])
+    described_files = described(Workflow(title="W"), run)
+
+    assert len(described_files) == 1
+    assert described_files[0].local_path is None
+
+
+def test_a_file_already_on_disk_is_reported_at_its_real_path(tmp_path, monkeypatch) -> None:
+    monkeypatch.setattr("backend.apps.workflows.cloud.run_files.downloads_root", lambda: str(tmp_path))
+    workflow = Workflow(title="W")
+    run = p_run(files=[{"id": "f1", "path": "summary.md", "size_bytes": 5}])
+    target = local_path_for(run_folder(workflow, run), run.files[0])
+    os.makedirs(os.path.dirname(target), exist_ok=True)
+    with open(target, "wb") as handle:
+        handle.write(b"hello")
+
+    assert described(workflow, run)[0].local_path == target
+
+
+def test_a_refused_file_is_listed_with_its_reason_and_never_fetched(tmp_path, monkeypatch) -> None:
+    monkeypatch.setattr("backend.apps.workflows.cloud.run_files.downloads_root", lambda: str(tmp_path))
+    workflow = Workflow(title="W")
+    run = p_run(files=[
+        {"id": "f1", "path": "render.mp4", "size_bytes": 999, "refusal": "it is 512.0 MB and the limit is 20.0 MB"},
+    ])
+
+    listed = described(workflow, run)
+    assert listed[0].refusal is not None
+    assert listed[0].local_path is None
+
+    asked: list = []
+
+    async def p_never(*args, **kwargs):
+        asked.append(args)
+        return b""
+
+    monkeypatch.setattr("backend.apps.workflows.cloud.run_files.cloud.download_run_file", p_never)
+    import asyncio
+
+    asyncio.run(fetch_missing("hosted-1", workflow, [run]))
+    assert asked == [], "a file that exists nowhere must never be requested"
+
+
+def test_a_download_that_fails_leaves_the_history_readable(tmp_path, monkeypatch) -> None:
+    monkeypatch.setattr("backend.apps.workflows.cloud.run_files.downloads_root", lambda: str(tmp_path))
+    workflow = Workflow(title="W")
+    run = p_run(files=[{"id": "f1", "path": "summary.md", "size_bytes": 5}])
+
+    async def p_boom(*args, **kwargs):
+        from backend.apps.workflows.cloud.client import CloudUnreachable
+
+        raise CloudUnreachable("offline")
+
+    monkeypatch.setattr("backend.apps.workflows.cloud.run_files.cloud.download_run_file", p_boom)
+    import asyncio
+
+    asyncio.run(fetch_missing("hosted-1", workflow, [run]))
+    assert described(workflow, run)[0].local_path is None
+
+
+def test_a_downloaded_file_is_written_whole_or_not_at_all(tmp_path, monkeypatch) -> None:
+    monkeypatch.setattr("backend.apps.workflows.cloud.run_files.downloads_root", lambda: str(tmp_path))
+    workflow = Workflow(title="W")
+    run = p_run(files=[{"id": "f1", "path": "notes/summary.md", "size_bytes": 5}])
+
+    async def p_ok(*args, **kwargs):
+        return b"hello"
+
+    monkeypatch.setattr("backend.apps.workflows.cloud.run_files.cloud.download_run_file", p_ok)
+    import asyncio
+
+    asyncio.run(fetch_missing("hosted-1", workflow, [run]))
+    target = local_path_for(run_folder(workflow, run), run.files[0])
+    with open(target, "rb") as handle:
+        assert handle.read() == b"hello"
+    # The .part file is the whole point of the atomic rename: no half file is ever visible.
+    assert not os.path.exists(f"{target}.part")