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