"""``muse coord sync`` — push/pull local coordination state to/from MuseHub. Enables distributed agent swarms to share coordination records (reservations, intents, releases, heartbeats, dependencies, tasks, claims) across machines via the MuseHub coordination bus. Sub-commands ------------ ``muse coord sync push`` Reads all local coordination records from ``.muse/coordination/`` and pushes them to MuseHub in batches. Records already on the hub are silently skipped (idempotent). Requires a valid signing identity. ``muse coord sync pull`` Pulls coordination records from MuseHub into local read-only JSON files under ``.muse/coordination/remote/``. Local agents can read these to discover reservations and tasks created by remote agents. Requires auth for private repos; public repos allow unauthenticated pull. Usage:: muse coord sync push --hub http://localhost:10003 --owner gabriel --slug myrepo muse coord sync pull --hub http://localhost:10003 --owner gabriel --slug myrepo muse coord sync pull --hub http://localhost:10003 --owner gabriel --slug myrepo \\ --since-id 42 --kinds reservation heartbeat Output (push):: ✅ Pushed 12 record(s) to gabriel/myrepo — inserted: 10, skipped: 2 Output (pull):: ✅ Pulled 8 new record(s) from gabriel/myrepo — cursor: 20 JSON output (push):: { "schema_version": "...", "inserted": 10, "skipped": 2, "total": 12, "failed": false, "elapsed_seconds": 0.123 } JSON output (pull):: { "schema_version": "...", "count": 8, "cursor": 20, "records": [...], "elapsed_seconds": 0.045 } Flags (common): ``--hub URL`` MuseHub base URL (default: read from ``[hub] url`` in ``config.toml``). ``--owner NAME`` Repository owner username. Maximum 256 characters. ``--slug NAME`` Repository URL slug. Maximum 256 characters. ``--json`` Emit results as JSON. Push-only flags: ``--kinds KIND [KIND ...]`` Limit push to specific record kinds (default: all kinds). Pull-only flags: ``--since-id N`` Return only records with id > N (default: 0 = all records). Must be >= 0. ``--kinds KIND [KIND ...]`` Filter by record kind. ``--limit N`` Maximum records to pull per call (1–1000, default: 500). Exit codes:: 0 — success 1 — bad arguments, auth failure, or hub communication error """ from __future__ import annotations import argparse import json import logging import pathlib import re import sys import time from datetime import datetime, timezone from typing import TYPE_CHECKING from muse.core.coord_bus import ( CoordBusError, MAX_PUSH_BATCH, pull_from_hub, push_to_hub, ) from muse.core.coord_bus import JsonDict from muse.core.errors import ExitCode from muse.core.repo import require_repo from muse.core.store import write_text_atomic from muse.core.validation import clamp_int, sanitize_display if TYPE_CHECKING: from muse.core.transport import SigningIdentity logger = logging.getLogger(__name__) # Coordination record kinds supported by the bus. _ALL_KINDS = ( "reservation", "intent", "release", "heartbeat", "dependency", "task", "claim", ) # Directory under .muse/coordination/ where pulled remote records are stored. _REMOTE_DIR_NAME = "remote" # ── Input constraints ───────────────────────────────────────────────────────── #: Maximum length of --owner and --slug to prevent oversized HTTP requests. _MAX_OWNER_LEN: int = 256 _MAX_SLUG_LEN: int = 256 #: Maximum records to pull per call. _MAX_PULL_LIMIT: int = 1000 #: Safe record UUID pattern — only alphanumeric, hyphens, underscores. #: Prevents path traversal when server-supplied record_uuid is used to #: construct filesystem paths under .muse/coordination/remote//.json. _SAFE_RECORD_UUID_RE: re.Pattern[str] = re.compile(r"^[a-zA-Z0-9_\-]{1,128}$") # ── Error helper ────────────────────────────────────────────────────────────── def _err(msg: str, as_json: bool = False, status: str = "error") -> None: """Print an error and return. Caller is responsible for raising SystemExit.""" if as_json: print(json.dumps({"error": msg, "status": status})) else: print(f"❌ {msg}", file=sys.stderr) def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register ``sync`` on *subparsers* (nested push/pull sub-subcommands). Wires ``push`` and ``pull`` sub-subcommands with their flags, defaults, and help text. Both sub-commands share ``--hub``, ``--owner``, ``--slug``, ``--format`` / ``--json``. Sets ``func`` to :func:`run_push` and :func:`run_pull` respectively. """ sync_parser = subparsers.add_parser( "sync", help="Push/pull coordination records to/from MuseHub.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) sync_subs = sync_parser.add_subparsers(dest="sync_command", metavar="SYNC_COMMAND") sync_subs.required = True # ── Common arguments shared by push and pull ────────────────────────────── def _add_common(p: argparse.ArgumentParser) -> None: p.add_argument( "--hub", default=None, metavar="URL", help="MuseHub base URL (default: from config.toml [hub] url).", ) p.add_argument( "--owner", required=True, metavar="NAME", help=f"Repository owner username (max {_MAX_OWNER_LEN} chars).", ) p.add_argument( "--slug", required=True, metavar="NAME", help=f"Repository URL slug (max {_MAX_SLUG_LEN} chars).", ) p.add_argument( "--format", "-f", default="text", dest="fmt", choices=("text", "json"), help="Output format: text (default) or json.", ) p.add_argument( "--json", "-j", action="store_const", const="json", dest="fmt", help="Shorthand for --format json.", ) # ── Pull ────────────────────────────────────────────────────────────────── pull_parser = sync_subs.add_parser( "pull", help="Pull coordination records from MuseHub.", description=( "Fetch coordination records from MuseHub and write them to\n" ".muse/coordination/remote//.json.\n" "Use --since-id for incremental pulls — pass the cursor\n" "from the previous JSON output back as --since-id on the next call.\n\n" "Agent quickstart\n" "----------------\n" " muse coord sync pull --owner gabriel --slug myrepo --json\n" " muse coord sync pull --owner gabriel --slug myrepo -j\n" " muse coord sync pull --owner gabriel --slug myrepo -j | jq .cursor\n" " muse coord sync pull --owner gabriel --slug myrepo \\\n" " --since-id 42 --kinds reservation intent -j\n\n" "JSON output schema\n" "------------------\n" ' {"schema_version": "", "count": , "cursor": ,\n' ' "records": [...], "elapsed_seconds": }\n\n' "Exit codes\n" "----------\n" " 0 — pull succeeded (0 records is a valid success)\n" " 1 — bad arguments, auth error, or hub communication error\n" " 2 — not inside a Muse repository\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) _add_common(pull_parser) pull_parser.add_argument( "--since-id", dest="since_id", type=int, default=0, metavar="N", help="Only pull records with id > N (default: 0 = all). Must be >= 0.", ) pull_parser.add_argument( "--kinds", nargs="+", choices=_ALL_KINDS, default=[], metavar="KIND", help=f"Filter by kind. Choices: {', '.join(_ALL_KINDS)}.", ) pull_parser.add_argument( "--limit", type=int, default=500, metavar="N", help=f"Maximum records to pull per call (1–{_MAX_PULL_LIMIT}, default: 500).", ) pull_parser.set_defaults(func=run_pull) # ── Push ────────────────────────────────────────────────────────────────── push_parser = sync_subs.add_parser( "push", help="Push local coordination records to MuseHub.", description=( "Read all local coordination files from .muse/coordination/ and\n" "POST them to MuseHub in batches. Records already on the hub are\n" "silently skipped (idempotent). Fails with exit 1 if any batch\n" "errors; other batches still complete.\n\n" "Agent quickstart\n" "----------------\n" " muse coord sync push --owner gabriel --slug myrepo --json\n" " muse coord sync push --owner gabriel --slug myrepo -j\n" " muse coord sync push --owner gabriel --slug myrepo -j | jq .inserted\n" " muse coord sync push --owner gabriel --slug myrepo \\\n" " --kinds reservation intent -j\n\n" "JSON output schema\n" "------------------\n" ' {"schema_version": "", "inserted": , "skipped": ,\n' ' "total": , "failed": , "elapsed_seconds": }\n\n' "Exit codes\n" "----------\n" " 0 — all batches pushed successfully (or nothing to push)\n" " 1 — bad arguments, auth error, or at least one batch failed\n" " 2 — not inside a Muse repository\n" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) _add_common(push_parser) push_parser.add_argument( "--kinds", nargs="+", choices=_ALL_KINDS, default=list(_ALL_KINDS), metavar="KIND", help=f"Limit push to these kinds (default: all). Choices: {', '.join(_ALL_KINDS)}.", ) push_parser.set_defaults(func=run_push) # ── Helpers ──────────────────────────────────────────────────────────────────── def _resolve_hub_and_signing( args: argparse.Namespace, as_json: bool = False, ) -> tuple[str, str | None]: """Return ``(hub_url, signing)`` from args or config. Reads ``[hub] url`` from ``config.toml`` and the signing identity from ``~/.muse/identity.toml`` when the corresponding CLI flags are omitted. Raises :class:`SystemExit` with exit code :attr:`~muse.core.errors.ExitCode.USER_ERROR` if no hub URL can be determined. Error messages go to *stdout* as JSON when *as_json* is ``True``, otherwise to *stderr*. Args: args: Parsed ``argparse.Namespace`` with ``hub`` and ``token`` attrs. as_json: Route error messages to stdout as JSON when ``True``. Returns: ``(hub_url, signing)`` where *signing* may be ``None`` for unauthenticated access to public repos. """ hub_url: str | None = args.hub signing: SigningIdentity | None = None if hub_url is None or signing is None: try: from muse.cli.config import get_hub_url, get_signing_identity if hub_url is None: hub_url = get_hub_url() if signing is None: signing = get_signing_identity(hub_url) except Exception as exc: # noqa: BLE001 if hub_url is None: msg = ( "no --hub URL supplied and could not read hub URL from config: " f"{exc}" ) _err(msg, as_json, "no_hub_url") raise SystemExit(ExitCode.USER_ERROR) if hub_url is None: _err("--hub URL is required", as_json, "no_hub_url") raise SystemExit(ExitCode.USER_ERROR) return hub_url, signing def _gather_local_records( root: pathlib.Path, kinds: list[str], ) -> list[JsonDict]: """Load local coordination records from ``.muse/coordination/``. Returns a flat list of serialized record dicts, each with the fields expected by the hub's push API: ``kind``, ``record_uuid``, ``run_id``, ``payload``, and ``expires_at``. Only kinds listed in *kinds* are included. Corrupt or unreadable files are silently skipped (logged at DEBUG level) so a single bad file never blocks a push. Args: root: Repository root directory (contains ``.muse/``). kinds: Subset of :data:`_ALL_KINDS` to include. Returns: List of record dicts ready for ``push_to_hub``. """ coord_dir = root / ".muse" / "coordination" records: list[JsonDict] = [] # ── Reservations ────────────────────────────────────────────────────────── if "reservation" in kinds: res_dir = coord_dir / "reservations" if res_dir.is_dir(): for fpath in res_dir.glob("*.json"): try: data = json.loads(fpath.read_text(encoding="utf-8")) records.append({ "kind": "reservation", "record_uuid": data.get("reservation_id", fpath.stem), "run_id": data.get("run_id", ""), "payload": data, "expires_at": data.get("expires_at"), }) except Exception: # noqa: BLE001 logger.debug("gather_local_records: skipping %s", fpath) # ── Heartbeats ──────────────────────────────────────────────────────────── if "heartbeat" in kinds: hb_dir = coord_dir / "heartbeats" if hb_dir.is_dir(): for fpath in hb_dir.glob("*.json"): try: data = json.loads(fpath.read_text(encoding="utf-8")) records.append({ "kind": "heartbeat", "record_uuid": data.get("run_id", fpath.stem), "run_id": data.get("run_id", ""), "payload": data, "expires_at": data.get("expires_at"), }) except Exception: # noqa: BLE001 logger.debug("gather_local_records: skipping heartbeat %s", fpath) # ── Intents ─────────────────────────────────────────────────────────────── if "intent" in kinds: intent_dir = coord_dir / "intents" if intent_dir.is_dir(): for fpath in intent_dir.glob("*.json"): try: data = json.loads(fpath.read_text(encoding="utf-8")) records.append({ "kind": "intent", "record_uuid": data.get("intent_id", fpath.stem), "run_id": data.get("run_id", ""), "payload": data, "expires_at": data.get("expires_at"), }) except Exception: # noqa: BLE001 logger.debug("gather_local_records: skipping intent %s", fpath) # ── Releases ────────────────────────────────────────────────────────────── if "release" in kinds: release_dir = coord_dir / "releases" if release_dir.is_dir(): for fpath in release_dir.glob("*.json"): try: data = json.loads(fpath.read_text(encoding="utf-8")) records.append({ "kind": "release", "record_uuid": data.get("release_id", fpath.stem), "run_id": data.get("run_id", ""), "payload": data, "expires_at": None, }) except Exception: # noqa: BLE001 logger.debug("gather_local_records: skipping release %s", fpath) # ── Dependencies ────────────────────────────────────────────────────────── if "dependency" in kinds: dep_dir = coord_dir / "dependencies" if dep_dir.is_dir(): for fpath in dep_dir.glob("*.json"): try: data = json.loads(fpath.read_text(encoding="utf-8")) records.append({ "kind": "dependency", "record_uuid": data.get("reservation_id", fpath.stem), "run_id": data.get("run_id", ""), "payload": data, "expires_at": None, }) except Exception: # noqa: BLE001 logger.debug("gather_local_records: skipping dependency %s", fpath) # ── Tasks ───────────────────────────────────────────────────────────────── if "task" in kinds: task_dir = coord_dir / "tasks" if task_dir.is_dir(): for fpath in task_dir.glob("*.json"): try: data = json.loads(fpath.read_text(encoding="utf-8")) records.append({ "kind": "task", "record_uuid": data.get("task_id", fpath.stem), "run_id": data.get("run_id", ""), "payload": data, "expires_at": None, }) except Exception: # noqa: BLE001 logger.debug("gather_local_records: skipping task %s", fpath) # ── Claims ──────────────────────────────────────────────────────────────── if "claim" in kinds: claim_dir = coord_dir / "claims" if claim_dir.is_dir(): for fpath in claim_dir.glob("*.json"): try: data = json.loads(fpath.read_text(encoding="utf-8")) records.append({ "kind": "claim", "record_uuid": data.get("task_id", fpath.stem), "run_id": data.get("claimer_run_id", ""), "payload": data, "expires_at": data.get("expires_at"), }) except Exception: # noqa: BLE001 logger.debug("gather_local_records: skipping claim %s", fpath) return records def _write_remote_records( root: pathlib.Path, records: list[JsonDict], ) -> None: """Write pulled remote records to ``.muse/coordination/remote/``. Each record is written to ``remote//.json``. Existing files are overwritten to keep remote state up to date. Security -------- Both *kind* and *record_uuid* come from the MuseHub response and must be sanitized before use in filesystem paths: * **kind** — validated against the :data:`_ALL_KINDS` allowlist. Records with an unrecognised kind are skipped; a server cannot write outside the ``remote/`` subtree by supplying ``"../evil"`` as the kind. * **record_uuid** — validated against :data:`_SAFE_RECORD_UUID_RE` (alphanumeric, hyphens, underscores, 1–128 chars). Records with an unsafe UUID are skipped; a server cannot escape the kind directory via ``"../../../etc/passwd"`` or similar. Args: root: Repository root directory (contains ``.muse/``). records: List of record dicts returned by :func:`pull_from_hub`. """ remote_dir = root / ".muse" / "coordination" / _REMOTE_DIR_NAME for rec in records: kind = rec.get("kind", "") record_uuid = rec.get("record_uuid", "") # Allowlist check — reject unknown kinds outright. if kind not in _ALL_KINDS: logger.warning("write_remote_records: rejected unknown kind %r", kind) continue # UUID safety check — reject traversal-capable strings. if not record_uuid or not _SAFE_RECORD_UUID_RE.match(record_uuid): logger.warning( "write_remote_records: rejected unsafe record_uuid %r (kind=%r)", record_uuid, kind, ) continue kind_dir = remote_dir / kind kind_dir.mkdir(parents=True, exist_ok=True) target = kind_dir / f"{record_uuid}.json" write_text_atomic(target, json.dumps(rec) + "\n") # ── Command handlers ─────────────────────────────────────────────────────────── def run_push(args: argparse.Namespace) -> None: """Push local coordination records to MuseHub. Loads all local coordination files from ``.muse/coordination/``, batches them into groups of at most :data:`~muse.core.coord_bus.MAX_PUSH_BATCH` records, and POSTs each batch to ``/{owner}/{slug}/coord/push``. Idempotent: records already on the hub are silently skipped. If any batch fails the remaining batches still run; ``failed`` is set to ``True`` in the output and the process exits 1. Execution order --------------- 1. **Validate inputs** — ``--owner`` and ``--slug`` length caps enforced before any file I/O. 2. **Resolve repo** — :func:`~muse.core.repo.require_repo`. 3. **Resolve hub + signing identity** — from CLI flags or config/identity files. 4. **Gather records** — read all local coordination files of the requested kinds. 5. **Push in batches** — POST each batch; track inserted/skipped counts. 6. **Emit output** — compact JSON or human-readable text. Security -------- The signing identity is read from ``~/.muse/identity.toml``. It is never included in output. ``--owner`` and ``--slug`` are capped at :data:`_MAX_OWNER_LEN` and :data:`_MAX_SLUG_LEN` respectively to prevent oversized HTTP requests. In text mode ``owner`` and ``slug`` are passed through ``sanitize_display()`` before printing so ANSI escape sequences cannot corrupt the terminal. JSON output fields (``--json`` / ``-j``) ----------------------------------------- ``schema_version`` Muse version string at call time. ``inserted`` Number of records the hub accepted as new. ``skipped`` Number of records the hub already had (idempotent duplicates). ``total`` Total local records gathered for the push attempt. ``failed`` ``true`` if any batch raised :class:`~muse.core.coord_bus.CoordBusError`. ``elapsed_seconds`` Wall-clock time for the full operation, rounded to 3 decimal places. Exit codes ---------- 0 — all batches pushed successfully (or nothing to push) 1 — bad arguments, auth error, or at least one batch failed 2 — not inside a Muse repository """ as_json: bool = args.fmt == "json" owner: str = args.owner slug: str = args.slug # ── Input validation (before any file I/O) ──────────────────────────────── if len(owner) > _MAX_OWNER_LEN: msg = f"--owner is too long ({len(owner)} chars; max {_MAX_OWNER_LEN})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) if len(slug) > _MAX_SLUG_LEN: msg = f"--slug is too long ({len(slug)} chars; max {_MAX_SLUG_LEN})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) root = require_repo() hub_url, signing = _resolve_hub_and_signing(args, as_json) kinds: list[str] = args.kinds t0 = time.monotonic() records = _gather_local_records(root, kinds) if not records: elapsed = time.monotonic() - t0 if as_json: from muse._version import __version__ print(json.dumps({ "schema_version": __version__, "inserted": 0, "skipped": 0, "total": 0, "failed": False, "errors": [], "elapsed_seconds": round(elapsed, 3), })) else: print(" (no local coordination records to push)") return # Push in batches. total_inserted = 0 total_skipped = 0 failed = False errors: list[str] = [] for i in range(0, len(records), MAX_PUSH_BATCH): batch = records[i : i + MAX_PUSH_BATCH] try: result = push_to_hub(hub_url, owner, slug, batch, signing) total_inserted += int(result.get("inserted", 0)) total_skipped += int(result.get("skipped", 0)) except CoordBusError as exc: logger.error("coord sync push: batch %d failed: %s", i // MAX_PUSH_BATCH + 1, exc) if not as_json: # Text mode: print error to stderr immediately so the operator # sees it. In JSON mode we accumulate errors and emit them in # the single final JSON object — printing here would produce # multiple JSON objects on stdout which breaks any caller doing # json.loads(stdout). _err(str(exc), as_json=False, status="hub_error") errors.append(str(exc)) failed = True elapsed = time.monotonic() - t0 if as_json: from muse._version import __version__ print(json.dumps({ "schema_version": __version__, "inserted": total_inserted, "skipped": total_skipped, "total": len(records), "failed": failed, "errors": errors, "elapsed_seconds": round(elapsed, 3), })) else: status_icon = "❌" if failed else "✅" print( f"\n{status_icon} Pushed {len(records)} record(s) to " f"{sanitize_display(owner)}/{sanitize_display(slug)}" f" — inserted: {total_inserted}, skipped: {total_skipped}" f" ({elapsed:.3f}s)" ) if failed: raise SystemExit(ExitCode.USER_ERROR) def run_pull(args: argparse.Namespace) -> None: """Pull coordination records from MuseHub. Writes pulled records to ``.muse/coordination/remote//.json``. The cursor (last seen record ``id``) is returned so callers can pass it back as ``--since-id`` on the next invocation for incremental pulls. Exit 0 is returned even when 0 records are available. Execution order --------------- 1. **Validate inputs** — ``--owner`` / ``--slug`` length caps, ``--since-id`` non-negative check, and ``--limit`` bounds (1–:data:`_MAX_PULL_LIMIT`) enforced before any file I/O. 2. **Resolve repo** — :func:`~muse.core.repo.require_repo`. 3. **Resolve hub + signing identity** — from CLI flags or config/identity files. 4. **Pull records** — POST ``/{owner}/{slug}/coord/pull``. 5. **Write records** — :func:`_write_remote_records` sanitizes both ``kind`` and ``record_uuid`` before constructing any filesystem path. 6. **Emit output** — compact JSON or human-readable text. Security -------- ``kind`` and ``record_uuid`` from the hub response are validated against an allowlist and a safe-UUID regex before use in filesystem paths (see :func:`_write_remote_records`); a malicious hub cannot escape ``.muse/coordination/remote/`` via path traversal. The signing identity is never included in output. In text mode ``owner`` and ``slug`` are passed through ``sanitize_display()`` so ANSI escape sequences cannot corrupt the terminal. JSON output fields (``--json`` / ``-j``) ----------------------------------------- ``schema_version`` Muse version string at call time. ``count`` Number of records returned by this pull. ``cursor`` The ``id`` of the last returned record. Pass as ``--since-id`` on the next pull to fetch only newer records. ``0`` when no records were returned. ``records`` Full list of record dicts as returned by the hub. ``elapsed_seconds`` Wall-clock time for the full operation, rounded to 3 decimal places. Exit codes ---------- 0 — pull succeeded (0 records is a valid success) 1 — bad arguments, auth error, or hub communication error 2 — not inside a Muse repository """ as_json: bool = args.fmt == "json" owner: str = args.owner slug: str = args.slug # ── Input validation (before any file I/O) ──────────────────────────────── if len(owner) > _MAX_OWNER_LEN: msg = f"--owner is too long ({len(owner)} chars; max {_MAX_OWNER_LEN})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) if len(slug) > _MAX_SLUG_LEN: msg = f"--slug is too long ({len(slug)} chars; max {_MAX_SLUG_LEN})" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) since_id: int = args.since_id if since_id < 0: msg = f"--since-id must be >= 0, got {since_id}" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) raw_limit: int = args.limit if not (1 <= raw_limit <= _MAX_PULL_LIMIT): msg = f"--limit must be 1–{_MAX_PULL_LIMIT}, got {raw_limit}" _err(msg, as_json, "bad_args") raise SystemExit(ExitCode.USER_ERROR) limit: int = clamp_int(raw_limit, 1, _MAX_PULL_LIMIT, "limit") kinds: list[str] = args.kinds or [] root = require_repo() hub_url, signing = _resolve_hub_and_signing(args, as_json) t0 = time.monotonic() try: result = pull_from_hub(hub_url, owner, slug, since_id, kinds, limit, signing) except CoordBusError as exc: _err(str(exc), as_json, "hub_error") raise SystemExit(ExitCode.USER_ERROR) pulled_records: list[JsonDict] = result.get("records", []) cursor: int = result.get("cursor", 0) # Write remote records to disk (path-safe: kind + uuid are sanitized inside). if pulled_records: _write_remote_records(root, pulled_records) elapsed = time.monotonic() - t0 if as_json: from muse._version import __version__ print(json.dumps({ "schema_version": __version__, "count": len(pulled_records), "cursor": cursor, "records": pulled_records, "elapsed_seconds": round(elapsed, 3), })) else: print( f"\n✅ Pulled {len(pulled_records)} new record(s) from " f"{sanitize_display(owner)}/{sanitize_display(slug)}" f" — cursor: {cursor} ({elapsed:.3f}s)" ) if pulled_records: print(f" Records written to .muse/coordination/remote/")