gabriel / muse public
transport.py python
1,568 lines 61.1 KB Cold
Raw
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 92 days ago
1 """Muse transport layer — presign + mpack remote communication.
2
3 The :class:`MuseTransport` Protocol defines the interface between the Muse CLI
4 and a remote host. Use :func:`make_transport` instead of constructing
5 :class:`HttpTransport` directly — it always returns an ``HttpTransport``.
6
7 Push flow
8 ---------
9 1. ``POST {url}/push/mpack-presign`` — get presigned PUT URL for the mpack.
10 2. ``PUT <presigned_url>`` — upload mpack to R2/MinIO directly.
11 3. ``POST {url}/push/unpack-mpack`` — notify server to index the uploaded mpack.
12
13 Fetch flow
14 ----------
15 1. ``POST {url}/fetch`` — receive presigned GET URL for the fetch mpack.
16 2. ``GET <mpack_url>`` — download mpack from R2/MinIO directly.
17 3. Verify ``sha256(bytes) == mpack_id`` before applying.
18
19 Authentication
20 --------------
21 All endpoints accept ``Authorization: MSign handle="<handle>" ts=<unix>
22 sig="<b64url>"`` for Ed25519 per-request signing. The signing identity is
23 never written to any log line.
24
25 Error codes
26 -----------
27 401 Unauthorized — invalid or missing signature
28 404 Not found — repo does not exist on the remote
29 409 Conflict — push rejected (non-fast-forward without ``--force``)
30 5xx Server error
31
32 Security model
33 --------------
34 - Refuses all HTTP redirects (prevents credential leakage to other hosts).
35 - Rejects non-HTTPS URLs when signing credentials are present.
36 - Caps response bodies at ``MAX_RESPONSE_BYTES`` (64 MiB) to prevent OOM.
37 """
38
39 import collections.abc
40 import contextlib
41 import json
42 import logging
43 import os
44 import pathlib
45 import signal
46 import threading
47 import time as _time_mod
48 import urllib.parse
49 from typing import TYPE_CHECKING, NamedTuple, Protocol, TypedDict
50
51 if TYPE_CHECKING:
52 import ssl
53 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
54
55 class SigningIdentity(NamedTuple):
56 """Ed25519 signing identity for MSign request authentication."""
57
58 handle: str # hub handle, e.g. "gabriel"
59 private_key: Ed25519PrivateKey
60
61 import msgpack
62
63 from muse.core.mpack import (
64 BlobPayload,
65 MPack,
66 PushResult,
67 RemoteInfo,
68 WireTag,
69 )
70 from muse.core.semver import ChangelogEntry, SemVerTag
71 from muse.core.io import safe_unpackb
72 from muse.core.types import (
73 BranchHeads,
74 Manifest,
75 Metadata,
76 MsgpackDict,
77 )
78 from muse.core.commits import CommitDict
79 from muse.core.snapshots import SnapshotDict
80 from muse.core.releases import ReleaseDict
81 from muse.core.validation import (
82 MAX_RESPONSE_BYTES,
83 sanitize_display,
84 )
85 from muse.core.types import SemVerBump, blob_id
86
87 logger = logging.getLogger(__name__)
88
89 import ssl as _ssl_mod
90 import urllib.error
91 import urllib.request
92
93 _TIMEOUT_SECONDS = 300
94
95 # Presigned mpack PUT has its own timeout budget — independent of the global
96 # API timeout — because it moves megabytes over user uplinks, not just JSON.
97 #
98 # Floor / max are intentionally generous: a 35 MB upload at 90 KB/s (seen in
99 # issue #40) takes ~390s, well above the old 300s ceiling. Max prevents an
100 # absurdly large value from masking a truly hung connection.
101 _PUT_TIMEOUT_FLOOR: int = 1800 # 30 min minimum
102 _PUT_TIMEOUT_MAX: int = 7200 # 2 hr hard cap
103 _PUT_MIN_BYTES_PER_SEC: int = 50_000 # 50 KB/s — floor throughput assumption
104
105
106 def _mpack_put_timeout(size_bytes: int) -> int:
107 """Return the PUT timeout (seconds) for a mpack of ``size_bytes``.
108
109 Scaled by size at a 50 KB/s floor, clamped to [_PUT_TIMEOUT_FLOOR,
110 _PUT_TIMEOUT_MAX]. MUSE_PUSH_TIMEOUT_SECONDS env var overrides everything.
111 """
112 import os as _os
113 env = _os.environ.get("MUSE_PUSH_TIMEOUT_SECONDS", "")
114 if env.strip():
115 try:
116 return int(env.strip())
117 except ValueError:
118 pass
119 scaled = size_bytes // _PUT_MIN_BYTES_PER_SEC
120 return max(_PUT_TIMEOUT_FLOOR, min(scaled, _PUT_TIMEOUT_MAX))
121
122 def _mkcert_ca() -> "pathlib.Path | None":
123 """Return the mkcert root CA path, or None if mkcert is not installed.
124
125 Tries ``mkcert -CAROOT`` first (cross-platform), then falls back to the
126 well-known platform paths.
127 """
128 import subprocess
129 try:
130 r = subprocess.run(["mkcert", "-CAROOT"], capture_output=True, text=True, timeout=3)
131 if r.returncode == 0:
132 ca = pathlib.Path(r.stdout.strip()) / "rootCA.pem"
133 if ca.exists():
134 return ca
135 except (FileNotFoundError, subprocess.TimeoutExpired):
136 pass
137 # Platform fallbacks
138 candidates = [
139 pathlib.Path.home() / "Library" / "Application Support" / "mkcert" / "rootCA.pem", # macOS
140 pathlib.Path.home() / ".local" / "share" / "mkcert" / "rootCA.pem", # Linux
141 ]
142 for p in candidates:
143 if p.exists():
144 return p
145 return None
146
147 _default_ssl_ctx: "_ssl_mod.SSLContext | None" = None
148 _localhost_ssl_ctx: "_ssl_mod.SSLContext | None" = None
149
150 def _make_ssl_ctx(url: str) -> "_ssl_mod.SSLContext":
151 """Return an SSLContext appropriate for *url*, cached after first use.
152
153 - localhost HTTPS → context loaded from the mkcert root CA
154 - everything else → default system CA context
155
156 Both branches are memoized process-wide: ``create_default_context()``
157 and (for localhost) shelling out to ``mkcert -CAROOT`` are non-trivial
158 work with no per-call variance for a fixed CA source, so repeating them
159 on every request (a real cost measured at 500 sequential calls) is pure
160 waste. Safe to share — an ``SSLContext`` has no per-request state.
161 """
162 global _default_ssl_ctx, _localhost_ssl_ctx
163 if url.startswith("https://localhost") or url.startswith("https://127.0.0.1"):
164 if _localhost_ssl_ctx is None:
165 ca = _mkcert_ca()
166 _localhost_ssl_ctx = (
167 _ssl_mod.create_default_context(cafile=str(ca)) if ca is not None
168 else _ssl_mod.create_default_context()
169 )
170 return _localhost_ssl_ctx
171 if _default_ssl_ctx is None:
172 _default_ssl_ctx = _ssl_mod.create_default_context()
173 return _default_ssl_ctx
174
175 class _Request(NamedTuple):
176 """Minimal HTTP request value passed between ``_build_request`` and ``_execute``."""
177
178 url: str
179 method: str
180 headers: dict[str, str]
181 data: bytes | None = None
182
183 # Recursive type alias for msgpack-serializable values.
184 # Covers every type that msgpack can encode/decode natively — no base64,
185 # no `object`, no `Any`. Python 3.12+ `type` statement creates a proper
186 # TypeAlias that mypy treats as a first-class recursive type.
187 type _MsgVal = (
188 str | int | float | bool | bytes | None
189 | list[_MsgVal]
190 | dict[str, _MsgVal]
191 )
192 type _MsgDict = dict[str, _MsgVal] # top-level msgpack response dict
193 type _WireVal = _MsgVal # wire-protocol value (same shape)
194 type _WireDict = dict[str, _WireVal] # wire-protocol response dict
195 type _HeadersDict = dict[str, str] # HTTP request/response headers
196
197 # Maximum number of objects to include in a single push batch.
198 # Must stay strictly below the server's MAX_OBJECTS_PER_PUSH limit (1 000).
199 # Large batches are split into sequential POSTs so each request body stays
200 # well within Cloudflare's body size limits.
201 CHUNK_OBJECTS: int = 500
202
203 # Maximum number of commits to include in a single POST /push call.
204 # Large histories are split into sequential chunked pushes so that each
205 # request stays well within Cloudflare's 100 MB upload limit.
206 # Each intermediate chunk uses force=True; the final chunk uses the caller's
207 # force flag.
208 #
209 # 500 commits/chunk means ~2 round-trips for a 1000-commit history vs the
210 # previous 5 at 200/chunk. The server now uses bulk INSERT ON CONFLICT DO
211 # NOTHING so per-chunk processing time is proportional to commit count, not
212 # the number of individual INSERT round-trips.
213 CHUNK_COMMITS: int = 500
214
215
216 # ---------------------------------------------------------------------------
217 # Result TypedDict
218 # ---------------------------------------------------------------------------
219
220 class FetchMPackResult(TypedDict):
221 """Result from ``fetch_mpack()`` — Phase 2 presigned-URL fetch.
222
223 ``blobs`` — raw blob dicts from the mpack; passed to ``apply_mpack``.
224 ``blobs_received`` — count of blobs in the mpack.
225 """
226
227 repo_id: str
228 domain: str
229 default_branch: str
230 branch_heads: BranchHeads
231 commits: list[CommitDict]
232 snapshots: list[SnapshotDict]
233 blobs: list[dict]
234 blobs_received: int
235 shallow_commits: list[str]
236
237 class _UnpackMPackResult(TypedDict):
238 """Response from POST /push/unpack-mpack."""
239 job_id: str
240 head: str
241 branch: str
242 blobs_in_mpack: int
243 commits_in_mpack: int
244
245
246 class _PresignResult(TypedDict):
247 """Response from POST /push/mpack-presign."""
248 upload_url: str
249 mpack_key: str
250
251
252 @contextlib.contextmanager
253 def _ignore_sigpipe() -> collections.abc.Generator[None, None, None]:
254 """Temporarily ignore SIGPIPE during socket I/O (main thread only).
255
256 ``muse/cli/app.py`` sets ``SIGPIPE = SIG_DFL`` at startup so that piping
257 ``muse`` output to ``head``/``grep``/``jq`` exits cleanly. But ``SIG_DFL``
258 also kills the process when a large HTTP request body is still in-flight
259 and the remote closes the connection early (auth failure, 409 conflict,
260 server crash, …). Without this guard the push command dies with exit code
261 141 instead of raising ``TransportError``.
262
263 Signal handlers may only be changed from the main interpreter thread.
264 Worker threads launched by ``ThreadPoolExecutor`` (used for parallel object
265 uploads) are already safe: Python blocks ``SIGPIPE`` in non-main threads,
266 so broken-socket writes raise ``BrokenPipeError`` rather than killing the
267 process. This context manager is therefore a no-op in non-main threads.
268
269 In the main thread the context manager saves the current SIGPIPE
270 disposition, sets it to ``SIG_IGN`` for the duration of the network call
271 (so that broken-pipe conditions surface as ``BrokenPipeError`` →
272 ``TransportError``), then restores the original disposition on exit. On
273 platforms without ``SIGPIPE`` (Windows) this is a no-op.
274 """
275 if not hasattr(signal, "SIGPIPE") or not threading.current_thread() is threading.main_thread():
276 yield
277 return
278 old_handler: signal.Handlers = signal.signal(signal.SIGPIPE, signal.SIG_IGN)
279 try:
280 yield
281 finally:
282 signal.signal(signal.SIGPIPE, old_handler)
283
284 # ---------------------------------------------------------------------------
285 # Exception
286 # ---------------------------------------------------------------------------
287
288 class TransportError(Exception):
289 """Raised when the remote returns a non-2xx response or is unreachable.
290
291 Attributes:
292 status_code: HTTP status code (e.g. ``401``, ``404``, ``409``, ``500``).
293 ``0`` for network-level failures (DNS, connection refused).
294 retry_after: Seconds the caller should wait before retrying, parsed from
295 the server's ``Retry-After`` header. ``None`` when absent.
296 """
297
298 def __init__(
299 self,
300 message: str,
301 status_code: int,
302 *,
303 retry_after: "int | None" = None,
304 ) -> None:
305 super().__init__(message)
306 self.status_code = status_code
307 self.retry_after = retry_after
308
309 # ---------------------------------------------------------------------------
310 # Protocol — the seam between CLI commands and the transport implementation
311 # ---------------------------------------------------------------------------
312
313 class MuseTransport(Protocol):
314 """Protocol for Muse remote transport implementations.
315
316 All methods are synchronous — the Muse CLI is synchronous by design.
317 """
318
319 def fetch_remote_info(self, url: str, signing: SigningIdentity | None) -> RemoteInfo:
320 """Return repository metadata from ``GET {url}/refs``.
321
322 Args:
323 url: Remote repository URL.
324 token: Ed25519 signing identity, or ``None`` for public repos.
325
326 Raises:
327 :class:`TransportError` on HTTP 4xx/5xx or network failure.
328 """
329 ...
330
331 def fetch_mpack(
332 self,
333 url: str,
334 signing: "SigningIdentity | None",
335 want: "list[str]",
336 have: "list[str]",
337 *,
338 on_object: "collections.abc.Callable[[BlobPayload], None] | None" = None,
339 ttl_seconds: int = 3600,
340 ) -> "FetchMPackResult":
341 """Fetch the delta as a single content-addressed MPack.
342
343 POSTs to ``{url}/fetch/mpack``. When the server returns
344 ``presign=False`` the mpack bytes are inline; when ``presign=True``
345 the client GETs the mpack from the presigned URL.
346
347 Either way ``sha256(mpack_bytes) == mpack_id`` is verified before
348 the mpack is unpacked — one hash covers all contents.
349
350 Args:
351 url: Canonical repo URL.
352 signing: Ed25519 signing identity; ``None`` for public repos.
353 want: Commit IDs the client wants.
354 have: Commit IDs already present locally (ancestry cut).
355 on_object: Callback invoked for each object in the mpack.
356 ttl_seconds: Presigned URL TTL forwarded to the server.
357
358 Raises:
359 :class:`TransportError` on non-200 response, integrity failure,
360 or network error.
361 """
362 ...
363
364 def push_tags(
365 self,
366 url: str,
367 signing: SigningIdentity | None,
368 tags: list[WireTag],
369 ) -> int:
370 """Push local tags to the remote via ``POST {url}/tags``.
371
372 Tags are immutable once created on the remote — the server skips any
373 it already holds. Returns the number of tags newly stored.
374
375 Args:
376 url: Remote repository URL.
377 token: Ed25519 signing identity, or ``None``.
378 tags: Tags to push.
379
380 Raises:
381 :class:`TransportError` on HTTP 4xx/5xx or network failure.
382 """
383 ...
384
385 def create_release(
386 self,
387 url: str,
388 signing: SigningIdentity | None,
389 release: ReleaseDict,
390 ) -> str:
391 """Create a release on the remote via ``POST {url}/releases``.
392
393 Returns the ``release_id`` assigned by the server.
394
395 Args:
396 url: Remote repository URL.
397 token: Ed25519 signing identity.
398 release: Fully-populated :class:`~muse.core.store.ReleaseDict`.
399
400 Raises:
401 :class:`TransportError` on HTTP 4xx/5xx or network failure.
402 """
403 ...
404
405 def list_releases_remote(
406 self,
407 url: str,
408 signing: SigningIdentity | None,
409 channel: str | None = None,
410 include_drafts: bool = False,
411 ) -> list[ReleaseDict]:
412 """Fetch releases from the remote via ``GET {url}/releases``.
413
414 Args:
415 url: Remote repository URL.
416 token: Ed25519 signing identity, or ``None``.
417 channel: Filter by release channel; ``None`` returns all.
418 include_drafts: Include draft releases when ``True``.
419
420 Raises:
421 :class:`TransportError` on HTTP 4xx/5xx or network failure.
422 """
423 ...
424
425 def delete_release_remote(
426 self,
427 url: str,
428 signing: SigningIdentity | None,
429 tag: str,
430 ) -> None:
431 """Retract a release from the remote via ``DELETE {url}/releases/{tag}``.
432
433 Removes only the named label from the remote registry. The underlying
434 commit and snapshot objects are **not** affected.
435
436 Args:
437 url: Remote repository URL.
438 token: Ed25519 signing identity (owner credentials required).
439 tag: Semver tag of the release to retract (e.g. ``"v1.2.0"``).
440
441 Raises:
442 :class:`TransportError` on HTTP 4xx/5xx, network failure, or if
443 the release does not exist on the remote.
444 """
445 ...
446
447 def delete_branch_remote(
448 self,
449 url: str,
450 signing: SigningIdentity | None,
451 branch: str,
452 ) -> None:
453 """Delete a branch on the remote via ``DELETE {url}/branches/{branch}``.
454
455 Equivalent to ``git push origin --delete <branch>``. The branch ref is
456 removed from the server; commits and objects are unaffected.
457
458 Args:
459 url: Remote repository URL.
460 token: Ed25519 signing identity (owner credentials required).
461 branch: Branch name to delete (e.g. ``"feat/my-thing"``).
462
463 Raises:
464 :class:`TransportError` on HTTP 4xx/5xx, network failure, or if
465 the branch does not exist on the remote.
466 """
467 ...
468
469 def push_version_tags(
470 self,
471 url: str,
472 signing: "SigningIdentity | None",
473 tags: "list[dict]",
474 *,
475 force: bool = False,
476 ) -> "dict":
477 """Push local version tags to the remote via ``POST {url}/version-tags``.
478
479 Tags are semantic version pointers (e.g. ``v1.0.0``). When *force* is
480 ``False`` the server rejects a push that would overwrite an existing tag
481 pointing to a different commit (HTTP 409). *force* instructs the server
482 to overwrite regardless.
483
484 Returns ``{"stored": N, "skipped": M}`` where *stored* is the number of
485 newly written tags and *skipped* is the number already present with the
486 same commit target.
487
488 Raises:
489 :class:`TransportError` on HTTP 4xx/5xx or network failure.
490 """
491 ...
492
493 def fetch_version_tags(
494 self,
495 url: str,
496 signing: "SigningIdentity | None",
497 ) -> "dict":
498 """Fetch all version tags from the remote via ``GET {url}/version-tags``.
499
500 Returns ``{"tags": [<VersionTagDict>, ...]}`` — the full list of version
501 tags stored on the remote. Empty list when no tags exist.
502
503 Raises:
504 :class:`TransportError` on HTTP 4xx/5xx or network failure.
505 """
506 ...
507
508 # ---------------------------------------------------------------------------
509 # HTTP/1.1 implementation (stdlib, zero extra dependencies)
510 # ---------------------------------------------------------------------------
511
512 def _http_error_message_from_bytes(status_code: int, body: bytes) -> str:
513 """Build a safe error message from a status code and raw response bytes."""
514 if status_code == 401:
515 return "Authentication failed (HTTP 401). Run 'muse auth register'."
516 raw = body[:300].decode("utf-8", errors="replace")
517 safe = sanitize_display(raw[:200])
518 if safe:
519 return f"HTTP {status_code}: {safe}"
520 return f"HTTP {status_code}"
521
522 def _http_error_message(err: "urllib.error.HTTPError") -> str:
523 """Return a safe, credential-free error message for an HTTP error.
524
525 HTTP 401 — NEVER includes the server body (which may echo the signing
526 credential). Returns a fixed generic string.
527
528 All other codes — include the body, capped at 200 chars, with all C0/C1
529 control characters stripped to prevent ANSI/OSC terminal injection.
530 """
531 if err.code == 401:
532 return "Authentication failed (HTTP 401). Run 'muse auth register'."
533 try:
534 raw_body = err.read(300).decode("utf-8", errors="replace")
535 except Exception:
536 raw_body = ""
537 safe_body = sanitize_display(raw_body[:200])
538 if safe_body:
539 return f"HTTP {err.code}: {safe_body}"
540 return f"HTTP {err.code}"
541
542 class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):
543 """urllib redirect handler that raises immediately on any 3xx response."""
544 def redirect_request( # type: ignore[override]
545 self,
546 req: urllib.request.Request,
547 fp: "Any",
548 code: int,
549 msg: str,
550 headers: "Any",
551 newurl: str,
552 ) -> None:
553 raise urllib.error.HTTPError(newurl, code, f"Redirect to {newurl}", headers, fp)
554
555
556 def _open_url(req: "urllib.request.Request") -> "Any":
557 """Thin seam around ``urllib.request.urlopen`` — patchable in tests."""
558 return urllib.request.urlopen(req) # noqa: S310
559
560
561 def _parse_retry_after(headers: "Any") -> "int | None":
562 """Parse an RFC 7231 Retry-After header value into whole seconds.
563
564 Accepts the delta-seconds form only (``"30"``, ``"60"``). The HTTP-date
565 form is intentionally treated as ``None`` — the server only emits
566 delta-seconds and brittle date parsing is not worth the risk.
567
568 Returns ``None`` on a missing, non-integer, or negative value. Never
569 raises.
570 """
571 if headers is None:
572 return None
573 raw = headers.get("Retry-After")
574 if raw is None:
575 return None
576 try:
577 secs = int(str(raw).strip())
578 except (TypeError, ValueError):
579 return None
580 return secs if secs >= 0 else None
581
582
583 # ---------------------------------------------------------------------------
584 # Retry primitives — bounded poll that honors Retry-After (MWP-5, RC-5)
585 # ---------------------------------------------------------------------------
586
587 _FETCH_RETRY_BUDGET_DEFAULT_S: float = 120.0
588 _RETRY_AFTER_CEILING_S: float = 60.0 # never sleep longer than this per attempt
589 _RETRY_AFTER_FLOOR_S: float = 1.0 # never busy-spin if server says 0
590 _RETRY_DEFAULT_BACKOFF_S: float = 5.0 # fallback when 503 carries no Retry-After
591
592
593 def _sleep(seconds: float) -> None:
594 """Thin wrapper over time.sleep — patched in tests to record wait schedule."""
595 _time_mod.sleep(seconds)
596
597
598 def _next_wait(retry_after: "int | None") -> float:
599 """Clamp the server's Retry-After value to a sane per-attempt sleep window.
600
601 Floors at ``_RETRY_AFTER_FLOOR_S`` (avoids busy-spin on 0 or missing
602 header) and ceilings at ``_RETRY_AFTER_CEILING_S`` (prevents runaway
603 waits if the server ever emits an unusually large value).
604 """
605 base = float(retry_after) if retry_after is not None else _RETRY_DEFAULT_BACKOFF_S
606 return min(max(base, _RETRY_AFTER_FLOOR_S), _RETRY_AFTER_CEILING_S)
607
608
609 def _resolve_retry_budget(explicit: "float | None") -> float:
610 """Return the effective fetch-retry wall-clock budget in seconds.
611
612 Priority: explicit arg > ``MUSE_FETCH_RETRY_BUDGET_S`` env var > module
613 default (120 s). Negative values are clamped to 0 (== no retry).
614 A junk env value (non-numeric) is silently ignored and the default is used.
615 """
616 if explicit is not None:
617 return max(0.0, explicit)
618 env = os.environ.get("MUSE_FETCH_RETRY_BUDGET_S")
619 if env:
620 try:
621 return max(0.0, float(env))
622 except ValueError:
623 pass
624 return _FETCH_RETRY_BUDGET_DEFAULT_S
625
626
627 def _urllib_do(
628 method: str,
629 url: str,
630 headers: "dict[str, str]",
631 data: "bytes | None" = None,
632 *,
633 timeout: int = _TIMEOUT_SECONDS,
634 follow_redirects: bool = False,
635 max_bytes: "int | None" = MAX_RESPONSE_BYTES,
636 ) -> bytes:
637 """Execute an HTTP request via urllib and return the response body.
638
639 Raises :class:`TransportError` on non-2xx or any network/SSL error.
640 Response body is capped at ``max_bytes`` (default: ``MAX_RESPONSE_BYTES``).
641 Pass ``max_bytes=None`` for uncapped streaming reads (e.g. mpack downloads).
642 """
643 ssl_ctx = _make_ssl_ctx(url)
644 opener_handlers: list[urllib.request.BaseHandler] = [urllib.request.HTTPSHandler(context=ssl_ctx)]
645 if not follow_redirects:
646 opener_handlers.append(_NoRedirectHandler())
647 opener = urllib.request.build_opener(*opener_handlers)
648 req = urllib.request.Request(url, data=data, headers=headers, method=method)
649 try:
650 with opener.open(req, timeout=timeout) as resp:
651 if max_bytes is None:
652 import io as _io
653 buf = _io.BytesIO()
654 _CHUNK = 1 << 20 # 1 MiB chunks
655 while True:
656 chunk = resp.read(_CHUNK)
657 if not chunk:
658 break
659 buf.write(chunk)
660 body = buf.getvalue()
661 else:
662 body = resp.read(max_bytes + 1)
663 if len(body) > max_bytes:
664 raise TransportError(
665 f"Response body exceeds the {max_bytes // (1024 * 1024)} MiB cap.", 0
666 )
667 return body
668 except urllib.error.HTTPError as err:
669 raise TransportError(
670 _http_error_message(err), err.code,
671 retry_after=_parse_retry_after(err.headers),
672 ) from err
673 except urllib.error.URLError as err:
674 raise TransportError(sanitize_display(str(err.reason)), 0) from err
675 except TransportError:
676 raise
677 except Exception as exc:
678 raise TransportError(sanitize_display(str(exc)), 0) from exc
679
680 class HttpTransport:
681 """HTTPS transport using urllib.
682
683 One short-lived connection per call. Signing identity values are
684 **never** written to any log line.
685 """
686
687 def _build_request(
688 self,
689 method: str,
690 url: str,
691 signing: SigningIdentity | None,
692 body_bytes: bytes | None = None,
693 content_type: str = "application/x-msgpack",
694 extra_headers: dict[str, str] | None = None,
695 ) -> _Request:
696 _parsed = urllib.parse.urlparse(url)
697 _is_loopback = _parsed.hostname in {
698 "localhost", "127.0.0.1", "::1", "host.docker.internal",
699 }
700 if signing and _parsed.scheme != "https" and not _is_loopback:
701 raise TransportError(
702 f"Refusing to send credentials to a non-HTTPS URL: {url!r}. "
703 "Ensure the remote URL uses https://.",
704 0,
705 )
706
707 # TOFU hub certificate fingerprint pinning — only for signed requests.
708 if not _is_loopback and signing is not None:
709 try:
710 from muse.core.hub_trust import check_and_pin
711 base_url = f"{_parsed.scheme}://{_parsed.netloc}"
712 check_and_pin(base_url)
713 except Exception as _hub_exc: # noqa: BLE001
714 from muse.core.errors import HubFingerprintMismatchError
715 if isinstance(_hub_exc, HubFingerprintMismatchError):
716 raise TransportError(str(_hub_exc), 0) from _hub_exc
717 logger.warning("⚠️ Hub trust check failed: %s", _hub_exc)
718
719 headers: _HeadersDict = {
720 "Accept": "application/x-msgpack, application/json",
721 }
722 if body_bytes is not None:
723 headers["Content-Type"] = content_type
724 if extra_headers:
725 headers.update(extra_headers)
726 if signing:
727 from muse.core.msign import build_msign_header
728 headers["Authorization"] = build_msign_header(signing, method, url, body_bytes)
729 return _Request(url=url, method=method, headers=headers, data=body_bytes)
730
731 @staticmethod
732 def _decode(raw: bytes) -> _MsgDict:
733 """Decode a msgpack server response into a plain dict.
734
735 Raises :class:`TransportError` if the payload is not valid msgpack.
736 """
737 if not raw:
738 return {}
739 try:
740 # MAX_RESPONSE_BYTES already caps the network read — safe_unpackb
741 # adds per-field limits as a second layer against billion-laughs payloads.
742 # allow_binary=True because pack/mpack responses carry raw blob content.
743 result: _MsgVal = safe_unpackb(raw, context="server response", allow_binary=True)
744 except Exception as exc:
745 raise TransportError(f"Server returned invalid msgpack: {exc}", 0) from exc
746 if not isinstance(result, dict):
747 return {}
748 return result
749
750 def _execute(self, req: "_Request | urllib.request.Request") -> bytes:
751 """Send *req* and return raw response bytes.
752
753 Accepts either an internal ``_Request`` or a ``urllib.request.Request``.
754 Both paths use urllib via ``_urllib_do`` / ``_execute_fetch``.
755
756 Raises:
757 :class:`TransportError` on non-2xx HTTP or any network error.
758 """
759 if isinstance(req, urllib.request.Request):
760 return self._execute_fetch(req)
761 with _ignore_sigpipe():
762 return _urllib_do(req.method, req.url, req.headers, req.data)
763
764 def _execute_fetch(self, req: "urllib.request.Request") -> bytes:
765 """Send *req* via urllib (HTTP/1.1) and return raw response bytes.
766
767 Uses ``_open_url`` as the network seam — patchable in tests.
768
769 Raises:
770 :class:`TransportError` on non-2xx HTTP or any network error.
771 """
772 try:
773 with _open_url(req) as resp:
774 body_bytes = resp.read(MAX_RESPONSE_BYTES + 1)
775 if len(body_bytes) > MAX_RESPONSE_BYTES:
776 raise TransportError(
777 f"Response body exceeds the {MAX_RESPONSE_BYTES // (1024 * 1024)} MiB cap.",
778 0,
779 )
780 return body_bytes
781 except urllib.error.HTTPError as err:
782 raise TransportError(
783 _http_error_message(err), err.code,
784 retry_after=_parse_retry_after(err.headers),
785 ) from err
786 except urllib.error.URLError as err:
787 raise TransportError(sanitize_display(str(err.reason)), 0) from err
788 except TransportError:
789 raise
790 except Exception as exc:
791 raise TransportError(sanitize_display(str(exc)), 0) from exc
792
793 def hub_json(
794 self,
795 method: str,
796 url: str,
797 signing: "SigningIdentity | None",
798 body: "dict[str, object] | None" = None,
799 ) -> "dict[str, object]":
800 """Make an authenticated JSON request and return the parsed response dict.
801
802 Uses urllib with the mkcert SSL context for localhost URLs — the same
803 path as all other transport calls. Raises :class:`TransportError` on
804 any HTTP or network error.
805
806 Args:
807 method: HTTP verb (``GET``, ``POST``, ``PATCH``, ``DELETE``).
808 url: Full URL including scheme and path.
809 signing: MSign identity, or ``None`` for unauthenticated calls.
810 body: Optional JSON-serializable dict sent as the request body.
811 """
812 import json as _json_mod
813 body_bytes = _json_mod.dumps(body).encode() if body is not None else None
814 req = self._build_request(
815 method, url, signing,
816 body_bytes=body_bytes,
817 content_type="application/json",
818 extra_headers={"Accept": "application/json"},
819 )
820 raw = self._execute(req)
821 try:
822 result = _json_mod.loads(raw.decode("utf-8"))
823 except Exception as exc:
824 raise TransportError(f"Server returned invalid JSON: {exc}", 0) from exc
825 if not isinstance(result, dict):
826 return {}
827 return result
828
829 def hub_bytes(
830 self,
831 url: str,
832 signing: "SigningIdentity | None",
833 ) -> bytes:
834 """Download raw bytes from *url* with optional MSign auth.
835
836 Uses urllib with the mkcert SSL context for localhost URLs.
837 Raises :class:`TransportError` on any HTTP or network error.
838
839 Args:
840 url: Full URL including scheme and path.
841 signing: MSign identity, or ``None`` for unauthenticated calls.
842 """
843 req = self._build_request(
844 "GET", url, signing,
845 extra_headers={"Accept": "*/*"},
846 )
847 return self._execute(req)
848
849 def fetch_remote_info(self, url: str, signing: SigningIdentity | None) -> RemoteInfo:
850 """Fetch repository metadata from ``GET {url}/refs``."""
851 endpoint = f"{url.rstrip('/')}/refs"
852 logger.debug("transport: GET %s", endpoint)
853 req = self._build_request("GET", endpoint, signing)
854 raw = self._execute(req)
855 return _parse_remote_info(raw)
856
857 def push_mpack_put(self, upload_url: str, mpack_bytes: bytes, mpack_key: str = "") -> None:
858 """PUT mpack_bytes directly to a presigned MinIO URL (Step 2).
859
860 No MuseHub API involvement. No Authorization header. The presigned
861 URL is the credential. Raises :class:`TransportError` on non-2xx.
862 """
863 import sys as _sys
864 from muse.core.types import blob_id as _blob_id_put
865 pre_put_hash = _blob_id_put(mpack_bytes)
866 print(f"[PUSH step 6] pre-PUT integrity check:", file=_sys.stderr)
867 print(f"[PUSH step 6] len(mpack_bytes) = {len(mpack_bytes)}", file=_sys.stderr)
868 print(f"[PUSH step 6] blob_id(mpack_bytes) = {pre_put_hash}", file=_sys.stderr)
869 if mpack_key:
870 match = pre_put_hash == mpack_key
871 print(f"[PUSH step 6] mpack_key = {mpack_key}", file=_sys.stderr)
872 print(f"[PUSH step 6] match = {match} {'✓' if match else '✗ MISMATCH'}", file=_sys.stderr)
873 body = _urllib_do(
874 "PUT", upload_url, {"Content-Type": "application/x-muse-pack"}, mpack_bytes,
875 follow_redirects=False,
876 timeout=_mpack_put_timeout(len(mpack_bytes)),
877 )
878 print(f"[PUSH step 6] PUT → HTTP 200 response_len={len(body)}", file=_sys.stderr)
879
880 def push_mpack_unpack(
881 self,
882 url: str,
883 signing: "SigningIdentity | None",
884 mpack_key: str,
885 *,
886 branch: str = "main",
887 head: str = "",
888 commits_count: int = 0,
889 blobs_count: int = 0,
890 force: bool = False,
891 ) -> _UnpackMPackResult:
892 """POST /push/unpack-mpack — notify server to index the uploaded mpack (Step 3).
893
894 Returns ``{"job_id": str, "head": str, "branch": str,
895 "blobs_in_mpack": int, "commits_in_mpack": int}``.
896 Raises :class:`TransportError` on non-2xx or missing job_id.
897 """
898 from muse.core.mpack import build_unpack_payload
899 endpoint = f"{url.rstrip('/')}/push/unpack-mpack"
900 payload = build_unpack_payload(
901 mpack_key, branch=branch, head=head,
902 commits_count=commits_count, blobs_count=blobs_count, force=force,
903 )
904 body_bytes = msgpack.packb(payload, use_bin_type=True)
905 req = self._build_request("POST", endpoint, signing, body_bytes)
906 raw = _urllib_do("POST", endpoint, dict(req.headers), body_bytes)
907 data = self._decode(raw)
908 return {
909 "job_id": str(data.get("job_id") or ""),
910 "head": str(data.get("head") or head),
911 "branch": str(data.get("branch") or branch),
912 "blobs_in_mpack": int(data.get("blobs_in_mpack") or 0),
913 "commits_in_mpack": int(data.get("commits_in_mpack") or 0),
914 }
915
916 def push_mpack_presign(
917 self,
918 url: str,
919 signing: "SigningIdentity | None",
920 mpack_bytes: bytes,
921 ttl_seconds: int = 3600,
922 ) -> _PresignResult:
923 """POST /push/mpack-presign — get a presigned PUT URL for the mpack.
924
925 Returns ``{"upload_url": str, "mpack_key": str}``.
926 Raises :class:`TransportError` on non-2xx or a missing upload_url.
927 """
928 from muse.core.mpack import build_presign_payload
929 endpoint = f"{url.rstrip('/')}/push/mpack-presign"
930 payload = build_presign_payload(mpack_bytes)
931 body_bytes = msgpack.packb(payload, use_bin_type=True)
932 req = self._build_request("POST", endpoint, signing, body_bytes)
933 raw = _urllib_do("POST", endpoint, dict(req.headers), body_bytes)
934 data = self._decode(raw)
935 upload_url = str(data.get("upload_url") or "")
936 if not upload_url:
937 raise TransportError("push/mpack-presign: server response missing upload_url", 0)
938 return {"upload_url": upload_url, "mpack_key": str(data.get("mpack_key") or payload["mpack_key"])}
939
940 def _fetch_mpack_once(
941 self,
942 url: str,
943 signing: "SigningIdentity | None",
944 want: "list[str]",
945 have: "list[str]",
946 *,
947 ttl_seconds: int = 3600,
948 ) -> "FetchMPackResult":
949 """Single-attempt fetch — no retry logic. Called by fetch_mpack."""
950 endpoint = f"{url.rstrip('/')}/fetch/mpack"
951 body_bytes = msgpack.packb(
952 {"want": want, "have": have, "ttl_seconds": ttl_seconds},
953 use_bin_type=True,
954 )
955 req = self._build_request("POST", endpoint, signing, body_bytes)
956
957 import sys as _sys
958 t0 = _time_mod.monotonic()
959 print(f"[transport] POST {endpoint} want={len(want)} have={len(have)}", file=_sys.stderr, flush=True)
960 try:
961 raw_post = _urllib_do("POST", endpoint, dict(req.headers), body_bytes, follow_redirects=True)
962 t_post = _time_mod.monotonic()
963 print(
964 f"[transport] POST done status=200 body={len(raw_post)}B t={int((t_post - t0)*1000)}ms",
965 file=_sys.stderr, flush=True,
966 )
967
968 data = msgpack.unpackb(raw_post, raw=False)
969 bundle_id: str = str(data.get("mpack_id") or data.get("bundle_id") or "")
970 mpack_url_raw: str = str(data.get("mpack_url") or "")
971 t_parse = _time_mod.monotonic()
972 print(
973 f"[transport] parsed bundle_id={bundle_id[:20] if bundle_id else 'none'}"
974 f" mpack_url={mpack_url_raw[:100] if mpack_url_raw else 'null'}"
975 f" t={int((t_parse - t0)*1000)}ms",
976 file=_sys.stderr, flush=True,
977 )
978
979 if mpack_url_raw:
980 print(f"[transport] GET mpack {mpack_url_raw[:120]}", file=_sys.stderr, flush=True)
981 mpack_bytes_dl = _urllib_do("GET", mpack_url_raw, {}, follow_redirects=True, max_bytes=None)
982 t_dl = _time_mod.monotonic()
983 dl_mb = len(mpack_bytes_dl) / (1024 * 1024)
984 print(
985 f"[transport] GET done status=200 size={dl_mb:.2f}MB"
986 f" t={int((t_dl - t_parse)*1000)}ms",
987 file=_sys.stderr, flush=True,
988 )
989 actual_id = blob_id(mpack_bytes_dl)
990 t_verify = _time_mod.monotonic()
991 print(
992 f"[transport] sha256 match={actual_id == bundle_id} t={int((t_verify - t_dl)*1000)}ms",
993 file=_sys.stderr, flush=True,
994 )
995 if actual_id != bundle_id:
996 raise TransportError(
997 f"fetch/mpack: integrity failure mpack_id={bundle_id!r} "
998 f"actual={actual_id!r}",
999 0,
1000 )
1001 if mpack_bytes_dl[:4] == b"MUSE":
1002 from muse.core.mpack import parse_wire_mpack
1003 mpack = parse_wire_mpack(mpack_bytes_dl)
1004 else:
1005 mpack = msgpack.unpackb(mpack_bytes_dl, raw=False)
1006 t_unpack = _time_mod.monotonic()
1007 print(
1008 f"[transport] unpackb commits={len(mpack.get('commits') or [])} snaps={len(mpack.get('snapshots') or [])}"
1009 f" blobs={len(mpack.get('blobs') or [])} t={int((t_unpack - t_verify)*1000)}ms",
1010 file=_sys.stderr, flush=True,
1011 )
1012 else:
1013 # mpack_url absent or null → all data is inline in the response body
1014 # (Phase 3 inline protocol — commits and snapshots in resp.content already parsed).
1015 print("[transport] no mpack_url — inline protocol (no S3 download)", file=_sys.stderr, flush=True)
1016 mpack = {"commits": [], "snapshots": [], "blobs": [], "branch_heads": {}}
1017 t_unpack = _time_mod.monotonic()
1018
1019 except TransportError:
1020 raise
1021 except Exception as exc:
1022 raise TransportError(sanitize_display(str(exc)), 0) from exc
1023
1024 raw_commits = mpack.get("commits") or []
1025 raw_snaps = mpack.get("snapshots") or []
1026 raw_blobs = mpack.get("blobs") or []
1027 raw_heads = data.get("branch_heads") or mpack.get("branch_heads") or {}
1028
1029 t_coerce0 = _time_mod.monotonic()
1030 commits: list[CommitDict] = [
1031 _coerce_commit_dict(dict(c)) for c in raw_commits if isinstance(c, dict)
1032 ]
1033 snapshots: list[SnapshotDict] = [
1034 _coerce_snapshot_dict(dict(s)) for s in raw_snaps if isinstance(s, dict)
1035 ]
1036 branch_heads: BranchHeads = {
1037 str(k): str(v) for k, v in raw_heads.items()
1038 if isinstance(k, str) and isinstance(v, str)
1039 }
1040 blobs: list[dict] = [obj for obj in raw_blobs if isinstance(obj, dict)]
1041 t_coerce1 = _time_mod.monotonic()
1042 print(
1043 f"[transport] coerce commits={len(commits)} snaps={len(snapshots)} blobs={len(blobs)} heads={len(branch_heads)}"
1044 f" t={int((t_coerce1 - t_coerce0)*1000)}ms TOTAL={int((t_coerce1 - t0)*1000)}ms",
1045 file=_sys.stderr, flush=True,
1046 )
1047
1048 return FetchMPackResult(
1049 repo_id=str(data.get("repo_id") or ""),
1050 domain=str(data.get("domain") or ""),
1051 default_branch=str(data.get("default_branch") or "main"),
1052 branch_heads=branch_heads,
1053 commits=commits,
1054 snapshots=snapshots,
1055 blobs=blobs,
1056 blobs_received=len(blobs),
1057 shallow_commits=[],
1058 )
1059
1060 def fetch_mpack(
1061 self,
1062 url: str,
1063 signing: "SigningIdentity | None",
1064 want: "list[str]",
1065 have: "list[str]",
1066 *,
1067 ttl_seconds: int = 3600,
1068 retry_budget_s: "float | None" = None,
1069 ) -> "FetchMPackResult":
1070 """Fetch the commit delta as one content-addressed MPack, with retry.
1071
1072 POSTs ``{url}/fetch/mpack``. On a ``503`` response the server is
1073 still building the fetch cache; the ``Retry-After`` header says how
1074 long to wait. This method retries within a configurable wall-clock
1075 budget rather than failing immediately.
1076
1077 Budget is tracked as accumulated sleep time (virtual time) so the
1078 check is correct whether ``_sleep`` is real or patched in tests.
1079
1080 Args:
1081 url: Canonical repo URL.
1082 signing: Ed25519 signing identity; ``None`` for public repos.
1083 want: Commit IDs the client wants.
1084 have: Commit IDs already present locally.
1085 ttl_seconds: Presigned URL TTL forwarded to the server.
1086 retry_budget_s: Max cumulative wait in seconds. ``None`` uses the
1087 module default (120 s) or ``MUSE_FETCH_RETRY_BUDGET_S``
1088 env var. Pass ``0`` to disable retry entirely.
1089
1090 Raises:
1091 :class:`TransportError` on non-200 (non-retryable) response,
1092 integrity failure, network error, or budget exhaustion.
1093 """
1094 import sys as _sys
1095 budget = _resolve_retry_budget(retry_budget_s)
1096 total_waited: float = 0.0
1097 attempt = 0
1098 while True:
1099 attempt += 1
1100 try:
1101 return self._fetch_mpack_once(
1102 url, signing, want, have, ttl_seconds=ttl_seconds
1103 )
1104 except TransportError as exc:
1105 if exc.status_code != 503:
1106 raise # non-retryable — fail immediately, zero added latency
1107 wait = _next_wait(exc.retry_after)
1108 if total_waited + wait > budget:
1109 raise # budget exhausted — surface the server's real 503
1110 print(
1111 f"⏳ remote preparing fetch data (server busy, attempt {attempt}); "
1112 f"retrying in {int(wait)}s … "
1113 f"(waited {int(total_waited)}s / {int(budget)}s budget)",
1114 file=_sys.stderr,
1115 flush=True,
1116 )
1117 _sleep(wait)
1118 total_waited += wait
1119
1120 def push_tags(
1121 self,
1122 url: str,
1123 signing: SigningIdentity | None,
1124 tags: list[WireTag],
1125 ) -> int:
1126 """Push tags via ``POST {url}/tags``."""
1127 endpoint = f"{url.rstrip('/')}/tags"
1128 logger.debug("transport: POST %s (tags=%d)", endpoint, len(tags))
1129 body_bytes: bytes = msgpack.packb({"tags": list(tags)}, use_bin_type=True)
1130 req = self._build_request("POST", endpoint, signing, body_bytes)
1131 raw = self._execute(req)
1132 parsed = self._decode(raw)
1133 stored_val = parsed.get("stored")
1134 return int(stored_val) if isinstance(stored_val, int) else 0
1135
1136 def create_release(
1137 self,
1138 url: str,
1139 signing: SigningIdentity | None,
1140 release: ReleaseDict,
1141 ) -> str:
1142 """Create a release via ``POST {url}/releases``."""
1143 endpoint = f"{url.rstrip('/')}/releases"
1144 logger.debug("transport: POST %s (tag=%s)", endpoint, release.get("tag", ""))
1145 # ReleaseDict contains no bytes fields so JSON encoding works directly.
1146 body_bytes: bytes = json.dumps(release).encode("utf-8")
1147 req = self._build_request("POST", endpoint, signing, body_bytes, content_type="application/json")
1148 raw = self._execute(req)
1149 parsed = self._decode(raw)
1150 release_id_val = parsed.get("release_id")
1151 return str(release_id_val) if isinstance(release_id_val, str) else ""
1152
1153 def list_releases_remote(
1154 self,
1155 url: str,
1156 signing: SigningIdentity | None,
1157 channel: str | None = None,
1158 include_drafts: bool = False,
1159 ) -> list[ReleaseDict]:
1160 """List releases via ``GET {url}/releases``."""
1161 qs_parts: list[str] = []
1162 if channel:
1163 qs_parts.append(f"channel={urllib.parse.quote(channel)}")
1164 if include_drafts:
1165 qs_parts.append("include_drafts=1")
1166 endpoint = f"{url.rstrip('/')}/releases"
1167 if qs_parts:
1168 endpoint = f"{endpoint}?{'&'.join(qs_parts)}"
1169 logger.debug("transport: GET %s", endpoint)
1170 req = self._build_request("GET", endpoint, signing)
1171 raw = self._execute(req)
1172 parsed = self._decode(raw)
1173 return _parse_releases_list(parsed)
1174
1175 def delete_release_remote(
1176 self,
1177 url: str,
1178 signing: SigningIdentity | None,
1179 tag: str,
1180 ) -> None:
1181 """Retract a release via ``DELETE {url}/releases/{tag}``."""
1182 endpoint = f"{url.rstrip('/')}/releases/{urllib.parse.quote(tag, safe='')}"
1183 logger.debug("transport: DELETE %s", endpoint)
1184 req = self._build_request("DELETE", endpoint, signing)
1185 self._execute(req)
1186
1187 def delete_branch_remote(
1188 self,
1189 url: str,
1190 signing: SigningIdentity | None,
1191 branch: str,
1192 ) -> None:
1193 """Delete a branch via ``DELETE {url}/branches/{branch}``."""
1194 endpoint = f"{url.rstrip('/')}/branches/{urllib.parse.quote(branch, safe='')}"
1195 logger.debug("transport: DELETE %s", endpoint)
1196 req = self._build_request("DELETE", endpoint, signing)
1197 self._execute(req)
1198
1199 def push_version_tags(
1200 self,
1201 url: str,
1202 signing: "SigningIdentity | None",
1203 tags: "list[dict]",
1204 *,
1205 force: bool = False,
1206 ) -> "dict":
1207 """Push version tags via ``POST {url}/version-tags``."""
1208 endpoint = f"{url.rstrip('/')}/version-tags"
1209 if force:
1210 endpoint += "?force=true"
1211 body: bytes = msgpack.packb({"tags": tags}, use_bin_type=True)
1212 logger.debug("transport: POST %s (%d tags)", endpoint, len(tags))
1213 req = self._build_request("POST", endpoint, signing, body)
1214 raw = self._execute(req)
1215 parsed = self._decode(raw)
1216 stored = parsed.get("stored", 0)
1217 skipped = parsed.get("skipped", 0)
1218 return {"stored": int(stored) if isinstance(stored, int) else 0,
1219 "skipped": int(skipped) if isinstance(skipped, int) else 0}
1220
1221 def fetch_version_tags(
1222 self,
1223 url: str,
1224 signing: "SigningIdentity | None",
1225 ) -> "dict":
1226 """Fetch version tags via ``GET {url}/version-tags``."""
1227 endpoint = f"{url.rstrip('/')}/version-tags"
1228 logger.debug("transport: GET %s", endpoint)
1229 req = self._build_request("GET", endpoint, signing)
1230 raw = self._execute(req)
1231 parsed = self._decode(raw)
1232 tags_raw = parsed.get("tags", [])
1233 tags: list[dict] = []
1234 if isinstance(tags_raw, list):
1235 for item in tags_raw:
1236 if isinstance(item, dict):
1237 tags.append({k: v for k, v in item.items()})
1238 return {"tags": tags}
1239
1240 # ---------------------------------------------------------------------------
1241 # Response parsers — JSON bytes → typed TypedDicts
1242 # ---------------------------------------------------------------------------
1243 # json.loads() returns Any (per typeshed), so we use isinstance narrowing
1244 # throughout. No explicit Any annotations appear in this file.
1245 # ---------------------------------------------------------------------------
1246
1247 def _parse_remote_info(raw: bytes) -> RemoteInfo:
1248 """Parse ``GET /refs`` response bytes into a :class:`~muse.core.mpack.RemoteInfo`."""
1249 parsed = HttpTransport._decode(raw)
1250 repo_id_val = parsed.get("repo_id")
1251 domain_val = parsed.get("domain")
1252 default_branch_val = parsed.get("default_branch")
1253 branch_heads_raw = parsed.get("branch_heads")
1254 branch_heads: BranchHeads = {}
1255 if isinstance(branch_heads_raw, dict):
1256 for k, v in branch_heads_raw.items():
1257 if isinstance(k, str) and isinstance(v, str):
1258 branch_heads[k] = v
1259 info = RemoteInfo(
1260 repo_id=str(repo_id_val) if isinstance(repo_id_val, str) else "",
1261 domain=str(domain_val) if isinstance(domain_val, str) else "midi",
1262 default_branch=(
1263 str(default_branch_val) if isinstance(default_branch_val, str) else "main"
1264 ),
1265 branch_heads=branch_heads,
1266 )
1267 return info
1268
1269 def _parse_mpack(raw: bytes) -> MPack:
1270 """Parse ``POST /fetch`` response bytes into a :class:`~muse.core.mpack.MPack`."""
1271 parsed = HttpTransport._decode(raw)
1272 mpack: MPack = {}
1273
1274 # Commits — each item is a raw dict that CommitRecord.from_dict() will validate.
1275 commits_raw = parsed.get("commits")
1276 if isinstance(commits_raw, list):
1277 commits: list[CommitDict] = []
1278 for item in commits_raw:
1279 if isinstance(item, dict):
1280 commits.append(_coerce_commit_dict(item))
1281 mpack["commits"] = commits
1282
1283 # Snapshots
1284 snapshots_raw = parsed.get("snapshots")
1285 if isinstance(snapshots_raw, list):
1286 snapshots: list[SnapshotDict] = []
1287 for item in snapshots_raw:
1288 if isinstance(item, dict):
1289 snapshots.append(_coerce_snapshot_dict(item))
1290 mpack["snapshots"] = snapshots
1291
1292 # Blobs — raw bytes in "content" (mpack wire format)
1293 blobs_raw = parsed.get("blobs")
1294 if isinstance(blobs_raw, list):
1295 blobs: list[BlobPayload] = []
1296 for item in blobs_raw:
1297 if not isinstance(item, dict):
1298 continue
1299 oid = item.get("object_id")
1300 if not isinstance(oid, str):
1301 continue
1302 content_raw = item.get("content")
1303 if isinstance(content_raw, (bytes, bytearray)):
1304 blobs.append(BlobPayload(object_id=oid, content=bytes(content_raw)))
1305 mpack["blobs"] = blobs
1306
1307 # Branch heads
1308 heads_raw = parsed.get("branch_heads")
1309 if isinstance(heads_raw, dict):
1310 branch_heads: BranchHeads = {}
1311 for k, v in heads_raw.items():
1312 if isinstance(k, str) and isinstance(v, str):
1313 branch_heads[k] = v
1314 mpack["branch_heads"] = branch_heads
1315
1316 return mpack
1317
1318 def _parse_push_result(raw: bytes) -> PushResult:
1319 """Parse ``POST /push`` response bytes into a :class:`~muse.core.mpack.PushResult`."""
1320 parsed = HttpTransport._decode(raw)
1321 ok_val = parsed.get("ok")
1322 msg_val = parsed.get("message")
1323 heads_raw = parsed.get("branch_heads")
1324 branch_heads: BranchHeads = {}
1325 if isinstance(heads_raw, dict):
1326 for k, v in heads_raw.items():
1327 if isinstance(k, str) and isinstance(v, str):
1328 branch_heads[k] = v
1329 return PushResult(
1330 ok=bool(ok_val) if isinstance(ok_val, bool) else False,
1331 message=str(msg_val) if isinstance(msg_val, str) else "",
1332 branch_heads=branch_heads,
1333 )
1334
1335 def _coerce_sem_ver_bump(raw: _MsgVal) -> SemVerBump:
1336 """Safely coerce a raw value to a :class:`~muse.domain.SemVerBump` literal."""
1337 if raw == "major":
1338 return "major"
1339 if raw == "minor":
1340 return "minor"
1341 if raw == "patch":
1342 return "patch"
1343 return "none"
1344
1345 def _parse_releases_list(parsed: _MsgDict) -> list[ReleaseDict]:
1346 """Extract a list of :class:`~muse.core.store.ReleaseDict` from a parsed response."""
1347 releases_raw = parsed.get("releases")
1348 if not isinstance(releases_raw, list):
1349 return []
1350 results: list[ReleaseDict] = []
1351 for item in releases_raw:
1352 if not isinstance(item, dict):
1353 continue
1354 try:
1355 semver_raw = item.get("semver")
1356 if isinstance(semver_raw, dict):
1357 sv_major = semver_raw.get("major")
1358 sv_minor = semver_raw.get("minor")
1359 sv_patch = semver_raw.get("patch")
1360 sv_pre = semver_raw.get("pre")
1361 sv_build = semver_raw.get("build")
1362 semver: SemVerTag = SemVerTag(
1363 major=int(sv_major) if isinstance(sv_major, int) else 0,
1364 minor=int(sv_minor) if isinstance(sv_minor, int) else 0,
1365 patch=int(sv_patch) if isinstance(sv_patch, int) else 0,
1366 pre=sv_pre if isinstance(sv_pre, str) else "",
1367 build=sv_build if isinstance(sv_build, str) else "",
1368 )
1369 else:
1370 semver = SemVerTag(major=0, minor=0, patch=0, pre="", build="")
1371 changelog_raw = item.get("changelog") or []
1372 changelog: list[ChangelogEntry] = []
1373 if isinstance(changelog_raw, list):
1374 for entry in changelog_raw:
1375 if not isinstance(entry, dict):
1376 continue
1377 bc_raw = entry.get("breaking_changes")
1378 bc_list: list[str] = [str(b) for b in bc_raw if isinstance(b, str)] if isinstance(bc_raw, list) else []
1379 changelog.append(ChangelogEntry(
1380 commit_id=str(entry.get("commit_id", "")),
1381 message=str(entry.get("message", "")),
1382 sem_ver_bump=_coerce_sem_ver_bump(entry.get("sem_ver_bump")),
1383 breaking_changes=bc_list,
1384 author=str(entry.get("author", "")),
1385 committed_at=str(entry.get("committed_at", "")),
1386 agent_id=str(entry.get("agent_id", "")),
1387 model_id=str(entry.get("model_id", "")),
1388 ))
1389 results.append(ReleaseDict(
1390 release_id=str(item.get("release_id", "")),
1391 repo_id=str(item.get("repo_id", "")),
1392 tag=str(item.get("tag", "")),
1393 semver=semver,
1394 channel=str(item.get("channel", "stable")),
1395 commit_id=str(item.get("commit_id", "")),
1396 snapshot_id=str(item.get("snapshot_id", "")),
1397 title=str(item.get("title", "")),
1398 body=str(item.get("body", "")),
1399 changelog=changelog,
1400 agent_id=str(item.get("agent_id", "")),
1401 model_id=str(item.get("model_id", "")),
1402 is_draft=bool(item.get("is_draft", False)),
1403 gpg_signature=str(item.get("gpg_signature", "")),
1404 created_at=str(item.get("created_at", "")),
1405 ))
1406 except (KeyError, TypeError, ValueError):
1407 continue
1408 return results
1409
1410 # ---------------------------------------------------------------------------
1411 # TypedDict coercion helpers — extract known string fields from raw JSON dicts
1412 # ---------------------------------------------------------------------------
1413 # CommitDict and SnapshotDict are total=False (all fields optional), so we
1414 # only extract the string/scalar fields we can safely validate here.
1415 # CommitRecord.from_dict() and SnapshotRecord.from_dict() re-validate
1416 # required fields when apply_mpack() calls them.
1417 # ---------------------------------------------------------------------------
1418
1419 def _str(val: _WireVal) -> str:
1420 """Return *val* as str, or empty string if not a str."""
1421 return val if isinstance(val, str) else ""
1422
1423 def _str_or_none(val: _WireVal) -> str | None:
1424 """Return *val* as str, or None if not a str."""
1425 return val if isinstance(val, str) else None
1426
1427 def _int_or(val: _WireVal, default: int) -> int:
1428 """Return *val* as int, or *default* if not an int."""
1429 return val if isinstance(val, int) else default
1430
1431 def _coerce_commit_dict(raw: _WireDict) -> CommitDict:
1432 """Extract typed scalar fields from *raw* into a :class:`~muse.core.store.CommitDict`.
1433
1434 Only primitive fields are validated here; ``structured_delta`` is
1435 preserved as-is because :class:`~muse.core.store.CommitRecord.from_dict`
1436 already handles it gracefully.
1437 """
1438 metadata_raw = raw.get("metadata")
1439 metadata: Metadata = {}
1440 if isinstance(metadata_raw, dict):
1441 for k, v in metadata_raw.items():
1442 if isinstance(k, str) and isinstance(v, str):
1443 metadata[k] = v
1444
1445 reviewed_by_raw = raw.get("reviewed_by")
1446 reviewed_by: list[str] = []
1447 if isinstance(reviewed_by_raw, list):
1448 for item in reviewed_by_raw:
1449 if isinstance(item, str):
1450 reviewed_by.append(item)
1451
1452 breaking_changes_raw = raw.get("breaking_changes")
1453 breaking_changes: list[str] = []
1454 if isinstance(breaking_changes_raw, list):
1455 for item in breaking_changes_raw:
1456 if isinstance(item, str):
1457 breaking_changes.append(item)
1458
1459 sem_ver_raw = raw.get("sem_ver_bump")
1460 sem_ver: SemVerBump
1461 if sem_ver_raw == "major":
1462 sem_ver = "major"
1463 elif sem_ver_raw == "minor":
1464 sem_ver = "minor"
1465 elif sem_ver_raw == "patch":
1466 sem_ver = "patch"
1467 else:
1468 sem_ver = "none"
1469
1470 return CommitDict(
1471 commit_id=_str(raw.get("commit_id")),
1472 repo_id=_str(raw.get("repo_id")),
1473 branch=_str(raw.get("branch") or raw.get("created_on_branch")),
1474 snapshot_id=_str(raw.get("snapshot_id")),
1475 message=_str(raw.get("message")),
1476 committed_at=_str(raw.get("committed_at")),
1477 parent_commit_id=_str_or_none(raw.get("parent_commit_id")),
1478 parent2_commit_id=_str_or_none(raw.get("parent2_commit_id")),
1479 author=_str(raw.get("author")),
1480 metadata=metadata,
1481 structured_delta=None,
1482 sem_ver_bump=sem_ver,
1483 breaking_changes=breaking_changes,
1484 agent_id=_str(raw.get("agent_id")),
1485 model_id=_str(raw.get("model_id")),
1486 toolchain_id=_str(raw.get("toolchain_id")),
1487 prompt_hash=_str(raw.get("prompt_hash")),
1488 signature=_str(raw.get("signature")),
1489 signer_public_key=_str(raw.get("signer_public_key")),
1490 signer_key_id=_str(raw.get("signer_key_id")),
1491 reviewed_by=reviewed_by,
1492 test_runs=_int_or(raw.get("test_runs"), 0),
1493 )
1494
1495 def _coerce_snapshot_dict(raw: _WireDict) -> SnapshotDict:
1496 """Extract typed fields from *raw* into a :class:`~muse.core.store.SnapshotDict`.
1497
1498 Two wire formats are accepted:
1499 1. Delta format (fetch mpack): {snapshot_id, parent_snapshot_id,
1500 delta_upsert, delta_remove} — pass through delta fields so
1501 _apply_snapshot_deltas can reconstruct the manifest.
1502 2. Full-manifest format: {snapshot_id, manifest} — coerce into
1503 SnapshotDict directly.
1504 """
1505 manifest_raw = raw.get("manifest")
1506 delta_upsert_raw = raw.get("delta_upsert")
1507
1508 # Delta format: no manifest, but delta_upsert present.
1509 # _apply_snapshot_deltas handles reconstruction; pass through fields as-is.
1510 if manifest_raw is None and isinstance(delta_upsert_raw, dict):
1511 directories_raw = raw.get("directories")
1512 return { # type: ignore[return-value]
1513 "snapshot_id": _str(raw.get("snapshot_id")),
1514 "parent_snapshot_id": _str_or_none(raw.get("parent_snapshot_id")),
1515 "delta_upsert": {
1516 str(k): str(v) for k, v in delta_upsert_raw.items()
1517 if isinstance(k, str) and isinstance(v, str)
1518 },
1519 "delta_remove": [
1520 str(p) for p in (raw.get("delta_remove") or [])
1521 if isinstance(p, str)
1522 ],
1523 "directories": [
1524 str(d) for d in (directories_raw or []) if isinstance(d, str)
1525 ] if isinstance(directories_raw, list) else [],
1526 "created_at": _str(raw.get("created_at")),
1527 "manifest": None, # type: ignore[typeddict-item]
1528 "note": "",
1529 }
1530
1531 # Full-manifest format.
1532 manifest: Manifest = {}
1533 if isinstance(manifest_raw, dict):
1534 for k, v in manifest_raw.items():
1535 if isinstance(k, str) and isinstance(v, str):
1536 manifest[k] = v
1537 # ``directories`` is hashed into snapshot_id — dropping it produces a
1538 # SnapshotRecord whose snapshot_id never matches the recomputed hash,
1539 # making every snapshot with non-empty directories permanently unreadable.
1540 directories_raw = raw.get("directories")
1541 directories: list[str] = []
1542 if isinstance(directories_raw, list):
1543 for item in directories_raw:
1544 if isinstance(item, str):
1545 directories.append(item)
1546 return SnapshotDict(
1547 snapshot_id=_str(raw.get("snapshot_id")),
1548 manifest=manifest,
1549 directories=directories,
1550 created_at=_str(raw.get("created_at")),
1551 note=_str(raw.get("note", "")),
1552 )
1553
1554
1555 # ---------------------------------------------------------------------------
1556 # Factory
1557 # ---------------------------------------------------------------------------
1558
1559 def make_transport(url: str) -> "HttpTransport":
1560 """Return an :class:`HttpTransport` for *url*.
1561
1562 Args:
1563 url: Remote repository URL.
1564
1565 Returns:
1566 An ``HttpTransport`` instance implementing :class:`MuseTransport`.
1567 """
1568 return HttpTransport()
File History 8 commits
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 92 days ago
sha256:6b91dc90e1c59c5209b764a276c1ee824bd369320a454a403d28ce79890103f2 fix: stream mpack downloads without size cap Sonnet 4.6 minor ⚠ 97 days ago
sha256:f6cd81bc71702f5c1c6890bd39aaba994fe58c75f019d7c03934724fa2739bb4 fix: carry dev changes harmony dropped in merge — detached … Sonnet 4.6 minor ⚠ 102 days ago
sha256:b37d33a9dbf176b32955b34e5c9e983c4d8c6e7bfa4e2714edc938f31e721561 update transport logs Human minor ⚠ 103 days ago
sha256:79ffe87f5fe2ec146e35f05521218bbf54dffdb0440c07f970bad05f16efb89f chore: merge main — carry all urllib/typing/test fixes from dev Sonnet 4.6 minor ⚠ 105 days ago
sha256:0bea7600d1eee83e87950be49933b1006fa9dc2c71e7c4ee748d324f61138156 chore: bump version to 0.2.0rc11; fix typing audit violatio… Sonnet 4.6 minor ⚠ 105 days ago
sha256:633dfa2940e97bf1a3d04996c772027a57d70d103f1693c96da04969613dba6c fix: urllib migration regressions — force flag, job_id, Con… Sonnet 4.6 minor ⚠ 105 days ago
sha256:00cec040ce5f70bf8191d2ce6a9f308fbde553911068f0c303217f4eb6d4e775 fix: migrate httpx → urllib in transport.py and push.py; fi… Sonnet 4.6 minor ⚠ 105 days ago