gabriel / muse public
coord_bus.py python
366 lines 12.6 KB
Raw
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ breaking 150 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
173 from muse.core.msign import build_msign_header
174
175 data = json.dumps(body).encode("utf-8")
176 req = urllib.request.Request(
177 url,
178 data=data,
179 method="POST",
180 headers={"Content-Type": "application/json"},
181 )
182 if isinstance(signing, SigningIdentity):
183 req.add_header("Authorization", build_msign_header(signing, "POST", url, data))
184
185 try:
186 with _STRICT_OPENER.open(req, timeout=_TIMEOUT_SECONDS) as resp:
187 raw = resp.read(MAX_COORD_RESPONSE_BYTES + 1)
188 if len(raw) > MAX_COORD_RESPONSE_BYTES:
189 raise CoordBusError(
190 f"Response body exceeded {MAX_COORD_RESPONSE_BYTES} bytes limit.",
191 status_code=0,
192 )
193 return json.loads(raw)
194 except urllib.error.HTTPError as exc:
195 raise CoordBusError(_http_error_message(exc), status_code=exc.code) from exc
196 except urllib.error.URLError as exc:
197 safe = sanitize_display(str(exc.reason)[:200])
198 raise CoordBusError(f"Network error: {safe}", status_code=0) from exc
199 except (json.JSONDecodeError, ValueError) as exc:
200 raise CoordBusError(f"Invalid JSON response: {exc}", status_code=0) from exc
201
202
203 # ── Public API ─────────────────────────────────────────────────────────────────
204
205
206 def push_to_hub(
207 hub_url: str,
208 owner: str,
209 slug: str,
210 records: list[JsonDict],
211 signing: SigningIdentity | None = None,
212 ) -> _IntMap:
213 """Push coordination records to MuseHub.
214
215 Args:
216 hub_url: Hub base URL (e.g. ``http://localhost:10003``).
217 owner: Repo owner username.
218 slug: Repo slug.
219 records: List of coordination record dicts. Each dict must have:
220 ``kind``, ``record_uuid``, ``run_id``, ``payload``.
221 ``expires_at`` is optional (ISO-8601 string or ``None``).
222 At most :data:`MAX_PUSH_BATCH` records per call.
223 signing: :class:`~muse.core.transport.SigningIdentity` (required — push always needs auth).
224
225 Returns:
226 Dict with ``"inserted"`` and ``"skipped"`` counts.
227
228 Raises:
229 :class:`CoordBusError` on failure.
230 :class:`ValueError` if *records* is empty or exceeds :data:`MAX_PUSH_BATCH`.
231 """
232 if not records:
233 raise ValueError("records must be non-empty")
234 if len(records) > MAX_PUSH_BATCH:
235 raise ValueError(
236 f"records exceeds maximum batch size of {MAX_PUSH_BATCH}; "
237 "split into smaller batches"
238 )
239
240 url = _build_url(hub_url, owner, slug, "coord/push")
241 logger.debug(
242 "coord_bus.push_to_hub: pushing %d record(s) to %s/%s",
243 len(records),
244 owner,
245 slug,
246 )
247 result = _post_json(url, {"records": records}, signing)
248
249 def _parse_count(field: str) -> int:
250 if field not in result:
251 return 0 # Key absent — treat as zero (hub may omit on partial success)
252 raw = result[field]
253 if raw is None:
254 raise CoordBusError(
255 f"Hub returned null {field!r} count", status_code=0
256 )
257 try:
258 n = int(raw)
259 except (TypeError, ValueError) as exc:
260 raise CoordBusError(
261 f"Hub returned non-integer {field!r} count", status_code=0
262 ) from exc
263 if n < 0:
264 raise CoordBusError(
265 f"Hub returned negative {field!r} count: {n}", status_code=0
266 )
267 if n > len(records):
268 raise CoordBusError(
269 f"Hub claimed {field!r}={n} but only {len(records)} records were sent",
270 status_code=0,
271 )
272 return n
273
274 return {
275 "inserted": _parse_count("inserted"),
276 "skipped": _parse_count("skipped"),
277 }
278
279
280 def pull_from_hub(
281 hub_url: str,
282 owner: str,
283 slug: str,
284 since_id: int = 0,
285 kinds: list[str] | None = None,
286 limit: int = 500,
287 signing: SigningIdentity | None = None,
288 ) -> _SummaryMap:
289 """Pull coordination records from MuseHub since *since_id*.
290
291 Args:
292 hub_url: Hub base URL.
293 owner: Repo owner username.
294 slug: Repo slug.
295 since_id: Return records with ``id > since_id``. ``0`` = all records.
296 kinds: Filter by record kind. ``None`` or ``[]`` = all kinds.
297 limit: Maximum records to return (1–:data:`MAX_PULL_LIMIT`).
298 signing: :class:`~muse.core.transport.SigningIdentity` (required for private repos).
299
300 Returns:
301 Dict with:
302 - ``"records"``: list of record dicts.
303 - ``"cursor"``: int — the ``id`` of the last returned record
304 (pass as ``since_id`` in the next call).
305
306 Raises:
307 :class:`CoordBusError` on failure.
308 :class:`ValueError` if *limit* is out of range.
309 """
310 if not (1 <= limit <= MAX_PULL_LIMIT):
311 raise ValueError(f"limit must be 1–{MAX_PULL_LIMIT}, got {limit}")
312
313 url = _build_url(hub_url, owner, slug, "coord/pull")
314 body: JsonDict = {
315 "since_id": since_id,
316 "kinds": kinds or [],
317 "limit": limit,
318 }
319 logger.debug(
320 "coord_bus.pull_from_hub: pulling from %s/%s since_id=%d",
321 owner,
322 slug,
323 since_id,
324 )
325 raw = _post_json(url, body, signing)
326
327 # --- Validate and normalise records ---
328 raw_records = raw.get("records")
329 if raw_records is None and "records" in raw:
330 raise CoordBusError("Hub returned null 'records' list", status_code=0)
331 if raw_records is not None and not isinstance(raw_records, list):
332 raise CoordBusError(
333 f"Hub returned non-list 'records': {type(raw_records).__name__}",
334 status_code=0,
335 )
336 records: list[JsonDict] = raw_records if raw_records is not None else []
337 for item in records:
338 if not isinstance(item, dict):
339 raise CoordBusError(
340 f"Hub returned non-dict record in 'records': {type(item).__name__}",
341 status_code=0,
342 )
343
344 # --- Validate and normalise cursor ---
345 raw_cursor = raw.get("cursor")
346 if raw_cursor is None and "cursor" in raw:
347 raise CoordBusError("Hub returned null 'cursor'", status_code=0)
348 if raw_cursor is None:
349 cursor: int = 0
350 else:
351 try:
352 cursor = int(raw_cursor)
353 except (TypeError, ValueError) as exc:
354 raise CoordBusError(
355 "Hub returned non-integer 'cursor'", status_code=0
356 ) from exc
357 if cursor < 0:
358 raise CoordBusError(
359 f"Hub returned negative 'cursor': {cursor}", status_code=0
360 )
361 if cursor > 2**53:
362 raise CoordBusError(
363 f"Hub returned implausibly large 'cursor': {cursor}", status_code=0
364 )
365
366 return {"records": records, "cursor": cursor}
File History 1 commit
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 150 days ago