diff --git a/backend/apps/marketplace/install_jobs.py b/backend/apps/marketplace/install_jobs.py new file mode 100644 index 00000000..b1f92fb8 --- /dev/null +++ b/backend/apps/marketplace/install_jobs.py @@ -0,0 +1,83 @@ +"""An install as a job the pill can watch: download (with byte progress), stage, then hand back the +same review a dropped .swarm gets. The job lives in memory for a few minutes; a restart forgets it and +the pill simply asks again. +""" + +import threading +import time +import uuid +from dataclasses import dataclass, field +from typing import Literal, Optional + +from typeguard import typechecked + +from backend.apps.marketplace.catalog import Listing +from backend.apps.marketplace.package_download import DownloadRefused, download_package, package_filename +from backend.apps.swarm.models import ImportPreflightResponse + +JobPhase = Literal["downloading", "staging", "ready", "failed"] +JOB_TTL_SECONDS = 600 + + +@dataclass +class InstallJob: + id: str + listing_id: str + phase: JobPhase = "downloading" + received: int = 0 + total: int = 0 + preflight: Optional[ImportPreflightResponse] = None + error: Optional[str] = None + started_at: float = field(default_factory=time.time) + + +P_JOBS: dict[str, InstallJob] = {} +P_LOCK = threading.Lock() + + +@typechecked +def start_job(listing_id: str) -> InstallJob: + sweep_expired() + job = InstallJob(id=uuid.uuid4().hex, listing_id=listing_id) + with P_LOCK: + P_JOBS[job.id] = job + return job + + +@typechecked +def get_job(job_id: str) -> Optional[InstallJob]: + with P_LOCK: + return P_JOBS.get(job_id) + + +@typechecked +def sweep_expired(now: Optional[float] = None) -> int: + cutoff = (now if now is not None else time.time()) - JOB_TTL_SECONDS + with P_LOCK: + stale = [job_id for job_id, job in P_JOBS.items() if job.started_at < cutoff] + for job_id in stale: + del P_JOBS[job_id] + return len(stale) + + +@typechecked +def run_job(job: InstallJob, listing: Listing) -> None: + """Runs on a worker thread. Every exit lands the job in ready or failed; nothing is left spinning.""" + + def on_progress(received: int, total: int) -> None: + job.received = received + job.total = total + + try: + raw = download_package(listing.download_url, on_progress) + job.phase = "staging" + # Staging is local and quick; the ring reads as full while it runs. + from backend.apps.swarm.swarm import stage_bundle_for_import + job.preflight = stage_bundle_for_import(raw, package_filename(listing.id, listing.title)) + job.phase = "ready" + except DownloadRefused as e: + job.error = str(e) + job.phase = "failed" + except Exception as e: + job.error = f"Couldn't install it: {e}" + job.phase = "failed" diff --git a/backend/apps/marketplace/marketplace.py b/backend/apps/marketplace/marketplace.py index 4f53574f..e28b7dbc 100644 --- a/backend/apps/marketplace/marketplace.py +++ b/backend/apps/marketplace/marketplace.py @@ -14,14 +14,9 @@ from fastapi import HTTPException from pydantic import BaseModel, ConfigDict from backend.apps.marketplace import catalog +from backend.apps.marketplace.install_jobs import InstallJob, JobPhase, get_job, run_job, start_job from backend.apps.marketplace.installs import InstallRecord, load_installs, record_install -from backend.apps.marketplace.package_download import ( - DownloadRefused, - download_package, - package_filename, -) from backend.apps.swarm.models import ImportPreflightResponse -from backend.apps.swarm.swarm import stage_bundle_for_import from backend.config.Apps import SubApp logger = logging.getLogger(__name__) @@ -48,20 +43,55 @@ async def get_listings(refresh: bool = False) -> catalog.CatalogResponse: return await asyncio.to_thread(catalog.load_catalog, refresh) -@marketplace.router.post("/install/preflight") -async def install_preflight(body: InstallRequest) -> ImportPreflightResponse: - """Download the listing's bundle and stage it for the ordinary import confirm flow.""" +class InstallStartResponse(BaseModel): + model_config = ConfigDict(validate_assignment=True) + + job_id: str + + +class InstallJobStatus(BaseModel): + model_config = ConfigDict(validate_assignment=True) + + job_id: str + phase: JobPhase + received: int + total: int + preflight: ImportPreflightResponse | None = None + error: str | None = None + + +def job_status(job: InstallJob) -> InstallJobStatus: + return InstallJobStatus(job_id=job.id, phase=job.phase, received=job.received, total=job.total, preflight=job.preflight, error=job.error) + + +# Fire-and-forget tasks need a reference or the loop may drop them mid-download. +P_RUNNING: set[asyncio.Task[None]] = set() + + +@marketplace.router.post("/install/start") +async def install_start(body: InstallRequest) -> InstallStartResponse: + """Start downloading the listing's bundle; the job stages it for the ordinary import confirm + flow and the pill polls /install/{job_id} for byte progress and the review.""" listing = await asyncio.to_thread(catalog.find_listing, body.id) if listing is None: raise HTTPException(status_code=404, detail="that package is not in the catalog any more") if not listing.download_url: raise HTTPException(status_code=400, detail="this listing has no package file yet") - try: - raw = await asyncio.to_thread(download_package, listing.download_url) - except DownloadRefused as e: - logger.warning("marketplace download refused for %s: %s", listing.id, e) - raise HTTPException(status_code=400, detail=str(e)) - return stage_bundle_for_import(raw, package_filename(listing.id, listing.title)) + job = start_job(listing.id) + task = asyncio.create_task(asyncio.to_thread(run_job, job, listing)) + P_RUNNING.add(task) + task.add_done_callback(P_RUNNING.discard) + return InstallStartResponse(job_id=job.id) + + +@marketplace.router.get("/install/{job_id}") +async def install_status(job_id: str) -> InstallJobStatus: + job = get_job(job_id) + if job is None: + raise HTTPException(status_code=404, detail="that install is no longer running; start it again") + if job.phase == "failed": + logger.warning("marketplace install failed for %s: %s", job.listing_id, job.error) + return job_status(job) class InstallsResponse(BaseModel): diff --git a/backend/apps/marketplace/package_download.py b/backend/apps/marketplace/package_download.py index d101378a..7c7e839c 100644 --- a/backend/apps/marketplace/package_download.py +++ b/backend/apps/marketplace/package_download.py @@ -5,8 +5,12 @@ otherwise make OUR backend fetch whatever it names, including localhost and clou hop of the redirect chain is checked, not just the first, because a permitted host may redirect. """ +import logging +import os +import time import urllib.error import urllib.request +from typing import Callable, Optional from urllib.parse import urlparse from typeguard import typechecked @@ -22,6 +26,14 @@ ALLOWED_DOWNLOAD_HOSTS = ( DOWNLOAD_TIMEOUT_SECONDS = 60 MAX_PACKAGE_BYTES = 200 * 1024 * 1024 +CHUNK_BYTES = 64 * 1024 +# Drill seam: cap the download at this many bytes per second so a human can watch the ring fill. Unset in real life. +THROTTLE_ENV = "OSW_MARKETPLACE_THROTTLE_BPS" + +logger = logging.getLogger(__name__) + +# (bytes received so far, total bytes or 0 when the server did not say) +ProgressCallback = Callable[[int, int], None] class DownloadRefused(Exception): @@ -48,15 +60,57 @@ class AllowlistRedirectHandler(urllib.request.HTTPRedirectHandler): @typechecked -def download_package(url: str) -> bytes: - """The bundle's bytes, or DownloadRefused. Never returns a partial or oversized body.""" +def throttle_bytes_per_second() -> int: + raw = (os.environ.get(THROTTLE_ENV) or "").strip() + if not raw: + return 0 + try: + bps = int(raw) + except ValueError: + return 0 + if bps > 0: + logger.warning("marketplace downloads throttled to %d bytes/s by %s (drill seam)", bps, THROTTLE_ENV) + return max(0, bps) + + +@typechecked +def declared_total(response: object) -> int: + """Content-Length when the server sent one, else 0 (the ring then spins instead of filling).""" + try: + return max(0, int(response.headers.get("Content-Length") or 0)) # type: ignore[attr-defined] + except (TypeError, ValueError): + return 0 + + +@typechecked +def download_package(url: str, on_progress: Optional[ProgressCallback] = None) -> bytes: + """The bundle's bytes, or DownloadRefused. Never returns a partial or oversized body. Reads in + chunks and reports (received, total) after each one so an install can show a real ring.""" if not host_allowed(url): raise DownloadRefused("this package is not hosted somewhere OpenSwarm will download from") opener = urllib.request.build_opener(AllowlistRedirectHandler()) request = urllib.request.Request(url, headers={"User-Agent": "OpenSwarm-Marketplace/1.0"}) try: with opener.open(request, timeout=DOWNLOAD_TIMEOUT_SECONDS) as response: - raw = response.read(MAX_PACKAGE_BYTES + 1) + total = declared_total(response) + if total > MAX_PACKAGE_BYTES: + raise DownloadRefused("the package is too large") + throttle = throttle_bytes_per_second() + chunks: list[bytes] = [] + received = 0 + while True: + chunk = response.read(CHUNK_BYTES) + if not chunk: + break + received += len(chunk) + if received > MAX_PACKAGE_BYTES: + raise DownloadRefused("the package is too large") + chunks.append(chunk) + if on_progress is not None: + on_progress(received, total) + if throttle: + time.sleep(len(chunk) / throttle) + raw = b"".join(chunks) except urllib.error.HTTPError as e: raise DownloadRefused(f"the download returned {e.code}") except DownloadRefused: diff --git a/backend/run.sh b/backend/run.sh index cf4c220a..11039d2f 100755 --- a/backend/run.sh +++ b/backend/run.sh @@ -82,7 +82,10 @@ done # single-process uvicorn. if [[ "${OPENSWARM_DEV:-}" == "1" ]]; then echo "OPENSWARM_DEV=1 detected, running uvicorn with --reload." + # A reload SIGTERMs the worker and waits for it; when the graceful shutdown wedges (2026-09-03: worker idle, + # listen queue 128/128, port dead for 13 minutes) the cap turns that wait into a real restart. python3 -m uvicorn backend.main:app --host 0.0.0.0 --port "$BACKEND_PORT" --reload \ + --timeout-graceful-shutdown 10 \ --reload-dir "$BACKEND_DIR_ABSPATH" \ "${UVICORN_EXCLUDE_ARGS[@]}" else diff --git a/backend/tests/test_marketplace_catalog.py b/backend/tests/test_marketplace_catalog.py index 612aa6ac..80888e76 100644 --- a/backend/tests/test_marketplace_catalog.py +++ b/backend/tests/test_marketplace_catalog.py @@ -121,10 +121,10 @@ def test_a_caller_names_a_listing_id_never_a_url(): def test_install_stages_through_the_same_door_as_a_dropped_file(): - from backend.apps.swarm import swarm as swarm_routes + from backend.apps.marketplace import install_jobs - assert marketplace.stage_bundle_for_import is swarm_routes.stage_bundle_for_import - body = inspect.getsource(marketplace.install_preflight) - assert "stage_bundle_for_import" in body + body = inspect.getsource(install_jobs.run_job) + assert "from backend.apps.swarm.swarm import stage_bundle_for_import" in body for forbidden in ("write_folder_skill", "closure.commit", "import_commit"): assert forbidden not in body, "install must never write; it stages and lets the user confirm" + assert forbidden not in inspect.getsource(marketplace), "the routes must never write either" diff --git a/backend/tests/test_marketplace_install_jobs.py b/backend/tests/test_marketplace_install_jobs.py new file mode 100644 index 00000000..97f341e5 --- /dev/null +++ b/backend/tests/test_marketplace_install_jobs.py @@ -0,0 +1,122 @@ +"""An install is a job the pill can watch: the download streams and reports bytes, the job walks +downloading -> staging -> ready (or failed) and nothing is left spinning, and the routes expose it.""" + +import io +import time +from typing import Any + +from backend.apps.marketplace import install_jobs, package_download +from backend.apps.marketplace.catalog import Listing +from backend.apps.marketplace.package_download import DownloadRefused, download_package + + +class P_Response: + def __init__(self, body: bytes, content_length: Any) -> None: + self.headers = {"Content-Length": content_length} + self._buf = io.BytesIO(body) + + def read(self, n: int) -> bytes: + return self._buf.read(n) + + def __enter__(self) -> "P_Response": + return self + + def __exit__(self, *exc: object) -> None: + return None + + +class P_Opener: + def __init__(self, response: P_Response) -> None: + self.response = response + + def open(self, request: object, timeout: float) -> P_Response: + return self.response + + +def p_fake_opener(monkeypatch: Any, body: bytes, content_length: Any) -> None: + monkeypatch.setattr(package_download.urllib.request, "build_opener", lambda *a, **k: P_Opener(P_Response(body, content_length))) + + +def test_the_download_reports_bytes_after_every_chunk_and_ends_on_the_total(monkeypatch: Any) -> None: + body = b"x" * (package_download.CHUNK_BYTES * 2 + 17) + p_fake_opener(monkeypatch, body, str(len(body))) + seen: list[tuple[int, int]] = [] + raw = download_package("https://github.com/o/r/releases/download/v1/p.swarm", lambda r, t: seen.append((r, t))) + assert raw == body + assert [r for r, _ in seen] == [package_download.CHUNK_BYTES, package_download.CHUNK_BYTES * 2, len(body)] + assert {t for _, t in seen} == {len(body)} + + +def test_no_content_length_means_total_zero_and_the_bytes_still_arrive(monkeypatch: Any) -> None: + p_fake_opener(monkeypatch, b"abc", None) + seen: list[tuple[int, int]] = [] + assert download_package("https://github.com/o/r/p.swarm", lambda r, t: seen.append((r, t))) == b"abc" + assert seen == [(3, 0)] + + +def test_an_oversized_declaration_is_refused_before_a_byte_is_read(monkeypatch: Any) -> None: + p_fake_opener(monkeypatch, b"", str(package_download.MAX_PACKAGE_BYTES + 1)) + try: + download_package("https://github.com/o/r/p.swarm") + except DownloadRefused as e: + assert "too large" in str(e) + else: + raise AssertionError("an oversized package was downloaded") + + +def test_a_job_walks_downloading_staging_ready_with_the_review(monkeypatch: Any) -> None: + listing = Listing(id="git-graph", title="Git Graph", download_url="https://github.com/o/r/p.swarm") + monkeypatch.setattr(install_jobs, "download_package", lambda url, cb: (cb(5, 10), cb(10, 10), b"bundle")[-1]) + import backend.apps.swarm.swarm as p_swarm + from backend.apps.swarm.models import BundleSummary, ImportPreflightResponse + monkeypatch.setattr(p_swarm, "stage_bundle_for_import", lambda raw, fn: ImportPreflightResponse.model_construct(summary=BundleSummary.model_construct(), staging_token="tok-" + fn)) + job = install_jobs.start_job(listing.id) + install_jobs.run_job(job, listing) + assert job.phase == "ready" + assert (job.received, job.total) == (10, 10) + assert job.preflight is not None and job.preflight.staging_token == "tok-git-graph.swarm" + assert install_jobs.get_job(job.id) is job + + +def test_a_refused_download_lands_the_job_in_failed_with_the_reason(monkeypatch: Any) -> None: + listing = Listing(id="x", title="X", download_url="https://github.com/o/r/p.swarm") + + def refuse(url: str, cb: Any) -> bytes: + raise DownloadRefused("the download returned 404") + + monkeypatch.setattr(install_jobs, "download_package", refuse) + job = install_jobs.start_job(listing.id) + install_jobs.run_job(job, listing) + assert job.phase == "failed" + assert job.error == "the download returned 404" + + +def test_jobs_expire_so_the_store_cannot_grow_forever() -> None: + job = install_jobs.start_job("old") + job.started_at = time.time() - install_jobs.JOB_TTL_SECONDS - 1 + assert install_jobs.sweep_expired() >= 1 + assert install_jobs.get_job(job.id) is None + + +def test_the_routes_start_a_job_and_report_it(monkeypatch: Any) -> None: + from fastapi.testclient import TestClient + from backend.apps.marketplace import marketplace as p_mp + + listing = Listing(id="git-graph", title="Git Graph", download_url="https://github.com/o/r/p.swarm") + monkeypatch.setattr(p_mp.catalog, "find_listing", lambda listing_id: listing if listing_id == "git-graph" else None) + monkeypatch.setattr(p_mp, "run_job", lambda job, lst: setattr(job, "phase", "failed") or setattr(job, "error", "the download returned 404")) + from fastapi import FastAPI + app = FastAPI() + app.include_router(p_mp.marketplace.router, prefix="/api/marketplace") + client = TestClient(app) + assert client.post("/api/marketplace/install/start", json={"id": "nope"}).status_code == 404 + started = client.post("/api/marketplace/install/start", json={"id": "git-graph"}) + assert started.status_code == 200 + job_id = started.json()["job_id"] + for _ in range(50): + status = client.get(f"/api/marketplace/install/{job_id}").json() + if status["phase"] == "failed": + break + time.sleep(0.02) + assert status["phase"] == "failed" and status["error"] == "the download returned 404" + assert client.get("/api/marketplace/install/does-not-exist").status_code == 404 diff --git a/backend/tests/test_stream_frames_skip_dashboard_socket.py b/backend/tests/test_stream_frames_skip_dashboard_socket.py index 2162100f..b021721c 100644 --- a/backend/tests/test_stream_frames_skip_dashboard_socket.py +++ b/backend/tests/test_stream_frames_skip_dashboard_socket.py @@ -5,7 +5,6 @@ Both directions pinned: deltas stay session-only, and every other event still re dashboard socket, or narrator pills and status chips would go blind.""" import pytest -from unittest.mock import AsyncMock from backend.apps.agents.core.ws_manager import ws_manager diff --git a/frontend/src/app/pages/Directory/DirectoryPackagesTab.tsx b/frontend/src/app/pages/Directory/DirectoryPackagesTab.tsx index 5e01ac17..b12058d8 100644 --- a/frontend/src/app/pages/Directory/DirectoryPackagesTab.tsx +++ b/frontend/src/app/pages/Directory/DirectoryPackagesTab.tsx @@ -21,7 +21,8 @@ import PackageCard from './packages/PackageCard'; import PackageDialog from './packages/detail/PackageDialog'; import PackageBundleCard from './packages/PackageBundleCard'; import PackageBundleDialog from './packages/detail/PackageBundleDialog'; -import { stagePackageInstall } from './packages/installPackage'; +import { stagePackageInstall, type InstallProgress } from './packages/installPackage'; +import { fractionOf } from './packages/installRing'; import { fetchInstalls, installState, recordFor, recordInstall, type InstallRecord, type PillState } from './packages/installs'; import { KIND_LABELS, isBundle, resolveBundleMembers, type Listing } from './packages/catalog'; @@ -43,6 +44,7 @@ const DirectoryPackagesTab: React.FC<{ onOpenSkill?: (skillId: string) => void } const [openListing, setOpenListing] = useState(null); const [openBundle, setOpenBundle] = useState(null); const [installingId, setInstallingId] = useState(null); + const [progress, setProgress] = useState(null); const [installs, setInstalls] = useState>({}); const [confirm, setConfirm] = useState<{ preflight: ImportPreflight; listingId: string } | null>(null); const [committing, setCommitting] = useState(false); @@ -130,16 +132,19 @@ const DirectoryPackagesTab: React.FC<{ onOpenSkill?: (skillId: string) => void } const install = async (listingId: string) => { setInstallingId(listingId); try { - const preflight = await stagePackageInstall(listingId); + const preflight = await stagePackageInstall(listingId, setProgress); if (marketplaceNeedsConfirm(preflight)) setConfirm({ preflight, listingId }); else await commit(preflight, listingId); } catch (e: unknown) { setToast({ message: e instanceof Error ? e.message : "Couldn't download it. Try again.", severity: 'error' }); } finally { setInstallingId(null); + setProgress(null); } }; + const progressFor = (listingId: string): number | null => (installingId === listingId ? fractionOf(progress) : null); + const installBundle = async (bundle: Listing) => { const members = resolveBundleMembers(bundle, listings).filter((m) => m.download_url); if (members.length === 0) { @@ -213,6 +218,7 @@ const DirectoryPackagesTab: React.FC<{ onOpenSkill?: (skillId: string) => void } key={listing.id} listing={listing} state={stateFor(listing)} + progress={progressFor(listing.id)} onOpen={() => setOpenListing(listing)} onGet={() => { void install(listing.id); }} onOpenInstalled={() => openInstalled(listing)} @@ -257,6 +263,7 @@ const DirectoryPackagesTab: React.FC<{ onOpenSkill?: (skillId: string) => void } { if (openListing) void install(openListing.id); }} onOpen={() => { if (openListing) openInstalled(openListing); }} onClose={() => setOpenListing(null)} @@ -265,6 +272,7 @@ const DirectoryPackagesTab: React.FC<{ onOpenSkill?: (skillId: string) => void } bundle={openBundle} members={openBundle ? resolveBundleMembers(openBundle, listings) : []} stateOf={(id) => { const l = listings.find((x) => x.id === id); return l ? stateFor(l) : 'get'; }} + progressOf={progressFor} onOpenInstalled={(member) => openInstalled(member)} installing={installingId !== null} onInstallAll={() => { if (openBundle) void installBundle(openBundle); }} diff --git a/frontend/src/app/pages/Directory/packages/InstallPill.tsx b/frontend/src/app/pages/Directory/packages/InstallPill.tsx index d3b01b3c..3e47e575 100644 --- a/frontend/src/app/pages/Directory/packages/InstallPill.tsx +++ b/frontend/src/app/pages/Directory/packages/InstallPill.tsx @@ -1,12 +1,16 @@ import React from 'react'; +import Box from '@mui/material/Box'; import Button from '@mui/material/Button'; import CircularProgress from '@mui/material/CircularProgress'; import CheckRoundedIcon from '@mui/icons-material/CheckRounded'; import { useClaudeTokens } from '@/shared/styles/ThemeContext'; import type { PillState } from './installs'; +import { ringFor } from './installRing'; interface Props { state: PillState; + // 0..1 while a download runs; null when the size is unknown. + progress?: number | null; disabled?: boolean; onGet: () => void; onOpen: () => void; @@ -14,7 +18,7 @@ interface Props { } // The one action a package has, the way the App Store draws it: Install, a spinner while it lands, then Open. It swallows the click so a card underneath never opens its sheet. -export default function InstallPill({ state, disabled, onGet, onOpen, size = 'md' }: Props) { +export default function InstallPill({ state, progress, disabled, onGet, onOpen, size = 'md' }: Props) { const c = useClaudeTokens(); const sm = size === 'sm'; const base = { @@ -43,15 +47,41 @@ export default function InstallPill({ state, disabled, onGet, onOpen, size = 'md ); } + if (state === 'installing') { + // The App Store's ring: the button gives way to a circle that fills with the bytes, in the same box so nothing shifts. + const ring = ringFor(progress); + const px = sm ? 18 : 22; + return ( + + + + + ); + } return ( ); } diff --git a/frontend/src/app/pages/Directory/packages/PackageCard.tsx b/frontend/src/app/pages/Directory/packages/PackageCard.tsx index 08cc3d01..17d1bb08 100644 --- a/frontend/src/app/pages/Directory/packages/PackageCard.tsx +++ b/frontend/src/app/pages/Directory/packages/PackageCard.tsx @@ -12,13 +12,14 @@ import type { PillState } from './installs'; interface Props { listing: Listing; state: PillState; + progress?: number | null; onOpen: () => void; onGet: () => void; onOpenInstalled: () => void; onTag: (tag: string) => void; } -export default function PackageCard({ listing, state, onOpen, onGet, onOpenInstalled, onTag }: Props) { +export default function PackageCard({ listing, state, progress, onOpen, onGet, onOpenInstalled, onTag }: Props) { const c = useClaudeTokens(); const tags = parseTags(listing.tags).slice(0, 3); @@ -78,7 +79,7 @@ export default function PackageCard({ listing, state, onOpen, onGet, onOpenInsta {listing.author ? ` · ${listing.author}` : ''} - + PillState; + progressOf: (listingId: string) => number | null; onOpenInstalled: (listing: Listing) => void; onClose: () => void; onOpenMember: (member: Listing) => void; @@ -30,7 +31,7 @@ interface Props { // A bundle has no package of its own: the sheet lists it, the dialog installs its members. There is no // file to download here on purpose; a store page installs, it does not hand out archives. -export default function PackageBundleDialog({ bundle, members, stateOf, onOpenInstalled, onClose, onOpenMember, onInstallAll, onInstallMember, installing }: Props) { +export default function PackageBundleDialog({ bundle, members, stateOf, progressOf, onOpenInstalled, onClose, onOpenMember, onInstallAll, onInstallMember, installing }: Props) { const c = useClaudeTokens(); if (!bundle) return null; const installable = members.filter((m) => m.download_url); @@ -116,7 +117,7 @@ export default function PackageBundleDialog({ bundle, members, stateOf, onOpenIn {KIND_LABELS[m.kind] || m.kind || 'Package'}{m.version ? ` · v${m.version}` : ''} - onInstallMember(m.id)} onOpen={() => onOpenInstalled(m)} size="sm" /> + onInstallMember(m.id)} onOpen={() => onOpenInstalled(m)} size="sm" /> ); })} diff --git a/frontend/src/app/pages/Directory/packages/detail/PackageDialog.tsx b/frontend/src/app/pages/Directory/packages/detail/PackageDialog.tsx index 721785bb..9f5af5a6 100644 --- a/frontend/src/app/pages/Directory/packages/detail/PackageDialog.tsx +++ b/frontend/src/app/pages/Directory/packages/detail/PackageDialog.tsx @@ -20,13 +20,14 @@ interface Props { listing: Listing | null; onClose: () => void; state: PillState; + progress?: number | null; onInstall: () => void; onOpen: () => void; } // The package sheet: one Install action, no file to download. The bundle is fetched and reviewed by the // same import path a dropped .swarm takes, so the raw archive has no job on a store page. -export default function PackageDialog({ listing, onClose, state, onInstall, onOpen }: Props) { +export default function PackageDialog({ listing, onClose, state, progress, onInstall, onOpen }: Props) { const c = useClaudeTokens(); if (!listing) return null; const details = detailsForListing(listing); @@ -69,7 +70,7 @@ export default function PackageDialog({ listing, onClose, state, onInstall, onOp {meta} - + {listing.description && ( diff --git a/frontend/src/app/pages/Directory/packages/installPackage.ts b/frontend/src/app/pages/Directory/packages/installPackage.ts index c978a09f..ceb6e722 100644 --- a/frontend/src/app/pages/Directory/packages/installPackage.ts +++ b/frontend/src/app/pages/Directory/packages/installPackage.ts @@ -4,22 +4,59 @@ import type { ImportPreflight } from '@/app/components/share/shareTypes'; // Installing a marketplace package is the ordinary bundle import with the download done for you: // the backend fetches the .swarm and stages it, then the SAME confirm surface and commit route a // dropped file uses take over. One door, so the secret review and the skill-confirm rule cannot -// hold on one path and not the other. -export async function stagePackageInstall(listingId: string): Promise { - const res = await fetch(`${API_BASE}/marketplace/install/preflight`, { +// hold on one path and not the other. The download runs as a job so the pill can show real bytes. + +export interface InstallProgress { + received: number; + total: number; +} + +interface JobStatus { + job_id: string; + phase: 'downloading' | 'staging' | 'ready' | 'failed'; + received: number; + total: number; + preflight?: ImportPreflight | null; + error?: string | null; +} + +export const POLL_MS = 150; +const DOWNLOAD_FAILED = "Couldn't download it. Try again."; + +async function detailOf(res: Response, fallback: string): Promise { + try { + const body = (await res.json()) as { detail?: string }; + return body.detail || fallback; + } catch { + // A non-JSON error body is still an error; the default sentence stands. + return fallback; + } +} + +const sleep = (ms: number) => new Promise((resolve) => { setTimeout(resolve, ms); }); + +export async function stagePackageInstall( + listingId: string, + onProgress?: (progress: InstallProgress) => void, + fetchImpl: typeof fetch = fetch, +): Promise { + const started = await fetchImpl(`${API_BASE}/marketplace/install/start`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ id: listingId }), }); - if (!res.ok) { - let detail = "Couldn't download it. Try again."; - try { - const body = (await res.json()) as { detail?: string }; - if (body.detail) detail = body.detail; - } catch { - // A non-JSON error body is still an error; the default sentence stands. - } - throw new Error(detail); + if (!started.ok) throw new Error(await detailOf(started, DOWNLOAD_FAILED)); + const { job_id } = (await started.json()) as { job_id: string }; + for (;;) { + const res = await fetchImpl(`${API_BASE}/marketplace/install/${job_id}`); + if (!res.ok) throw new Error(await detailOf(res, DOWNLOAD_FAILED)); + const status = (await res.json()) as JobStatus; + if (status.phase === 'failed') throw new Error(status.error || DOWNLOAD_FAILED); + if (status.phase === 'ready' && status.preflight) return status.preflight; + // Staging is local and quick; the ring reads as full while it runs. + onProgress?.(status.phase === 'staging' && status.total > 0 + ? { received: status.total, total: status.total } + : { received: status.received, total: status.total }); + await sleep(POLL_MS); } - return res.json(); } diff --git a/frontend/src/app/pages/Directory/packages/installProgress.test.ts b/frontend/src/app/pages/Directory/packages/installProgress.test.ts new file mode 100644 index 00000000..789e6499 --- /dev/null +++ b/frontend/src/app/pages/Directory/packages/installProgress.test.ts @@ -0,0 +1,65 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import fs from 'node:fs'; +import path from 'node:path'; +import { ringFor, fractionOf, RING_FLOOR_PERCENT } from './installRing'; +import { stagePackageInstall, POLL_MS } from './installPackage'; + +// The App Store's ring: while a package downloads the pill becomes a circle that fills with the +// bytes the backend has received. The size comes from Content-Length; without it the ring spins. + +test('a fraction fills the ring, an unknown total spins it, and a fresh start shows a sliver', () => { + assert.deepEqual(ringFor(0.5), { variant: 'determinate', value: 50 }); + assert.deepEqual(ringFor(0), { variant: 'determinate', value: RING_FLOOR_PERCENT }); + assert.deepEqual(ringFor(1.7), { variant: 'determinate', value: 100 }); + assert.deepEqual(ringFor(null), { variant: 'indeterminate', value: 0 }); + assert.equal(fractionOf({ received: 250, total: 1000 }), 0.25); + assert.equal(fractionOf({ received: 9, total: 0 }), null); + assert.equal(fractionOf(null), null); +}); + +function fakeFetch(statuses: Array>): { fetch: typeof fetch; calls: string[] } { + const calls: string[] = []; + const fetchImpl = (async (input: RequestInfo | URL) => { + const url = String(input); + calls.push(url); + if (url.endsWith('/install/start')) return new Response(JSON.stringify({ job_id: 'job-1' }), { status: 200 }); + const next = statuses.shift() ?? statuses[statuses.length - 1]; + return new Response(JSON.stringify(next), { status: 200 }); + }) as typeof fetch; + return { fetch: fetchImpl, calls }; +} + +test('the install polls the job, reports bytes as they land, and resolves with the review', async () => { + const preflight = { ok: true, summary: {}, staging_token: 'tok', conflicts: [], warnings: [] }; + const { fetch: f, calls } = fakeFetch([ + { job_id: 'job-1', phase: 'downloading', received: 300, total: 1000 }, + { job_id: 'job-1', phase: 'downloading', received: 1000, total: 1000 }, + { job_id: 'job-1', phase: 'staging', received: 1000, total: 1000 }, + { job_id: 'job-1', phase: 'ready', received: 1000, total: 1000, preflight }, + ]); + const seen: Array<{ received: number; total: number }> = []; + const t0 = Date.now(); + const res = await stagePackageInstall('git-graph', (p) => seen.push(p), f); + assert.equal(res.staging_token, 'tok'); + assert.deepEqual(seen, [{ received: 300, total: 1000 }, { received: 1000, total: 1000 }, { received: 1000, total: 1000 }]); + assert.equal(calls[0].endsWith('/marketplace/install/start'), true); + assert.equal(calls.filter((u) => u.endsWith('/install/job-1')).length, 4); + assert.ok(Date.now() - t0 >= POLL_MS * 3 - 5, 'the poll waits between reads'); +}); + +test('a failed job rejects with the backend\'s reason', async () => { + const { fetch: f } = fakeFetch([{ job_id: 'job-1', phase: 'failed', received: 0, total: 0, error: 'the download returned 404' }]); + await assert.rejects(stagePackageInstall('x', undefined, f), /the download returned 404/); +}); + +test('the pill draws the ring in the installing state and every surface passes progress', () => { + const pill = fs.readFileSync(path.join(process.cwd(), 'src/app/pages/Directory/packages/InstallPill.tsx'), 'utf8'); + assert.match(pill, /state === 'installing'/); + assert.match(pill, /role="progressbar"/); + assert.match(pill, /data-install-ring=\{ring\.variant\}/); + for (const file of ['src/app/pages/Directory/packages/PackageCard.tsx', 'src/app/pages/Directory/packages/detail/PackageDialog.tsx', 'src/app/pages/Directory/packages/detail/PackageBundleDialog.tsx']) { + const src = fs.readFileSync(path.join(process.cwd(), file), 'utf8'); + assert.match(src, /progress=\{/, file); + } +}); diff --git a/frontend/src/app/pages/Directory/packages/installRing.ts b/frontend/src/app/pages/Directory/packages/installRing.ts new file mode 100644 index 00000000..9f42fbf1 --- /dev/null +++ b/frontend/src/app/pages/Directory/packages/installRing.ts @@ -0,0 +1,14 @@ +// The ring on an installing pill: a fraction fills it, an unknown total spins it. A freshly started +// download shows a sliver rather than an empty ring, because an empty ring reads as nothing happening. +export const RING_FLOOR_PERCENT = 4; + +export function ringFor(progress: number | null | undefined): { variant: 'determinate' | 'indeterminate'; value: number } { + if (progress == null || !Number.isFinite(progress)) return { variant: 'indeterminate', value: 0 }; + const pct = Math.round(Math.min(1, Math.max(0, progress)) * 100); + return { variant: 'determinate', value: Math.max(RING_FLOOR_PERCENT, pct) }; +} + +export function fractionOf(progress: { received: number; total: number } | null | undefined): number | null { + if (!progress || progress.total <= 0) return null; + return progress.received / progress.total; +}