# Package source archive only; requires the full hugging-bay package. # Install: python -m pip install 'hugging-bay==0.4.12' # Guide: https://huggingbay.xyz/python-sdk (no standalone CLI). """Client for the Hugging Bay HTTP API. The client is deliberately thin: one method per endpoint, no client-side business logic, no response reshaping beyond returning parsed JSON. The server is the source of truth for verdicts, fits, and identities. """ from __future__ import annotations import math import random from email.utils import parsedate_to_datetime import errno import hashlib import os import re import stat import sys import tempfile import time from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, TypedDict, cast from urllib.parse import parse_qsl, quote, urlparse import httpx from .generated import ( PublicApi, SearchQuery, SearchResponse, GetArtifactResponse, FindRunnableQuery, FindRunnableResponse as WireFindRunnableResponse, DownloadPlanResponse, DeploymentPlanResponse, ) from .resume import ResumeTransfer from ._proxy_env import make_http_client from .finder_params import normalize_device_family, validate_find_model_hardware_params from ._version import __version__ class FindRunnableResponse(WireFindRunnableResponse, total=False): """Canonical verdict passes through unchanged when the server flag is enabled.""" oneAnswer: Dict[str, Any] __all__ = [ "Client", "HuggingBayError", "DownloadError", "CaseCollisionError", "DownloadCleanupWarning", ] _DEFAULT_BASE_URL = "https://huggingbay.xyz" _USER_AGENT = f"hugging-bay-python/{__version__}" _SHA256_HEX = re.compile(r"^[a-f0-9]{64}$") class HuggingBayError(Exception): """Raised when the API returns a >=400 status. Every Hugging Bay error is the same typed envelope on every surface (``{error, code, status, message, requestId, cause?, next}``), so the exception surfaces those fields directly for status-based error handling. Attributes: status: HTTP status code. body: parsed JSON body (dict) when available, else the raw text. error: the server's ``error`` field (machine code) when present. code: the server's ``code`` field (alias of ``error``) when present. message: the server's human-readable ``message`` when present. request_id: the server's ``requestId`` (also on the ``X-Request-Id`` response header) for support correlation. next: the server's ``next`` recovery step ``{action, href}`` when present. """ def __init__(self, status: int, body: Any, retry_after_seconds: Optional[float] = None) -> None: self.retry_after_seconds = retry_after_seconds self.status = status self.body = body self.note = (body.get("note") or body.get("containment")) if isinstance(body, dict) else None self.error: Optional[str] = None self.code: Optional[str] = None self.message: Optional[str] = None self.request_id: Optional[str] = None self.next: Optional[Dict[str, Any]] = None if isinstance(body, dict): if isinstance(body.get("error"), str): self.error = body["error"] if isinstance(body.get("code"), str): self.code = body["code"] if isinstance(body.get("message"), str): self.message = body["message"] if isinstance(body.get("requestId"), str): self.request_id = body["requestId"] if isinstance(body.get("next"), dict): self.next = body["next"] message = self.message or self.error or (body if isinstance(body, str) else "request failed") super().__init__(f"HTTP {status}: {message}") class DownloadError(RuntimeError): """Raised when a hosted download cannot be verified and attested.""" class DownloadCleanupWarning(UserWarning): """Typed outcome describing a leftover temporary download name. A successful ``download_file`` result may contain this warning under ``cleanupWarning`` when the destination is ready but the temporary name could not be removed. On a primary failure it is attached to that exception as ``cleanup_warning`` when the exception permits attributes. """ code = "temporary_file_cleanup_failed" def __init__( self, path: Path, error: OSError, *, destination: Optional[str] = None, published: bool = False, ) -> None: self.path = str(path) self.error = error self.destination = destination self.published = published if destination is None: message = f"temporary download cleanup failed; residual path {self.path!r}: {error}" elif published: message = ( f"download was published to {destination!r}, but temporary cleanup failed; " f"residual path {self.path!r}: {error}" ) else: message = ( f"verified destination {destination!r} is ready, but temporary cleanup failed; " f"residual path {self.path!r}: {error}" ) super().__init__(message) class CaseCollisionError(DownloadError): """Raised when a download would alias another path on a case-insensitive filesystem.""" def __init__(self, path_a: str, path_b: str) -> None: self.path_a = str(path_a) self.path_b = str(path_b) super().__init__( f"case-insensitive path collision between {self.path_a!r} and {self.path_b!r}; " "pass on_case_collision='suffix' to use deterministic " ".caseN names" ) class DownloadFileResultRequired(TypedDict): status: str destination: str sha256: str sizeBytes: int completion: Dict[str, Any] class DownloadFileResult(DownloadFileResultRequired, total=False): pathMapping: Dict[str, str] cleanupWarning: DownloadCleanupWarning class Client: """Synchronous Hugging Bay API client. Args: base_url: API origin. Defaults to the public production host. token: optional bearer token; sent as ``Authorization: Bearer ``. The token is never included in ``repr()`` / ``str()`` and never logged. timeout: per-request timeout in seconds. """ def __init__( self, base_url: str = _DEFAULT_BASE_URL, token: Optional[str] = None, timeout: float = 20, *, transport: Optional[httpx.BaseTransport] = None, ) -> None: self.api = PublicApi(self._get) self.base_url = base_url.rstrip("/") self._token = token self._next_read_at = 0.0 headers = {"Accept": "application/json", "User-Agent": _USER_AGENT} if token: headers["Authorization"] = f"Bearer {token}" # `transport` is an injection point for tests (httpx.MockTransport). self._http = make_http_client( base_url=self.base_url, headers=headers, timeout=timeout, transport=transport, ) # -- lifecycle --------------------------------------------------------- def close(self) -> None: self._http.close() def __enter__(self) -> "Client": return self def __exit__(self, *exc: object) -> None: self.close() def __repr__(self) -> str: # Never leak the token. return f"Client(base_url={self.base_url!r}, token={'set' if self._token else None})" __str__ = __repr__ # -- transport --------------------------------------------------------- def _pace_read(self) -> None: pending = self._next_read_at - time.monotonic() if pending > 0: if pending > 60: raise HuggingBayError(429, {"code": "public_read_backoff_pending"}, pending) time.sleep(pending) self._next_read_at = 0.0 def _request(self, method: str, path: str, *, params: Optional[Dict[str, Any]] = None, json: Optional[Dict[str, Any]] = None, _retried: bool = False) -> Any: clean_params = None if params is not None: clean_params = {k: v for k, v in params.items() if v is not None} read = method.upper() in ("GET", "HEAD") if read: self._pace_read() resp = self._http.request(method, path, params=clean_params, json=json) body = _safe_body(resp) if resp.status_code >= 400: data = body if isinstance(body, dict) else {} hints = [_parse_retry_after(resp.headers.get("Retry-After"))] hints.extend(_parse_retry_after(str(data[key])) for key in ("retry_after_s", "retryAfterSeconds", "retry_after_seconds") if key in data) valid = [value for value in hints if value is not None] delay = max(valid) if valid else None jitter = random.random() * 0.25 if delay is not None and delay > 0 else 0 if read and resp.status_code in (429, 503) and not _retried and delay is not None and delay + jitter <= 60: resp.close() time.sleep(delay + jitter) return self._request(method, path, params=params, json=json, _retried=True) raise HuggingBayError(resp.status_code, body, delay) if read and resp.headers.get("X-HB-Backoff") == "v1": delay = _parse_retry_after(resp.headers.get("Retry-After")) if delay is not None: self._next_read_at = max(self._next_read_at, time.monotonic() + delay) return body def _get(self, path: str, **params: Any) -> Any: return self._request("GET", path, params=params) # -- endpoints --------------------------------------------------------- def search(self, query: str = "", *, limit: int = 20, cursor: Optional[str] = None, sort: str = "best", semantic: bool = False, **filters: Any) -> SearchResponse: """Search public v1 rows; pass nextCursor as cursor until it is None.""" # Preserve the SDK's rich browse rows when the REST no-query default is # compact. Callers can still request summary=1 or detail="summary". if not query.strip() and "summary" not in filters and "detail" not in filters: filters["summary"] = "0" params = dict(q=query, limit=limit, cursor=cursor, sort=sort, semantic="1" if semantic else None, **filters) return self.api.search(cast(SearchQuery, params)) def report(self, id: str) -> Dict[str, Any]: return self._get(f"/api/v1/artifacts/{quote(str(id), safe='')}/report") def artifact_versions(self, id: str) -> Dict[str, Any]: return self._get(f"/api/v1/artifacts/{quote(str(id), safe='')}/versions") def find_model( self, task: str, gpu: Optional[str] = None, ctx: Optional[int] = None, device_family: Optional[str] = None, commercial: bool = False, limit: int = 3, *, snapshot: Optional[str] = None, hardware: Optional[str] = None, custom_memory_gb: Optional[float] = None, ) -> Dict[str, Any]: """GET /api/agents/find-model — catalog shortlist for a supported task/rig. Use ``hardware="8gb-laptop"`` for a laptop. ``device_family`` accepts GPU vendors or ``cpu``; a machine type such as ``laptop`` is not a family. """ normalized_family = normalize_device_family(device_family) validate_find_model_hardware_params( gpu=gpu, hardware=hardware, device_family=normalized_family ) return self._get( "/api/agents/find-model", task=task, gpu=gpu, ctx=ctx, deviceFamily=normalized_family, snapshot=snapshot, hardware=hardware, customMemoryGb=custom_memory_gb, commercial="true" if commercial else None, limit=limit, ) def artifact(self, id: str) -> GetArtifactResponse: """GET /api/v1/artifacts/{quote(str(id), safe='')} — one public artifact.""" return self.api.get_artifact(id) def safety(self, id: str) -> Dict[str, Any]: """GET /api/v1/artifacts/{quote(str(id), safe='')}/safety — per-artifact safe-to-run verdict.""" return self._get(f"/api/v1/artifacts/{quote(str(id), safe='')}/safety") def bundle(self, id: str) -> Dict[str, Any]: """GET /api/v1/artifacts/{quote(str(id), safe='')}/bundle — trust bundle state.""" return self._get(f"/api/v1/artifacts/{quote(str(id), safe='')}/bundle") def passport(self, id: str) -> Dict[str, Any]: """GET /api/v1/artifacts/{quote(str(id), safe='')}/passport — artifact identity passport.""" return self._get(f"/api/v1/artifacts/{quote(str(id), safe='')}/passport") def resolve(self, repo_or_url: str) -> Dict[str, Any]: """GET /api/v1/resolve?repo= — resolve an owner/name or upstream URL. A resolved (``found``/``found_stale``) reference returns the document. A well-formed reference with no catalog match raises ``HuggingBayError`` with status 404; malformed / unsupported input raises status 400 (``error='invalid_query'``). Both error bodies carry ``resolution`` plus a ``fallback.searchUrl`` recovery pointer. """ return self._get("/api/v1/resolve", repo=repo_or_url) def create_reader_token( self, *, display_name: Optional[str] = None, purpose: Optional[str] = None, ) -> Dict[str, Any]: """POST /api/account/reader-token — mint a self-serve reader token. This is the on-ramp for account workflows and needs NO prior authentication and no signup (it is public and rate-limited per client; a 429 carries the unified envelope with ``Retry-After``). The plaintext token is returned once under ``apiKey.token``; Hugging Bay stores only a hash, so save it immediately. Reader tokens persist saves, collections, watchlists, and saved-search alerts but cannot publish artifacts or approve mirrors. Public discovery (search / resolve / download-plan / download) needs no token at all. Example:: issued = Client().create_reader_token(display_name="my app") token = issued["apiKey"]["token"] authed = Client(token=token) """ body: Dict[str, Any] = {} if display_name is not None: body["displayName"] = display_name if purpose is not None: body["purpose"] = purpose return self._request("POST", "/api/account/reader-token", json=body) def me(self) -> Dict[str, Any]: """GET /api/me — the principal (role, scopes) for the configured token. Use it to validate a token issued by :meth:`create_reader_token`. """ return self._get("/api/me") def find_runnable( self, task: str, *, limit: int = 3, commercial: bool = False, **filters: Any, ) -> FindRunnableResponse: """GET /api/agents/find-runnable — the execution-oriented finder. Returns candidates whose winner carries a ``next_call`` (Bay Run Pin or direct job). Pass that ``next_call`` to :meth:`hugging_bay.BayRunClient.run_next_call` to execute it. """ params = dict(task=task, limit=limit, commercial="true" if commercial else None, **filters) return self.api.find_runnable(cast(FindRunnableQuery, params)) def download_plan(self, id: str) -> DownloadPlanResponse: """GET /api/v1/artifacts/{quote(str(id), safe='')}/download-plan — hosted download plan + hashes.""" return self.api.download_plan(id) def deployment_plan(self, id: str) -> DeploymentPlanResponse: """Read hosted files, hashes and any tested run steps for one artifact.""" return self.api.deployment_plan({"artifact": id}) # -- swarm helpers (0.4.8) --------------------------------------------- def snapshot(self) -> Dict[str, Any]: """GET /api/v1/snapshot — the current ``snapshotId`` for pinning reads. Pass it as ``snapshot=`` to :meth:`download_plans` / :meth:`find_runnable_many` (or ``?snapshot=`` on single reads) so every agent in a swarm gets the same bytes for the same question. """ return self._get("/api/v1/snapshot") def key_usage(self) -> Dict[str, Any]: """GET /api/v1/key — your rate-limit window and in-memory usage counters.""" return self._get("/api/v1/key") def resolve_many(self, repos: Sequence[str], *, snapshot: Optional[str] = None) -> Dict[str, Any]: """POST /api/v1/resolve/batch — resolve up to 100 references in one call. Returns ``{"results": [{"input", "artifactId", "state", "hostedFiles", ...}]}`` in input order; rows that could not be checked in time have ``state == "unknown"`` (retry those) instead of a guess. """ items = [str(repo) for repo in repos] if not 1 <= len(items) <= 100: raise ValueError("resolve_many takes 1-100 references; split larger lists") return self._request("POST", "/api/v1/resolve/batch", json={"repos": items, **({"snapshot": snapshot} if snapshot else {})}) def download_plans( self, ids: Sequence[str], *, snapshot: Optional[str] = None, detail: str = "full", tool: Optional[str] = None, wait: float = 0, ) -> Dict[str, Any]: """POST /api/v1/download-plans — file lists for up to 50 artifact ids. The server answers from memory only. Each row's ``state`` is ``hit``, ``stale`` (labelled ``stale: true``), ``pending`` (being computed in the background; retry after ``retryAfterSeconds``), ``error`` or ``deferred``. With ``wait > 0`` the batch is re-sent, honoring ``retryAfterSeconds``, until no row is pending or ``wait`` seconds pass. """ items = [str(item) for item in ids] if not 1 <= len(items) <= 50: raise ValueError("download_plans takes 1-50 artifact ids; split larger lists") body: Dict[str, Any] = {"artifacts": items, "detail": detail} if snapshot: body["snapshot"] = snapshot if tool: body["tool"] = tool return self._batch_until_settled("/api/v1/download-plans", body, wait) def find_runnable_many( self, rows: Sequence[Dict[str, Any]], *, snapshot: Optional[str] = None, wait: float = 0, ) -> Dict[str, Any]: """POST /api/agents/find-runnable/batch — up to 20 finder questions. Each row is ``{"task": ..., "hardware"?, "commercial"?, "ctx"?, "gpu"?, "model"?}``. Rows follow the same ``hit`` / ``stale`` / ``pending`` states as :meth:`download_plans`. """ items: List[Dict[str, Any]] = [dict(row) for row in rows] if not 1 <= len(items) <= 20: raise ValueError("find_runnable_many takes 1-20 rows; split larger lists") body: Dict[str, Any] = {"rows": items} if snapshot: body["snapshot"] = snapshot return self._batch_until_settled("/api/agents/find-runnable/batch", body, wait) def download_lease(self, id: str, paths: Optional[Sequence[str]] = None) -> Dict[str, Any]: """POST /api/v1/artifacts/{id}/download-lease — sign every hosted file at once. Each ``files[]`` row carries ``url`` + ``expiresAt`` (the same short single-file lifetime) or an ``error``. Start transfers before ``expiresAt`` and verify each file's ``sha256``. """ body: Dict[str, Any] = {} if paths is None else {"paths": [str(path) for path in paths]} return self._request("POST", f"/api/v1/artifacts/{quote(str(id), safe='')}/download-lease", json=body) def _batch_until_settled(self, path: str, body: Dict[str, Any], wait: float) -> Dict[str, Any]: deadline = time.monotonic() + max(0.0, float(wait)) while True: document = self._request("POST", path, json=body) pending = int(((document or {}).get("counts") or {}).get("pending") or 0) if not pending or wait <= 0: return document delay = float(document.get("retryAfterSeconds") or 5) if time.monotonic() + delay > deadline: return document time.sleep(delay) # Keep later polls on the same snapshot so the swarm stays consistent. if document.get("snapshotId") and "snapshot" not in body: body = {**body, "snapshot": document["snapshotId"]} def download_file( self, id: str, path: str, destination: str, *, sha256: str, size_bytes: int, on_case_collision: str = "error", resume: bool = False, ) -> DownloadFileResult: """Stream, verify, and attest one hosted file. The initial request is stopped at Hugging Bay's 302 so its one-time completion capability can be captured. The signed GCS request is then made without the API bearer, streamed to a same-directory temporary file, checked against the plan's exact size and SHA-256, and attested before the temporary file is atomically published. A successful return therefore means the completion endpoint accepted (or replayed) the capability; redirects and partial bodies never count. Expired or network- rebound capabilities are renewed once without fetching the bytes again. ``on_case_collision`` is ``"error"`` (the default) or ``"suffix"``. The default refuses model paths or existing destination files that alias after case folding. Suffix mode computes names from the sorted full plan, skips every planned basename, and returns the selected mapping in ``pathMapping``. Existing unrelated destination names can advance a suffix; a same-name file is reused only when its digest still matches. If publication or reuse succeeds but the temporary name cannot be removed, the result includes a ``cleanupWarning`` value of type :class:`DownloadCleanupWarning`; its ``path`` identifies the residual temporary file. Cleanup failures on an otherwise failing download are attached to the primary exception as ``cleanup_warning`` where possible. """ if on_case_collision not in {"error", "suffix"}: raise ValueError("on_case_collision must be 'error' or 'suffix'") expected_sha256 = str(sha256 or "").strip().lower().removeprefix("sha256:") if not _SHA256_HEX.fullmatch(expected_sha256): raise ValueError("sha256 must be a 64-character hexadecimal digest") try: expected_size = int(size_bytes) except (TypeError, ValueError) as error: raise ValueError("size_bytes must be a non-negative integer") from error if expected_size < 0 or expected_size != size_bytes: raise ValueError("size_bytes must be a non-negative integer") destination_path = Path(destination).expanduser() if destination_path.is_dir(): raise DownloadError("destination is a directory; pass a file path") plan = self.download_plan(id) model_paths = _download_plan_paths(plan) normalized_path = _normalize_model_path(path) matching_path = next( (planned for planned in model_paths if _normalize_model_path(planned) == normalized_path), None, ) if matching_path is None: raise DownloadError( f"download path {str(path)!r} is not listed in the current download plan for {id!r}" ) matching_row = next(row for row in plan["files"] if row.get("path") == matching_path) if matching_row.get("sha256") is not None and matching_row["sha256"] != expected_sha256: raise DownloadError("SHA-256 differs from the current download plan") if matching_row.get("sizeBytes") is not None and matching_row["sizeBytes"] != expected_size: raise DownloadError("size differs from the current download plan") colliding_paths = sorted({ planned for planned in model_paths if _normalize_model_path(planned).casefold() == normalized_path.casefold() }) if len(colliding_paths) > 1 and on_case_collision == "error": other = next(planned for planned in colliding_paths if planned != matching_path) raise CaseCollisionError(matching_path, other) encoded_path = "/".join(quote(part, safe="") for part in str(path).split("/")) encoded_id = quote(str(id), safe="") download_path = f"/api/downloads/{encoded_id}/{encoded_path}" def download_capability(): for attempt in range(2): self._pace_read() initial_request = self._http.build_request("GET", download_path) initial = self._http.send(initial_request, stream=True, follow_redirects=False) if initial.status_code not in (429, 503) or attempt == 1: break initial.read() body = _safe_body(initial) data = body if isinstance(body, dict) else {} hints = [_parse_retry_after(initial.headers.get("Retry-After"))] hints.extend(_parse_retry_after(str(data[key])) for key in ("retry_after_s", "retryAfterSeconds", "retry_after_seconds") if key in data) valid = [value for value in hints if value is not None] delay = max(valid) if valid else None if delay is None: break delay += random.random() * 0.25 if delay > 0 else 0 if delay > 60: break initial.close() time.sleep(delay) try: if initial.status_code != 302: error = DownloadError(f"download redirect required, got HTTP {initial.status_code}") if initial.status_code in (429, 503): initial.read() body = _safe_body(initial) data = body if isinstance(body, dict) else {} hints = [_parse_retry_after(initial.headers.get("Retry-After"))] hints.extend(_parse_retry_after(str(data[key])) for key in ("retry_after_s", "retryAfterSeconds", "retry_after_seconds") if key in data) valid = [value for value in hints if value is not None] error.retry_after_seconds = max(valid) if valid else None raise error location = initial.headers.get("location", "") completion_token = initial.headers.get("x-hugging-bay-completion-token", "") finally: initial.close() if not completion_token: raise DownloadError("download completion capability missing from the initial GET redirect; check server completion configuration or telemetry opt-out (a reader token cannot replace it)") if not _trusted_gcs_signed_url(location): raise DownloadError("untrusted GCS download redirect") return location, completion_token target = Path(destination).expanduser() target.parent.mkdir(parents=True, exist_ok=True) destination_target = target planned_target, reserved_case_names, next_case_index = _plan_case_collision_target( destination_target, model_paths, matching_path ) target = _resolve_case_collision_target( destination_target, expected_sha256, on_case_collision, preferred_case_index=planned_target[1], reserved_case_names=reserved_case_names, ) location, completion_token = download_capability() temporary_path: Optional[Path] = None result: Optional[DownloadFileResult] = None published = False resume_state = None try: if resume: resume_state = ResumeTransfer(target, [self.base_url, id, path, expected_sha256, expected_size]) temporary_path = resume_state.path actual_sha256, received = resume_state.transfer( self._http, location, self._http.headers.get("User-Agent", _USER_AGENT), expected_sha256, expected_size, ) else: # Build this request explicitly so the Client's API bearer cannot # cross the origin boundary to GCS. gcs_request = httpx.Request( "GET", location, headers={ "Accept": "application/octet-stream", "User-Agent": self._http.headers.get("User-Agent", _USER_AGENT), }, ) gcs = self._http.send(gcs_request, stream=True, follow_redirects=False) try: if gcs.status_code != 200: raise DownloadError(f"GCS download failed with HTTP {gcs.status_code}") declared_size = gcs.headers.get("content-length") if declared_size is not None and declared_size.isdigit() and int(declared_size) != expected_size: raise DownloadError("GCS content length does not match the download plan") file_descriptor, temporary_name = tempfile.mkstemp( dir=target.parent, prefix=f".{target.name}.hbay-", suffix=".part", ) os.close(file_descriptor) temporary_path = Path(temporary_name) digest = hashlib.sha256() received = 0 with temporary_path.open("wb") as output: for chunk in gcs.iter_bytes(): if not chunk: continue received += len(chunk) if received > expected_size: raise DownloadError("GCS response exceeded the download plan size") digest.update(chunk) output.write(chunk) output.flush() os.fsync(output.fileno()) finally: gcs.close() actual_sha256 = digest.hexdigest() if received != expected_size or actual_sha256 != expected_sha256: raise DownloadError( f"download verification failed: expected {expected_sha256}/{expected_size}, " f"got {actual_sha256}/{received}" ) # Recheck before attestation as another process may have created a # case-folded alias while the bytes were streaming. target = _resolve_case_collision_target( destination_target, expected_sha256, on_case_collision, preferred_case_index=planned_target[1], reserved_case_names=reserved_case_names, ) try: completion = self.download_completion( id, path, sha256=actual_sha256, size_bytes=received, completion_token=completion_token, ) except HuggingBayError as error: # Renew once after a slow transfer or a changed client network. # Re-enter the normal eligibility gate, close the 302, and never # fetch the bytes twice. The POST still checks the exact digest, # size and generation; unrelated failures must not be retried. if error.status != 403 or (error.error or error.code) not in { "completion_token_expired", "completion_token_clientBucket_mismatch", }: raise _, completion_token = download_capability() completion = self.download_completion( id, path, sha256=actual_sha256, size_bytes=received, completion_token=completion_token, ) completion_status = completion.get("status") if isinstance(completion, dict) else None if completion_status not in {"client-attested", "already-recorded"} \ or not isinstance(completion, dict) \ or completion.get("localHashAttested") is not True: raise DownloadError("download completion was not recorded") if temporary_path is None: raise DownloadError("download temporary file missing") # Resolve once more, then claim the name with an atomic hard link. # Unlike os.replace, link creation fails if any writer has claimed # the exact name (or a case alias on a case-insensitive filesystem). target = _resolve_case_collision_target( destination_target, expected_sha256, on_case_collision, preferred_case_index=planned_target[1], reserved_case_names=reserved_case_names, ) while True: try: published = _publish_verified_file_no_clobber( temporary_path, target, expected_sha256 ) break except CaseCollisionError: if on_case_collision != "suffix": raise next_case_index = _next_unreserved_case_index( destination_target, max(next_case_index, planned_target[1] + 1), reserved_case_names, ) target = _resolve_case_collision_target( destination_target, expected_sha256, on_case_collision, preferred_case_index=next_case_index, reserved_case_names=reserved_case_names, ) next_case_index = _next_unreserved_case_index( destination_target, _case_suffix_number(target) + 1, reserved_case_names, ) result = { "status": completion_status, "destination": str(target), "sha256": actual_sha256, "sizeBytes": received, "completion": completion, } if on_case_collision == "suffix": result["pathMapping"] = {str(matching_path): str(target)} return result finally: if resume_state is not None: resume_state.file.close() if temporary_path is not None and not (resume_state and resume_state.keep): cleanup_error = _cleanup_temporary_file(temporary_path) if cleanup_error is not None: warning = DownloadCleanupWarning( temporary_path, cleanup_error, destination=(str(target) if result is not None or published else None), published=published, ) if result is not None: result["cleanupWarning"] = warning else: active_error = sys.exc_info()[1] if active_error is not None: try: setattr(active_error, "cleanup_warning", warning) except (AttributeError, TypeError): pass add_note = getattr(active_error, "add_note", None) if callable(add_note): try: add_note(str(warning)) except Exception: pass if resume_state is not None: resume_state.release() def download_completion( self, id: str, path: str, *, sha256: str, size_bytes: int, completion_token: str, ) -> Dict[str, Any]: """POST /api/download-completions/{quote(str(id), safe='')}/{path} — attest a completed pull. Call ONLY after the file has been downloaded in full and its SHA-256 recomputed locally to equal the download-plan's file.sha256. ``completion_token`` is the single-use capability from the ``X-Hugging-Bay-Completion-Token`` header of the download redirect; without it the pull is not counted. Redirect issuance is a download initiation, not a completed pull, and probes never count. """ encoded_path = "/".join(quote(part, safe="") for part in str(path).split("/")) return self._request( "POST", f"/api/download-completions/{quote(str(id), safe='')}/{encoded_path}", json={ "sha256": sha256, "sizeBytes": int(size_bytes), "completionToken": completion_token, }, ) def doctor(self, log_text: str, runtime: Optional[str] = None) -> Dict[str, Any]: """POST /api/doctor — diagnose a failing run from its log output.""" body: Dict[str, Any] = {"log": log_text} if runtime is not None: body["runner"] = runtime return self._request("POST", "/api/doctor", json=body) def recipes(self) -> Dict[str, Any]: """GET /api/recipes — index of locked task recipes.""" return self._get("/api/recipes") def recipe(self, slug: str) -> Dict[str, Any]: """GET /api/recipes/{slug} — one locked recipe by slug.""" return self._get(f"/api/recipes/{slug}") def _safe_body(resp: httpx.Response) -> Any: try: return resp.json() except ValueError: return resp.text def _download_plan_paths(plan: Any) -> list[str]: """Return the exact paths in a well-formed download plan.""" files = plan.get("files") if isinstance(plan, dict) else None if not isinstance(files, list): raise DownloadError("current download plan is missing its files list; refusing to publish") paths = [] for row in files: if not isinstance(row, dict) or not isinstance(row.get("path"), str) or not row["path"]: raise DownloadError("current download plan contains a file without a valid path") paths.append(row["path"]) return paths def _normalize_model_path(path: str) -> str: """Normalize manifest separators while preserving exact path case.""" return str(path).replace("\\", "/") def _case_suffix_path(path: Path, number: int) -> Path: # Keep the final extension only: model filenames often contain version # dots (e.g. Qwen3.8-27B), which pathlib's multi-suffix heuristic treats as # extensions even though they are part of the stem. stem, extension = os.path.splitext(path.name) return path.with_name(f"{stem}.case{number}{extension}") def _case_suffix_number(path: Path) -> int: match = re.search(r"\.case(\d+)(?=\.[^.]+$|$)", path.name) return int(match.group(1)) if match else 1 def _next_unreserved_case_index(target: Path, number: int, reserved: set[str]) -> int: while _case_suffix_path(target, number).name.casefold() in reserved: number += 1 return number def _plan_case_collision_target( target: Path, model_paths: list[str], selected_path: str, ) -> tuple[tuple[Path, int], set[str], int]: """Allocate suffixes from the complete, sorted plan before any transfer. Generated names skip every planned basename, so a suffix for one row can never occupy another row's requested filename. The selected row uses the caller's destination basename; other rows use their manifest basenames. """ paths = sorted(set(_normalize_model_path(item) for item in model_paths)) selected = _normalize_model_path(selected_path) groups: dict[str, list[str]] = {} for item in paths: groups.setdefault(item.casefold(), []).append(item) planned_names = { item.rsplit("/", 1)[-1].casefold() for item in paths } planned_names.add(target.name.casefold()) used_generated: set[str] = set() mapped_suffixes: dict[str, tuple[str, int]] = {} for item in paths: members = groups[item.casefold()] if len(members) < 2: continue rank = members.index(item) + 1 if rank == 1: continue base = target if item == selected else target.with_name(item.rsplit("/", 1)[-1]) number = rank while True: candidate = _case_suffix_path(base, number) folded_name = candidate.name.casefold() if folded_name not in planned_names and folded_name not in used_generated: break number += 1 mapped_suffixes[item] = (candidate.name, number) used_generated.add(folded_name) if selected in mapped_suffixes: name, number = mapped_suffixes[selected] selected_target = target.with_name(name) next_number = number + 1 else: selected_target = target next_number = 2 reserved_names = planned_names | used_generated selected_number = _case_suffix_number(selected_target) if selected_target != target else 1 return (selected_target, selected_number), reserved_names, next_number def _find_existing_case_alias(target: Path) -> Optional[Path]: """Find an existing file that aliases ``target`` on case-insensitive filesystems.""" try: entries = sorted(target.parent.iterdir(), key=lambda item: item.name) except OSError as error: raise DownloadError(f"could not inspect destination directory {str(target.parent)!r}: {error}") from error for entry in entries: try: is_file = entry.is_file() or entry.is_symlink() except OSError: is_file = False if not is_file: continue names_fold_together = entry.name.casefold() == target.name.casefold() same_file = False try: same_file = os.path.samefile(str(entry), str(target)) except (FileNotFoundError, OSError): pass if names_fold_together or same_file: return entry # Some filesystems or fake filesystem layers may not expose the target in # directory enumeration, so retain a direct-path fallback. if target.is_file() or target.is_symlink(): return target return None def _existing_file_has_digest(path: Path, expected_sha256: str) -> bool: return _stable_file_has_digest(path, expected_sha256) def _file_snapshot(info: os.stat_result) -> tuple[int, int, int, int, int]: return ( info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns, info.st_ctime_ns, ) def _stable_file_has_digest( path: Path, expected_sha256: str, *, expected_identity: Optional[tuple[int, int]] = None, ) -> bool: """Hash one regular file and ensure its directory entry stayed the same.""" digest = hashlib.sha256() try: before = path.lstat() if not stat.S_ISREG(before.st_mode): return False with path.open("rb") as source: opened_before = os.fstat(source.fileno()) for chunk in iter(lambda: source.read(1024 * 1024), b""): digest.update(chunk) opened_after = os.fstat(source.fileno()) after = path.lstat() except FileNotFoundError: return False except OSError as error: raise DownloadError(f"could not verify existing destination {str(path)!r}: {error}") from error snapshot = _file_snapshot(before) return ( stat.S_ISREG(after.st_mode) and snapshot == _file_snapshot(opened_before) == _file_snapshot(opened_after) == _file_snapshot(after) and (expected_identity is None or (before.st_dev, before.st_ino) == expected_identity) and digest.hexdigest() == expected_sha256 ) _TEMP_CLEANUP_MAX_ATTEMPTS = 3 _TEMP_CLEANUP_RETRY_DELAY_SECONDS = 0.05 _TRANSIENT_TEMP_CLEANUP_ERRNOS = { errno.EACCES, errno.EBUSY, errno.EPERM, errno.EAGAIN, } def _is_transient_temp_cleanup_error(error: OSError) -> bool: """Recognize bounded-retry cases for sharing and temporary lock failures.""" if getattr(error, "winerror", None) in {32, 33}: return True return error.errno in _TRANSIENT_TEMP_CLEANUP_ERRNOS def _cleanup_temporary_file(path: Path) -> Optional[OSError]: """Remove our temporary name, retrying transient sharing failures finitely. Returns the final cleanup error rather than raising so callers can preserve a primary download exception or report a successful publication accurately. """ for attempt in range(_TEMP_CLEANUP_MAX_ATTEMPTS): try: path.unlink() return None except FileNotFoundError: return None except OSError as error: if ( attempt + 1 >= _TEMP_CLEANUP_MAX_ATTEMPTS or not _is_transient_temp_cleanup_error(error) ): return error time.sleep(_TEMP_CLEANUP_RETRY_DELAY_SECONDS * (attempt + 1)) return None def _publish_verified_file_no_clobber( temporary_path: Path, target: Path, expected_sha256: str, ) -> bool: """Atomically publish by linking, never replacing an occupied name. Returns ``False`` when an exact-name file already exists with the verified digest. Filesystems without hard-link support fail closed with DownloadError. """ temporary_stat = temporary_path.stat() temporary_identity = (temporary_stat.st_dev, temporary_stat.st_ino) try: os.link(temporary_path, target) except OSError as error: existing = _find_existing_case_alias(target) if existing is not None: if existing.name == target.name and _stable_file_has_digest(existing, expected_sha256): return False raise CaseCollisionError(str(target), str(existing)) from error raise DownloadError( f"filesystem could not atomically claim destination {str(target)!r}; refusing to overwrite" ) from error if not _stable_file_has_digest( target, expected_sha256, expected_identity=temporary_identity, ): raise DownloadError( f"published destination identity or digest changed at {str(target)!r}" ) return True def _resolve_case_collision_target( target: Path, expected_sha256: str, policy: str, *, preferred_case_index: int, reserved_case_names: Optional[set[str]] = None, ) -> Path: """Resolve planned suffixes and existing filesystem aliases deterministically.""" reserved = reserved_case_names or set() if preferred_case_index <= 1: candidate = target else: candidate = _case_suffix_path(target, preferred_case_index) next_index = max(2, preferred_case_index + 1) while True: existing = _find_existing_case_alias(candidate) if existing is None or ( existing.name == candidate.name and _existing_file_has_digest(existing, expected_sha256) ): return candidate if policy == "error": raise CaseCollisionError(str(candidate), str(existing)) while True: candidate = _case_suffix_path(target, next_index) if candidate.name.casefold() not in reserved: break next_index += 1 next_index += 1 def _parse_retry_after(value: Optional[str]) -> Optional[float]: """Parse seconds or HTTP-date without shortening the server delay.""" if not value or not value.strip(): return None clean = value.strip() if re.fullmatch(r"\d+(?:\.\d+)?", clean): seconds = float(clean) return seconds if math.isfinite(seconds) else sys.float_info.max try: at = parsedate_to_datetime(clean) if at.tzinfo is None: return None return max(0.0, at.timestamp() - time.time()) except (TypeError, ValueError, OverflowError): return None def _trusted_gcs_signed_url(value: str) -> bool: """Accept only short-lived, HTTPS GCS V4 signed URLs.""" try: parsed = urlparse(str(value)) host = (parsed.hostname or "").lower() if ( parsed.scheme != "https" or not (host == "storage.googleapis.com" or host.endswith(".storage.googleapis.com")) or parsed.username or parsed.password or parsed.fragment ): return False params = {key.lower(): value for key, value in parse_qsl(parsed.query, keep_blank_values=True)} expires = float(params.get("x-goog-expires", "nan")) return ( all(params.get(key) for key in ( "x-goog-algorithm", "x-goog-credential", "x-goog-date", "x-goog-signature", "generation", )) and expires > 0 and expires <= 300 ) except (TypeError, ValueError): return False