gabriel / muse public
transport.py python
1,312 lines 51.5 KB
Raw
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 44 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 pathlib
44 import signal
45 import threading
46 import time as _time_mod
47 import urllib.parse
48 from typing import TYPE_CHECKING, NamedTuple, Protocol, TypedDict
49
50 if TYPE_CHECKING:
51 import ssl
52 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
53
54 class SigningIdentity(NamedTuple):
55 """Ed25519 signing identity for MSign request authentication."""
56
57 handle: str # hub handle, e.g. "gabriel"
58 private_key: Ed25519PrivateKey
59
60 import msgpack
61
62 from muse.core.mpack import (
63 BlobPayload,
64 MPack,
65 PushResult,
66 RemoteInfo,
67 WireTag,
68 )
69 from muse.core.semver import ChangelogEntry, SemVerTag
70 from muse.core.io import safe_unpackb
71 from muse.core.types import (
72 BranchHeads,
73 Manifest,
74 Metadata,
75 MsgpackDict,
76 )
77 from muse.core.commits import CommitDict
78 from muse.core.snapshots import SnapshotDict
79 from muse.core.releases import ReleaseDict
80 from muse.core.validation import (
81 MAX_RESPONSE_BYTES,
82 sanitize_display,
83 )
84 from muse.core.types import SemVerBump, blob_id
85
86 logger = logging.getLogger(__name__)
87
88 import httpx as _httpx_mod
89
90 _TIMEOUT_SECONDS = 300
91
92 def _mkcert_ca() -> "pathlib.Path | None":
93 """Return the mkcert root CA path, or None if mkcert is not installed.
94
95 Tries ``mkcert -CAROOT`` first (cross-platform), then falls back to the
96 well-known platform paths.
97 """
98 import subprocess
99 try:
100 r = subprocess.run(["mkcert", "-CAROOT"], capture_output=True, text=True, timeout=3)
101 if r.returncode == 0:
102 ca = pathlib.Path(r.stdout.strip()) / "rootCA.pem"
103 if ca.exists():
104 return ca
105 except (FileNotFoundError, subprocess.TimeoutExpired):
106 pass
107 # Platform fallbacks
108 candidates = [
109 pathlib.Path.home() / "Library" / "Application Support" / "mkcert" / "rootCA.pem", # macOS
110 pathlib.Path.home() / ".local" / "share" / "mkcert" / "rootCA.pem", # Linux
111 ]
112 for p in candidates:
113 if p.exists():
114 return p
115 return None
116
117 def _httpx_verify(url: str) -> "ssl.SSLContext | bool":
118 """Return the httpx `verify` value for *url*.
119
120 - localhost HTTPS → SSLContext loaded from the mkcert root CA
121 - everything else → True (system CA store)
122 """
123 import ssl
124 if url.startswith("https://localhost") or url.startswith("https://127.0.0.1"):
125 ca = _mkcert_ca()
126 if ca is not None:
127 ctx = ssl.create_default_context(cafile=str(ca))
128 return ctx
129 return True
130
131 class _Request(NamedTuple):
132 """Minimal HTTP request value passed between ``_build_request`` and ``_execute``."""
133
134 url: str
135 method: str
136 headers: dict[str, str]
137 data: bytes | None = None
138
139 # Recursive type alias for msgpack-serializable values.
140 # Covers every type that msgpack can encode/decode natively — no base64,
141 # no `object`, no `Any`. Python 3.12+ `type` statement creates a proper
142 # TypeAlias that mypy treats as a first-class recursive type.
143 type _MsgVal = (
144 str | int | float | bool | bytes | None
145 | list[_MsgVal]
146 | dict[str, _MsgVal]
147 )
148 type _MsgDict = dict[str, _MsgVal] # top-level msgpack response dict
149 type _WireVal = _MsgVal # wire-protocol value (same shape)
150 type _WireDict = dict[str, _WireVal] # wire-protocol response dict
151 type _HeadersDict = dict[str, str] # HTTP request/response headers
152
153 # Maximum number of objects to include in a single push batch.
154 # Must stay strictly below the server's MAX_OBJECTS_PER_PUSH limit (1 000).
155 # Large batches are split into sequential POSTs so each request body stays
156 # well within Cloudflare's body size limits.
157 CHUNK_OBJECTS: int = 500
158
159 # Maximum number of commits to include in a single POST /push call.
160 # Large histories are split into sequential chunked pushes so that each
161 # request stays well within Cloudflare's 100 MB upload limit.
162 # Each intermediate chunk uses force=True; the final chunk uses the caller's
163 # force flag.
164 #
165 # 500 commits/chunk means ~2 round-trips for a 1000-commit history vs the
166 # previous 5 at 200/chunk. The server now uses bulk INSERT ON CONFLICT DO
167 # NOTHING so per-chunk processing time is proportional to commit count, not
168 # the number of individual INSERT round-trips.
169 CHUNK_COMMITS: int = 500
170
171
172 # ---------------------------------------------------------------------------
173 # Result TypedDict
174 # ---------------------------------------------------------------------------
175
176 class FetchMPackResult(TypedDict):
177 """Result from ``fetch_mpack()`` — Phase 2 presigned-URL fetch.
178
179 ``blobs`` — raw blob dicts from the mpack; passed to ``apply_mpack``.
180 ``blobs_received`` — count of blobs in the mpack.
181 """
182
183 repo_id: str
184 domain: str
185 default_branch: str
186 branch_heads: BranchHeads
187 commits: list[CommitDict]
188 snapshots: list[SnapshotDict]
189 blobs: list[dict]
190 blobs_received: int
191 shallow_commits: list[str]
192
193 class _UnpackMPackResult(TypedDict):
194 """Response from POST /push/unpack-mpack."""
195 job_id: str
196 head: str
197 branch: str
198 blobs_in_mpack: int
199 commits_in_mpack: int
200
201
202 class _PresignResult(TypedDict):
203 """Response from POST /push/mpack-presign."""
204 upload_url: str
205 mpack_key: str
206
207
208 @contextlib.contextmanager
209 def _ignore_sigpipe() -> collections.abc.Generator[None, None, None]:
210 """Temporarily ignore SIGPIPE during socket I/O (main thread only).
211
212 ``muse/cli/app.py`` sets ``SIGPIPE = SIG_DFL`` at startup so that piping
213 ``muse`` output to ``head``/``grep``/``jq`` exits cleanly. But ``SIG_DFL``
214 also kills the process when a large HTTP request body is still in-flight
215 and the remote closes the connection early (auth failure, 409 conflict,
216 server crash, …). Without this guard the push command dies with exit code
217 141 instead of raising ``TransportError``.
218
219 Signal handlers may only be changed from the main interpreter thread.
220 Worker threads launched by ``ThreadPoolExecutor`` (used for parallel object
221 uploads) are already safe: Python blocks ``SIGPIPE`` in non-main threads,
222 so broken-socket writes raise ``BrokenPipeError`` rather than killing the
223 process. This context manager is therefore a no-op in non-main threads.
224
225 In the main thread the context manager saves the current SIGPIPE
226 disposition, sets it to ``SIG_IGN`` for the duration of the network call
227 (so that broken-pipe conditions surface as ``BrokenPipeError`` →
228 ``TransportError``), then restores the original disposition on exit. On
229 platforms without ``SIGPIPE`` (Windows) this is a no-op.
230 """
231 if not hasattr(signal, "SIGPIPE") or not threading.current_thread() is threading.main_thread():
232 yield
233 return
234 old_handler: signal.Handlers = signal.signal(signal.SIGPIPE, signal.SIG_IGN)
235 try:
236 yield
237 finally:
238 signal.signal(signal.SIGPIPE, old_handler)
239
240 # ---------------------------------------------------------------------------
241 # Exception
242 # ---------------------------------------------------------------------------
243
244 class TransportError(Exception):
245 """Raised when the remote returns a non-2xx response or is unreachable.
246
247 Attributes:
248 status_code: HTTP status code (e.g. ``401``, ``404``, ``409``, ``500``).
249 ``0`` for network-level failures (DNS, connection refused).
250 """
251
252 def __init__(self, message: str, status_code: int) -> None:
253 super().__init__(message)
254 self.status_code = status_code
255
256 # ---------------------------------------------------------------------------
257 # Protocol — the seam between CLI commands and the transport implementation
258 # ---------------------------------------------------------------------------
259
260 class MuseTransport(Protocol):
261 """Protocol for Muse remote transport implementations.
262
263 All methods are synchronous — the Muse CLI is synchronous by design.
264 """
265
266 def fetch_remote_info(self, url: str, signing: SigningIdentity | None) -> RemoteInfo:
267 """Return repository metadata from ``GET {url}/refs``.
268
269 Args:
270 url: Remote repository URL.
271 token: Ed25519 signing identity, or ``None`` for public repos.
272
273 Raises:
274 :class:`TransportError` on HTTP 4xx/5xx or network failure.
275 """
276 ...
277
278 def fetch_mpack(
279 self,
280 url: str,
281 signing: "SigningIdentity | None",
282 want: "list[str]",
283 have: "list[str]",
284 *,
285 on_object: "collections.abc.Callable[[BlobPayload], None] | None" = None,
286 ttl_seconds: int = 3600,
287 ) -> "FetchMPackResult":
288 """Fetch the delta as a single content-addressed MPack.
289
290 POSTs to ``{url}/fetch/mpack``. When the server returns
291 ``presign=False`` the mpack bytes are inline; when ``presign=True``
292 the client GETs the mpack from the presigned URL.
293
294 Either way ``sha256(mpack_bytes) == mpack_id`` is verified before
295 the mpack is unpacked — one hash covers all contents.
296
297 Args:
298 url: Canonical repo URL.
299 signing: Ed25519 signing identity; ``None`` for public repos.
300 want: Commit IDs the client wants.
301 have: Commit IDs already present locally (ancestry cut).
302 on_object: Callback invoked for each object in the mpack.
303 ttl_seconds: Presigned URL TTL forwarded to the server.
304
305 Raises:
306 :class:`TransportError` on non-200 response, integrity failure,
307 or network error.
308 """
309 ...
310
311 def push_tags(
312 self,
313 url: str,
314 signing: SigningIdentity | None,
315 tags: list[WireTag],
316 ) -> int:
317 """Push local tags to the remote via ``POST {url}/tags``.
318
319 Tags are immutable once created on the remote — the server skips any
320 it already holds. Returns the number of tags newly stored.
321
322 Args:
323 url: Remote repository URL.
324 token: Ed25519 signing identity, or ``None``.
325 tags: Tags to push.
326
327 Raises:
328 :class:`TransportError` on HTTP 4xx/5xx or network failure.
329 """
330 ...
331
332 def create_release(
333 self,
334 url: str,
335 signing: SigningIdentity | None,
336 release: ReleaseDict,
337 ) -> str:
338 """Create a release on the remote via ``POST {url}/releases``.
339
340 Returns the ``release_id`` assigned by the server.
341
342 Args:
343 url: Remote repository URL.
344 token: Ed25519 signing identity.
345 release: Fully-populated :class:`~muse.core.store.ReleaseDict`.
346
347 Raises:
348 :class:`TransportError` on HTTP 4xx/5xx or network failure.
349 """
350 ...
351
352 def list_releases_remote(
353 self,
354 url: str,
355 signing: SigningIdentity | None,
356 channel: str | None = None,
357 include_drafts: bool = False,
358 ) -> list[ReleaseDict]:
359 """Fetch releases from the remote via ``GET {url}/releases``.
360
361 Args:
362 url: Remote repository URL.
363 token: Ed25519 signing identity, or ``None``.
364 channel: Filter by release channel; ``None`` returns all.
365 include_drafts: Include draft releases when ``True``.
366
367 Raises:
368 :class:`TransportError` on HTTP 4xx/5xx or network failure.
369 """
370 ...
371
372 def delete_release_remote(
373 self,
374 url: str,
375 signing: SigningIdentity | None,
376 tag: str,
377 ) -> None:
378 """Retract a release from the remote via ``DELETE {url}/releases/{tag}``.
379
380 Removes only the named label from the remote registry. The underlying
381 commit and snapshot objects are **not** affected.
382
383 Args:
384 url: Remote repository URL.
385 token: Ed25519 signing identity (owner credentials required).
386 tag: Semver tag of the release to retract (e.g. ``"v1.2.0"``).
387
388 Raises:
389 :class:`TransportError` on HTTP 4xx/5xx, network failure, or if
390 the release does not exist on the remote.
391 """
392 ...
393
394 def delete_branch_remote(
395 self,
396 url: str,
397 signing: SigningIdentity | None,
398 branch: str,
399 ) -> None:
400 """Delete a branch on the remote via ``DELETE {url}/branches/{branch}``.
401
402 Equivalent to ``git push origin --delete <branch>``. The branch ref is
403 removed from the server; commits and objects are unaffected.
404
405 Args:
406 url: Remote repository URL.
407 token: Ed25519 signing identity (owner credentials required).
408 branch: Branch name to delete (e.g. ``"feat/my-thing"``).
409
410 Raises:
411 :class:`TransportError` on HTTP 4xx/5xx, network failure, or if
412 the branch does not exist on the remote.
413 """
414 ...
415
416 # ---------------------------------------------------------------------------
417 # HTTP/1.1 implementation (stdlib, zero extra dependencies)
418 # ---------------------------------------------------------------------------
419
420 def _http_error_message_from_bytes(status_code: int, body: bytes) -> str:
421 """Build a safe error message from a status code and raw response bytes."""
422 if status_code == 401:
423 return "Authentication failed (HTTP 401). Run 'muse auth register'."
424 raw = body[:300].decode("utf-8", errors="replace")
425 safe = sanitize_display(raw[:200])
426 if safe:
427 return f"HTTP {status_code}: {safe}"
428 return f"HTTP {status_code}"
429
430 def _http_error_message(err: "urllib.error.HTTPError") -> str:
431 """Return a safe, credential-free error message for an HTTP error.
432
433 HTTP 401 — NEVER includes the server body (which may echo the signing
434 credential). Returns a fixed generic string.
435
436 All other codes — include the body, capped at 200 chars, with all C0/C1
437 control characters stripped to prevent ANSI/OSC terminal injection.
438 """
439 if err.code == 401:
440 return "Authentication failed (HTTP 401). Run 'muse auth register'."
441 try:
442 raw_body = err.read(300).decode("utf-8", errors="replace")
443 except Exception:
444 raw_body = ""
445 safe_body = sanitize_display(raw_body[:200])
446 if safe_body:
447 return f"HTTP {err.code}: {safe_body}"
448 return f"HTTP {err.code}"
449
450 def _open_url(req: "urllib.request.Request") -> "Any":
451 """Thin seam around ``urllib.request.urlopen`` — patchable in tests."""
452 return urllib.request.urlopen(req) # noqa: S310
453
454 class HttpTransport:
455 """HTTPS transport using ``httpx`` (HTTP/2).
456
457 One short-lived connection per call. Signing identity values are
458 **never** written to any log line.
459 """
460
461 def _build_request(
462 self,
463 method: str,
464 url: str,
465 signing: SigningIdentity | None,
466 body_bytes: bytes | None = None,
467 content_type: str = "application/x-msgpack",
468 extra_headers: dict[str, str] | None = None,
469 ) -> _Request:
470 _parsed = urllib.parse.urlparse(url)
471 _is_loopback = _parsed.hostname in {
472 "localhost", "127.0.0.1", "::1", "host.docker.internal",
473 }
474 if signing and _parsed.scheme != "https" and not _is_loopback:
475 raise TransportError(
476 f"Refusing to send credentials to a non-HTTPS URL: {url!r}. "
477 "Ensure the remote URL uses https://.",
478 0,
479 )
480
481 # TOFU hub certificate fingerprint pinning — only for signed requests.
482 if not _is_loopback and signing is not None:
483 try:
484 from muse.core.hub_trust import check_and_pin
485 base_url = f"{_parsed.scheme}://{_parsed.netloc}"
486 check_and_pin(base_url)
487 except Exception as _hub_exc: # noqa: BLE001
488 from muse.core.errors import HubFingerprintMismatchError
489 if isinstance(_hub_exc, HubFingerprintMismatchError):
490 raise TransportError(str(_hub_exc), 0) from _hub_exc
491 logger.warning("⚠️ Hub trust check failed: %s", _hub_exc)
492
493 headers: _HeadersDict = {
494 "Accept": "application/x-msgpack, application/json",
495 }
496 if body_bytes is not None:
497 headers["Content-Type"] = content_type
498 if extra_headers:
499 headers.update(extra_headers)
500 if signing:
501 from muse.core.msign import build_msign_header
502 headers["Authorization"] = build_msign_header(signing, method, url, body_bytes)
503 return _Request(url=url, method=method, headers=headers, data=body_bytes)
504
505 @staticmethod
506 def _decode(raw: bytes) -> _MsgDict:
507 """Decode a msgpack server response into a plain dict.
508
509 Raises :class:`TransportError` if the payload is not valid msgpack.
510 """
511 if not raw:
512 return {}
513 try:
514 # MAX_RESPONSE_BYTES already caps the network read — safe_unpackb
515 # adds per-field limits as a second layer against billion-laughs payloads.
516 # allow_binary=True because pack/mpack responses carry raw blob content.
517 result: _MsgVal = safe_unpackb(raw, context="server response", allow_binary=True)
518 except Exception as exc:
519 raise TransportError(f"Server returned invalid msgpack: {exc}", 0) from exc
520 if not isinstance(result, dict):
521 return {}
522 return result
523
524 def _execute(self, req: "_Request | urllib.request.Request") -> bytes:
525 """Send *req* and return raw response bytes.
526
527 Accepts either an internal ``_Request`` (httpx path) or a
528 ``urllib.request.Request`` (urllib path via ``_open_url``).
529
530 Raises:
531 :class:`TransportError` on non-2xx HTTP or any network error.
532 """
533 if isinstance(req, urllib.request.Request):
534 return self._execute_fetch(req)
535 try:
536 with _ignore_sigpipe():
537 with _httpx_mod.Client(
538 http2=True,
539 timeout=_TIMEOUT_SECONDS,
540 follow_redirects=False,
541 verify=_httpx_verify(req.url),
542 ) as client:
543 resp = client.request(req.method, req.url, content=req.data, headers=req.headers)
544 body_bytes: bytes = resp.content
545 if resp.status_code >= 400:
546 raise TransportError(
547 _http_error_message_from_bytes(resp.status_code, body_bytes),
548 resp.status_code,
549 )
550 if len(body_bytes) > MAX_RESPONSE_BYTES:
551 raise TransportError(
552 f"Response body exceeds the {MAX_RESPONSE_BYTES // (1024 * 1024)} MiB cap.",
553 0,
554 )
555 return body_bytes
556 except TransportError:
557 raise
558 except Exception as exc:
559 raise TransportError(sanitize_display(str(exc)), 0) from exc
560
561 def _execute_fetch(self, req: "urllib.request.Request") -> bytes:
562 """Send *req* via urllib (HTTP/1.1) and return raw response bytes.
563
564 Uses ``_open_url`` as the network seam — patchable in tests.
565
566 Raises:
567 :class:`TransportError` on non-2xx HTTP or any network error.
568 """
569 try:
570 with _open_url(req) as resp:
571 body_bytes = resp.read(MAX_RESPONSE_BYTES + 1)
572 if len(body_bytes) > MAX_RESPONSE_BYTES:
573 raise TransportError(
574 f"Response body exceeds the {MAX_RESPONSE_BYTES // (1024 * 1024)} MiB cap.",
575 0,
576 )
577 return body_bytes
578 except urllib.error.HTTPError as err:
579 raise TransportError(_http_error_message(err), err.code) from err
580 except urllib.error.URLError as err:
581 raise TransportError(sanitize_display(str(err.reason)), 0) from err
582 except TransportError:
583 raise
584 except Exception as exc:
585 raise TransportError(sanitize_display(str(exc)), 0) from exc
586
587 def hub_json(
588 self,
589 method: str,
590 url: str,
591 signing: "SigningIdentity | None",
592 body: "dict[str, object] | None" = None,
593 ) -> "dict[str, object]":
594 """Make an authenticated JSON request and return the parsed response dict.
595
596 Uses httpx with the mkcert SSL context for localhost URLs — the same
597 path as all other transport calls. Raises :class:`TransportError` on
598 any HTTP or network error.
599
600 Args:
601 method: HTTP verb (``GET``, ``POST``, ``PATCH``, ``DELETE``).
602 url: Full URL including scheme and path.
603 signing: MSign identity, or ``None`` for unauthenticated calls.
604 body: Optional JSON-serializable dict sent as the request body.
605 """
606 import json as _json_mod
607 body_bytes = _json_mod.dumps(body).encode() if body is not None else None
608 req = self._build_request(
609 method, url, signing,
610 body_bytes=body_bytes,
611 content_type="application/json",
612 extra_headers={"Accept": "application/json"},
613 )
614 raw = self._execute(req)
615 try:
616 result = _json_mod.loads(raw.decode("utf-8"))
617 except Exception as exc:
618 raise TransportError(f"Server returned invalid JSON: {exc}", 0) from exc
619 if not isinstance(result, dict):
620 return {}
621 return result
622
623 def hub_bytes(
624 self,
625 url: str,
626 signing: "SigningIdentity | None",
627 ) -> bytes:
628 """Download raw bytes from *url* with optional MSign auth.
629
630 Uses httpx with the mkcert SSL context for localhost URLs.
631 Raises :class:`TransportError` on any HTTP or network error.
632
633 Args:
634 url: Full URL including scheme and path.
635 signing: MSign identity, or ``None`` for unauthenticated calls.
636 """
637 req = self._build_request(
638 "GET", url, signing,
639 extra_headers={"Accept": "*/*"},
640 )
641 return self._execute(req)
642
643 def fetch_remote_info(self, url: str, signing: SigningIdentity | None) -> RemoteInfo:
644 """Fetch repository metadata from ``GET {url}/refs``."""
645 endpoint = f"{url.rstrip('/')}/refs"
646 logger.debug("transport: GET %s", endpoint)
647 req = self._build_request("GET", endpoint, signing)
648 raw = self._execute(req)
649 return _parse_remote_info(raw)
650
651 def push_mpack_put(self, upload_url: str, mpack_bytes: bytes) -> None:
652 """PUT mpack_bytes directly to a presigned MinIO URL (Step 2).
653
654 No MuseHub API involvement. No Authorization header. The presigned
655 URL is the credential. Raises :class:`TransportError` on non-2xx.
656 """
657 with _httpx_mod.Client(
658 http2=True,
659 timeout=_TIMEOUT_SECONDS,
660 follow_redirects=False,
661 verify=_httpx_verify(upload_url),
662 ) as client:
663 resp = client.put(upload_url, content=mpack_bytes,
664 headers={"Content-Type": "application/x-muse-pack"})
665 if resp.status_code >= 400:
666 raise TransportError(
667 _http_error_message_from_bytes(resp.status_code, resp.content),
668 resp.status_code,
669 )
670
671 def push_mpack_unpack(
672 self,
673 url: str,
674 signing: "SigningIdentity | None",
675 mpack_key: str,
676 *,
677 branch: str = "main",
678 head: str = "",
679 commits_count: int = 0,
680 blobs_count: int = 0,
681 ) -> _UnpackMPackResult:
682 """POST /push/unpack-mpack — notify server to index the uploaded mpack (Step 3).
683
684 Returns ``{"job_id": str, "head": str, "branch": str,
685 "blobs_in_mpack": int, "commits_in_mpack": int}``.
686 Raises :class:`TransportError` on non-2xx or missing job_id.
687 """
688 from muse.core.mpack import build_unpack_payload
689 endpoint = f"{url.rstrip('/')}/push/unpack-mpack"
690 payload = build_unpack_payload(
691 mpack_key, branch=branch, head=head,
692 commits_count=commits_count, blobs_count=blobs_count,
693 )
694 body_bytes = msgpack.packb(payload, use_bin_type=True)
695 req = self._build_request("POST", endpoint, signing, body_bytes)
696 with _httpx_mod.Client(
697 http2=True,
698 timeout=_TIMEOUT_SECONDS,
699 follow_redirects=False,
700 verify=_httpx_verify(endpoint),
701 ) as client:
702 resp = client.post(endpoint, content=body_bytes, headers=dict(req.headers))
703 if resp.status_code >= 400:
704 raise TransportError(
705 _http_error_message_from_bytes(resp.status_code, resp.content),
706 resp.status_code,
707 )
708 data = self._decode(resp.content)
709 job_id = str(data.get("job_id") or "")
710 if not job_id:
711 raise TransportError("push/unpack-mpack: server response missing job_id", 0)
712 return {
713 "job_id": job_id,
714 "head": str(data.get("head") or head),
715 "branch": str(data.get("branch") or branch),
716 "blobs_in_mpack": int(data.get("blobs_in_mpack") or 0),
717 "commits_in_mpack": int(data.get("commits_in_mpack") or 0),
718 }
719
720 def push_mpack_presign(
721 self,
722 url: str,
723 signing: "SigningIdentity | None",
724 mpack_bytes: bytes,
725 ttl_seconds: int = 3600,
726 ) -> _PresignResult:
727 """POST /push/mpack-presign — get a presigned PUT URL for the mpack.
728
729 Returns ``{"upload_url": str, "mpack_key": str}``.
730 Raises :class:`TransportError` on non-2xx or a missing upload_url.
731 """
732 from muse.core.mpack import build_presign_payload
733 endpoint = f"{url.rstrip('/')}/push/mpack-presign"
734 payload = build_presign_payload(mpack_bytes)
735 body_bytes = msgpack.packb(payload, use_bin_type=True)
736 req = self._build_request("POST", endpoint, signing, body_bytes)
737 with _httpx_mod.Client(
738 http2=True,
739 timeout=_TIMEOUT_SECONDS,
740 follow_redirects=False,
741 verify=_httpx_verify(endpoint),
742 ) as client:
743 resp = client.post(endpoint, content=body_bytes, headers=dict(req.headers))
744 if resp.status_code >= 400:
745 raise TransportError(
746 _http_error_message_from_bytes(resp.status_code, resp.content),
747 resp.status_code,
748 )
749 data = self._decode(resp.content)
750 upload_url = str(data.get("upload_url") or "")
751 if not upload_url:
752 raise TransportError("push/mpack-presign: server response missing upload_url", 0)
753 return {"upload_url": upload_url, "mpack_key": str(data.get("mpack_key") or payload["mpack_key"])}
754
755 def fetch_mpack(
756 self,
757 url: str,
758 signing: "SigningIdentity | None",
759 want: "list[str]",
760 have: "list[str]",
761 *,
762 ttl_seconds: int = 3600,
763 ) -> "FetchMPackResult":
764 """Fetch the commit delta as one content-addressed MPack.
765
766 POSTs ``{url}/fetch/mpack``. Server returns a presigned GET URL.
767 Client GETs the full mpack bytes directly from MinIO, verifies
768 ``blob_id(bytes) == mpack_id``, then returns all content for the
769 caller to pass to ``apply_mpack``. No per-object streaming.
770 """
771 endpoint = f"{url.rstrip('/')}/fetch/mpack"
772 body_bytes = msgpack.packb(
773 {"want": want, "have": have, "ttl_seconds": ttl_seconds},
774 use_bin_type=True,
775 )
776 req = self._build_request("POST", endpoint, signing, body_bytes)
777
778 import sys as _sys
779 t0 = _time_mod.monotonic()
780 print(f"[transport] POST {endpoint} want={len(want)} have={len(have)}", file=_sys.stderr, flush=True)
781 try:
782 with _httpx_mod.Client(
783 http2=True,
784 timeout=_TIMEOUT_SECONDS,
785 follow_redirects=True,
786 verify=_httpx_verify(endpoint),
787 ) as client:
788 resp = client.post(endpoint, content=body_bytes, headers=dict(req.headers))
789 t_post = _time_mod.monotonic()
790 print(
791 f"[transport] POST done status={resp.status_code} body={len(resp.content)}B t={int((t_post - t0)*1000)}ms",
792 file=_sys.stderr, flush=True,
793 )
794 if resp.status_code >= 400:
795 raise TransportError(
796 _http_error_message_from_bytes(resp.status_code, resp.content),
797 resp.status_code,
798 )
799
800 data = msgpack.unpackb(resp.content, raw=False)
801 bundle_id: str = str(data.get("mpack_id") or data.get("bundle_id") or "")
802 mpack_url_raw: str = str(data.get("mpack_url") or "")
803 t_parse = _time_mod.monotonic()
804 print(
805 f"[transport] parsed bundle_id={bundle_id[:20] if bundle_id else 'none'}"
806 f" mpack_url={mpack_url_raw[:100] if mpack_url_raw else 'null'}"
807 f" t={int((t_parse - t0)*1000)}ms",
808 file=_sys.stderr, flush=True,
809 )
810
811 if mpack_url_raw:
812 print(f"[transport] GET mpack {mpack_url_raw[:120]}", file=_sys.stderr, flush=True)
813 # Use a fresh client with follow_redirects=True for the direct S3/MinIO download.
814 with _httpx_mod.Client(
815 http2=False,
816 timeout=_TIMEOUT_SECONDS,
817 follow_redirects=True,
818 verify=_httpx_verify(mpack_url_raw),
819 ) as dl_client:
820 get_resp = dl_client.get(mpack_url_raw)
821 t_dl = _time_mod.monotonic()
822 dl_mb = len(get_resp.content) / (1024 * 1024)
823 print(
824 f"[transport] GET done status={get_resp.status_code} size={dl_mb:.2f}MB"
825 f" t={int((t_dl - t_parse)*1000)}ms",
826 file=_sys.stderr, flush=True,
827 )
828 if get_resp.status_code >= 400:
829 raise TransportError(
830 f"fetch/mpack: GET HTTP {get_resp.status_code}",
831 get_resp.status_code,
832 )
833 mpack_bytes_dl = get_resp.content
834 actual_id = blob_id(mpack_bytes_dl)
835 t_verify = _time_mod.monotonic()
836 print(
837 f"[transport] sha256 match={actual_id == bundle_id} t={int((t_verify - t_dl)*1000)}ms",
838 file=_sys.stderr, flush=True,
839 )
840 if actual_id != bundle_id:
841 raise TransportError(
842 f"fetch/mpack: integrity failure mpack_id={bundle_id!r} "
843 f"actual={actual_id!r}",
844 0,
845 )
846 if mpack_bytes_dl[:4] == b"MUSE":
847 from muse.core.mpack import parse_wire_mpack
848 mpack = parse_wire_mpack(mpack_bytes_dl)
849 else:
850 mpack = msgpack.unpackb(mpack_bytes_dl, raw=False)
851 t_unpack = _time_mod.monotonic()
852 print(
853 f"[transport] unpackb commits={len(mpack.get('commits') or [])} snaps={len(mpack.get('snapshots') or [])}"
854 f" blobs={len(mpack.get('blobs') or [])} t={int((t_unpack - t_verify)*1000)}ms",
855 file=_sys.stderr, flush=True,
856 )
857 else:
858 # mpack_url absent or null → all data is inline in the response body
859 # (Phase 3 inline protocol — commits and snapshots in resp.content already parsed).
860 print("[transport] no mpack_url — inline protocol (no S3 download)", file=_sys.stderr, flush=True)
861 mpack = {"commits": [], "snapshots": [], "blobs": [], "branch_heads": {}}
862 t_unpack = _time_mod.monotonic()
863
864 except TransportError:
865 raise
866 except Exception as exc:
867 raise TransportError(sanitize_display(str(exc)), 0) from exc
868
869 raw_commits = mpack.get("commits") or []
870 raw_snaps = mpack.get("snapshots") or []
871 raw_blobs = mpack.get("blobs") or []
872 raw_heads = data.get("branch_heads") or mpack.get("branch_heads") or {}
873
874 t_coerce0 = _time_mod.monotonic()
875 commits: list[CommitDict] = [
876 _coerce_commit_dict(dict(c)) for c in raw_commits if isinstance(c, dict)
877 ]
878 snapshots: list[SnapshotDict] = [
879 _coerce_snapshot_dict(dict(s)) for s in raw_snaps if isinstance(s, dict)
880 ]
881 branch_heads: BranchHeads = {
882 str(k): str(v) for k, v in raw_heads.items()
883 if isinstance(k, str) and isinstance(v, str)
884 }
885 blobs: list[dict] = [obj for obj in raw_blobs if isinstance(obj, dict)]
886 t_coerce1 = _time_mod.monotonic()
887 print(
888 f"[transport] coerce commits={len(commits)} snaps={len(snapshots)} blobs={len(blobs)} heads={len(branch_heads)}"
889 f" t={int((t_coerce1 - t_coerce0)*1000)}ms TOTAL={int((t_coerce1 - t0)*1000)}ms",
890 file=_sys.stderr, flush=True,
891 )
892
893 return FetchMPackResult(
894 repo_id=str(data.get("repo_id") or ""),
895 domain=str(data.get("domain") or ""),
896 default_branch=str(data.get("default_branch") or "main"),
897 branch_heads=branch_heads,
898 commits=commits,
899 snapshots=snapshots,
900 blobs=blobs,
901 blobs_received=len(blobs),
902 shallow_commits=[],
903 )
904
905 def push_tags(
906 self,
907 url: str,
908 signing: SigningIdentity | None,
909 tags: list[WireTag],
910 ) -> int:
911 """Push tags via ``POST {url}/tags``."""
912 endpoint = f"{url.rstrip('/')}/tags"
913 logger.debug("transport: POST %s (tags=%d)", endpoint, len(tags))
914 body_bytes: bytes = msgpack.packb({"tags": list(tags)}, use_bin_type=True)
915 req = self._build_request("POST", endpoint, signing, body_bytes)
916 raw = self._execute(req)
917 parsed = self._decode(raw)
918 stored_val = parsed.get("stored")
919 return int(stored_val) if isinstance(stored_val, int) else 0
920
921 def create_release(
922 self,
923 url: str,
924 signing: SigningIdentity | None,
925 release: ReleaseDict,
926 ) -> str:
927 """Create a release via ``POST {url}/releases``."""
928 endpoint = f"{url.rstrip('/')}/releases"
929 logger.debug("transport: POST %s (tag=%s)", endpoint, release.get("tag", ""))
930 # ReleaseDict contains no bytes fields so JSON encoding works directly.
931 body_bytes: bytes = json.dumps(release).encode("utf-8")
932 req = self._build_request("POST", endpoint, signing, body_bytes, content_type="application/json")
933 raw = self._execute(req)
934 parsed = self._decode(raw)
935 release_id_val = parsed.get("release_id")
936 return str(release_id_val) if isinstance(release_id_val, str) else ""
937
938 def list_releases_remote(
939 self,
940 url: str,
941 signing: SigningIdentity | None,
942 channel: str | None = None,
943 include_drafts: bool = False,
944 ) -> list[ReleaseDict]:
945 """List releases via ``GET {url}/releases``."""
946 qs_parts: list[str] = []
947 if channel:
948 qs_parts.append(f"channel={urllib.parse.quote(channel)}")
949 if include_drafts:
950 qs_parts.append("include_drafts=1")
951 endpoint = f"{url.rstrip('/')}/releases"
952 if qs_parts:
953 endpoint = f"{endpoint}?{'&'.join(qs_parts)}"
954 logger.debug("transport: GET %s", endpoint)
955 req = self._build_request("GET", endpoint, signing)
956 raw = self._execute(req)
957 parsed = self._decode(raw)
958 return _parse_releases_list(parsed)
959
960 def delete_release_remote(
961 self,
962 url: str,
963 signing: SigningIdentity | None,
964 tag: str,
965 ) -> None:
966 """Retract a release via ``DELETE {url}/releases/{tag}``."""
967 endpoint = f"{url.rstrip('/')}/releases/{urllib.parse.quote(tag, safe='')}"
968 logger.debug("transport: DELETE %s", endpoint)
969 req = self._build_request("DELETE", endpoint, signing)
970 self._execute(req)
971
972 def delete_branch_remote(
973 self,
974 url: str,
975 signing: SigningIdentity | None,
976 branch: str,
977 ) -> None:
978 """Delete a branch via ``DELETE {url}/branches/{branch}``."""
979 endpoint = f"{url.rstrip('/')}/branches/{urllib.parse.quote(branch, safe='')}"
980 logger.debug("transport: DELETE %s", endpoint)
981 req = self._build_request("DELETE", endpoint, signing)
982 self._execute(req)
983
984 # ---------------------------------------------------------------------------
985 # Response parsers — JSON bytes → typed TypedDicts
986 # ---------------------------------------------------------------------------
987 # json.loads() returns Any (per typeshed), so we use isinstance narrowing
988 # throughout. No explicit Any annotations appear in this file.
989 # ---------------------------------------------------------------------------
990
991 def _parse_remote_info(raw: bytes) -> RemoteInfo:
992 """Parse ``GET /refs`` response bytes into a :class:`~muse.core.mpack.RemoteInfo`."""
993 parsed = HttpTransport._decode(raw)
994 repo_id_val = parsed.get("repo_id")
995 domain_val = parsed.get("domain")
996 default_branch_val = parsed.get("default_branch")
997 branch_heads_raw = parsed.get("branch_heads")
998 branch_heads: BranchHeads = {}
999 if isinstance(branch_heads_raw, dict):
1000 for k, v in branch_heads_raw.items():
1001 if isinstance(k, str) and isinstance(v, str):
1002 branch_heads[k] = v
1003 info = RemoteInfo(
1004 repo_id=str(repo_id_val) if isinstance(repo_id_val, str) else "",
1005 domain=str(domain_val) if isinstance(domain_val, str) else "midi",
1006 default_branch=(
1007 str(default_branch_val) if isinstance(default_branch_val, str) else "main"
1008 ),
1009 branch_heads=branch_heads,
1010 )
1011 return info
1012
1013 def _parse_mpack(raw: bytes) -> MPack:
1014 """Parse ``POST /fetch`` response bytes into a :class:`~muse.core.mpack.MPack`."""
1015 parsed = HttpTransport._decode(raw)
1016 mpack: MPack = {}
1017
1018 # Commits — each item is a raw dict that CommitRecord.from_dict() will validate.
1019 commits_raw = parsed.get("commits")
1020 if isinstance(commits_raw, list):
1021 commits: list[CommitDict] = []
1022 for item in commits_raw:
1023 if isinstance(item, dict):
1024 commits.append(_coerce_commit_dict(item))
1025 mpack["commits"] = commits
1026
1027 # Snapshots
1028 snapshots_raw = parsed.get("snapshots")
1029 if isinstance(snapshots_raw, list):
1030 snapshots: list[SnapshotDict] = []
1031 for item in snapshots_raw:
1032 if isinstance(item, dict):
1033 snapshots.append(_coerce_snapshot_dict(item))
1034 mpack["snapshots"] = snapshots
1035
1036 # Blobs — raw bytes in "content" (mpack wire format)
1037 blobs_raw = parsed.get("blobs")
1038 if isinstance(blobs_raw, list):
1039 blobs: list[BlobPayload] = []
1040 for item in blobs_raw:
1041 if not isinstance(item, dict):
1042 continue
1043 oid = item.get("object_id")
1044 if not isinstance(oid, str):
1045 continue
1046 content_raw = item.get("content")
1047 if isinstance(content_raw, (bytes, bytearray)):
1048 blobs.append(BlobPayload(object_id=oid, content=bytes(content_raw)))
1049 mpack["blobs"] = blobs
1050
1051 # Branch heads
1052 heads_raw = parsed.get("branch_heads")
1053 if isinstance(heads_raw, dict):
1054 branch_heads: BranchHeads = {}
1055 for k, v in heads_raw.items():
1056 if isinstance(k, str) and isinstance(v, str):
1057 branch_heads[k] = v
1058 mpack["branch_heads"] = branch_heads
1059
1060 return mpack
1061
1062 def _parse_push_result(raw: bytes) -> PushResult:
1063 """Parse ``POST /push`` response bytes into a :class:`~muse.core.mpack.PushResult`."""
1064 parsed = HttpTransport._decode(raw)
1065 ok_val = parsed.get("ok")
1066 msg_val = parsed.get("message")
1067 heads_raw = parsed.get("branch_heads")
1068 branch_heads: BranchHeads = {}
1069 if isinstance(heads_raw, dict):
1070 for k, v in heads_raw.items():
1071 if isinstance(k, str) and isinstance(v, str):
1072 branch_heads[k] = v
1073 return PushResult(
1074 ok=bool(ok_val) if isinstance(ok_val, bool) else False,
1075 message=str(msg_val) if isinstance(msg_val, str) else "",
1076 branch_heads=branch_heads,
1077 )
1078
1079 def _coerce_sem_ver_bump(raw: _MsgVal) -> SemVerBump:
1080 """Safely coerce a raw value to a :class:`~muse.domain.SemVerBump` literal."""
1081 if raw == "major":
1082 return "major"
1083 if raw == "minor":
1084 return "minor"
1085 if raw == "patch":
1086 return "patch"
1087 return "none"
1088
1089 def _parse_releases_list(parsed: _MsgDict) -> list[ReleaseDict]:
1090 """Extract a list of :class:`~muse.core.store.ReleaseDict` from a parsed response."""
1091 releases_raw = parsed.get("releases")
1092 if not isinstance(releases_raw, list):
1093 return []
1094 results: list[ReleaseDict] = []
1095 for item in releases_raw:
1096 if not isinstance(item, dict):
1097 continue
1098 try:
1099 semver_raw = item.get("semver")
1100 if isinstance(semver_raw, dict):
1101 sv_major = semver_raw.get("major")
1102 sv_minor = semver_raw.get("minor")
1103 sv_patch = semver_raw.get("patch")
1104 sv_pre = semver_raw.get("pre")
1105 sv_build = semver_raw.get("build")
1106 semver: SemVerTag = SemVerTag(
1107 major=int(sv_major) if isinstance(sv_major, int) else 0,
1108 minor=int(sv_minor) if isinstance(sv_minor, int) else 0,
1109 patch=int(sv_patch) if isinstance(sv_patch, int) else 0,
1110 pre=sv_pre if isinstance(sv_pre, str) else "",
1111 build=sv_build if isinstance(sv_build, str) else "",
1112 )
1113 else:
1114 semver = SemVerTag(major=0, minor=0, patch=0, pre="", build="")
1115 changelog_raw = item.get("changelog") or []
1116 changelog: list[ChangelogEntry] = []
1117 if isinstance(changelog_raw, list):
1118 for entry in changelog_raw:
1119 if not isinstance(entry, dict):
1120 continue
1121 bc_raw = entry.get("breaking_changes")
1122 bc_list: list[str] = [str(b) for b in bc_raw if isinstance(b, str)] if isinstance(bc_raw, list) else []
1123 changelog.append(ChangelogEntry(
1124 commit_id=str(entry.get("commit_id", "")),
1125 message=str(entry.get("message", "")),
1126 sem_ver_bump=_coerce_sem_ver_bump(entry.get("sem_ver_bump")),
1127 breaking_changes=bc_list,
1128 author=str(entry.get("author", "")),
1129 committed_at=str(entry.get("committed_at", "")),
1130 agent_id=str(entry.get("agent_id", "")),
1131 model_id=str(entry.get("model_id", "")),
1132 ))
1133 results.append(ReleaseDict(
1134 release_id=str(item.get("release_id", "")),
1135 repo_id=str(item.get("repo_id", "")),
1136 tag=str(item.get("tag", "")),
1137 semver=semver,
1138 channel=str(item.get("channel", "stable")),
1139 commit_id=str(item.get("commit_id", "")),
1140 snapshot_id=str(item.get("snapshot_id", "")),
1141 title=str(item.get("title", "")),
1142 body=str(item.get("body", "")),
1143 changelog=changelog,
1144 agent_id=str(item.get("agent_id", "")),
1145 model_id=str(item.get("model_id", "")),
1146 is_draft=bool(item.get("is_draft", False)),
1147 gpg_signature=str(item.get("gpg_signature", "")),
1148 created_at=str(item.get("created_at", "")),
1149 ))
1150 except (KeyError, TypeError, ValueError):
1151 continue
1152 return results
1153
1154 # ---------------------------------------------------------------------------
1155 # TypedDict coercion helpers — extract known string fields from raw JSON dicts
1156 # ---------------------------------------------------------------------------
1157 # CommitDict and SnapshotDict are total=False (all fields optional), so we
1158 # only extract the string/scalar fields we can safely validate here.
1159 # CommitRecord.from_dict() and SnapshotRecord.from_dict() re-validate
1160 # required fields when apply_mpack() calls them.
1161 # ---------------------------------------------------------------------------
1162
1163 def _str(val: _WireVal) -> str:
1164 """Return *val* as str, or empty string if not a str."""
1165 return val if isinstance(val, str) else ""
1166
1167 def _str_or_none(val: _WireVal) -> str | None:
1168 """Return *val* as str, or None if not a str."""
1169 return val if isinstance(val, str) else None
1170
1171 def _int_or(val: _WireVal, default: int) -> int:
1172 """Return *val* as int, or *default* if not an int."""
1173 return val if isinstance(val, int) else default
1174
1175 def _coerce_commit_dict(raw: _WireDict) -> CommitDict:
1176 """Extract typed scalar fields from *raw* into a :class:`~muse.core.store.CommitDict`.
1177
1178 Only primitive fields are validated here; ``structured_delta`` is
1179 preserved as-is because :class:`~muse.core.store.CommitRecord.from_dict`
1180 already handles it gracefully.
1181 """
1182 metadata_raw = raw.get("metadata")
1183 metadata: Metadata = {}
1184 if isinstance(metadata_raw, dict):
1185 for k, v in metadata_raw.items():
1186 if isinstance(k, str) and isinstance(v, str):
1187 metadata[k] = v
1188
1189 reviewed_by_raw = raw.get("reviewed_by")
1190 reviewed_by: list[str] = []
1191 if isinstance(reviewed_by_raw, list):
1192 for item in reviewed_by_raw:
1193 if isinstance(item, str):
1194 reviewed_by.append(item)
1195
1196 breaking_changes_raw = raw.get("breaking_changes")
1197 breaking_changes: list[str] = []
1198 if isinstance(breaking_changes_raw, list):
1199 for item in breaking_changes_raw:
1200 if isinstance(item, str):
1201 breaking_changes.append(item)
1202
1203 sem_ver_raw = raw.get("sem_ver_bump")
1204 sem_ver: SemVerBump
1205 if sem_ver_raw == "major":
1206 sem_ver = "major"
1207 elif sem_ver_raw == "minor":
1208 sem_ver = "minor"
1209 elif sem_ver_raw == "patch":
1210 sem_ver = "patch"
1211 else:
1212 sem_ver = "none"
1213
1214 return CommitDict(
1215 commit_id=_str(raw.get("commit_id")),
1216 repo_id=_str(raw.get("repo_id")),
1217 branch=_str(raw.get("branch") or raw.get("created_on_branch")),
1218 snapshot_id=_str(raw.get("snapshot_id")),
1219 message=_str(raw.get("message")),
1220 committed_at=_str(raw.get("committed_at")),
1221 parent_commit_id=_str_or_none(raw.get("parent_commit_id")),
1222 parent2_commit_id=_str_or_none(raw.get("parent2_commit_id")),
1223 author=_str(raw.get("author")),
1224 metadata=metadata,
1225 structured_delta=None,
1226 sem_ver_bump=sem_ver,
1227 breaking_changes=breaking_changes,
1228 agent_id=_str(raw.get("agent_id")),
1229 model_id=_str(raw.get("model_id")),
1230 toolchain_id=_str(raw.get("toolchain_id")),
1231 prompt_hash=_str(raw.get("prompt_hash")),
1232 signature=_str(raw.get("signature")),
1233 signer_public_key=_str(raw.get("signer_public_key")),
1234 signer_key_id=_str(raw.get("signer_key_id")),
1235 reviewed_by=reviewed_by,
1236 test_runs=_int_or(raw.get("test_runs"), 0),
1237 )
1238
1239 def _coerce_snapshot_dict(raw: _WireDict) -> SnapshotDict:
1240 """Extract typed fields from *raw* into a :class:`~muse.core.store.SnapshotDict`.
1241
1242 Two wire formats are accepted:
1243 1. Delta format (fetch mpack): {snapshot_id, parent_snapshot_id,
1244 delta_upsert, delta_remove} — pass through delta fields so
1245 _apply_snapshot_deltas can reconstruct the manifest.
1246 2. Full-manifest format: {snapshot_id, manifest} — coerce into
1247 SnapshotDict directly.
1248 """
1249 manifest_raw = raw.get("manifest")
1250 delta_upsert_raw = raw.get("delta_upsert")
1251
1252 # Delta format: no manifest, but delta_upsert present.
1253 # _apply_snapshot_deltas handles reconstruction; pass through fields as-is.
1254 if manifest_raw is None and isinstance(delta_upsert_raw, dict):
1255 directories_raw = raw.get("directories")
1256 return { # type: ignore[return-value]
1257 "snapshot_id": _str(raw.get("snapshot_id")),
1258 "parent_snapshot_id": _str_or_none(raw.get("parent_snapshot_id")),
1259 "delta_upsert": {
1260 str(k): str(v) for k, v in delta_upsert_raw.items()
1261 if isinstance(k, str) and isinstance(v, str)
1262 },
1263 "delta_remove": [
1264 str(p) for p in (raw.get("delta_remove") or [])
1265 if isinstance(p, str)
1266 ],
1267 "directories": [
1268 str(d) for d in (directories_raw or []) if isinstance(d, str)
1269 ] if isinstance(directories_raw, list) else [],
1270 "created_at": _str(raw.get("created_at")),
1271 "manifest": None, # type: ignore[typeddict-item]
1272 "note": "",
1273 }
1274
1275 # Full-manifest format.
1276 manifest: Manifest = {}
1277 if isinstance(manifest_raw, dict):
1278 for k, v in manifest_raw.items():
1279 if isinstance(k, str) and isinstance(v, str):
1280 manifest[k] = v
1281 # ``directories`` is hashed into snapshot_id — dropping it produces a
1282 # SnapshotRecord whose snapshot_id never matches the recomputed hash,
1283 # making every snapshot with non-empty directories permanently unreadable.
1284 directories_raw = raw.get("directories")
1285 directories: list[str] = []
1286 if isinstance(directories_raw, list):
1287 for item in directories_raw:
1288 if isinstance(item, str):
1289 directories.append(item)
1290 return SnapshotDict(
1291 snapshot_id=_str(raw.get("snapshot_id")),
1292 manifest=manifest,
1293 directories=directories,
1294 created_at=_str(raw.get("created_at")),
1295 note=_str(raw.get("note", "")),
1296 )
1297
1298
1299 # ---------------------------------------------------------------------------
1300 # Factory
1301 # ---------------------------------------------------------------------------
1302
1303 def make_transport(url: str) -> "HttpTransport":
1304 """Return an :class:`HttpTransport` for *url*.
1305
1306 Args:
1307 url: Remote repository URL.
1308
1309 Returns:
1310 An ``HttpTransport`` instance implementing :class:`MuseTransport`.
1311 """
1312 return HttpTransport()
File History 8 commits
sha256:2eaa5d95f9d9383498e76947410a26e5a3ba23d182f339910c424cf88fad412b fix: try fetch/presign before fetch/mpack to avoid Cloudfla… Sonnet 4.6 patch 44 days ago
sha256:6b91dc90e1c59c5209b764a276c1ee824bd369320a454a403d28ce79890103f2 fix: stream mpack downloads without size cap Sonnet 4.6 minor ⚠ 48 days ago
sha256:f6cd81bc71702f5c1c6890bd39aaba994fe58c75f019d7c03934724fa2739bb4 fix: carry dev changes harmony dropped in merge — detached … Sonnet 4.6 minor ⚠ 53 days ago
sha256:b37d33a9dbf176b32955b34e5c9e983c4d8c6e7bfa4e2714edc938f31e721561 update transport logs Human minor ⚠ 55 days ago
sha256:79ffe87f5fe2ec146e35f05521218bbf54dffdb0440c07f970bad05f16efb89f chore: merge main — carry all urllib/typing/test fixes from dev Sonnet 4.6 minor ⚠ 57 days ago
sha256:0bea7600d1eee83e87950be49933b1006fa9dc2c71e7c4ee748d324f61138156 chore: bump version to 0.2.0rc11; fix typing audit violatio… Sonnet 4.6 minor ⚠ 57 days ago
sha256:633dfa2940e97bf1a3d04996c772027a57d70d103f1693c96da04969613dba6c fix: urllib migration regressions — force flag, job_id, Con… Sonnet 4.6 minor ⚠ 57 days ago
sha256:00cec040ce5f70bf8191d2ce6a9f308fbde553911068f0c303217f4eb6d4e775 fix: migrate httpx → urllib in transport.py and push.py; fi… Sonnet 4.6 minor ⚠ 57 days ago