coord_sync.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
32 days ago
| 1 | """``muse coord sync`` — push/pull local coordination state to/from MuseHub. |
| 2 | |
| 3 | Enables distributed agent swarms to share coordination records (reservations, |
| 4 | intents, releases, heartbeats, dependencies, tasks, claims) across machines |
| 5 | via the MuseHub coordination bus. |
| 6 | |
| 7 | Sub-commands |
| 8 | ------------ |
| 9 | ``muse coord sync push`` |
| 10 | Reads all local coordination records from ``.muse/coordination/`` and |
| 11 | pushes them to MuseHub in batches. Records already on the hub are |
| 12 | silently skipped (idempotent). Requires a valid signing identity. |
| 13 | |
| 14 | ``muse coord sync pull`` |
| 15 | Pulls coordination records from MuseHub into local read-only JSON files |
| 16 | under ``.muse/coordination/remote/``. Local agents can read these to |
| 17 | discover reservations and tasks created by remote agents. Requires auth |
| 18 | for private repos; public repos allow unauthenticated pull. |
| 19 | |
| 20 | Usage:: |
| 21 | |
| 22 | muse coord sync push --hub http://localhost:10003 --owner gabriel --slug myrepo |
| 23 | muse coord sync pull --hub http://localhost:10003 --owner gabriel --slug myrepo |
| 24 | muse coord sync pull --hub http://localhost:10003 --owner gabriel --slug myrepo \\ |
| 25 | --since-id 42 --kinds reservation heartbeat |
| 26 | |
| 27 | Output (push):: |
| 28 | |
| 29 | ✅ Pushed 12 record(s) to gabriel/myrepo — inserted: 10, skipped: 2 |
| 30 | |
| 31 | Output (pull):: |
| 32 | |
| 33 | ✅ Pulled 8 new record(s) from gabriel/myrepo — cursor: 20 |
| 34 | |
| 35 | JSON output (push):: |
| 36 | |
| 37 | { |
| 38 | "schema_version": "...", |
| 39 | "inserted": 10, |
| 40 | "skipped": 2, |
| 41 | "total": 12, |
| 42 | "failed": false, |
| 43 | "elapsed_seconds": 0.123 |
| 44 | } |
| 45 | |
| 46 | JSON output (pull):: |
| 47 | |
| 48 | { |
| 49 | "schema_version": "...", |
| 50 | "count": 8, |
| 51 | "cursor": 20, |
| 52 | "records": [...], |
| 53 | "elapsed_seconds": 0.045 |
| 54 | } |
| 55 | |
| 56 | Flags (common): |
| 57 | |
| 58 | ``--hub URL`` |
| 59 | MuseHub base URL (default: read from ``[hub] url`` in ``config.toml``). |
| 60 | |
| 61 | ``--owner NAME`` |
| 62 | Repository owner username. Maximum 256 characters. |
| 63 | |
| 64 | ``--slug NAME`` |
| 65 | Repository URL slug. Maximum 256 characters. |
| 66 | |
| 67 | |
| 68 | ``--json`` |
| 69 | Emit results as JSON. |
| 70 | |
| 71 | Push-only flags: |
| 72 | |
| 73 | ``--kinds KIND [KIND ...]`` |
| 74 | Limit push to specific record kinds (default: all kinds). |
| 75 | |
| 76 | Pull-only flags: |
| 77 | |
| 78 | ``--since-id N`` |
| 79 | Return only records with id > N (default: 0 = all records). Must be >= 0. |
| 80 | |
| 81 | ``--kinds KIND [KIND ...]`` |
| 82 | Filter by record kind. |
| 83 | |
| 84 | ``--limit N`` |
| 85 | Maximum records to pull per call (1–1000, default: 500). |
| 86 | |
| 87 | Exit codes:: |
| 88 | |
| 89 | 0 — success |
| 90 | 1 — bad arguments, auth failure, or hub communication error |
| 91 | """ |
| 92 | |
| 93 | from __future__ import annotations |
| 94 | |
| 95 | import argparse |
| 96 | import json |
| 97 | import logging |
| 98 | import pathlib |
| 99 | import re |
| 100 | import sys |
| 101 | import time |
| 102 | from datetime import datetime, timezone |
| 103 | from typing import TYPE_CHECKING |
| 104 | |
| 105 | from muse.core.coord_bus import ( |
| 106 | CoordBusError, |
| 107 | MAX_PUSH_BATCH, |
| 108 | pull_from_hub, |
| 109 | push_to_hub, |
| 110 | ) |
| 111 | from muse.core.coord_bus import JsonDict |
| 112 | from muse.core.errors import ExitCode |
| 113 | from muse.core.repo import require_repo |
| 114 | from muse.core.store import write_text_atomic |
| 115 | from muse.core.validation import clamp_int, sanitize_display |
| 116 | |
| 117 | if TYPE_CHECKING: |
| 118 | from muse.core.transport import SigningIdentity |
| 119 | |
| 120 | logger = logging.getLogger(__name__) |
| 121 | |
| 122 | # Coordination record kinds supported by the bus. |
| 123 | _ALL_KINDS = ( |
| 124 | "reservation", |
| 125 | "intent", |
| 126 | "release", |
| 127 | "heartbeat", |
| 128 | "dependency", |
| 129 | "task", |
| 130 | "claim", |
| 131 | ) |
| 132 | |
| 133 | # Directory under .muse/coordination/ where pulled remote records are stored. |
| 134 | _REMOTE_DIR_NAME = "remote" |
| 135 | |
| 136 | # ── Input constraints ───────────────────────────────────────────────────────── |
| 137 | |
| 138 | #: Maximum length of --owner and --slug to prevent oversized HTTP requests. |
| 139 | _MAX_OWNER_LEN: int = 256 |
| 140 | _MAX_SLUG_LEN: int = 256 |
| 141 | |
| 142 | #: Maximum records to pull per call. |
| 143 | _MAX_PULL_LIMIT: int = 1000 |
| 144 | |
| 145 | #: Safe record UUID pattern — only alphanumeric, hyphens, underscores. |
| 146 | #: Prevents path traversal when server-supplied record_uuid is used to |
| 147 | #: construct filesystem paths under .muse/coordination/remote/<kind>/<uuid>.json. |
| 148 | _SAFE_RECORD_UUID_RE: re.Pattern[str] = re.compile(r"^[a-zA-Z0-9_\-]{1,128}$") |
| 149 | |
| 150 | |
| 151 | # ── Error helper ────────────────────────────────────────────────────────────── |
| 152 | |
| 153 | |
| 154 | def _err(msg: str, as_json: bool = False, status: str = "error") -> None: |
| 155 | """Print an error and return. Caller is responsible for raising SystemExit.""" |
| 156 | if as_json: |
| 157 | print(json.dumps({"error": msg, "status": status})) |
| 158 | else: |
| 159 | print(f"❌ {msg}", file=sys.stderr) |
| 160 | |
| 161 | |
| 162 | def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: |
| 163 | """Register ``sync`` on *subparsers* (nested push/pull sub-subcommands). |
| 164 | |
| 165 | Wires ``push`` and ``pull`` sub-subcommands with their flags, defaults, |
| 166 | and help text. Both sub-commands share ``--hub``, ``--owner``, ``--slug``, |
| 167 | ``--format`` / ``--json``. Sets ``func`` to |
| 168 | :func:`run_push` and :func:`run_pull` respectively. |
| 169 | """ |
| 170 | sync_parser = subparsers.add_parser( |
| 171 | "sync", |
| 172 | help="Push/pull coordination records to/from MuseHub.", |
| 173 | description=__doc__, |
| 174 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 175 | ) |
| 176 | sync_subs = sync_parser.add_subparsers(dest="sync_command", metavar="SYNC_COMMAND") |
| 177 | sync_subs.required = True |
| 178 | |
| 179 | # ── Common arguments shared by push and pull ────────────────────────────── |
| 180 | def _add_common(p: argparse.ArgumentParser) -> None: |
| 181 | p.add_argument( |
| 182 | "--hub", |
| 183 | default=None, |
| 184 | metavar="URL", |
| 185 | help="MuseHub base URL (default: from config.toml [hub] url).", |
| 186 | ) |
| 187 | p.add_argument( |
| 188 | "--owner", |
| 189 | required=True, |
| 190 | metavar="NAME", |
| 191 | help=f"Repository owner username (max {_MAX_OWNER_LEN} chars).", |
| 192 | ) |
| 193 | p.add_argument( |
| 194 | "--slug", |
| 195 | required=True, |
| 196 | metavar="NAME", |
| 197 | help=f"Repository URL slug (max {_MAX_SLUG_LEN} chars).", |
| 198 | ) |
| 199 | p.add_argument( |
| 200 | "--format", "-f", |
| 201 | default="text", |
| 202 | dest="fmt", |
| 203 | choices=("text", "json"), |
| 204 | help="Output format: text (default) or json.", |
| 205 | ) |
| 206 | p.add_argument( |
| 207 | "--json", "-j", |
| 208 | action="store_const", |
| 209 | const="json", |
| 210 | dest="fmt", |
| 211 | help="Shorthand for --format json.", |
| 212 | ) |
| 213 | |
| 214 | # ── Pull ────────────────────────────────────────────────────────────────── |
| 215 | pull_parser = sync_subs.add_parser( |
| 216 | "pull", |
| 217 | help="Pull coordination records from MuseHub.", |
| 218 | description=( |
| 219 | "Fetch coordination records from MuseHub and write them to\n" |
| 220 | ".muse/coordination/remote/<kind>/<uuid>.json.\n" |
| 221 | "Use --since-id <cursor> for incremental pulls — pass the cursor\n" |
| 222 | "from the previous JSON output back as --since-id on the next call.\n\n" |
| 223 | "Agent quickstart\n" |
| 224 | "----------------\n" |
| 225 | " muse coord sync pull --owner gabriel --slug myrepo --json\n" |
| 226 | " muse coord sync pull --owner gabriel --slug myrepo -j\n" |
| 227 | " muse coord sync pull --owner gabriel --slug myrepo -j | jq .cursor\n" |
| 228 | " muse coord sync pull --owner gabriel --slug myrepo \\\n" |
| 229 | " --since-id 42 --kinds reservation intent -j\n\n" |
| 230 | "JSON output schema\n" |
| 231 | "------------------\n" |
| 232 | ' {"schema_version": "<str>", "count": <int>, "cursor": <int>,\n' |
| 233 | ' "records": [...], "elapsed_seconds": <float>}\n\n' |
| 234 | "Exit codes\n" |
| 235 | "----------\n" |
| 236 | " 0 — pull succeeded (0 records is a valid success)\n" |
| 237 | " 1 — bad arguments, auth error, or hub communication error\n" |
| 238 | " 2 — not inside a Muse repository\n" |
| 239 | ), |
| 240 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 241 | ) |
| 242 | _add_common(pull_parser) |
| 243 | pull_parser.add_argument( |
| 244 | "--since-id", |
| 245 | dest="since_id", |
| 246 | type=int, |
| 247 | default=0, |
| 248 | metavar="N", |
| 249 | help="Only pull records with id > N (default: 0 = all). Must be >= 0.", |
| 250 | ) |
| 251 | pull_parser.add_argument( |
| 252 | "--kinds", |
| 253 | nargs="+", |
| 254 | choices=_ALL_KINDS, |
| 255 | default=[], |
| 256 | metavar="KIND", |
| 257 | help=f"Filter by kind. Choices: {', '.join(_ALL_KINDS)}.", |
| 258 | ) |
| 259 | pull_parser.add_argument( |
| 260 | "--limit", |
| 261 | type=int, |
| 262 | default=500, |
| 263 | metavar="N", |
| 264 | help=f"Maximum records to pull per call (1–{_MAX_PULL_LIMIT}, default: 500).", |
| 265 | ) |
| 266 | pull_parser.set_defaults(func=run_pull) |
| 267 | |
| 268 | # ── Push ────────────────────────────────────────────────────────────────── |
| 269 | push_parser = sync_subs.add_parser( |
| 270 | "push", |
| 271 | help="Push local coordination records to MuseHub.", |
| 272 | description=( |
| 273 | "Read all local coordination files from .muse/coordination/ and\n" |
| 274 | "POST them to MuseHub in batches. Records already on the hub are\n" |
| 275 | "silently skipped (idempotent). Fails with exit 1 if any batch\n" |
| 276 | "errors; other batches still complete.\n\n" |
| 277 | "Agent quickstart\n" |
| 278 | "----------------\n" |
| 279 | " muse coord sync push --owner gabriel --slug myrepo --json\n" |
| 280 | " muse coord sync push --owner gabriel --slug myrepo -j\n" |
| 281 | " muse coord sync push --owner gabriel --slug myrepo -j | jq .inserted\n" |
| 282 | " muse coord sync push --owner gabriel --slug myrepo \\\n" |
| 283 | " --kinds reservation intent -j\n\n" |
| 284 | "JSON output schema\n" |
| 285 | "------------------\n" |
| 286 | ' {"schema_version": "<str>", "inserted": <int>, "skipped": <int>,\n' |
| 287 | ' "total": <int>, "failed": <bool>, "elapsed_seconds": <float>}\n\n' |
| 288 | "Exit codes\n" |
| 289 | "----------\n" |
| 290 | " 0 — all batches pushed successfully (or nothing to push)\n" |
| 291 | " 1 — bad arguments, auth error, or at least one batch failed\n" |
| 292 | " 2 — not inside a Muse repository\n" |
| 293 | ), |
| 294 | formatter_class=argparse.RawDescriptionHelpFormatter, |
| 295 | ) |
| 296 | _add_common(push_parser) |
| 297 | push_parser.add_argument( |
| 298 | "--kinds", |
| 299 | nargs="+", |
| 300 | choices=_ALL_KINDS, |
| 301 | default=list(_ALL_KINDS), |
| 302 | metavar="KIND", |
| 303 | help=f"Limit push to these kinds (default: all). Choices: {', '.join(_ALL_KINDS)}.", |
| 304 | ) |
| 305 | push_parser.set_defaults(func=run_push) |
| 306 | |
| 307 | |
| 308 | # ── Helpers ──────────────────────────────────────────────────────────────────── |
| 309 | |
| 310 | |
| 311 | def _resolve_hub_and_signing( |
| 312 | args: argparse.Namespace, |
| 313 | as_json: bool = False, |
| 314 | ) -> tuple[str, str | None]: |
| 315 | """Return ``(hub_url, signing)`` from args or config. |
| 316 | |
| 317 | Reads ``[hub] url`` from ``config.toml`` and the signing identity from |
| 318 | ``~/.muse/identity.toml`` when the corresponding CLI flags are omitted. |
| 319 | |
| 320 | Raises :class:`SystemExit` with exit code |
| 321 | :attr:`~muse.core.errors.ExitCode.USER_ERROR` if no hub URL can be |
| 322 | determined. Error messages go to *stdout* as JSON when *as_json* is |
| 323 | ``True``, otherwise to *stderr*. |
| 324 | |
| 325 | Args: |
| 326 | args: Parsed ``argparse.Namespace`` with ``hub`` and ``token`` attrs. |
| 327 | as_json: Route error messages to stdout as JSON when ``True``. |
| 328 | |
| 329 | Returns: |
| 330 | ``(hub_url, signing)`` where *signing* may be ``None`` for unauthenticated |
| 331 | access to public repos. |
| 332 | """ |
| 333 | hub_url: str | None = args.hub |
| 334 | signing: SigningIdentity | None = None |
| 335 | |
| 336 | if hub_url is None or signing is None: |
| 337 | try: |
| 338 | from muse.cli.config import get_hub_url, get_signing_identity |
| 339 | if hub_url is None: |
| 340 | hub_url = get_hub_url() |
| 341 | if signing is None: |
| 342 | signing = get_signing_identity(hub_url) |
| 343 | except Exception as exc: # noqa: BLE001 |
| 344 | if hub_url is None: |
| 345 | msg = ( |
| 346 | "no --hub URL supplied and could not read hub URL from config: " |
| 347 | f"{exc}" |
| 348 | ) |
| 349 | _err(msg, as_json, "no_hub_url") |
| 350 | raise SystemExit(ExitCode.USER_ERROR) |
| 351 | |
| 352 | if hub_url is None: |
| 353 | _err("--hub URL is required", as_json, "no_hub_url") |
| 354 | raise SystemExit(ExitCode.USER_ERROR) |
| 355 | |
| 356 | return hub_url, signing |
| 357 | |
| 358 | |
| 359 | def _gather_local_records( |
| 360 | root: pathlib.Path, |
| 361 | kinds: list[str], |
| 362 | ) -> list[JsonDict]: |
| 363 | """Load local coordination records from ``.muse/coordination/``. |
| 364 | |
| 365 | Returns a flat list of serialized record dicts, each with the fields |
| 366 | expected by the hub's push API: ``kind``, ``record_uuid``, ``run_id``, |
| 367 | ``payload``, and ``expires_at``. |
| 368 | |
| 369 | Only kinds listed in *kinds* are included. Corrupt or unreadable files |
| 370 | are silently skipped (logged at DEBUG level) so a single bad file never |
| 371 | blocks a push. |
| 372 | |
| 373 | Args: |
| 374 | root: Repository root directory (contains ``.muse/``). |
| 375 | kinds: Subset of :data:`_ALL_KINDS` to include. |
| 376 | |
| 377 | Returns: |
| 378 | List of record dicts ready for ``push_to_hub``. |
| 379 | """ |
| 380 | coord_dir = root / ".muse" / "coordination" |
| 381 | records: list[JsonDict] = [] |
| 382 | |
| 383 | # ── Reservations ────────────────────────────────────────────────────────── |
| 384 | if "reservation" in kinds: |
| 385 | res_dir = coord_dir / "reservations" |
| 386 | if res_dir.is_dir(): |
| 387 | for fpath in res_dir.glob("*.json"): |
| 388 | try: |
| 389 | data = json.loads(fpath.read_text(encoding="utf-8")) |
| 390 | records.append({ |
| 391 | "kind": "reservation", |
| 392 | "record_uuid": data.get("reservation_id", fpath.stem), |
| 393 | "run_id": data.get("run_id", ""), |
| 394 | "payload": data, |
| 395 | "expires_at": data.get("expires_at"), |
| 396 | }) |
| 397 | except Exception: # noqa: BLE001 |
| 398 | logger.debug("gather_local_records: skipping %s", fpath) |
| 399 | |
| 400 | # ── Heartbeats ──────────────────────────────────────────────────────────── |
| 401 | if "heartbeat" in kinds: |
| 402 | hb_dir = coord_dir / "heartbeats" |
| 403 | if hb_dir.is_dir(): |
| 404 | for fpath in hb_dir.glob("*.json"): |
| 405 | try: |
| 406 | data = json.loads(fpath.read_text(encoding="utf-8")) |
| 407 | records.append({ |
| 408 | "kind": "heartbeat", |
| 409 | "record_uuid": data.get("run_id", fpath.stem), |
| 410 | "run_id": data.get("run_id", ""), |
| 411 | "payload": data, |
| 412 | "expires_at": data.get("expires_at"), |
| 413 | }) |
| 414 | except Exception: # noqa: BLE001 |
| 415 | logger.debug("gather_local_records: skipping heartbeat %s", fpath) |
| 416 | |
| 417 | # ── Intents ─────────────────────────────────────────────────────────────── |
| 418 | if "intent" in kinds: |
| 419 | intent_dir = coord_dir / "intents" |
| 420 | if intent_dir.is_dir(): |
| 421 | for fpath in intent_dir.glob("*.json"): |
| 422 | try: |
| 423 | data = json.loads(fpath.read_text(encoding="utf-8")) |
| 424 | records.append({ |
| 425 | "kind": "intent", |
| 426 | "record_uuid": data.get("intent_id", fpath.stem), |
| 427 | "run_id": data.get("run_id", ""), |
| 428 | "payload": data, |
| 429 | "expires_at": data.get("expires_at"), |
| 430 | }) |
| 431 | except Exception: # noqa: BLE001 |
| 432 | logger.debug("gather_local_records: skipping intent %s", fpath) |
| 433 | |
| 434 | # ── Releases ────────────────────────────────────────────────────────────── |
| 435 | if "release" in kinds: |
| 436 | release_dir = coord_dir / "releases" |
| 437 | if release_dir.is_dir(): |
| 438 | for fpath in release_dir.glob("*.json"): |
| 439 | try: |
| 440 | data = json.loads(fpath.read_text(encoding="utf-8")) |
| 441 | records.append({ |
| 442 | "kind": "release", |
| 443 | "record_uuid": data.get("release_id", fpath.stem), |
| 444 | "run_id": data.get("run_id", ""), |
| 445 | "payload": data, |
| 446 | "expires_at": None, |
| 447 | }) |
| 448 | except Exception: # noqa: BLE001 |
| 449 | logger.debug("gather_local_records: skipping release %s", fpath) |
| 450 | |
| 451 | # ── Dependencies ────────────────────────────────────────────────────────── |
| 452 | if "dependency" in kinds: |
| 453 | dep_dir = coord_dir / "dependencies" |
| 454 | if dep_dir.is_dir(): |
| 455 | for fpath in dep_dir.glob("*.json"): |
| 456 | try: |
| 457 | data = json.loads(fpath.read_text(encoding="utf-8")) |
| 458 | records.append({ |
| 459 | "kind": "dependency", |
| 460 | "record_uuid": data.get("reservation_id", fpath.stem), |
| 461 | "run_id": data.get("run_id", ""), |
| 462 | "payload": data, |
| 463 | "expires_at": None, |
| 464 | }) |
| 465 | except Exception: # noqa: BLE001 |
| 466 | logger.debug("gather_local_records: skipping dependency %s", fpath) |
| 467 | |
| 468 | # ── Tasks ───────────────────────────────────────────────────────────────── |
| 469 | if "task" in kinds: |
| 470 | task_dir = coord_dir / "tasks" |
| 471 | if task_dir.is_dir(): |
| 472 | for fpath in task_dir.glob("*.json"): |
| 473 | try: |
| 474 | data = json.loads(fpath.read_text(encoding="utf-8")) |
| 475 | records.append({ |
| 476 | "kind": "task", |
| 477 | "record_uuid": data.get("task_id", fpath.stem), |
| 478 | "run_id": data.get("run_id", ""), |
| 479 | "payload": data, |
| 480 | "expires_at": None, |
| 481 | }) |
| 482 | except Exception: # noqa: BLE001 |
| 483 | logger.debug("gather_local_records: skipping task %s", fpath) |
| 484 | |
| 485 | # ── Claims ──────────────────────────────────────────────────────────────── |
| 486 | if "claim" in kinds: |
| 487 | claim_dir = coord_dir / "claims" |
| 488 | if claim_dir.is_dir(): |
| 489 | for fpath in claim_dir.glob("*.json"): |
| 490 | try: |
| 491 | data = json.loads(fpath.read_text(encoding="utf-8")) |
| 492 | records.append({ |
| 493 | "kind": "claim", |
| 494 | "record_uuid": data.get("task_id", fpath.stem), |
| 495 | "run_id": data.get("claimer_run_id", ""), |
| 496 | "payload": data, |
| 497 | "expires_at": data.get("expires_at"), |
| 498 | }) |
| 499 | except Exception: # noqa: BLE001 |
| 500 | logger.debug("gather_local_records: skipping claim %s", fpath) |
| 501 | |
| 502 | return records |
| 503 | |
| 504 | |
| 505 | def _write_remote_records( |
| 506 | root: pathlib.Path, |
| 507 | records: list[JsonDict], |
| 508 | ) -> None: |
| 509 | """Write pulled remote records to ``.muse/coordination/remote/``. |
| 510 | |
| 511 | Each record is written to ``remote/<kind>/<record_uuid>.json``. |
| 512 | Existing files are overwritten to keep remote state up to date. |
| 513 | |
| 514 | Security |
| 515 | -------- |
| 516 | Both *kind* and *record_uuid* come from the MuseHub response and must be |
| 517 | sanitized before use in filesystem paths: |
| 518 | |
| 519 | * **kind** — validated against the :data:`_ALL_KINDS` allowlist. Records |
| 520 | with an unrecognised kind are skipped; a server cannot write outside the |
| 521 | ``remote/`` subtree by supplying ``"../evil"`` as the kind. |
| 522 | * **record_uuid** — validated against :data:`_SAFE_RECORD_UUID_RE` |
| 523 | (alphanumeric, hyphens, underscores, 1–128 chars). Records with an |
| 524 | unsafe UUID are skipped; a server cannot escape the kind directory via |
| 525 | ``"../../../etc/passwd"`` or similar. |
| 526 | |
| 527 | Args: |
| 528 | root: Repository root directory (contains ``.muse/``). |
| 529 | records: List of record dicts returned by :func:`pull_from_hub`. |
| 530 | """ |
| 531 | remote_dir = root / ".muse" / "coordination" / _REMOTE_DIR_NAME |
| 532 | for rec in records: |
| 533 | kind = rec.get("kind", "") |
| 534 | record_uuid = rec.get("record_uuid", "") |
| 535 | |
| 536 | # Allowlist check — reject unknown kinds outright. |
| 537 | if kind not in _ALL_KINDS: |
| 538 | logger.warning("write_remote_records: rejected unknown kind %r", kind) |
| 539 | continue |
| 540 | |
| 541 | # UUID safety check — reject traversal-capable strings. |
| 542 | if not record_uuid or not _SAFE_RECORD_UUID_RE.match(record_uuid): |
| 543 | logger.warning( |
| 544 | "write_remote_records: rejected unsafe record_uuid %r (kind=%r)", |
| 545 | record_uuid, |
| 546 | kind, |
| 547 | ) |
| 548 | continue |
| 549 | |
| 550 | kind_dir = remote_dir / kind |
| 551 | kind_dir.mkdir(parents=True, exist_ok=True) |
| 552 | target = kind_dir / f"{record_uuid}.json" |
| 553 | write_text_atomic(target, json.dumps(rec) + "\n") |
| 554 | |
| 555 | |
| 556 | # ── Command handlers ─────────────────────────────────────────────────────────── |
| 557 | |
| 558 | |
| 559 | def run_push(args: argparse.Namespace) -> None: |
| 560 | """Push local coordination records to MuseHub. |
| 561 | |
| 562 | Loads all local coordination files from ``.muse/coordination/``, batches |
| 563 | them into groups of at most :data:`~muse.core.coord_bus.MAX_PUSH_BATCH` |
| 564 | records, and POSTs each batch to ``/{owner}/{slug}/coord/push``. |
| 565 | |
| 566 | Idempotent: records already on the hub are silently skipped. If any batch |
| 567 | fails the remaining batches still run; ``failed`` is set to ``True`` in |
| 568 | the output and the process exits 1. |
| 569 | |
| 570 | Execution order |
| 571 | --------------- |
| 572 | 1. **Validate inputs** — ``--owner`` and ``--slug`` length caps enforced |
| 573 | before any file I/O. |
| 574 | 2. **Resolve repo** — :func:`~muse.core.repo.require_repo`. |
| 575 | 3. **Resolve hub + signing identity** — from CLI flags or config/identity files. |
| 576 | 4. **Gather records** — read all local coordination files of the requested |
| 577 | kinds. |
| 578 | 5. **Push in batches** — POST each batch; track inserted/skipped counts. |
| 579 | 6. **Emit output** — compact JSON or human-readable text. |
| 580 | |
| 581 | Security |
| 582 | -------- |
| 583 | The signing identity is read from ``~/.muse/identity.toml``. |
| 584 | It is never included in output. ``--owner`` and |
| 585 | ``--slug`` are capped at :data:`_MAX_OWNER_LEN` and :data:`_MAX_SLUG_LEN` |
| 586 | respectively to prevent oversized HTTP requests. In text mode ``owner`` |
| 587 | and ``slug`` are passed through ``sanitize_display()`` before printing so |
| 588 | ANSI escape sequences cannot corrupt the terminal. |
| 589 | |
| 590 | JSON output fields (``--json`` / ``-j``) |
| 591 | ----------------------------------------- |
| 592 | ``schema_version`` |
| 593 | Muse version string at call time. |
| 594 | ``inserted`` |
| 595 | Number of records the hub accepted as new. |
| 596 | ``skipped`` |
| 597 | Number of records the hub already had (idempotent duplicates). |
| 598 | ``total`` |
| 599 | Total local records gathered for the push attempt. |
| 600 | ``failed`` |
| 601 | ``true`` if any batch raised :class:`~muse.core.coord_bus.CoordBusError`. |
| 602 | ``elapsed_seconds`` |
| 603 | Wall-clock time for the full operation, rounded to 3 decimal places. |
| 604 | |
| 605 | Exit codes |
| 606 | ---------- |
| 607 | 0 — all batches pushed successfully (or nothing to push) |
| 608 | 1 — bad arguments, auth error, or at least one batch failed |
| 609 | 2 — not inside a Muse repository |
| 610 | """ |
| 611 | as_json: bool = args.fmt == "json" |
| 612 | owner: str = args.owner |
| 613 | slug: str = args.slug |
| 614 | |
| 615 | # ── Input validation (before any file I/O) ──────────────────────────────── |
| 616 | |
| 617 | if len(owner) > _MAX_OWNER_LEN: |
| 618 | msg = f"--owner is too long ({len(owner)} chars; max {_MAX_OWNER_LEN})" |
| 619 | _err(msg, as_json, "bad_args") |
| 620 | raise SystemExit(ExitCode.USER_ERROR) |
| 621 | |
| 622 | if len(slug) > _MAX_SLUG_LEN: |
| 623 | msg = f"--slug is too long ({len(slug)} chars; max {_MAX_SLUG_LEN})" |
| 624 | _err(msg, as_json, "bad_args") |
| 625 | raise SystemExit(ExitCode.USER_ERROR) |
| 626 | |
| 627 | root = require_repo() |
| 628 | hub_url, signing = _resolve_hub_and_signing(args, as_json) |
| 629 | kinds: list[str] = args.kinds |
| 630 | |
| 631 | t0 = time.monotonic() |
| 632 | records = _gather_local_records(root, kinds) |
| 633 | |
| 634 | if not records: |
| 635 | elapsed = time.monotonic() - t0 |
| 636 | if as_json: |
| 637 | from muse._version import __version__ |
| 638 | print(json.dumps({ |
| 639 | "schema_version": __version__, |
| 640 | "inserted": 0, |
| 641 | "skipped": 0, |
| 642 | "total": 0, |
| 643 | "failed": False, |
| 644 | "errors": [], |
| 645 | "elapsed_seconds": round(elapsed, 3), |
| 646 | })) |
| 647 | else: |
| 648 | print(" (no local coordination records to push)") |
| 649 | return |
| 650 | |
| 651 | # Push in batches. |
| 652 | total_inserted = 0 |
| 653 | total_skipped = 0 |
| 654 | failed = False |
| 655 | errors: list[str] = [] |
| 656 | |
| 657 | for i in range(0, len(records), MAX_PUSH_BATCH): |
| 658 | batch = records[i : i + MAX_PUSH_BATCH] |
| 659 | try: |
| 660 | result = push_to_hub(hub_url, owner, slug, batch, signing) |
| 661 | total_inserted += int(result.get("inserted", 0)) |
| 662 | total_skipped += int(result.get("skipped", 0)) |
| 663 | except CoordBusError as exc: |
| 664 | logger.error("coord sync push: batch %d failed: %s", i // MAX_PUSH_BATCH + 1, exc) |
| 665 | if not as_json: |
| 666 | # Text mode: print error to stderr immediately so the operator |
| 667 | # sees it. In JSON mode we accumulate errors and emit them in |
| 668 | # the single final JSON object — printing here would produce |
| 669 | # multiple JSON objects on stdout which breaks any caller doing |
| 670 | # json.loads(stdout). |
| 671 | _err(str(exc), as_json=False, status="hub_error") |
| 672 | errors.append(str(exc)) |
| 673 | failed = True |
| 674 | |
| 675 | elapsed = time.monotonic() - t0 |
| 676 | |
| 677 | if as_json: |
| 678 | from muse._version import __version__ |
| 679 | print(json.dumps({ |
| 680 | "schema_version": __version__, |
| 681 | "inserted": total_inserted, |
| 682 | "skipped": total_skipped, |
| 683 | "total": len(records), |
| 684 | "failed": failed, |
| 685 | "errors": errors, |
| 686 | "elapsed_seconds": round(elapsed, 3), |
| 687 | })) |
| 688 | else: |
| 689 | status_icon = "❌" if failed else "✅" |
| 690 | print( |
| 691 | f"\n{status_icon} Pushed {len(records)} record(s) to " |
| 692 | f"{sanitize_display(owner)}/{sanitize_display(slug)}" |
| 693 | f" — inserted: {total_inserted}, skipped: {total_skipped}" |
| 694 | f" ({elapsed:.3f}s)" |
| 695 | ) |
| 696 | |
| 697 | if failed: |
| 698 | raise SystemExit(ExitCode.USER_ERROR) |
| 699 | |
| 700 | |
| 701 | def run_pull(args: argparse.Namespace) -> None: |
| 702 | """Pull coordination records from MuseHub. |
| 703 | |
| 704 | Writes pulled records to ``.muse/coordination/remote/<kind>/<uuid>.json``. |
| 705 | The cursor (last seen record ``id``) is returned so callers can pass it |
| 706 | back as ``--since-id`` on the next invocation for incremental pulls. |
| 707 | Exit 0 is returned even when 0 records are available. |
| 708 | |
| 709 | Execution order |
| 710 | --------------- |
| 711 | 1. **Validate inputs** — ``--owner`` / ``--slug`` length caps, ``--since-id`` |
| 712 | non-negative check, and ``--limit`` bounds (1–:data:`_MAX_PULL_LIMIT`) |
| 713 | enforced before any file I/O. |
| 714 | 2. **Resolve repo** — :func:`~muse.core.repo.require_repo`. |
| 715 | 3. **Resolve hub + signing identity** — from CLI flags or config/identity files. |
| 716 | 4. **Pull records** — POST ``/{owner}/{slug}/coord/pull``. |
| 717 | 5. **Write records** — :func:`_write_remote_records` sanitizes both |
| 718 | ``kind`` and ``record_uuid`` before constructing any filesystem path. |
| 719 | 6. **Emit output** — compact JSON or human-readable text. |
| 720 | |
| 721 | Security |
| 722 | -------- |
| 723 | ``kind`` and ``record_uuid`` from the hub response are validated against |
| 724 | an allowlist and a safe-UUID regex before use in filesystem paths (see |
| 725 | :func:`_write_remote_records`); a malicious hub cannot escape |
| 726 | ``.muse/coordination/remote/`` via path traversal. The signing identity is |
| 727 | never included in output. In text mode ``owner`` and ``slug`` are passed |
| 728 | through ``sanitize_display()`` so ANSI escape sequences cannot corrupt |
| 729 | the terminal. |
| 730 | |
| 731 | JSON output fields (``--json`` / ``-j``) |
| 732 | ----------------------------------------- |
| 733 | ``schema_version`` |
| 734 | Muse version string at call time. |
| 735 | ``count`` |
| 736 | Number of records returned by this pull. |
| 737 | ``cursor`` |
| 738 | The ``id`` of the last returned record. Pass as ``--since-id`` on |
| 739 | the next pull to fetch only newer records. ``0`` when no records |
| 740 | were returned. |
| 741 | ``records`` |
| 742 | Full list of record dicts as returned by the hub. |
| 743 | ``elapsed_seconds`` |
| 744 | Wall-clock time for the full operation, rounded to 3 decimal places. |
| 745 | |
| 746 | Exit codes |
| 747 | ---------- |
| 748 | 0 — pull succeeded (0 records is a valid success) |
| 749 | 1 — bad arguments, auth error, or hub communication error |
| 750 | 2 — not inside a Muse repository |
| 751 | """ |
| 752 | as_json: bool = args.fmt == "json" |
| 753 | owner: str = args.owner |
| 754 | slug: str = args.slug |
| 755 | |
| 756 | # ── Input validation (before any file I/O) ──────────────────────────────── |
| 757 | |
| 758 | if len(owner) > _MAX_OWNER_LEN: |
| 759 | msg = f"--owner is too long ({len(owner)} chars; max {_MAX_OWNER_LEN})" |
| 760 | _err(msg, as_json, "bad_args") |
| 761 | raise SystemExit(ExitCode.USER_ERROR) |
| 762 | |
| 763 | if len(slug) > _MAX_SLUG_LEN: |
| 764 | msg = f"--slug is too long ({len(slug)} chars; max {_MAX_SLUG_LEN})" |
| 765 | _err(msg, as_json, "bad_args") |
| 766 | raise SystemExit(ExitCode.USER_ERROR) |
| 767 | |
| 768 | since_id: int = args.since_id |
| 769 | if since_id < 0: |
| 770 | msg = f"--since-id must be >= 0, got {since_id}" |
| 771 | _err(msg, as_json, "bad_args") |
| 772 | raise SystemExit(ExitCode.USER_ERROR) |
| 773 | |
| 774 | raw_limit: int = args.limit |
| 775 | if not (1 <= raw_limit <= _MAX_PULL_LIMIT): |
| 776 | msg = f"--limit must be 1–{_MAX_PULL_LIMIT}, got {raw_limit}" |
| 777 | _err(msg, as_json, "bad_args") |
| 778 | raise SystemExit(ExitCode.USER_ERROR) |
| 779 | |
| 780 | limit: int = clamp_int(raw_limit, 1, _MAX_PULL_LIMIT, "limit") |
| 781 | kinds: list[str] = args.kinds or [] |
| 782 | |
| 783 | root = require_repo() |
| 784 | hub_url, signing = _resolve_hub_and_signing(args, as_json) |
| 785 | |
| 786 | t0 = time.monotonic() |
| 787 | try: |
| 788 | result = pull_from_hub(hub_url, owner, slug, since_id, kinds, limit, signing) |
| 789 | except CoordBusError as exc: |
| 790 | _err(str(exc), as_json, "hub_error") |
| 791 | raise SystemExit(ExitCode.USER_ERROR) |
| 792 | |
| 793 | pulled_records: list[JsonDict] = result.get("records", []) |
| 794 | cursor: int = result.get("cursor", 0) |
| 795 | |
| 796 | # Write remote records to disk (path-safe: kind + uuid are sanitized inside). |
| 797 | if pulled_records: |
| 798 | _write_remote_records(root, pulled_records) |
| 799 | |
| 800 | elapsed = time.monotonic() - t0 |
| 801 | |
| 802 | if as_json: |
| 803 | from muse._version import __version__ |
| 804 | print(json.dumps({ |
| 805 | "schema_version": __version__, |
| 806 | "count": len(pulled_records), |
| 807 | "cursor": cursor, |
| 808 | "records": pulled_records, |
| 809 | "elapsed_seconds": round(elapsed, 3), |
| 810 | })) |
| 811 | else: |
| 812 | print( |
| 813 | f"\n✅ Pulled {len(pulled_records)} new record(s) from " |
| 814 | f"{sanitize_display(owner)}/{sanitize_display(slug)}" |
| 815 | f" — cursor: {cursor} ({elapsed:.3f}s)" |
| 816 | ) |
| 817 | if pulled_records: |
| 818 | print(f" Records written to .muse/coordination/remote/") |
File History
2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
32 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
33 days ago