"""Muse transport layer — presign + mpack remote communication. The :class:`MuseTransport` Protocol defines the interface between the Muse CLI and a remote host. Use :func:`make_transport` instead of constructing :class:`HttpTransport` directly — it always returns an ``HttpTransport``. Push flow --------- 1. ``POST {url}/push/mpack-presign`` — get presigned PUT URL for the mpack. 2. ``PUT `` — upload mpack to R2/MinIO directly. 3. ``POST {url}/push/unpack-mpack`` — notify server to index the uploaded mpack. Fetch flow ---------- 1. ``POST {url}/fetch`` — receive presigned GET URL for the fetch mpack. 2. ``GET `` — download mpack from R2/MinIO directly. 3. Verify ``sha256(bytes) == mpack_id`` before applying. Authentication -------------- All endpoints accept ``Authorization: MSign handle="" ts= sig=""`` for Ed25519 per-request signing. The signing identity is never written to any log line. Error codes ----------- 401 Unauthorized — invalid or missing signature 404 Not found — repo does not exist on the remote 409 Conflict — push rejected (non-fast-forward without ``--force``) 5xx Server error Security model -------------- - Refuses all HTTP redirects (prevents credential leakage to other hosts). - Rejects non-HTTPS URLs when signing credentials are present. - Caps response bodies at ``MAX_RESPONSE_BYTES`` (64 MiB) to prevent OOM. """ import collections.abc import contextlib import json import logging import os import pathlib import signal import threading import time as _time_mod import urllib.parse from typing import TYPE_CHECKING, NamedTuple, Protocol, TypedDict if TYPE_CHECKING: import ssl from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey class SigningIdentity(NamedTuple): """Ed25519 signing identity for MSign request authentication.""" handle: str # hub handle, e.g. "gabriel" private_key: Ed25519PrivateKey import msgpack from muse.core.mpack import ( BlobPayload, MPack, PushResult, RemoteInfo, WireTag, ) from muse.core.semver import ChangelogEntry, SemVerTag from muse.core.io import safe_unpackb from muse.core.types import ( BranchHeads, Manifest, Metadata, MsgpackDict, ) from muse.core.commits import CommitDict from muse.core.snapshots import SnapshotDict from muse.core.releases import ReleaseDict from muse.core.validation import ( MAX_RESPONSE_BYTES, sanitize_display, ) from muse.core.types import SemVerBump, blob_id logger = logging.getLogger(__name__) import ssl as _ssl_mod import urllib.error import urllib.request _TIMEOUT_SECONDS = 300 # Presigned mpack PUT has its own timeout budget — independent of the global # API timeout — because it moves megabytes over user uplinks, not just JSON. # # Floor / max are intentionally generous: a 35 MB upload at 90 KB/s (seen in # issue #40) takes ~390s, well above the old 300s ceiling. Max prevents an # absurdly large value from masking a truly hung connection. _PUT_TIMEOUT_FLOOR: int = 1800 # 30 min minimum _PUT_TIMEOUT_MAX: int = 7200 # 2 hr hard cap _PUT_MIN_BYTES_PER_SEC: int = 50_000 # 50 KB/s — floor throughput assumption def _mpack_put_timeout(size_bytes: int) -> int: """Return the PUT timeout (seconds) for a mpack of ``size_bytes``. Scaled by size at a 50 KB/s floor, clamped to [_PUT_TIMEOUT_FLOOR, _PUT_TIMEOUT_MAX]. MUSE_PUSH_TIMEOUT_SECONDS env var overrides everything. """ import os as _os env = _os.environ.get("MUSE_PUSH_TIMEOUT_SECONDS", "") if env.strip(): try: return int(env.strip()) except ValueError: pass scaled = size_bytes // _PUT_MIN_BYTES_PER_SEC return max(_PUT_TIMEOUT_FLOOR, min(scaled, _PUT_TIMEOUT_MAX)) def _mkcert_ca() -> "pathlib.Path | None": """Return the mkcert root CA path, or None if mkcert is not installed. Tries ``mkcert -CAROOT`` first (cross-platform), then falls back to the well-known platform paths. """ import subprocess try: r = subprocess.run(["mkcert", "-CAROOT"], capture_output=True, text=True, timeout=3) if r.returncode == 0: ca = pathlib.Path(r.stdout.strip()) / "rootCA.pem" if ca.exists(): return ca except (FileNotFoundError, subprocess.TimeoutExpired): pass # Platform fallbacks candidates = [ pathlib.Path.home() / "Library" / "Application Support" / "mkcert" / "rootCA.pem", # macOS pathlib.Path.home() / ".local" / "share" / "mkcert" / "rootCA.pem", # Linux ] for p in candidates: if p.exists(): return p return None _default_ssl_ctx: "_ssl_mod.SSLContext | None" = None _localhost_ssl_ctx: "_ssl_mod.SSLContext | None" = None def _make_ssl_ctx(url: str) -> "_ssl_mod.SSLContext": """Return an SSLContext appropriate for *url*, cached after first use. - localhost HTTPS → context loaded from the mkcert root CA - everything else → default system CA context Both branches are memoized process-wide: ``create_default_context()`` and (for localhost) shelling out to ``mkcert -CAROOT`` are non-trivial work with no per-call variance for a fixed CA source, so repeating them on every request (a real cost measured at 500 sequential calls) is pure waste. Safe to share — an ``SSLContext`` has no per-request state. """ global _default_ssl_ctx, _localhost_ssl_ctx if url.startswith("https://localhost") or url.startswith("https://127.0.0.1"): if _localhost_ssl_ctx is None: ca = _mkcert_ca() _localhost_ssl_ctx = ( _ssl_mod.create_default_context(cafile=str(ca)) if ca is not None else _ssl_mod.create_default_context() ) return _localhost_ssl_ctx if _default_ssl_ctx is None: _default_ssl_ctx = _ssl_mod.create_default_context() return _default_ssl_ctx class _Request(NamedTuple): """Minimal HTTP request value passed between ``_build_request`` and ``_execute``.""" url: str method: str headers: dict[str, str] data: bytes | None = None # Recursive type alias for msgpack-serializable values. # Covers every type that msgpack can encode/decode natively — no base64, # no `object`, no `Any`. Python 3.12+ `type` statement creates a proper # TypeAlias that mypy treats as a first-class recursive type. type _MsgVal = ( str | int | float | bool | bytes | None | list[_MsgVal] | dict[str, _MsgVal] ) type _MsgDict = dict[str, _MsgVal] # top-level msgpack response dict type _WireVal = _MsgVal # wire-protocol value (same shape) type _WireDict = dict[str, _WireVal] # wire-protocol response dict type _HeadersDict = dict[str, str] # HTTP request/response headers # Maximum number of objects to include in a single push batch. # Must stay strictly below the server's MAX_OBJECTS_PER_PUSH limit (1 000). # Large batches are split into sequential POSTs so each request body stays # well within Cloudflare's body size limits. CHUNK_OBJECTS: int = 500 # Maximum number of commits to include in a single POST /push call. # Large histories are split into sequential chunked pushes so that each # request stays well within Cloudflare's 100 MB upload limit. # Each intermediate chunk uses force=True; the final chunk uses the caller's # force flag. # # 500 commits/chunk means ~2 round-trips for a 1000-commit history vs the # previous 5 at 200/chunk. The server now uses bulk INSERT ON CONFLICT DO # NOTHING so per-chunk processing time is proportional to commit count, not # the number of individual INSERT round-trips. CHUNK_COMMITS: int = 500 # --------------------------------------------------------------------------- # Result TypedDict # --------------------------------------------------------------------------- class FetchMPackResult(TypedDict): """Result from ``fetch_mpack()`` — Phase 2 presigned-URL fetch. ``blobs`` — raw blob dicts from the mpack; passed to ``apply_mpack``. ``blobs_received`` — count of blobs in the mpack. """ repo_id: str domain: str default_branch: str branch_heads: BranchHeads commits: list[CommitDict] snapshots: list[SnapshotDict] blobs: list[dict] blobs_received: int shallow_commits: list[str] class _UnpackMPackResult(TypedDict): """Response from POST /push/unpack-mpack.""" job_id: str head: str branch: str blobs_in_mpack: int commits_in_mpack: int class _PresignResult(TypedDict): """Response from POST /push/mpack-presign.""" upload_url: str mpack_key: str @contextlib.contextmanager def _ignore_sigpipe() -> collections.abc.Generator[None, None, None]: """Temporarily ignore SIGPIPE during socket I/O (main thread only). ``muse/cli/app.py`` sets ``SIGPIPE = SIG_DFL`` at startup so that piping ``muse`` output to ``head``/``grep``/``jq`` exits cleanly. But ``SIG_DFL`` also kills the process when a large HTTP request body is still in-flight and the remote closes the connection early (auth failure, 409 conflict, server crash, …). Without this guard the push command dies with exit code 141 instead of raising ``TransportError``. Signal handlers may only be changed from the main interpreter thread. Worker threads launched by ``ThreadPoolExecutor`` (used for parallel object uploads) are already safe: Python blocks ``SIGPIPE`` in non-main threads, so broken-socket writes raise ``BrokenPipeError`` rather than killing the process. This context manager is therefore a no-op in non-main threads. In the main thread the context manager saves the current SIGPIPE disposition, sets it to ``SIG_IGN`` for the duration of the network call (so that broken-pipe conditions surface as ``BrokenPipeError`` → ``TransportError``), then restores the original disposition on exit. On platforms without ``SIGPIPE`` (Windows) this is a no-op. """ if not hasattr(signal, "SIGPIPE") or not threading.current_thread() is threading.main_thread(): yield return old_handler: signal.Handlers = signal.signal(signal.SIGPIPE, signal.SIG_IGN) try: yield finally: signal.signal(signal.SIGPIPE, old_handler) # --------------------------------------------------------------------------- # Exception # --------------------------------------------------------------------------- class TransportError(Exception): """Raised when the remote returns a non-2xx response or is unreachable. Attributes: status_code: HTTP status code (e.g. ``401``, ``404``, ``409``, ``500``). ``0`` for network-level failures (DNS, connection refused). retry_after: Seconds the caller should wait before retrying, parsed from the server's ``Retry-After`` header. ``None`` when absent. """ def __init__( self, message: str, status_code: int, *, retry_after: "int | None" = None, ) -> None: super().__init__(message) self.status_code = status_code self.retry_after = retry_after # --------------------------------------------------------------------------- # Protocol — the seam between CLI commands and the transport implementation # --------------------------------------------------------------------------- class MuseTransport(Protocol): """Protocol for Muse remote transport implementations. All methods are synchronous — the Muse CLI is synchronous by design. """ def fetch_remote_info(self, url: str, signing: SigningIdentity | None) -> RemoteInfo: """Return repository metadata from ``GET {url}/refs``. Args: url: Remote repository URL. token: Ed25519 signing identity, or ``None`` for public repos. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def fetch_mpack( self, url: str, signing: "SigningIdentity | None", want: "list[str]", have: "list[str]", *, on_object: "collections.abc.Callable[[BlobPayload], None] | None" = None, ttl_seconds: int = 3600, ) -> "FetchMPackResult": """Fetch the delta as a single content-addressed MPack. POSTs to ``{url}/fetch/mpack``. When the server returns ``presign=False`` the mpack bytes are inline; when ``presign=True`` the client GETs the mpack from the presigned URL. Either way ``sha256(mpack_bytes) == mpack_id`` is verified before the mpack is unpacked — one hash covers all contents. Args: url: Canonical repo URL. signing: Ed25519 signing identity; ``None`` for public repos. want: Commit IDs the client wants. have: Commit IDs already present locally (ancestry cut). on_object: Callback invoked for each object in the mpack. ttl_seconds: Presigned URL TTL forwarded to the server. Raises: :class:`TransportError` on non-200 response, integrity failure, or network error. """ ... def push_tags( self, url: str, signing: SigningIdentity | None, tags: list[WireTag], ) -> int: """Push local tags to the remote via ``POST {url}/tags``. Tags are immutable once created on the remote — the server skips any it already holds. Returns the number of tags newly stored. Args: url: Remote repository URL. token: Ed25519 signing identity, or ``None``. tags: Tags to push. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def create_release( self, url: str, signing: SigningIdentity | None, release: ReleaseDict, ) -> str: """Create a release on the remote via ``POST {url}/releases``. Returns the ``release_id`` assigned by the server. Args: url: Remote repository URL. token: Ed25519 signing identity. release: Fully-populated :class:`~muse.core.store.ReleaseDict`. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def list_releases_remote( self, url: str, signing: SigningIdentity | None, channel: str | None = None, include_drafts: bool = False, ) -> list[ReleaseDict]: """Fetch releases from the remote via ``GET {url}/releases``. Args: url: Remote repository URL. token: Ed25519 signing identity, or ``None``. channel: Filter by release channel; ``None`` returns all. include_drafts: Include draft releases when ``True``. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def delete_release_remote( self, url: str, signing: SigningIdentity | None, tag: str, ) -> None: """Retract a release from the remote via ``DELETE {url}/releases/{tag}``. Removes only the named label from the remote registry. The underlying commit and snapshot objects are **not** affected. Args: url: Remote repository URL. token: Ed25519 signing identity (owner credentials required). tag: Semver tag of the release to retract (e.g. ``"v1.2.0"``). Raises: :class:`TransportError` on HTTP 4xx/5xx, network failure, or if the release does not exist on the remote. """ ... def delete_branch_remote( self, url: str, signing: SigningIdentity | None, branch: str, ) -> None: """Delete a branch on the remote via ``DELETE {url}/branches/{branch}``. Equivalent to ``git push origin --delete ``. The branch ref is removed from the server; commits and objects are unaffected. Args: url: Remote repository URL. token: Ed25519 signing identity (owner credentials required). branch: Branch name to delete (e.g. ``"feat/my-thing"``). Raises: :class:`TransportError` on HTTP 4xx/5xx, network failure, or if the branch does not exist on the remote. """ ... def push_version_tags( self, url: str, signing: "SigningIdentity | None", tags: "list[dict]", *, force: bool = False, ) -> "dict": """Push local version tags to the remote via ``POST {url}/version-tags``. Tags are semantic version pointers (e.g. ``v1.0.0``). When *force* is ``False`` the server rejects a push that would overwrite an existing tag pointing to a different commit (HTTP 409). *force* instructs the server to overwrite regardless. Returns ``{"stored": N, "skipped": M}`` where *stored* is the number of newly written tags and *skipped* is the number already present with the same commit target. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def fetch_version_tags( self, url: str, signing: "SigningIdentity | None", ) -> "dict": """Fetch all version tags from the remote via ``GET {url}/version-tags``. Returns ``{"tags": [, ...]}`` — the full list of version tags stored on the remote. Empty list when no tags exist. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def push_reflog( self, url: str, signing: "SigningIdentity | None", branch: str, entries: "list[dict]", ) -> "dict": """Push branch reflog entries via ``POST {url}/reflog?branch=``. Entries are immutable history (``{old_id, new_id, author, timestamp, operation}`` dicts) — there is no *force* flag, since there is nothing to conflict: re-pushing an entry already stored on the remote is a true no-op, matched by content hash server-side. Returns ``{"stored": N, "skipped": M}`` where *stored* is the number of newly written entries and *skipped* is the number already present. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def fetch_reflog( self, url: str, signing: "SigningIdentity | None", branch: str, ) -> "dict": """Fetch branch reflog entries via ``GET {url}/reflog?branch=``. Returns ``{"entries": [, ...]}`` ordered oldest-first — the full reflog the remote has stored for *branch*. Empty list when no entries exist. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def push_symlog( self, url: str, signing: "SigningIdentity | None", entries_by_symbol: "dict[str, list[dict]]", ) -> "dict": """Push symbol-journal entries via ``POST {url}/symlog``. Unlike reflog (per-branch), symlog is per-symbol, so a single request batches every changed symbol's entries: ``{"symbols": {addr: [entry, ...], ...}}``. Entries are immutable history — there is no *force* flag, since there is nothing to conflict: re-pushing an entry already stored on the remote is a true no-op, matched by content hash server-side. Returns ``{"stored": N, "skipped": M}`` totalled across every symbol in the batch. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... def fetch_symlog( self, url: str, signing: "SigningIdentity | None", ) -> "dict": """Fetch every symbol-journal entry via ``GET {url}/symlog``. Returns ``{"symbols": {addr: [, ...], ...}}`` — every symbol the remote has journal entries for, each list oldest-first. Empty dict when no entries exist. Raises: :class:`TransportError` on HTTP 4xx/5xx or network failure. """ ... # --------------------------------------------------------------------------- # HTTP/1.1 implementation (stdlib, zero extra dependencies) # --------------------------------------------------------------------------- def _http_error_message_from_bytes(status_code: int, body: bytes) -> str: """Build a safe error message from a status code and raw response bytes.""" if status_code == 401: return "Authentication failed (HTTP 401). Run 'muse auth register'." raw = body[:300].decode("utf-8", errors="replace") safe = sanitize_display(raw[:200]) if safe: return f"HTTP {status_code}: {safe}" return f"HTTP {status_code}" def _http_error_message(err: "urllib.error.HTTPError") -> str: """Return a safe, credential-free error message for an HTTP error. HTTP 401 — NEVER includes the server body (which may echo the signing credential). Returns a fixed generic string. All other codes — include the body, capped at 200 chars, with all C0/C1 control characters stripped to prevent ANSI/OSC terminal injection. """ if err.code == 401: return "Authentication failed (HTTP 401). Run 'muse auth register'." try: raw_body = err.read(300).decode("utf-8", errors="replace") except Exception: raw_body = "" safe_body = sanitize_display(raw_body[:200]) if safe_body: return f"HTTP {err.code}: {safe_body}" return f"HTTP {err.code}" class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): """urllib redirect handler that raises immediately on any 3xx response.""" def redirect_request( # type: ignore[override] self, req: urllib.request.Request, fp: "Any", code: int, msg: str, headers: "Any", newurl: str, ) -> None: raise urllib.error.HTTPError(newurl, code, f"Redirect to {newurl}", headers, fp) def _open_url(req: "urllib.request.Request") -> "Any": """Thin seam around ``urllib.request.urlopen`` — patchable in tests.""" return urllib.request.urlopen(req) # noqa: S310 def _parse_retry_after(headers: "Any") -> "int | None": """Parse an RFC 7231 Retry-After header value into whole seconds. Accepts the delta-seconds form only (``"30"``, ``"60"``). The HTTP-date form is intentionally treated as ``None`` — the server only emits delta-seconds and brittle date parsing is not worth the risk. Returns ``None`` on a missing, non-integer, or negative value. Never raises. """ if headers is None: return None raw = headers.get("Retry-After") if raw is None: return None try: secs = int(str(raw).strip()) except (TypeError, ValueError): return None return secs if secs >= 0 else None # --------------------------------------------------------------------------- # Retry primitives — bounded poll that honors Retry-After (MWP-5, RC-5) # --------------------------------------------------------------------------- _FETCH_RETRY_BUDGET_DEFAULT_S: float = 120.0 _RETRY_AFTER_CEILING_S: float = 60.0 # never sleep longer than this per attempt _RETRY_AFTER_FLOOR_S: float = 1.0 # never busy-spin if server says 0 _RETRY_DEFAULT_BACKOFF_S: float = 5.0 # fallback when 503 carries no Retry-After def _sleep(seconds: float) -> None: """Thin wrapper over time.sleep — patched in tests to record wait schedule.""" _time_mod.sleep(seconds) def _next_wait(retry_after: "int | None") -> float: """Clamp the server's Retry-After value to a sane per-attempt sleep window. Floors at ``_RETRY_AFTER_FLOOR_S`` (avoids busy-spin on 0 or missing header) and ceilings at ``_RETRY_AFTER_CEILING_S`` (prevents runaway waits if the server ever emits an unusually large value). """ base = float(retry_after) if retry_after is not None else _RETRY_DEFAULT_BACKOFF_S return min(max(base, _RETRY_AFTER_FLOOR_S), _RETRY_AFTER_CEILING_S) def _resolve_retry_budget(explicit: "float | None") -> float: """Return the effective fetch-retry wall-clock budget in seconds. Priority: explicit arg > ``MUSE_FETCH_RETRY_BUDGET_S`` env var > module default (120 s). Negative values are clamped to 0 (== no retry). A junk env value (non-numeric) is silently ignored and the default is used. """ if explicit is not None: return max(0.0, explicit) env = os.environ.get("MUSE_FETCH_RETRY_BUDGET_S") if env: try: return max(0.0, float(env)) except ValueError: pass return _FETCH_RETRY_BUDGET_DEFAULT_S def _urllib_do( method: str, url: str, headers: "dict[str, str]", data: "bytes | None" = None, *, timeout: int = _TIMEOUT_SECONDS, follow_redirects: bool = False, max_bytes: "int | None" = MAX_RESPONSE_BYTES, ) -> bytes: """Execute an HTTP request via urllib and return the response body. Raises :class:`TransportError` on non-2xx or any network/SSL error. Response body is capped at ``max_bytes`` (default: ``MAX_RESPONSE_BYTES``). Pass ``max_bytes=None`` for uncapped streaming reads (e.g. mpack downloads). """ ssl_ctx = _make_ssl_ctx(url) opener_handlers: list[urllib.request.BaseHandler] = [urllib.request.HTTPSHandler(context=ssl_ctx)] if not follow_redirects: opener_handlers.append(_NoRedirectHandler()) opener = urllib.request.build_opener(*opener_handlers) req = urllib.request.Request(url, data=data, headers=headers, method=method) try: with opener.open(req, timeout=timeout) as resp: if max_bytes is None: import io as _io buf = _io.BytesIO() _CHUNK = 1 << 20 # 1 MiB chunks while True: chunk = resp.read(_CHUNK) if not chunk: break buf.write(chunk) body = buf.getvalue() else: body = resp.read(max_bytes + 1) if len(body) > max_bytes: raise TransportError( f"Response body exceeds the {max_bytes // (1024 * 1024)} MiB cap.", 0 ) return body except urllib.error.HTTPError as err: raise TransportError( _http_error_message(err), err.code, retry_after=_parse_retry_after(err.headers), ) from err except urllib.error.URLError as err: raise TransportError(sanitize_display(str(err.reason)), 0) from err except TransportError: raise except Exception as exc: raise TransportError(sanitize_display(str(exc)), 0) from exc class HttpTransport: """HTTPS transport using urllib. One short-lived connection per call. Signing identity values are **never** written to any log line. """ def _build_request( self, method: str, url: str, signing: SigningIdentity | None, body_bytes: bytes | None = None, content_type: str = "application/x-msgpack", extra_headers: dict[str, str] | None = None, ) -> _Request: _parsed = urllib.parse.urlparse(url) _is_loopback = _parsed.hostname in { "localhost", "127.0.0.1", "::1", "host.docker.internal", } if signing and _parsed.scheme != "https" and not _is_loopback: raise TransportError( f"Refusing to send credentials to a non-HTTPS URL: {url!r}. " "Ensure the remote URL uses https://.", 0, ) # TOFU hub certificate fingerprint pinning — only for signed requests. if not _is_loopback and signing is not None: try: from muse.core.hub_trust import check_and_pin base_url = f"{_parsed.scheme}://{_parsed.netloc}" check_and_pin(base_url) except Exception as _hub_exc: # noqa: BLE001 from muse.core.errors import HubFingerprintMismatchError if isinstance(_hub_exc, HubFingerprintMismatchError): raise TransportError(str(_hub_exc), 0) from _hub_exc logger.warning("⚠️ Hub trust check failed: %s", _hub_exc) headers: _HeadersDict = { "Accept": "application/x-msgpack, application/json", } if body_bytes is not None: headers["Content-Type"] = content_type if extra_headers: headers.update(extra_headers) if signing: from muse.core.msign import build_msign_header headers["Authorization"] = build_msign_header(signing, method, url, body_bytes) return _Request(url=url, method=method, headers=headers, data=body_bytes) @staticmethod def _decode(raw: bytes) -> _MsgDict: """Decode a msgpack server response into a plain dict. Raises :class:`TransportError` if the payload is not valid msgpack. """ if not raw: return {} try: # MAX_RESPONSE_BYTES already caps the network read — safe_unpackb # adds per-field limits as a second layer against billion-laughs payloads. # allow_binary=True because pack/mpack responses carry raw blob content. result: _MsgVal = safe_unpackb(raw, context="server response", allow_binary=True) except Exception as exc: raise TransportError(f"Server returned invalid msgpack: {exc}", 0) from exc if not isinstance(result, dict): return {} return result def _execute(self, req: "_Request | urllib.request.Request") -> bytes: """Send *req* and return raw response bytes. Accepts either an internal ``_Request`` or a ``urllib.request.Request``. Both paths use urllib via ``_urllib_do`` / ``_execute_fetch``. Raises: :class:`TransportError` on non-2xx HTTP or any network error. """ if isinstance(req, urllib.request.Request): return self._execute_fetch(req) with _ignore_sigpipe(): return _urllib_do(req.method, req.url, req.headers, req.data) def _execute_fetch(self, req: "urllib.request.Request") -> bytes: """Send *req* via urllib (HTTP/1.1) and return raw response bytes. Uses ``_open_url`` as the network seam — patchable in tests. Raises: :class:`TransportError` on non-2xx HTTP or any network error. """ try: with _open_url(req) as resp: body_bytes = resp.read(MAX_RESPONSE_BYTES + 1) if len(body_bytes) > MAX_RESPONSE_BYTES: raise TransportError( f"Response body exceeds the {MAX_RESPONSE_BYTES // (1024 * 1024)} MiB cap.", 0, ) return body_bytes except urllib.error.HTTPError as err: raise TransportError( _http_error_message(err), err.code, retry_after=_parse_retry_after(err.headers), ) from err except urllib.error.URLError as err: raise TransportError(sanitize_display(str(err.reason)), 0) from err except TransportError: raise except Exception as exc: raise TransportError(sanitize_display(str(exc)), 0) from exc def hub_json( self, method: str, url: str, signing: "SigningIdentity | None", body: "dict[str, object] | None" = None, ) -> "dict[str, object] | list[object]": """Make an authenticated JSON request and return the parsed response. Uses urllib with the mkcert SSL context for localhost URLs — the same path as all other transport calls. Raises :class:`TransportError` on any HTTP or network error. Args: method: HTTP verb (``GET``, ``POST``, ``PATCH``, ``DELETE``). url: Full URL including scheme and path. signing: MSign identity, or ``None`` for unauthenticated calls. body: Optional JSON-serializable dict sent as the request body. Returns: The parsed top-level JSON value — a ``dict`` for the common case of an object response, or a ``list`` for an endpoint whose response body is a top-level array (e.g. ``GET /api/mists/{id}/ forks``) (musehub#213). Callers decide how to handle whichever shape they expect; this function does not coerce one into the other. """ import json as _json_mod body_bytes = _json_mod.dumps(body).encode() if body is not None else None req = self._build_request( method, url, signing, body_bytes=body_bytes, content_type="application/json", extra_headers={"Accept": "application/json"}, ) raw = self._execute(req) try: result = _json_mod.loads(raw.decode("utf-8")) except Exception as exc: raise TransportError(f"Server returned invalid JSON: {exc}", 0) from exc if not isinstance(result, (dict, list)): raise TransportError( f"Server returned unexpected JSON shape: {type(result).__name__}", 0 ) return result def hub_bytes( self, url: str, signing: "SigningIdentity | None", ) -> bytes: """Download raw bytes from *url* with optional MSign auth. Uses urllib with the mkcert SSL context for localhost URLs. Raises :class:`TransportError` on any HTTP or network error. Args: url: Full URL including scheme and path. signing: MSign identity, or ``None`` for unauthenticated calls. """ req = self._build_request( "GET", url, signing, extra_headers={"Accept": "*/*"}, ) return self._execute(req) def fetch_remote_info(self, url: str, signing: SigningIdentity | None) -> RemoteInfo: """Fetch repository metadata from ``GET {url}/refs``.""" endpoint = f"{url.rstrip('/')}/refs" logger.debug("transport: GET %s", endpoint) req = self._build_request("GET", endpoint, signing) raw = self._execute(req) return _parse_remote_info(raw) def push_mpack_put(self, upload_url: str, mpack_bytes: bytes, mpack_key: str = "") -> None: """PUT mpack_bytes directly to a presigned MinIO URL (Step 2). No MuseHub API involvement. No Authorization header. The presigned URL is the credential. Raises :class:`TransportError` on non-2xx. """ import sys as _sys from muse.core.types import blob_id as _blob_id_put pre_put_hash = _blob_id_put(mpack_bytes) print(f"[PUSH step 6] pre-PUT integrity check:", file=_sys.stderr) print(f"[PUSH step 6] len(mpack_bytes) = {len(mpack_bytes)}", file=_sys.stderr) print(f"[PUSH step 6] blob_id(mpack_bytes) = {pre_put_hash}", file=_sys.stderr) if mpack_key: match = pre_put_hash == mpack_key print(f"[PUSH step 6] mpack_key = {mpack_key}", file=_sys.stderr) print(f"[PUSH step 6] match = {match} {'✓' if match else '✗ MISMATCH'}", file=_sys.stderr) body = _urllib_do( "PUT", upload_url, {"Content-Type": "application/x-muse-pack"}, mpack_bytes, follow_redirects=False, timeout=_mpack_put_timeout(len(mpack_bytes)), ) print(f"[PUSH step 6] PUT → HTTP 200 response_len={len(body)}", file=_sys.stderr) def push_mpack_unpack( self, url: str, signing: "SigningIdentity | None", mpack_key: str, *, branch: str = "main", head: str = "", commits_count: int = 0, blobs_count: int = 0, force: bool = False, ) -> _UnpackMPackResult: """POST /push/unpack-mpack — notify server to index the uploaded mpack (Step 3). Returns ``{"job_id": str, "head": str, "branch": str, "blobs_in_mpack": int, "commits_in_mpack": int}``. Raises :class:`TransportError` on non-2xx or missing job_id. """ from muse.core.mpack import build_unpack_payload endpoint = f"{url.rstrip('/')}/push/unpack-mpack" payload = build_unpack_payload( mpack_key, branch=branch, head=head, commits_count=commits_count, blobs_count=blobs_count, force=force, ) body_bytes = msgpack.packb(payload, use_bin_type=True) req = self._build_request("POST", endpoint, signing, body_bytes) raw = _urllib_do("POST", endpoint, dict(req.headers), body_bytes) data = self._decode(raw) return { "job_id": str(data.get("job_id") or ""), "head": str(data.get("head") or head), "branch": str(data.get("branch") or branch), "blobs_in_mpack": int(data.get("blobs_in_mpack") or 0), "commits_in_mpack": int(data.get("commits_in_mpack") or 0), } def push_mpack_presign( self, url: str, signing: "SigningIdentity | None", mpack_bytes: bytes, ttl_seconds: int = 3600, ) -> _PresignResult: """POST /push/mpack-presign — get a presigned PUT URL for the mpack. Returns ``{"upload_url": str, "mpack_key": str}``. Raises :class:`TransportError` on non-2xx or a missing upload_url. """ from muse.core.mpack import build_presign_payload endpoint = f"{url.rstrip('/')}/push/mpack-presign" payload = build_presign_payload(mpack_bytes) body_bytes = msgpack.packb(payload, use_bin_type=True) req = self._build_request("POST", endpoint, signing, body_bytes) raw = _urllib_do("POST", endpoint, dict(req.headers), body_bytes) data = self._decode(raw) upload_url = str(data.get("upload_url") or "") if not upload_url: raise TransportError("push/mpack-presign: server response missing upload_url", 0) return {"upload_url": upload_url, "mpack_key": str(data.get("mpack_key") or payload["mpack_key"])} def _fetch_mpack_once( self, url: str, signing: "SigningIdentity | None", want: "list[str]", have: "list[str]", *, ttl_seconds: int = 3600, ) -> "FetchMPackResult": """Single-attempt fetch — no retry logic. Called by fetch_mpack.""" endpoint = f"{url.rstrip('/')}/fetch/mpack" body_bytes = msgpack.packb( {"want": want, "have": have, "ttl_seconds": ttl_seconds}, use_bin_type=True, ) req = self._build_request("POST", endpoint, signing, body_bytes) import sys as _sys t0 = _time_mod.monotonic() print(f"[transport] POST {endpoint} want={len(want)} have={len(have)}", file=_sys.stderr, flush=True) try: raw_post = _urllib_do("POST", endpoint, dict(req.headers), body_bytes, follow_redirects=True) t_post = _time_mod.monotonic() print( f"[transport] POST done status=200 body={len(raw_post)}B t={int((t_post - t0)*1000)}ms", file=_sys.stderr, flush=True, ) data = msgpack.unpackb(raw_post, raw=False) bundle_id: str = str(data.get("mpack_id") or data.get("bundle_id") or "") mpack_url_raw: str = str(data.get("mpack_url") or "") t_parse = _time_mod.monotonic() print( f"[transport] parsed bundle_id={bundle_id[:20] if bundle_id else 'none'}" f" mpack_url={mpack_url_raw[:100] if mpack_url_raw else 'null'}" f" t={int((t_parse - t0)*1000)}ms", file=_sys.stderr, flush=True, ) if mpack_url_raw: print(f"[transport] GET mpack {mpack_url_raw[:120]}", file=_sys.stderr, flush=True) mpack_bytes_dl = _urllib_do("GET", mpack_url_raw, {}, follow_redirects=True, max_bytes=None) t_dl = _time_mod.monotonic() dl_mb = len(mpack_bytes_dl) / (1024 * 1024) print( f"[transport] GET done status=200 size={dl_mb:.2f}MB" f" t={int((t_dl - t_parse)*1000)}ms", file=_sys.stderr, flush=True, ) actual_id = blob_id(mpack_bytes_dl) t_verify = _time_mod.monotonic() print( f"[transport] sha256 match={actual_id == bundle_id} t={int((t_verify - t_dl)*1000)}ms", file=_sys.stderr, flush=True, ) if actual_id != bundle_id: raise TransportError( f"fetch/mpack: integrity failure mpack_id={bundle_id!r} " f"actual={actual_id!r}", 0, ) if mpack_bytes_dl[:4] == b"MUSE": from muse.core.mpack import parse_wire_mpack mpack = parse_wire_mpack(mpack_bytes_dl) else: mpack = msgpack.unpackb(mpack_bytes_dl, raw=False) t_unpack = _time_mod.monotonic() print( f"[transport] unpackb commits={len(mpack.get('commits') or [])} snaps={len(mpack.get('snapshots') or [])}" f" blobs={len(mpack.get('blobs') or [])} t={int((t_unpack - t_verify)*1000)}ms", file=_sys.stderr, flush=True, ) else: # mpack_url absent or null → all data is inline in the response body # (Phase 3 inline protocol — commits and snapshots in resp.content already parsed). print("[transport] no mpack_url — inline protocol (no S3 download)", file=_sys.stderr, flush=True) mpack = {"commits": [], "snapshots": [], "blobs": [], "branch_heads": {}} t_unpack = _time_mod.monotonic() except TransportError: raise except Exception as exc: raise TransportError(sanitize_display(str(exc)), 0) from exc raw_commits = mpack.get("commits") or [] raw_snaps = mpack.get("snapshots") or [] raw_blobs = mpack.get("blobs") or [] raw_heads = data.get("branch_heads") or mpack.get("branch_heads") or {} t_coerce0 = _time_mod.monotonic() commits: list[CommitDict] = [ _coerce_commit_dict(dict(c)) for c in raw_commits if isinstance(c, dict) ] snapshots: list[SnapshotDict] = [ _coerce_snapshot_dict(dict(s)) for s in raw_snaps if isinstance(s, dict) ] branch_heads: BranchHeads = { str(k): str(v) for k, v in raw_heads.items() if isinstance(k, str) and isinstance(v, str) } blobs: list[dict] = [obj for obj in raw_blobs if isinstance(obj, dict)] t_coerce1 = _time_mod.monotonic() print( f"[transport] coerce commits={len(commits)} snaps={len(snapshots)} blobs={len(blobs)} heads={len(branch_heads)}" f" t={int((t_coerce1 - t_coerce0)*1000)}ms TOTAL={int((t_coerce1 - t0)*1000)}ms", file=_sys.stderr, flush=True, ) return FetchMPackResult( repo_id=str(data.get("repo_id") or ""), domain=str(data.get("domain") or ""), default_branch=str(data.get("default_branch") or "main"), branch_heads=branch_heads, commits=commits, snapshots=snapshots, blobs=blobs, blobs_received=len(blobs), shallow_commits=[], ) def fetch_mpack( self, url: str, signing: "SigningIdentity | None", want: "list[str]", have: "list[str]", *, ttl_seconds: int = 3600, retry_budget_s: "float | None" = None, ) -> "FetchMPackResult": """Fetch the commit delta as one content-addressed MPack, with retry. POSTs ``{url}/fetch/mpack``. On a ``503`` response the server is still building the fetch cache; the ``Retry-After`` header says how long to wait. This method retries within a configurable wall-clock budget rather than failing immediately. Budget is tracked as accumulated sleep time (virtual time) so the check is correct whether ``_sleep`` is real or patched in tests. Args: url: Canonical repo URL. signing: Ed25519 signing identity; ``None`` for public repos. want: Commit IDs the client wants. have: Commit IDs already present locally. ttl_seconds: Presigned URL TTL forwarded to the server. retry_budget_s: Max cumulative wait in seconds. ``None`` uses the module default (120 s) or ``MUSE_FETCH_RETRY_BUDGET_S`` env var. Pass ``0`` to disable retry entirely. Raises: :class:`TransportError` on non-200 (non-retryable) response, integrity failure, network error, or budget exhaustion. """ import sys as _sys budget = _resolve_retry_budget(retry_budget_s) total_waited: float = 0.0 attempt = 0 while True: attempt += 1 try: return self._fetch_mpack_once( url, signing, want, have, ttl_seconds=ttl_seconds ) except TransportError as exc: if exc.status_code != 503: raise # non-retryable — fail immediately, zero added latency wait = _next_wait(exc.retry_after) if total_waited + wait > budget: raise # budget exhausted — surface the server's real 503 print( f"⏳ remote preparing fetch data (server busy, attempt {attempt}); " f"retrying in {int(wait)}s … " f"(waited {int(total_waited)}s / {int(budget)}s budget)", file=_sys.stderr, flush=True, ) _sleep(wait) total_waited += wait def push_tags( self, url: str, signing: SigningIdentity | None, tags: list[WireTag], ) -> int: """Push tags via ``POST {url}/tags``.""" endpoint = f"{url.rstrip('/')}/tags" logger.debug("transport: POST %s (tags=%d)", endpoint, len(tags)) body_bytes: bytes = msgpack.packb({"tags": list(tags)}, use_bin_type=True) req = self._build_request("POST", endpoint, signing, body_bytes) raw = self._execute(req) parsed = self._decode(raw) stored_val = parsed.get("stored") return int(stored_val) if isinstance(stored_val, int) else 0 def create_release( self, url: str, signing: SigningIdentity | None, release: ReleaseDict, ) -> str: """Create a release via ``POST {url}/releases``.""" endpoint = f"{url.rstrip('/')}/releases" logger.debug("transport: POST %s (tag=%s)", endpoint, release.get("tag", "")) # ReleaseDict contains no bytes fields so JSON encoding works directly. body_bytes: bytes = json.dumps(release).encode("utf-8") req = self._build_request("POST", endpoint, signing, body_bytes, content_type="application/json") raw = self._execute(req) parsed = self._decode(raw) release_id_val = parsed.get("release_id") return str(release_id_val) if isinstance(release_id_val, str) else "" def list_releases_remote( self, url: str, signing: SigningIdentity | None, channel: str | None = None, include_drafts: bool = False, ) -> list[ReleaseDict]: """List releases via ``GET {url}/releases``.""" qs_parts: list[str] = [] if channel: qs_parts.append(f"channel={urllib.parse.quote(channel)}") if include_drafts: qs_parts.append("include_drafts=1") endpoint = f"{url.rstrip('/')}/releases" if qs_parts: endpoint = f"{endpoint}?{'&'.join(qs_parts)}" logger.debug("transport: GET %s", endpoint) req = self._build_request("GET", endpoint, signing) raw = self._execute(req) parsed = self._decode(raw) return _parse_releases_list(parsed) def delete_release_remote( self, url: str, signing: SigningIdentity | None, tag: str, ) -> None: """Retract a release via ``DELETE {url}/releases/{tag}``.""" endpoint = f"{url.rstrip('/')}/releases/{urllib.parse.quote(tag, safe='')}" logger.debug("transport: DELETE %s", endpoint) req = self._build_request("DELETE", endpoint, signing) self._execute(req) def delete_branch_remote( self, url: str, signing: SigningIdentity | None, branch: str, ) -> None: """Delete a branch via ``DELETE {url}/branches/{branch}``.""" endpoint = f"{url.rstrip('/')}/branches/{urllib.parse.quote(branch, safe='')}" logger.debug("transport: DELETE %s", endpoint) req = self._build_request("DELETE", endpoint, signing) self._execute(req) def push_version_tags( self, url: str, signing: "SigningIdentity | None", tags: "list[dict]", *, force: bool = False, ) -> "dict": """Push version tags via ``POST {url}/version-tags``.""" endpoint = f"{url.rstrip('/')}/version-tags" if force: endpoint += "?force=true" body: bytes = msgpack.packb({"tags": tags}, use_bin_type=True) logger.debug("transport: POST %s (%d tags)", endpoint, len(tags)) req = self._build_request("POST", endpoint, signing, body) raw = self._execute(req) parsed = self._decode(raw) stored = parsed.get("stored", 0) skipped = parsed.get("skipped", 0) return {"stored": int(stored) if isinstance(stored, int) else 0, "skipped": int(skipped) if isinstance(skipped, int) else 0} def fetch_version_tags( self, url: str, signing: "SigningIdentity | None", ) -> "dict": """Fetch version tags via ``GET {url}/version-tags``.""" endpoint = f"{url.rstrip('/')}/version-tags" logger.debug("transport: GET %s", endpoint) req = self._build_request("GET", endpoint, signing) raw = self._execute(req) parsed = self._decode(raw) tags_raw = parsed.get("tags", []) tags: list[dict] = [] if isinstance(tags_raw, list): for item in tags_raw: if isinstance(item, dict): tags.append({k: v for k, v in item.items()}) return {"tags": tags} def push_reflog( self, url: str, signing: "SigningIdentity | None", branch: str, entries: "list[dict]", ) -> "dict": """Push branch reflog entries via ``POST {url}/reflog?branch=``.""" endpoint = f"{url.rstrip('/')}/reflog?branch={urllib.parse.quote(branch, safe='')}" body: bytes = msgpack.packb({"entries": entries}, use_bin_type=True) logger.debug("transport: POST %s (%d entries)", endpoint, len(entries)) req = self._build_request("POST", endpoint, signing, body) raw = self._execute(req) parsed = self._decode(raw) stored = parsed.get("stored", 0) skipped = parsed.get("skipped", 0) return {"stored": int(stored) if isinstance(stored, int) else 0, "skipped": int(skipped) if isinstance(skipped, int) else 0} def fetch_reflog( self, url: str, signing: "SigningIdentity | None", branch: str, ) -> "dict": """Fetch branch reflog entries via ``GET {url}/reflog?branch=``.""" endpoint = f"{url.rstrip('/')}/reflog?branch={urllib.parse.quote(branch, safe='')}" logger.debug("transport: GET %s", endpoint) req = self._build_request("GET", endpoint, signing) raw = self._execute(req) parsed = self._decode(raw) entries_raw = parsed.get("entries", []) entries: list[dict] = [] if isinstance(entries_raw, list): for item in entries_raw: if isinstance(item, dict): entries.append({k: v for k, v in item.items()}) return {"entries": entries} def push_symlog( self, url: str, signing: "SigningIdentity | None", entries_by_symbol: "dict[str, list[dict]]", ) -> "dict": """Push symbol-journal entries via ``POST {url}/symlog``.""" endpoint = f"{url.rstrip('/')}/symlog" body: bytes = msgpack.packb({"symbols": entries_by_symbol}, use_bin_type=True) total_entries = sum(len(v) for v in entries_by_symbol.values()) logger.debug( "transport: POST %s (%d symbols, %d entries)", endpoint, len(entries_by_symbol), total_entries, ) req = self._build_request("POST", endpoint, signing, body) raw = self._execute(req) parsed = self._decode(raw) stored = parsed.get("stored", 0) skipped = parsed.get("skipped", 0) return {"stored": int(stored) if isinstance(stored, int) else 0, "skipped": int(skipped) if isinstance(skipped, int) else 0} def fetch_symlog( self, url: str, signing: "SigningIdentity | None", ) -> "dict": """Fetch every symbol-journal entry via ``GET {url}/symlog``.""" endpoint = f"{url.rstrip('/')}/symlog" logger.debug("transport: GET %s", endpoint) req = self._build_request("GET", endpoint, signing) raw = self._execute(req) parsed = self._decode(raw) symbols_raw = parsed.get("symbols", {}) symbols: dict[str, list[dict]] = {} if isinstance(symbols_raw, dict): for addr, entries_raw in symbols_raw.items(): if not isinstance(addr, str) or not isinstance(entries_raw, list): continue entries: list[dict] = [] for item in entries_raw: if isinstance(item, dict): entries.append({k: v for k, v in item.items()}) symbols[addr] = entries return {"symbols": symbols} # --------------------------------------------------------------------------- # Response parsers — JSON bytes → typed TypedDicts # --------------------------------------------------------------------------- # json.loads() returns Any (per typeshed), so we use isinstance narrowing # throughout. No explicit Any annotations appear in this file. # --------------------------------------------------------------------------- def _parse_remote_info(raw: bytes) -> RemoteInfo: """Parse ``GET /refs`` response bytes into a :class:`~muse.core.mpack.RemoteInfo`.""" parsed = HttpTransport._decode(raw) repo_id_val = parsed.get("repo_id") domain_val = parsed.get("domain") default_branch_val = parsed.get("default_branch") branch_heads_raw = parsed.get("branch_heads") branch_heads: BranchHeads = {} if isinstance(branch_heads_raw, dict): for k, v in branch_heads_raw.items(): if isinstance(k, str) and isinstance(v, str): branch_heads[k] = v info = RemoteInfo( repo_id=str(repo_id_val) if isinstance(repo_id_val, str) else "", domain=str(domain_val) if isinstance(domain_val, str) else "midi", default_branch=( str(default_branch_val) if isinstance(default_branch_val, str) else "main" ), branch_heads=branch_heads, ) return info def _parse_mpack(raw: bytes) -> MPack: """Parse ``POST /fetch`` response bytes into a :class:`~muse.core.mpack.MPack`.""" parsed = HttpTransport._decode(raw) mpack: MPack = {} # Commits — each item is a raw dict that CommitRecord.from_dict() will validate. commits_raw = parsed.get("commits") if isinstance(commits_raw, list): commits: list[CommitDict] = [] for item in commits_raw: if isinstance(item, dict): commits.append(_coerce_commit_dict(item)) mpack["commits"] = commits # Snapshots snapshots_raw = parsed.get("snapshots") if isinstance(snapshots_raw, list): snapshots: list[SnapshotDict] = [] for item in snapshots_raw: if isinstance(item, dict): snapshots.append(_coerce_snapshot_dict(item)) mpack["snapshots"] = snapshots # Blobs — raw bytes in "content" (mpack wire format) blobs_raw = parsed.get("blobs") if isinstance(blobs_raw, list): blobs: list[BlobPayload] = [] for item in blobs_raw: if not isinstance(item, dict): continue oid = item.get("object_id") if not isinstance(oid, str): continue content_raw = item.get("content") if isinstance(content_raw, (bytes, bytearray)): blobs.append(BlobPayload(object_id=oid, content=bytes(content_raw))) mpack["blobs"] = blobs # Branch heads heads_raw = parsed.get("branch_heads") if isinstance(heads_raw, dict): branch_heads: BranchHeads = {} for k, v in heads_raw.items(): if isinstance(k, str) and isinstance(v, str): branch_heads[k] = v mpack["branch_heads"] = branch_heads return mpack def _parse_push_result(raw: bytes) -> PushResult: """Parse ``POST /push`` response bytes into a :class:`~muse.core.mpack.PushResult`.""" parsed = HttpTransport._decode(raw) ok_val = parsed.get("ok") msg_val = parsed.get("message") heads_raw = parsed.get("branch_heads") branch_heads: BranchHeads = {} if isinstance(heads_raw, dict): for k, v in heads_raw.items(): if isinstance(k, str) and isinstance(v, str): branch_heads[k] = v return PushResult( ok=bool(ok_val) if isinstance(ok_val, bool) else False, message=str(msg_val) if isinstance(msg_val, str) else "", branch_heads=branch_heads, ) def _coerce_sem_ver_bump(raw: _MsgVal) -> SemVerBump: """Safely coerce a raw value to a :class:`~muse.domain.SemVerBump` literal.""" if raw == "major": return "major" if raw == "minor": return "minor" if raw == "patch": return "patch" return "none" def _parse_releases_list(parsed: _MsgDict) -> list[ReleaseDict]: """Extract a list of :class:`~muse.core.store.ReleaseDict` from a parsed response.""" releases_raw = parsed.get("releases") if not isinstance(releases_raw, list): return [] results: list[ReleaseDict] = [] for item in releases_raw: if not isinstance(item, dict): continue try: semver_raw = item.get("semver") if isinstance(semver_raw, dict): sv_major = semver_raw.get("major") sv_minor = semver_raw.get("minor") sv_patch = semver_raw.get("patch") sv_pre = semver_raw.get("pre") sv_build = semver_raw.get("build") semver: SemVerTag = SemVerTag( major=int(sv_major) if isinstance(sv_major, int) else 0, minor=int(sv_minor) if isinstance(sv_minor, int) else 0, patch=int(sv_patch) if isinstance(sv_patch, int) else 0, pre=sv_pre if isinstance(sv_pre, str) else "", build=sv_build if isinstance(sv_build, str) else "", ) else: semver = SemVerTag(major=0, minor=0, patch=0, pre="", build="") changelog_raw = item.get("changelog") or [] changelog: list[ChangelogEntry] = [] if isinstance(changelog_raw, list): for entry in changelog_raw: if not isinstance(entry, dict): continue bc_raw = entry.get("breaking_changes") bc_list: list[str] = [str(b) for b in bc_raw if isinstance(b, str)] if isinstance(bc_raw, list) else [] changelog.append(ChangelogEntry( commit_id=str(entry.get("commit_id", "")), message=str(entry.get("message", "")), sem_ver_bump=_coerce_sem_ver_bump(entry.get("sem_ver_bump")), breaking_changes=bc_list, author=str(entry.get("author", "")), committed_at=str(entry.get("committed_at", "")), agent_id=str(entry.get("agent_id", "")), model_id=str(entry.get("model_id", "")), )) results.append(ReleaseDict( release_id=str(item.get("release_id", "")), repo_id=str(item.get("repo_id", "")), tag=str(item.get("tag", "")), semver=semver, channel=str(item.get("channel", "stable")), commit_id=str(item.get("commit_id", "")), snapshot_id=str(item.get("snapshot_id", "")), title=str(item.get("title", "")), body=str(item.get("body", "")), changelog=changelog, agent_id=str(item.get("agent_id", "")), model_id=str(item.get("model_id", "")), is_draft=bool(item.get("is_draft", False)), gpg_signature=str(item.get("gpg_signature", "")), created_at=str(item.get("created_at", "")), )) except (KeyError, TypeError, ValueError): continue return results # --------------------------------------------------------------------------- # TypedDict coercion helpers — extract known string fields from raw JSON dicts # --------------------------------------------------------------------------- # CommitDict and SnapshotDict are total=False (all fields optional), so we # only extract the string/scalar fields we can safely validate here. # CommitRecord.from_dict() and SnapshotRecord.from_dict() re-validate # required fields when apply_mpack() calls them. # --------------------------------------------------------------------------- def _str(val: _WireVal) -> str: """Return *val* as str, or empty string if not a str.""" return val if isinstance(val, str) else "" def _str_or_none(val: _WireVal) -> str | None: """Return *val* as str, or None if not a str.""" return val if isinstance(val, str) else None def _int_or(val: _WireVal, default: int) -> int: """Return *val* as int, or *default* if not an int.""" return val if isinstance(val, int) else default def _coerce_commit_dict(raw: _WireDict) -> CommitDict: """Extract typed scalar fields from *raw* into a :class:`~muse.core.store.CommitDict`. ``structured_delta`` is passed through when the wire value is a dict, and coerced to ``None`` otherwise (missing, or a malformed non-dict value) -- it is advisory presentation data, not covered by the commit signature, so there is no integrity check to fail here, only a type guard against propagating garbage. """ metadata_raw = raw.get("metadata") metadata: Metadata = {} if isinstance(metadata_raw, dict): for k, v in metadata_raw.items(): if isinstance(k, str) and isinstance(v, str): metadata[k] = v reviewed_by_raw = raw.get("reviewed_by") reviewed_by: list[str] = [] if isinstance(reviewed_by_raw, list): for item in reviewed_by_raw: if isinstance(item, str): reviewed_by.append(item) breaking_changes_raw = raw.get("breaking_changes") breaking_changes: list[str] = [] if isinstance(breaking_changes_raw, list): for item in breaking_changes_raw: if isinstance(item, str): breaking_changes.append(item) sem_ver_raw = raw.get("sem_ver_bump") sem_ver: SemVerBump if sem_ver_raw == "major": sem_ver = "major" elif sem_ver_raw == "minor": sem_ver = "minor" elif sem_ver_raw == "patch": sem_ver = "patch" else: sem_ver = "none" return CommitDict( commit_id=_str(raw.get("commit_id")), repo_id=_str(raw.get("repo_id")), branch=_str(raw.get("branch") or raw.get("created_on_branch")), snapshot_id=_str(raw.get("snapshot_id")), message=_str(raw.get("message")), committed_at=_str(raw.get("committed_at")), parent_commit_id=_str_or_none(raw.get("parent_commit_id")), parent2_commit_id=_str_or_none(raw.get("parent2_commit_id")), author=_str(raw.get("author")), metadata=metadata, structured_delta=( raw.get("structured_delta") if isinstance(raw.get("structured_delta"), dict) else None ), sem_ver_bump=sem_ver, breaking_changes=breaking_changes, agent_id=_str(raw.get("agent_id")), model_id=_str(raw.get("model_id")), toolchain_id=_str(raw.get("toolchain_id")), prompt_hash=_str(raw.get("prompt_hash")), signature=_str(raw.get("signature")), signer_public_key=_str(raw.get("signer_public_key")), signer_key_id=_str(raw.get("signer_key_id")), reviewed_by=reviewed_by, test_runs=_int_or(raw.get("test_runs"), 0), ) def _coerce_snapshot_dict(raw: _WireDict) -> SnapshotDict: """Extract typed fields from *raw* into a :class:`~muse.core.store.SnapshotDict`. Two wire formats are accepted: 1. Delta format (fetch mpack): {snapshot_id, parent_snapshot_id, delta_upsert, delta_remove} — pass through delta fields so _apply_snapshot_deltas can reconstruct the manifest. 2. Full-manifest format: {snapshot_id, manifest} — coerce into SnapshotDict directly. """ manifest_raw = raw.get("manifest") delta_upsert_raw = raw.get("delta_upsert") # Delta format: no manifest, but delta_upsert present. # _apply_snapshot_deltas handles reconstruction; pass through fields as-is. if manifest_raw is None and isinstance(delta_upsert_raw, dict): directories_raw = raw.get("directories") return { # type: ignore[return-value] "snapshot_id": _str(raw.get("snapshot_id")), "parent_snapshot_id": _str_or_none(raw.get("parent_snapshot_id")), "delta_upsert": { str(k): str(v) for k, v in delta_upsert_raw.items() if isinstance(k, str) and isinstance(v, str) }, "delta_remove": [ str(p) for p in (raw.get("delta_remove") or []) if isinstance(p, str) ], "directories": [ str(d) for d in (directories_raw or []) if isinstance(d, str) ] if isinstance(directories_raw, list) else [], "created_at": _str(raw.get("created_at")), "manifest": None, # type: ignore[typeddict-item] "note": "", } # Full-manifest format. manifest: Manifest = {} if isinstance(manifest_raw, dict): for k, v in manifest_raw.items(): if isinstance(k, str) and isinstance(v, str): manifest[k] = v # ``directories`` is hashed into snapshot_id — dropping it produces a # SnapshotRecord whose snapshot_id never matches the recomputed hash, # making every snapshot with non-empty directories permanently unreadable. directories_raw = raw.get("directories") directories: list[str] = [] if isinstance(directories_raw, list): for item in directories_raw: if isinstance(item, str): directories.append(item) return SnapshotDict( snapshot_id=_str(raw.get("snapshot_id")), manifest=manifest, directories=directories, created_at=_str(raw.get("created_at")), note=_str(raw.get("note", "")), ) # --------------------------------------------------------------------------- # Factory # --------------------------------------------------------------------------- def make_transport(url: str) -> "HttpTransport": """Return an :class:`HttpTransport` for *url*. Args: url: Remote repository URL. Returns: An ``HttpTransport`` instance implementing :class:`MuseTransport`. """ return HttpTransport()