From 6ab95132406f0ac0db6a0140cdaf4c80e9ee0a0e Mon Sep 17 00:00:00 2001 From: ciregenz Date: Sat, 1 Aug 2026 01:20:10 -0700 Subject: [PATCH] [eric] backend: a finished cloud run's files land in Downloads, where a user would actually look --- backend/apps/workflows/cloud/client.py | 74 ++++++++++- backend/apps/workflows/cloud/routes.py | 19 ++- backend/apps/workflows/cloud/run_files.py | 143 ++++++++++++++++++++ backend/tests/test_cloud_run_files.py | 152 ++++++++++++++++++++++ 4 files changed, 383 insertions(+), 5 deletions(-) create mode 100644 backend/apps/workflows/cloud/run_files.py create mode 100644 backend/tests/test_cloud_run_files.py diff --git a/backend/apps/workflows/cloud/client.py b/backend/apps/workflows/cloud/client.py index 57cc82ab..a1ffabc4 100644 --- a/backend/apps/workflows/cloud/client.py +++ b/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 diff --git a/backend/apps/workflows/cloud/routes.py b/backend/apps/workflows/cloud/routes.py index 8671a6c5..2d664671 100644 --- a/backend/apps/workflows/cloud/routes.py +++ b/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 - ] + ], ) diff --git a/backend/apps/workflows/cloud/run_files.py b/backend/apps/workflows/cloud/run_files.py new file mode 100644 index 00000000..2695c805 --- /dev/null +++ b/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) diff --git a/backend/tests/test_cloud_run_files.py b/backend/tests/test_cloud_run_files.py new file mode 100644 index 00000000..0af62106 --- /dev/null +++ b/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")