coord_bus.py
python
sha256:649be9ce77a127cd847c4f798d60104dab7fe830cbc928762fafe3deadf38839
chore: point muse-git-backup.sh's MUSE_COPY at the fresh 'm…
Sonnet 5
patch
13 days ago
| 1 | """Muse coordination bus client — push/pull coordination records to/from MuseHub. |
| 2 | |
| 3 | Provides synchronous HTTP operations for syncing local coordination state with |
| 4 | a MuseHub remote so that agent swarms on different machines share state. |
| 5 | |
| 6 | The client re-uses the transport layer's security model: |
| 7 | - All HTTP redirects are refused (credentials must not follow redirects). |
| 8 | - The signing identity is never logged. |
| 9 | - Responses are capped at :data:`MAX_COORD_RESPONSE_BYTES` to prevent OOM. |
| 10 | - HTTP is accepted only when no signing identity is supplied (public repos); |
| 11 | HTTPS is required when a signing identity is present. |
| 12 | |
| 13 | Protocol |
| 14 | -------- |
| 15 | Push (POST ``/{owner}/{slug}/coord/push``): |
| 16 | - Request body: JSON ``{"records": [...]}`` |
| 17 | - Response: JSON ``{"inserted": N, "skipped": M}`` |
| 18 | |
| 19 | Pull (POST ``/{owner}/{slug}/coord/pull``): |
| 20 | - Request body: JSON ``{"since_id": N, "kinds": [...], "limit": M}`` |
| 21 | - Response: JSON ``{"records": [...], "cursor": N}`` |
| 22 | |
| 23 | Both endpoints use JSON (not msgpack) — coordination records are small and |
| 24 | infrequent compared to object-store traffic. stdlib ``json`` is sufficient. |
| 25 | |
| 26 | Error handling |
| 27 | -------------- |
| 28 | All errors raise :class:`CoordBusError` with a human-readable message. The |
| 29 | caller (CLI command) is responsible for printing the error and exiting with a |
| 30 | non-zero status code. |
| 31 | |
| 32 | Security |
| 33 | -------- |
| 34 | - ``owner`` and ``slug`` are URL-path components — they are %-encoded by |
| 35 | ``urllib.parse.quote`` to prevent path traversal. |
| 36 | - The signing identity is used in the ``Authorization`` header only, never in the URL. |
| 37 | - Response bodies are read into memory only up to ``MAX_COORD_RESPONSE_BYTES``. |
| 38 | """ |
| 39 | |
| 40 | import http.client |
| 41 | import json |
| 42 | import logging |
| 43 | import urllib.error |
| 44 | import urllib.parse |
| 45 | import urllib.request |
| 46 | from typing import TYPE_CHECKING |
| 47 | |
| 48 | from muse.core.validation import sanitize_display |
| 49 | |
| 50 | type _SummaryMap = dict[str, int | list["JsonDict"]] |
| 51 | type _IntMap = dict[str, int] |
| 52 | if TYPE_CHECKING: |
| 53 | from muse.core.transport import SigningIdentity |
| 54 | |
| 55 | # JSON-compatible value type for coord request/response bodies. |
| 56 | JsonValue = str | int | float | bool | None | list["JsonValue"] | "JsonDict" |
| 57 | JsonDict = dict[str, JsonValue] |
| 58 | |
| 59 | logger = logging.getLogger(__name__) |
| 60 | |
| 61 | # Maximum response body size accepted from the hub (4 MiB). |
| 62 | # Coordination records are small; 4 MiB accommodates ≈ 8,000 full records. |
| 63 | MAX_COORD_RESPONSE_BYTES: int = 4 * 1024 * 1024 |
| 64 | |
| 65 | # Request/response timeout in seconds for coordination HTTP calls. |
| 66 | _TIMEOUT_SECONDS: int = 30 |
| 67 | |
| 68 | # Maximum records per push call (must stay ≤ server limit of 500). |
| 69 | MAX_PUSH_BATCH: int = 500 |
| 70 | |
| 71 | # Maximum records per pull call (must stay ≤ server limit of 1000). |
| 72 | MAX_PULL_LIMIT: int = 1000 |
| 73 | |
| 74 | # ── Exceptions ───────────────────────────────────────────────────────────────── |
| 75 | |
| 76 | class CoordBusError(Exception): |
| 77 | """Raised when a coordination bus HTTP operation fails. |
| 78 | |
| 79 | Attributes: |
| 80 | status_code: HTTP status code, or 0 for network-level errors. |
| 81 | """ |
| 82 | |
| 83 | def __init__(self, message: str, status_code: int = 0) -> None: |
| 84 | super().__init__(message) |
| 85 | self.status_code = status_code |
| 86 | |
| 87 | # ── Transport helpers (reuse pattern from transport.py) ──────────────────────── |
| 88 | |
| 89 | class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): |
| 90 | """Refuse all HTTP redirects to prevent credential leakage.""" |
| 91 | |
| 92 | def redirect_request( |
| 93 | self, |
| 94 | req: urllib.request.Request, |
| 95 | fp: http.client.HTTPResponse | None, |
| 96 | code: int, |
| 97 | msg: str, |
| 98 | headers: http.client.HTTPMessage, |
| 99 | newurl: str, |
| 100 | ) -> urllib.request.Request | None: |
| 101 | raise urllib.error.HTTPError( |
| 102 | req.full_url, |
| 103 | code, |
| 104 | ( |
| 105 | f"Redirect refused ({code}): server tried to redirect to {newurl!r}. " |
| 106 | "Update the configured remote URL to the final destination." |
| 107 | ), |
| 108 | headers, |
| 109 | fp, |
| 110 | ) |
| 111 | |
| 112 | _STRICT_OPENER = urllib.request.build_opener(_NoRedirectHandler()) |
| 113 | |
| 114 | def _http_error_message(exc: urllib.error.HTTPError) -> str: |
| 115 | """Build a safe, sanitized error message — never exposes response body on 401.""" |
| 116 | if exc.code == 401: |
| 117 | return "Authentication failed (HTTP 401). Run 'muse auth register'." |
| 118 | try: |
| 119 | raw_body = exc.read().decode("utf-8", errors="replace") |
| 120 | except Exception: # noqa: BLE001 |
| 121 | raw_body = "" |
| 122 | safe_body = sanitize_display(raw_body[:200]) |
| 123 | return f"HTTP {exc.code}: {safe_body}" if safe_body else f"HTTP {exc.code}" |
| 124 | |
| 125 | def _build_url(hub_url: str, owner: str, slug: str, endpoint: str) -> str: |
| 126 | """Build a %-encoded URL for a coord endpoint. |
| 127 | |
| 128 | Args: |
| 129 | hub_url: Hub base URL (e.g. ``https://localhost:1337``). |
| 130 | owner: Repo owner — %-encoded to prevent path traversal. |
| 131 | slug: Repo slug — %-encoded to prevent path traversal. |
| 132 | endpoint: Endpoint suffix (``coord/push``, ``coord/pull``). |
| 133 | |
| 134 | Returns: |
| 135 | Full URL string. |
| 136 | """ |
| 137 | safe_owner = urllib.parse.quote(owner, safe="") |
| 138 | safe_slug = urllib.parse.quote(slug, safe="") |
| 139 | return f"{hub_url.rstrip('/')}/{safe_owner}/{safe_slug}/{endpoint}" |
| 140 | |
| 141 | def _post_json( |
| 142 | url: str, |
| 143 | body: JsonDict, |
| 144 | signing: "SigningIdentity | None", |
| 145 | ) -> JsonDict: |
| 146 | """POST a JSON body to *url* and return the parsed JSON response. |
| 147 | |
| 148 | Args: |
| 149 | url: Full target URL. |
| 150 | body: Request body dict — encoded as UTF-8 JSON. |
| 151 | signing: :class:`~muse.core.transport.SigningIdentity`, or ``None`` |
| 152 | for unauthenticated calls. |
| 153 | |
| 154 | Returns: |
| 155 | Parsed JSON response dict. |
| 156 | |
| 157 | Raises: |
| 158 | :class:`CoordBusError` on HTTP error or network failure. |
| 159 | """ |
| 160 | from muse.core.transport import SigningIdentity |
| 161 | from muse.core.msign import build_msign_header |
| 162 | |
| 163 | data = json.dumps(body).encode("utf-8") |
| 164 | req = urllib.request.Request( |
| 165 | url, |
| 166 | data=data, |
| 167 | method="POST", |
| 168 | headers={"Content-Type": "application/json"}, |
| 169 | ) |
| 170 | if isinstance(signing, SigningIdentity): |
| 171 | req.add_header("Authorization", build_msign_header(signing, "POST", url, data)) |
| 172 | |
| 173 | try: |
| 174 | with _STRICT_OPENER.open(req, timeout=_TIMEOUT_SECONDS) as resp: |
| 175 | raw = resp.read(MAX_COORD_RESPONSE_BYTES + 1) |
| 176 | if len(raw) > MAX_COORD_RESPONSE_BYTES: |
| 177 | raise CoordBusError( |
| 178 | f"Response body exceeded {MAX_COORD_RESPONSE_BYTES} bytes limit.", |
| 179 | status_code=0, |
| 180 | ) |
| 181 | return json.loads(raw) |
| 182 | except urllib.error.HTTPError as exc: |
| 183 | raise CoordBusError(_http_error_message(exc), status_code=exc.code) from exc |
| 184 | except urllib.error.URLError as exc: |
| 185 | safe = sanitize_display(str(exc.reason)[:200]) |
| 186 | raise CoordBusError(f"Network error: {safe}", status_code=0) from exc |
| 187 | except (json.JSONDecodeError, ValueError) as exc: |
| 188 | raise CoordBusError(f"Invalid JSON response: {exc}", status_code=0) from exc |
| 189 | |
| 190 | # ── Public API ───────────────────────────────────────────────────────────────── |
| 191 | |
| 192 | def push_to_hub( |
| 193 | hub_url: str, |
| 194 | owner: str, |
| 195 | slug: str, |
| 196 | records: list[JsonDict], |
| 197 | signing: SigningIdentity | None = None, |
| 198 | ) -> _IntMap: |
| 199 | """Push coordination records to MuseHub. |
| 200 | |
| 201 | Args: |
| 202 | hub_url: Hub base URL (e.g. ``https://localhost:1337``). |
| 203 | owner: Repo owner username. |
| 204 | slug: Repo slug. |
| 205 | records: List of coordination record dicts. Each dict must have: |
| 206 | ``kind``, ``record_id``, ``run_id``, ``payload``. |
| 207 | ``expires_at`` is optional (ISO-8601 string or ``None``). |
| 208 | At most :data:`MAX_PUSH_BATCH` records per call. |
| 209 | signing: :class:`~muse.core.transport.SigningIdentity` (required — push always needs auth). |
| 210 | |
| 211 | Returns: |
| 212 | Dict with ``"inserted"`` and ``"skipped"`` counts. |
| 213 | |
| 214 | Raises: |
| 215 | :class:`CoordBusError` on failure. |
| 216 | :class:`ValueError` if *records* is empty or exceeds :data:`MAX_PUSH_BATCH`. |
| 217 | """ |
| 218 | if not records: |
| 219 | raise ValueError("records must be non-empty") |
| 220 | if len(records) > MAX_PUSH_BATCH: |
| 221 | raise ValueError( |
| 222 | f"records exceeds maximum batch size of {MAX_PUSH_BATCH}; " |
| 223 | "split into smaller batches" |
| 224 | ) |
| 225 | |
| 226 | url = _build_url(hub_url, owner, slug, "coord/push") |
| 227 | logger.debug( |
| 228 | "coord_bus.push_to_hub: pushing %d record(s) to %s/%s", |
| 229 | len(records), |
| 230 | owner, |
| 231 | slug, |
| 232 | ) |
| 233 | result = _post_json(url, {"records": records}, signing) |
| 234 | |
| 235 | def _parse_count(field: str) -> int: |
| 236 | if field not in result: |
| 237 | return 0 # Key absent — treat as zero (hub may omit on partial success) |
| 238 | raw = result[field] |
| 239 | if raw is None: |
| 240 | raise CoordBusError( |
| 241 | f"Hub returned null {field!r} count", status_code=0 |
| 242 | ) |
| 243 | try: |
| 244 | n = int(raw) |
| 245 | except (TypeError, ValueError) as exc: |
| 246 | raise CoordBusError( |
| 247 | f"Hub returned non-integer {field!r} count", status_code=0 |
| 248 | ) from exc |
| 249 | if n < 0: |
| 250 | raise CoordBusError( |
| 251 | f"Hub returned negative {field!r} count: {n}", status_code=0 |
| 252 | ) |
| 253 | if n > len(records): |
| 254 | raise CoordBusError( |
| 255 | f"Hub claimed {field!r}={n} but only {len(records)} records were sent", |
| 256 | status_code=0, |
| 257 | ) |
| 258 | return n |
| 259 | |
| 260 | return { |
| 261 | "inserted": _parse_count("inserted"), |
| 262 | "skipped": _parse_count("skipped"), |
| 263 | } |
| 264 | |
| 265 | def pull_from_hub( |
| 266 | hub_url: str, |
| 267 | owner: str, |
| 268 | slug: str, |
| 269 | since_id: int = 0, |
| 270 | kinds: list[str] | None = None, |
| 271 | limit: int = 500, |
| 272 | signing: SigningIdentity | None = None, |
| 273 | ) -> _SummaryMap: |
| 274 | """Pull coordination records from MuseHub since *since_id*. |
| 275 | |
| 276 | Args: |
| 277 | hub_url: Hub base URL. |
| 278 | owner: Repo owner username. |
| 279 | slug: Repo slug. |
| 280 | since_id: Return records with ``id > since_id``. ``0`` = all records. |
| 281 | kinds: Filter by record kind. ``None`` or ``[]`` = all kinds. |
| 282 | limit: Maximum records to return (1–:data:`MAX_PULL_LIMIT`). |
| 283 | signing: :class:`~muse.core.transport.SigningIdentity` (required for private repos). |
| 284 | |
| 285 | Returns: |
| 286 | Dict with: |
| 287 | - ``"records"``: list of record dicts. |
| 288 | - ``"cursor"``: int — the ``id`` of the last returned record |
| 289 | (pass as ``since_id`` in the next call). |
| 290 | |
| 291 | Raises: |
| 292 | :class:`CoordBusError` on failure. |
| 293 | :class:`ValueError` if *limit* is out of range. |
| 294 | """ |
| 295 | if not (1 <= limit <= MAX_PULL_LIMIT): |
| 296 | raise ValueError(f"limit must be 1–{MAX_PULL_LIMIT}, got {limit}") |
| 297 | |
| 298 | url = _build_url(hub_url, owner, slug, "coord/pull") |
| 299 | body: JsonDict = { |
| 300 | "since_id": since_id, |
| 301 | "kinds": kinds or [], |
| 302 | "limit": limit, |
| 303 | } |
| 304 | logger.debug( |
| 305 | "coord_bus.pull_from_hub: pulling from %s/%s since_id=%d", |
| 306 | owner, |
| 307 | slug, |
| 308 | since_id, |
| 309 | ) |
| 310 | raw = _post_json(url, body, signing) |
| 311 | |
| 312 | # --- Validate and normalise records --- |
| 313 | raw_records = raw.get("records") |
| 314 | if raw_records is None and "records" in raw: |
| 315 | raise CoordBusError("Hub returned null 'records' list", status_code=0) |
| 316 | if raw_records is not None and not isinstance(raw_records, list): |
| 317 | raise CoordBusError( |
| 318 | f"Hub returned non-list 'records': {type(raw_records).__name__}", |
| 319 | status_code=0, |
| 320 | ) |
| 321 | records: list[JsonDict] = raw_records if raw_records is not None else [] |
| 322 | for item in records: |
| 323 | if not isinstance(item, dict): |
| 324 | raise CoordBusError( |
| 325 | f"Hub returned non-dict record in 'records': {type(item).__name__}", |
| 326 | status_code=0, |
| 327 | ) |
| 328 | |
| 329 | # --- Validate and normalise cursor --- |
| 330 | raw_cursor = raw.get("cursor") |
| 331 | if raw_cursor is None and "cursor" in raw: |
| 332 | raise CoordBusError("Hub returned null 'cursor'", status_code=0) |
| 333 | if raw_cursor is None: |
| 334 | cursor: int = 0 |
| 335 | else: |
| 336 | try: |
| 337 | cursor = int(raw_cursor) |
| 338 | except (TypeError, ValueError) as exc: |
| 339 | raise CoordBusError( |
| 340 | "Hub returned non-integer 'cursor'", status_code=0 |
| 341 | ) from exc |
| 342 | if cursor < 0: |
| 343 | raise CoordBusError( |
| 344 | f"Hub returned negative 'cursor': {cursor}", status_code=0 |
| 345 | ) |
| 346 | if cursor > 2**53: |
| 347 | raise CoordBusError( |
| 348 | f"Hub returned implausibly large 'cursor': {cursor}", status_code=0 |
| 349 | ) |
| 350 | |
| 351 | return {"records": records, "cursor": cursor} |
File History
1 commit
sha256:649be9ce77a127cd847c4f798d60104dab7fe830cbc928762fafe3deadf38839
chore: point muse-git-backup.sh's MUSE_COPY at the fresh 'm…
Sonnet 5
patch
13 days ago