"""Muse coordination bus client — push/pull coordination records to/from MuseHub. Provides synchronous HTTP operations for syncing local coordination state with a MuseHub remote so that agent swarms on different machines share state. The client re-uses the transport layer's security model: - All HTTP redirects are refused (credentials must not follow redirects). - The signing identity is never logged. - Responses are capped at :data:`MAX_COORD_RESPONSE_BYTES` to prevent OOM. - HTTP is accepted only when no signing identity is supplied (public repos); HTTPS is required when a signing identity is present. Protocol -------- Push (POST ``/{owner}/{slug}/coord/push``): - Request body: JSON ``{"records": [...]}`` - Response: JSON ``{"inserted": N, "skipped": M}`` Pull (POST ``/{owner}/{slug}/coord/pull``): - Request body: JSON ``{"since_id": N, "kinds": [...], "limit": M}`` - Response: JSON ``{"records": [...], "cursor": N}`` Both endpoints use JSON (not msgpack) — coordination records are small and infrequent compared to object-store traffic. stdlib ``json`` is sufficient. Error handling -------------- All errors raise :class:`CoordBusError` with a human-readable message. The caller (CLI command) is responsible for printing the error and exiting with a non-zero status code. Security -------- - ``owner`` and ``slug`` are URL-path components — they are %-encoded by ``urllib.parse.quote`` to prevent path traversal. - The signing identity is used in the ``Authorization`` header only, never in the URL. - Response bodies are read into memory only up to ``MAX_COORD_RESPONSE_BYTES``. """ from __future__ import annotations import http.client import json import logging import urllib.error import urllib.parse import urllib.request from typing import TYPE_CHECKING from muse.core.validation import sanitize_display type _SummaryMap = dict[str, int | list["JsonDict"]] type _IntMap = dict[str, int] if TYPE_CHECKING: from muse.core.transport import SigningIdentity # JSON-compatible value type for coord request/response bodies. JsonValue = str | int | float | bool | None | list["JsonValue"] | "JsonDict" JsonDict = dict[str, JsonValue] logger = logging.getLogger(__name__) # Maximum response body size accepted from the hub (4 MiB). # Coordination records are small; 4 MiB accommodates ≈ 8,000 full records. MAX_COORD_RESPONSE_BYTES: int = 4 * 1024 * 1024 # Request/response timeout in seconds for coordination HTTP calls. _TIMEOUT_SECONDS: int = 30 # Maximum records per push call (must stay ≤ server limit of 500). MAX_PUSH_BATCH: int = 500 # Maximum records per pull call (must stay ≤ server limit of 1000). MAX_PULL_LIMIT: int = 1000 # ── Exceptions ───────────────────────────────────────────────────────────────── class CoordBusError(Exception): """Raised when a coordination bus HTTP operation fails. Attributes: status_code: HTTP status code, or 0 for network-level errors. """ def __init__(self, message: str, status_code: int = 0) -> None: super().__init__(message) self.status_code = status_code # ── Transport helpers (reuse pattern from transport.py) ──────────────────────── class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): """Refuse all HTTP redirects to prevent credential leakage.""" def redirect_request( self, req: urllib.request.Request, fp: http.client.HTTPResponse | None, code: int, msg: str, headers: http.client.HTTPMessage, newurl: str, ) -> urllib.request.Request | None: raise urllib.error.HTTPError( req.full_url, code, ( f"Redirect refused ({code}): server tried to redirect to {newurl!r}. " "Update the configured remote URL to the final destination." ), headers, fp, ) _STRICT_OPENER = urllib.request.build_opener(_NoRedirectHandler()) def _http_error_message(exc: urllib.error.HTTPError) -> str: """Build a safe, sanitized error message — never exposes response body on 401.""" if exc.code == 401: return "Authentication failed (HTTP 401). Run 'muse auth register'." try: raw_body = exc.read().decode("utf-8", errors="replace") except Exception: # noqa: BLE001 raw_body = "" safe_body = sanitize_display(raw_body[:200]) return f"HTTP {exc.code}: {safe_body}" if safe_body else f"HTTP {exc.code}" def _build_url(hub_url: str, owner: str, slug: str, endpoint: str) -> str: """Build a %-encoded URL for a coord endpoint. Args: hub_url: Hub base URL (e.g. ``http://localhost:10003``). owner: Repo owner — %-encoded to prevent path traversal. slug: Repo slug — %-encoded to prevent path traversal. endpoint: Endpoint suffix (``coord/push``, ``coord/pull``). Returns: Full URL string. """ safe_owner = urllib.parse.quote(owner, safe="") safe_slug = urllib.parse.quote(slug, safe="") return f"{hub_url.rstrip('/')}/{safe_owner}/{safe_slug}/{endpoint}" def _post_json( url: str, body: JsonDict, signing: "SigningIdentity | None", ) -> JsonDict: """POST a JSON body to *url* and return the parsed JSON response. Args: url: Full target URL. body: Request body dict — encoded as UTF-8 JSON. signing: :class:`~muse.core.transport.SigningIdentity`, or ``None`` for unauthenticated calls. Returns: Parsed JSON response dict. Raises: :class:`CoordBusError` on HTTP error or network failure. """ from muse.core.transport import SigningIdentity, build_msign_header data = json.dumps(body).encode("utf-8") req = urllib.request.Request( url, data=data, method="POST", headers={"Content-Type": "application/json"}, ) if isinstance(signing, SigningIdentity): req.add_header("Authorization", build_msign_header(signing, "POST", url, data)) try: with _STRICT_OPENER.open(req, timeout=_TIMEOUT_SECONDS) as resp: raw = resp.read(MAX_COORD_RESPONSE_BYTES + 1) if len(raw) > MAX_COORD_RESPONSE_BYTES: raise CoordBusError( f"Response body exceeded {MAX_COORD_RESPONSE_BYTES} bytes limit.", status_code=0, ) return json.loads(raw) except urllib.error.HTTPError as exc: raise CoordBusError(_http_error_message(exc), status_code=exc.code) from exc except urllib.error.URLError as exc: safe = sanitize_display(str(exc.reason)[:200]) raise CoordBusError(f"Network error: {safe}", status_code=0) from exc except (json.JSONDecodeError, ValueError) as exc: raise CoordBusError(f"Invalid JSON response: {exc}", status_code=0) from exc # ── Public API ───────────────────────────────────────────────────────────────── def push_to_hub( hub_url: str, owner: str, slug: str, records: list[JsonDict], signing: SigningIdentity | None = None, ) -> _IntMap: """Push coordination records to MuseHub. Args: hub_url: Hub base URL (e.g. ``http://localhost:10003``). owner: Repo owner username. slug: Repo slug. records: List of coordination record dicts. Each dict must have: ``kind``, ``record_uuid``, ``run_id``, ``payload``. ``expires_at`` is optional (ISO-8601 string or ``None``). At most :data:`MAX_PUSH_BATCH` records per call. signing: :class:`~muse.core.transport.SigningIdentity` (required — push always needs auth). Returns: Dict with ``"inserted"`` and ``"skipped"`` counts. Raises: :class:`CoordBusError` on failure. :class:`ValueError` if *records* is empty or exceeds :data:`MAX_PUSH_BATCH`. """ if not records: raise ValueError("records must be non-empty") if len(records) > MAX_PUSH_BATCH: raise ValueError( f"records exceeds maximum batch size of {MAX_PUSH_BATCH}; " "split into smaller batches" ) url = _build_url(hub_url, owner, slug, "coord/push") logger.debug( "coord_bus.push_to_hub: pushing %d record(s) to %s/%s", len(records), owner, slug, ) result = _post_json(url, {"records": records}, signing) def _parse_count(field: str) -> int: if field not in result: return 0 # Key absent — treat as zero (hub may omit on partial success) raw = result[field] if raw is None: raise CoordBusError( f"Hub returned null {field!r} count", status_code=0 ) try: n = int(raw) except (TypeError, ValueError) as exc: raise CoordBusError( f"Hub returned non-integer {field!r} count", status_code=0 ) from exc if n < 0: raise CoordBusError( f"Hub returned negative {field!r} count: {n}", status_code=0 ) if n > len(records): raise CoordBusError( f"Hub claimed {field!r}={n} but only {len(records)} records were sent", status_code=0, ) return n return { "inserted": _parse_count("inserted"), "skipped": _parse_count("skipped"), } def pull_from_hub( hub_url: str, owner: str, slug: str, since_id: int = 0, kinds: list[str] | None = None, limit: int = 500, signing: SigningIdentity | None = None, ) -> _SummaryMap: """Pull coordination records from MuseHub since *since_id*. Args: hub_url: Hub base URL. owner: Repo owner username. slug: Repo slug. since_id: Return records with ``id > since_id``. ``0`` = all records. kinds: Filter by record kind. ``None`` or ``[]`` = all kinds. limit: Maximum records to return (1–:data:`MAX_PULL_LIMIT`). signing: :class:`~muse.core.transport.SigningIdentity` (required for private repos). Returns: Dict with: - ``"records"``: list of record dicts. - ``"cursor"``: int — the ``id`` of the last returned record (pass as ``since_id`` in the next call). Raises: :class:`CoordBusError` on failure. :class:`ValueError` if *limit* is out of range. """ if not (1 <= limit <= MAX_PULL_LIMIT): raise ValueError(f"limit must be 1–{MAX_PULL_LIMIT}, got {limit}") url = _build_url(hub_url, owner, slug, "coord/pull") body: JsonDict = { "since_id": since_id, "kinds": kinds or [], "limit": limit, } logger.debug( "coord_bus.pull_from_hub: pulling from %s/%s since_id=%d", owner, slug, since_id, ) raw = _post_json(url, body, signing) # --- Validate and normalise records --- raw_records = raw.get("records") if raw_records is None and "records" in raw: raise CoordBusError("Hub returned null 'records' list", status_code=0) if raw_records is not None and not isinstance(raw_records, list): raise CoordBusError( f"Hub returned non-list 'records': {type(raw_records).__name__}", status_code=0, ) records: list[JsonDict] = raw_records if raw_records is not None else [] for item in records: if not isinstance(item, dict): raise CoordBusError( f"Hub returned non-dict record in 'records': {type(item).__name__}", status_code=0, ) # --- Validate and normalise cursor --- raw_cursor = raw.get("cursor") if raw_cursor is None and "cursor" in raw: raise CoordBusError("Hub returned null 'cursor'", status_code=0) if raw_cursor is None: cursor: int = 0 else: try: cursor = int(raw_cursor) except (TypeError, ValueError) as exc: raise CoordBusError( "Hub returned non-integer 'cursor'", status_code=0 ) from exc if cursor < 0: raise CoordBusError( f"Hub returned negative 'cursor': {cursor}", status_code=0 ) if cursor > 2**53: raise CoordBusError( f"Hub returned implausibly large 'cursor': {cursor}", status_code=0 ) return {"records": records, "cursor": cursor}