wire.py
python
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a
feat(intel): standardize headers, gauge icon, velocity card…
Sonnet 4.6
minor
⚠ breaking
143 days ago
| 1 | """Wire protocol Pydantic models — Muse CLI native format (MWP — Muse Wire Protocol). |
| 2 | |
| 3 | These models match the Muse CLI ``HttpTransport`` wire format exactly. |
| 4 | All fields are snake_case to match Muse's internal CommitDict/SnapshotDict/ |
| 5 | ObjectPayload TypedDicts. |
| 6 | |
| 7 | The wire protocol is intentionally separate from the REST API's CamelModel: |
| 8 | Wire protocol /wire/repos/{repo_id}/ ← Muse CLI speaks here (MWP, msgpack) |
| 9 | REST API /api/repos/{id}/ ← agents and integrations speak here |
| 10 | MCP /mcp ← agents speak here too |
| 11 | |
| 12 | Encoding |
| 13 | -------- |
| 14 | All wire endpoints accept and return ``application/x-msgpack`` binary. |
| 15 | Objects are transported as raw ``bytes`` under the ``content`` key — no |
| 16 | base64 encoding overhead. |
| 17 | |
| 18 | Denial-of-Service limits |
| 19 | ------------------------ |
| 20 | All list fields that arrive over the network are capped so a single large |
| 21 | request cannot exhaust memory or DB connections: |
| 22 | |
| 23 | MAX_COMMITS_PER_PUSH = 10 000 — one push should carry at most 10k commits |
| 24 | MAX_OBJECTS_PER_PUSH = 1 000 — ditto for binary blobs per chunk |
| 25 | MAX_SNAPSHOTS_PER_PUSH = 10 000 — ditto for snapshot manifests |
| 26 | MAX_WANT_PER_FETCH = 1 000 — fetch want/have lists |
| 27 | MAX_OBJECT_BYTES = 38_000_000 — ~38 MB raw; objects above this limit are rejected |
| 28 | """ |
| 29 | from __future__ import annotations |
| 30 | |
| 31 | import re |
| 32 | |
| 33 | from pydantic import BaseModel, Field, field_validator, model_validator |
| 34 | |
| 35 | from musehub.types.json_types import JSONObject, StrDict |
| 36 | from musehub.types.pydantic_types import PydanticJson |
| 37 | |
| 38 | type _SizeMap = dict[str, int] |
| 39 | |
| 40 | # ── Per-request DoS limits ──────────────────────────────────────────────────── |
| 41 | MAX_COMMITS_PER_PUSH: int = 10_000 |
| 42 | MAX_OBJECTS_PER_PUSH: int = 1_000 |
| 43 | |
| 44 | # ── Object ID validation ────────────────────────────────────────────────────── |
| 45 | # object_id values arrive from untrusted clients and are used to construct |
| 46 | # storage keys (S3/R2 object paths). A malicious value containing '/' or '..' |
| 47 | # could escape the objects/ key namespace and overwrite arbitrary R2 keys. |
| 48 | # |
| 49 | # Valid format: <algo>:<lowercase hex digest> |
| 50 | # - algo : lower-case alphanumeric only (e.g. "sha256", "blake3") — no slashes |
| 51 | # - digest: lowercase hex, at least 32 chars (128-bit minimum) |
| 52 | # Raw hex (no prefix) is rejected — the algo: prefix is mandatory everywhere. |
| 53 | # The pattern is intentionally algo-agnostic so future hash upgrades (blake3, |
| 54 | # sha3-256, …) require no validator change. The hex-only digest ensures no |
| 55 | # path-traversal characters ('.', '/') can appear in the storage key. |
| 56 | _OBJECT_ID_RE: re.Pattern[str] = re.compile(r"^[a-z][a-z0-9]*:[0-9a-f]{32,}$") |
| 57 | _OBJECT_ID_MAX_LEN: int = 200 # generous cap; algo(<=16) + ":" + digest(<=128) = 145 |
| 58 | |
| 59 | |
| 60 | def _validate_object_ids(ids: list[str]) -> list[str]: |
| 61 | """Raise ValueError for any object_id that contains unsafe characters.""" |
| 62 | for oid in ids: |
| 63 | if not _OBJECT_ID_RE.match(oid): |
| 64 | raise ValueError( |
| 65 | f"invalid object_id {oid!r}: only [a-zA-Z0-9:_-] characters are allowed" |
| 66 | ) |
| 67 | if len(oid) > _OBJECT_ID_MAX_LEN: |
| 68 | raise ValueError( |
| 69 | f"object_id exceeds maximum length ({_OBJECT_ID_MAX_LEN}): {oid[:40]!r}…" |
| 70 | ) |
| 71 | return ids |
| 72 | MAX_SNAPSHOTS_PER_PUSH: int = 10_000 |
| 73 | MAX_WANT_PER_FETCH: int = 1_000 |
| 74 | # Raw bytes limit per object — objects above this are rejected at the wire layer. |
| 75 | MAX_OBJECT_BYTES: int = 38_000_000 |
| 76 | |
| 77 | |
| 78 | class WireCommit(BaseModel): |
| 79 | """Muse native commit record — mirrors CommitDict from muse.core.store.""" |
| 80 | |
| 81 | commit_id: str |
| 82 | repo_id: str = "" |
| 83 | branch: str = "" |
| 84 | snapshot_id: str | None = None |
| 85 | message: str = "" |
| 86 | committed_at: str = "" # ISO-8601 UTC string |
| 87 | parent_commit_id: str | None = None # first parent (linear history) |
| 88 | parent2_commit_id: str | None = None # second parent (merge commits) |
| 89 | author: str = "" |
| 90 | metadata: StrDict = Field(default_factory=dict) |
| 91 | structured_delta: PydanticJson | None = None # domain-specific delta blob |
| 92 | sem_ver_bump: str = "none" # "none" | "patch" | "minor" | "major" |
| 93 | breaking_changes: list[str] = Field(default_factory=list) |
| 94 | agent_id: str = "" |
| 95 | model_id: str = "" |
| 96 | toolchain_id: str = "" |
| 97 | prompt_hash: str = "" |
| 98 | signature: str = "" |
| 99 | signer_public_key: str = "" |
| 100 | signer_key_id: str = "" |
| 101 | format_version: int = 7 |
| 102 | reviewed_by: list[str] = Field(default_factory=list) |
| 103 | test_runs: int = 0 |
| 104 | |
| 105 | model_config = {"extra": "ignore"} # tolerate future Muse fields gracefully |
| 106 | |
| 107 | @field_validator("commit_id") |
| 108 | @classmethod |
| 109 | def _check_commit_id(cls, v: str) -> str: |
| 110 | if not _OBJECT_ID_RE.match(v): |
| 111 | raise ValueError( |
| 112 | f"invalid commit_id {v!r}: must be 'sha256:<64 lowercase hex chars>'" |
| 113 | ) |
| 114 | return v |
| 115 | |
| 116 | @field_validator("snapshot_id") |
| 117 | @classmethod |
| 118 | def _check_snapshot_id(cls, v: str | None) -> str | None: |
| 119 | if v is not None and not _OBJECT_ID_RE.match(v): |
| 120 | raise ValueError( |
| 121 | f"invalid snapshot_id {v!r}: must be 'sha256:<64 lowercase hex chars>'" |
| 122 | ) |
| 123 | return v |
| 124 | |
| 125 | @field_validator("parent_commit_id") |
| 126 | @classmethod |
| 127 | def _check_parent_commit_id(cls, v: str | None) -> str | None: |
| 128 | if v is not None and not _OBJECT_ID_RE.match(v): |
| 129 | raise ValueError( |
| 130 | f"invalid parent_commit_id {v!r}: must be 'sha256:<64 lowercase hex chars>'" |
| 131 | ) |
| 132 | return v |
| 133 | |
| 134 | |
| 135 | class WireSnapshot(BaseModel): |
| 136 | """Muse native snapshot — mirrors SnapshotDict from muse.core.store. |
| 137 | |
| 138 | The manifest maps file paths to content-addressed object IDs, |
| 139 | e.g. ``{"muse/core/pack.py": "sha256:abc123..."}``. |
| 140 | |
| 141 | ``directories`` is the sorted list of workspace-relative directory paths |
| 142 | that were explicitly tracked at snapshot time. It is included in the |
| 143 | snapshot ID hash — omitting it causes every snapshot with non-empty |
| 144 | directories to fail content-hash verification on clone. |
| 145 | """ |
| 146 | |
| 147 | snapshot_id: str |
| 148 | # max_length caps the number of manifest entries — a 10 000-file snapshot |
| 149 | # would already be pathologically large; prevent unbounded dict parsing. |
| 150 | manifest: StrDict = Field(default_factory=dict, max_length=10_000) |
| 151 | directories: list[str] = Field(default_factory=list, max_length=10_000) |
| 152 | created_at: str = "" |
| 153 | |
| 154 | model_config = {"extra": "ignore"} |
| 155 | |
| 156 | @field_validator("snapshot_id") |
| 157 | @classmethod |
| 158 | def _check_snapshot_id(cls, v: str) -> str: |
| 159 | if not _OBJECT_ID_RE.match(v): |
| 160 | raise ValueError( |
| 161 | f"invalid snapshot_id {v!r}: must be 'sha256:<64 lowercase hex chars>'" |
| 162 | ) |
| 163 | return v |
| 164 | |
| 165 | |
| 166 | class WireSnapshotDelta(BaseModel): |
| 167 | """Delta-encoded snapshot — reconstructed by applying added/removed to a base manifest. |
| 168 | |
| 169 | Sent in C frames as ``snapshot_deltas`` alongside (or instead of) full |
| 170 | ``snapshots``. The server reconstructs the full manifest by looking up |
| 171 | ``base_id`` in its stream-local cache and applying the delta. |
| 172 | |
| 173 | Wire format:: |
| 174 | |
| 175 | { |
| 176 | "snapshot_id": "sha256:<64hex>", |
| 177 | "base_id": "sha256:<64hex>", # snapshot already seen in this stream |
| 178 | "added": {"path": "sha256:<64hex>", ...}, # new + modified files |
| 179 | "removed": ["path", ...], # deleted files |
| 180 | "directories": [...], |
| 181 | "created_at": "...", |
| 182 | } |
| 183 | """ |
| 184 | |
| 185 | snapshot_id: str |
| 186 | base_id: str |
| 187 | added: StrDict = Field(default_factory=dict, max_length=10_000) |
| 188 | removed: list[str] = Field(default_factory=list, max_length=10_000) |
| 189 | directories: list[str] = Field(default_factory=list, max_length=10_000) |
| 190 | created_at: str = "" |
| 191 | |
| 192 | model_config = {"extra": "ignore"} |
| 193 | |
| 194 | @field_validator("snapshot_id", "base_id") |
| 195 | @classmethod |
| 196 | def _check_id(cls, v: str) -> str: |
| 197 | if not _OBJECT_ID_RE.match(v): |
| 198 | raise ValueError( |
| 199 | f"invalid snapshot id {v!r}: must be 'sha256:<64 lowercase hex chars>'" |
| 200 | ) |
| 201 | return v |
| 202 | |
| 203 | |
| 204 | class WireObject(BaseModel): |
| 205 | """Content-addressed object payload — mirrors ObjectPayload from muse.core.pack. |
| 206 | |
| 207 | MWP encodes ``content`` as raw bytes (msgpack bin type) — no base64 overhead. |
| 208 | |
| 209 | Encoding field controls how the server interprets ``content``: |
| 210 | ``"raw"`` — plain bytes; store as-is after hash verification. |
| 211 | ``"zlib"`` — zlib-compressed; decompress then verify hash. |
| 212 | ``"delta+zlib"`` — delta-encoded relative to ``base_id``, then zlib-compressed; |
| 213 | fetch base, apply delta, then verify hash. |
| 214 | """ |
| 215 | |
| 216 | object_id: str |
| 217 | content: bytes = Field(max_length=MAX_OBJECT_BYTES) |
| 218 | path: str = Field(default="", max_length=4096) |
| 219 | encoding: str = Field(default="raw") |
| 220 | base_id: str | None = Field(default=None) |
| 221 | |
| 222 | model_config = {"extra": "ignore"} |
| 223 | |
| 224 | @field_validator("object_id") |
| 225 | @classmethod |
| 226 | def _check_object_id(cls, v: str) -> str: |
| 227 | if not _OBJECT_ID_RE.match(v): |
| 228 | raise ValueError( |
| 229 | f"invalid object_id {v!r}: must be 'sha256:<64 lowercase hex chars>'" |
| 230 | ) |
| 231 | return v |
| 232 | |
| 233 | @field_validator("content") |
| 234 | @classmethod |
| 235 | def _check_content_size(cls, v: bytes) -> bytes: |
| 236 | if len(v) > MAX_OBJECT_BYTES: |
| 237 | raise ValueError( |
| 238 | f"content exceeds maximum size ({MAX_OBJECT_BYTES} bytes)." |
| 239 | ) |
| 240 | return v |
| 241 | |
| 242 | |
| 243 | class WireBundle(BaseModel): |
| 244 | """A pack bundle sent in a push request. |
| 245 | |
| 246 | Mirrors PackBundle from muse.core.pack. All fields are optional because |
| 247 | a minimal push may only contain commits (no new objects). |
| 248 | |
| 249 | List lengths are capped to prevent DoS via an oversized single request. |
| 250 | See the module-level ``MAX_*`` constants for the exact limits. |
| 251 | """ |
| 252 | |
| 253 | commits: list[WireCommit] = Field(default_factory=list, max_length=MAX_COMMITS_PER_PUSH) |
| 254 | snapshots: list[WireSnapshot] = Field(default_factory=list, max_length=MAX_SNAPSHOTS_PER_PUSH) |
| 255 | objects: list[WireObject] = Field(default_factory=list, max_length=MAX_OBJECTS_PER_PUSH) |
| 256 | branch_heads: StrDict = Field(default_factory=dict) |
| 257 | |
| 258 | |
| 259 | class WireFetchRequest(BaseModel): |
| 260 | """Body for ``POST /wire/repos/{repo_id}/fetch``. |
| 261 | |
| 262 | Matches HttpTransport.fetch_pack() payload: |
| 263 | ``{"want": [...sha...], "have": [...sha...]}`` |
| 264 | |
| 265 | ``want`` — commit SHAs the client wants. |
| 266 | ``have`` — commit SHAs the client already has (exclusion list). |
| 267 | """ |
| 268 | |
| 269 | want: list[str] = Field(default_factory=list, max_length=MAX_WANT_PER_FETCH) |
| 270 | have: list[str] = Field(default_factory=list, max_length=MAX_WANT_PER_FETCH) |
| 271 | depth: int | None = Field(default=None, ge=1) |
| 272 | |
| 273 | @field_validator("want", "have") |
| 274 | @classmethod |
| 275 | def _check_commit_ids(cls, v: list[str]) -> list[str]: |
| 276 | return _validate_object_ids(v) |
| 277 | |
| 278 | |
| 279 | class WireRefsResponse(BaseModel): |
| 280 | """Response for ``GET /wire/repos/{repo_id}/refs``. |
| 281 | |
| 282 | Parsed by HttpTransport._parse_remote_info() into RemoteInfo. |
| 283 | """ |
| 284 | |
| 285 | repo_id: str |
| 286 | domain: str |
| 287 | default_branch: str |
| 288 | branch_heads: StrDict |
| 289 | |
| 290 | |
| 291 | class WireNegotiateRequest(BaseModel): |
| 292 | """Body for ``POST /{owner}/{slug}/negotiate`` — commit negotiation.""" |
| 293 | have: list[str] = [] |
| 294 | want: list[str] = [] |
| 295 | |
| 296 | @field_validator("have", "want") |
| 297 | @classmethod |
| 298 | def _check_commit_ids(cls, v: list[str]) -> list[str]: |
| 299 | return _validate_object_ids(v) |
| 300 | |
| 301 | |
| 302 | class WireNegotiateResponse(BaseModel): |
| 303 | """Response for ``POST /{owner}/{slug}/negotiate`` — commit negotiation.""" |
| 304 | ack: list[str] |
| 305 | common_base: str | None |
| 306 | ready: bool |
| 307 | |
| 308 | |
| 309 | # ───────────────────────────────────────────────────────────────────────────── |
| 310 | # MWP — Streaming Push Protocol |
| 311 | # ───────────────────────────────────────────────────────────────────────────── |
| 312 | # |
| 313 | # Two push paths: stream (small) and presigned R2 (large). |
| 314 | # This is the server-side equivalent of |
| 315 | # git-receive-pack's smart HTTP protocol: |
| 316 | # |
| 317 | # git smart HTTP: Muse Wire Protocol (MWP): |
| 318 | # ───────────────────── ────────────────────────────── |
| 319 | # GET /info/refs GET /{owner}/{slug}/refs (unchanged) |
| 320 | # POST /git-receive-pack POST /{owner}/{slug}/push/stream (single streaming request) |
| 321 | # |
| 322 | # Request body — concatenated self-delimiting msgpack values (no length prefix; |
| 323 | # msgpack.Unpacker handles framing exactly like the existing /fetch/objects stream): |
| 324 | # |
| 325 | # frame 1: HEADER frame — branch, force, have-list, object/commit counts |
| 326 | # frame 2…N: OBJECT frames — content-addressed binary objects (zlib-compressed) |
| 327 | # frame N+1: COMMIT_PACK — commits and snapshot manifests |
| 328 | # frame N+2: END — stream terminator |
| 329 | # |
| 330 | # Response body — same wire format (concatenated msgpack dicts): |
| 331 | # |
| 332 | # frame(s): PROGRESS — human-readable progress (print to stderr) |
| 333 | # frame: ERROR — fatal; client aborts, push failed |
| 334 | # frame: RESULT — final push result; stream ends |
| 335 | # |
| 336 | # Content-Type: application/x-muse-wire (both request and response) |
| 337 | # |
| 338 | # ── Frame type tag constants ────────────────────────────────────────────────── |
| 339 | |
| 340 | # Request frame types (client → server) — single-char tags keep wire overhead minimal. |
| 341 | SFRAME_HEADER: str = "H" # opens stream; branch, force, have, object/commit count |
| 342 | SFRAME_OBJECT: str = "O" # one content-addressed binary object |
| 343 | SFRAME_OBJECT_CHUNK: str = "OC" # one chunk of a large object (Phase 14B) |
| 344 | SFRAME_COMMIT_PACK: str = "C" # final batch of commits + snapshot manifests |
| 345 | SFRAME_END: str = "E" # stream terminator |
| 346 | |
| 347 | # Response frame types (server → client) |
| 348 | SFRAME_PROGRESS: str = "P" # human-readable progress (sideband 2 — stderr) |
| 349 | SFRAME_ERROR: str = "X" # fatal error — client aborts; no RESULT follows |
| 350 | SFRAME_RESULT: str = "R" # final push result — always the last frame on success |
| 351 | |
| 352 | # ── Per-stream DoS limits ───────────────────────────────────────────────────── |
| 353 | # |
| 354 | # These are deliberately generous: a single streaming push replaces what used to |
| 355 | # be dozens of individual HTTP requests. The server enforces hard caps so an |
| 356 | # adversarial client cannot exhaust memory or storage by sending an unbounded stream. |
| 357 | |
| 358 | STREAM_MAX_OBJECTS: int = 500_000 # 500k objects per single push stream |
| 359 | STREAM_MAX_COMMITS: int = 1_000_000 # 1M commits per single push stream |
| 360 | |
| 361 | # Maximum wire bytes the server accepts per OBJECT frame's content field. |
| 362 | # Applies to the *compressed* wire bytes; the server decompresses before storing. |
| 363 | STREAM_MAX_OBJECT_WIRE_BYTES: int = 50_000_000 # 50 MB compressed per object |
| 364 | |
| 365 | # R2 PUT batch size during stream ingestion — flush every N incoming OBJECT frames. |
| 366 | # Must not exceed _R2_PUT_SEM capacity (100) in musehub_wire.py. |
| 367 | STREAM_OBJECT_BATCH_SIZE: int = 64 |
File History
1 commit
sha256:bb2baaabdd19320bde50cb69d447fd1c1e571467df729be1a23b5e93064b046a
feat(intel): standardize headers, gauge icon, velocity card…
Sonnet 4.6
minor
⚠
143 days ago