"""MuseHub sync service — push and pull protocol implementation. Implements the two core data-movement operations: - ``ingest_push``: stores commits and objects from a client push, enforcing fast-forward semantics and updating the branch head. - ``compute_pull_delta``: returns commits and objects the client does not yet have, keyed by their ``have_commits`` / ``have_objects`` exclusion lists. Object content is written to the blob store (R2/MinIO) and metadata (path, size, storage_uri) is persisted to Postgres. Boundary rules (same as musehub_repository): - Must NOT import state stores, SSE queues, or LLM clients. - May import ORM models from musehub.db domain-specific modules. - May import Pydantic models from musehub.models.musehub. """ import base64 import datetime import hashlib import logging from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from muse.core.types import blob_id from musehub.core.genesis import compute_branch_id from musehub.db.musehub_repo_models import MusehubBranch, MusehubCommit, MusehubCommitRef, MusehubObject, MusehubObjectRef, MusehubRepo, MusehubSnapshot from musehub.models.musehub import ( CommitInput, CommitResponse, ObjectInput, ObjectResponse, PullResponse, PushResponse, SnapshotInput, ) logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _to_commit_response(row: MusehubCommit) -> CommitResponse: return CommitResponse( commit_id=row.commit_id, branch=row.branch, parent_ids=list(row.parent_ids or []), message=row.message, author=row.author, timestamp=row.timestamp, snapshot_id=row.snapshot_id, ) async def _to_object_response(row: MusehubObject) -> ObjectResponse: """Fetch object bytes from the blob store and return as base64-encoded response.""" from musehub.storage.backends import get_backend data = await get_backend().get(row.object_id) if data is None: logger.warning("⚠️ Object missing from blob store: object_id=%s", row.object_id) data = b"" return ObjectResponse( object_id=row.object_id, path=row.path, content_b64=base64.b64encode(data).decode(), ) def _is_fast_forward( remote_head: str | None, head_commit_id: str, commits: list[CommitInput], ) -> bool: """Return True if the push is a fast-forward update. A push is fast-forward when: - the remote branch has no head yet (first push), or - the new head_commit_id equals the remote head (no-op), or - the remote head appears somewhere in the ancestry graph of the pushed commits (meaning the client built on top of the remote head). We build a local graph from the pushed commits and walk parents. This does NOT query the DB for previously stored commits — for MVP the client is expected to include all commits since the common ancestor. """ if remote_head is None: return True if head_commit_id == remote_head: return True from muse.core.graph import walk_dag parent_map = {c.commit_id: c.parent_ids for c in commits} for cid in walk_dag(head_commit_id, lambda cid: parent_map.get(cid, [])): if cid == remote_head: return True return False # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- async def ingest_push( session: AsyncSession, *, repo_id: str, branch: str, head_commit_id: str, commits: list[CommitInput], snapshots: list[SnapshotInput] | None = None, objects: list[ObjectInput], force: bool, author: str, ) -> PushResponse: """Store commits, snapshot manifests, and objects from a push; update the branch head. Execution steps (in order): 1. **Resolve / create branch** — upsert the branch row; first push initialises ``head_commit_id = None`` and ``default_branch`` is set only when this is the repo's inaugural branch. 2. **Fast-forward check** — BFS traverses both ``parent_commit_id`` and ``parent2_commit_id`` to support merge commits. Rejected with ``ValueError("non_fast_forward")`` when the current tip is not an ancestor of the new head and ``force`` is False. The route handler maps this to HTTP 409. 3. **Upsert commits** — existing commit IDs are skipped; new rows are bulk-inserted. 4. **Upsert snapshots** — each :class:`SnapshotInput` is stored with ``(repo_id, snapshot_id)`` as a composite key. Re-pushing an identical snapshot is a safe no-op (idempotent via ``merge`` / ``ON CONFLICT DO NOTHING``). 5. **Upsert objects** — binary blobs are written to the configured storage backend and their metadata rows are inserted or skipped if already present. 6. **Update branch head** — the branch row's ``head_commit_id`` is set to ``head_commit_id`` and ``pushed_at`` is refreshed. Args: session: Active async SQLAlchemy session. repo_id: Repository ID; must already exist in the DB. branch: Target branch name (created on first push). head_commit_id: SHA of the new branch tip after the push. commits: Ordered list of :class:`CommitInput` objects to store. snapshots: Optional list of :class:`SnapshotInput` objects; ``None`` or ``[]`` are both treated as "no snapshots in this push". objects: Content-addressed blob payloads. force: When ``True``, skip the fast-forward check (destructive). author: Username performing the push (used for the response summary). Returns: :class:`PushResponse` with commit / snapshot / object counts and the new branch head commit ID. Raises: ValueError: with key ``"non_fast_forward"`` when the update would create a non-linear history and ``force`` is False. """ # ------------------------------------------------------------------ # 1. Resolve (or create) the branch # ------------------------------------------------------------------ branch_row = await _get_or_create_branch(session, repo_id=repo_id, branch=branch) # ------------------------------------------------------------------ # 2. Fast-forward check # ------------------------------------------------------------------ if not force and not _is_fast_forward(branch_row.head_commit_id, head_commit_id, commits): logger.warning( "⚠️ Non-fast-forward push rejected for repo=%s branch=%s remote_head=%s new_head=%s", repo_id, branch, branch_row.head_commit_id, head_commit_id, ) raise ValueError("non_fast_forward") # ------------------------------------------------------------------ # 3. Upsert commits # ------------------------------------------------------------------ existing_commit_ids: set[str] = set() if commits: stmt = select(MusehubCommitRef.commit_id).where( MusehubCommitRef.repo_id == repo_id, MusehubCommitRef.commit_id.in_([c.commit_id for c in commits]), ) result = await session.execute(stmt) existing_commit_ids = set(result.scalars().all()) # ------------------------------------------------------------------ # 3a. Parent existence validation (Phase 8 / invariant 8 parity) # # Every parent_id referenced by an incoming commit must either: # (a) be in this push mpack itself, or # (b) already exist in the DB for this repo. # # Without this guard, a client can push a commit whose parent is a # fabricated or missing ID. `muse log` then walks off the end of # history trying to read the parent — silent data corruption. # # Mirrors the same validation in wire_push_unpack_mpack() so the ingest # path is equally safe. # ------------------------------------------------------------------ mpack_commit_ids: set[str] = {c.commit_id for c in commits} external_parent_ids: set[str] = set() for c in commits: if c.commit_id in existing_commit_ids: continue for pid in (c.parent_ids or []): if pid and pid not in mpack_commit_ids: external_parent_ids.add(pid) if external_parent_ids: db_parent_ids: set[str] = set( (await session.execute( select(MusehubCommitRef.commit_id).where( MusehubCommitRef.commit_id.in_(external_parent_ids), MusehubCommitRef.repo_id == repo_id, ) )).scalars().all() ) missing_parents = external_parent_ids - db_parent_ids if missing_parents: short = ", ".join(p[:16] for p in sorted(missing_parents)) raise ValueError(f"missing_parent_commits: {short}") new_commits: list[MusehubCommit] = [] new_commit_refs: list[MusehubCommitRef] = [] for c in commits: if c.commit_id in existing_commit_ids: new_commit_refs.append(MusehubCommitRef(repo_id=repo_id, commit_id=c.commit_id)) continue row = MusehubCommit( commit_id=c.commit_id, branch=branch, parent_ids=c.parent_ids, message=c.message, author=c.author if c.author else author, timestamp=c.timestamp, committed_at_raw=c.timestamp.isoformat() if c.timestamp else None, snapshot_id=c.snapshot_id, ) new_commits.append(row) new_commit_refs.append(MusehubCommitRef(repo_id=repo_id, commit_id=c.commit_id)) if new_commits: session.add_all(new_commits) logger.info("✅ Ingested %d new commits for repo=%s", len(new_commits), repo_id) if new_commit_refs: session.add_all(new_commit_refs) # ------------------------------------------------------------------ # 4. Upsert snapshots (idempotent — always writes entries) # ------------------------------------------------------------------ from musehub.services.musehub_snapshot import upsert_snapshot_entries ingest_snapshots = snapshots or [] for s in ingest_snapshots: if not s.snapshot_id: continue manifest = s.manifest if isinstance(s.manifest, dict) else {} await upsert_snapshot_entries(session, repo_id, s.snapshot_id, manifest) if ingest_snapshots: logger.info("✅ Ingested %d snapshots for repo=%s", len(ingest_snapshots), repo_id) # ------------------------------------------------------------------ # 5. Upsert objects (write bytes to per-repo store, metadata to DB) # ------------------------------------------------------------------ existing_object_ids: set[str] = set() if objects: stmt_obj = select(MusehubObjectRef.object_id).where( MusehubObjectRef.repo_id == repo_id, MusehubObjectRef.object_id.in_([o.object_id for o in objects]), ) res_obj = await session.execute(stmt_obj) existing_object_ids = set(res_obj.scalars().all()) repo_row = await session.get(MusehubRepo, repo_id) if objects else None for obj in objects: if obj.object_id in existing_object_ids: continue await _write_object( session, repo_id=repo_id, obj=obj, ) # ------------------------------------------------------------------ # 6. Update branch head # ------------------------------------------------------------------ branch_row.head_commit_id = head_commit_id logger.info( "✅ Branch '%s' head updated to %s for repo=%s", branch, head_commit_id, repo_id, ) await session.flush() return PushResponse(ok=True, remote_head=head_commit_id) _PULL_OBJECTS_PAGE_SIZE: int = 500 # max objects returned per pull response async def compute_pull_delta( session: AsyncSession, *, repo_id: str, branch: str, have_commits: list[str], have_objects: list[str], cursor: str | None = None, ) -> PullResponse: """Return commits and objects the caller does not have. Objects are paginated at ``_PULL_OBJECTS_PAGE_SIZE`` items per response to prevent a single pull from returning an unbounded payload (OOM / timeout). Pagination protocol: - First call: ``cursor=None`` - When ``has_more=True``, re-issue with ``cursor=response.next_cursor`` - The ``next_cursor`` is the ``object_id`` of the last item in this page; the next page starts *after* that ID (keyset pagination, stable sort). Commits are never paginated — a repo's commit graph is O(kB per commit), so even 10 000 commits stay well within a single JSON response. """ branch_row = await _get_branch(session, repo_id=repo_id, branch=branch) remote_head = branch_row.head_commit_id if branch_row else None # ------------------------------------------------------------------ # Missing commits (no pagination — commit metadata is small) # ------------------------------------------------------------------ commit_stmt = ( select(MusehubCommit) .join(MusehubCommitRef, MusehubCommitRef.commit_id == MusehubCommit.commit_id) .where(MusehubCommitRef.repo_id == repo_id, MusehubCommit.branch == branch) ) if have_commits: commit_stmt = commit_stmt.where( MusehubCommit.commit_id.notin_(have_commits) ) commit_rows = (await session.execute(commit_stmt)).scalars().all() missing_commits = [_to_commit_response(r) for r in commit_rows] # ------------------------------------------------------------------ # Missing objects — keyset-paginated, capped at _PULL_OBJECTS_PAGE_SIZE # ------------------------------------------------------------------ obj_stmt = ( select(MusehubObject) .join(MusehubObjectRef, MusehubObject.object_id == MusehubObjectRef.object_id) .where(MusehubObjectRef.repo_id == repo_id) .order_by(MusehubObject.object_id) # stable keyset sort .limit(_PULL_OBJECTS_PAGE_SIZE + 1) # fetch one extra to detect next page ) if have_objects: obj_stmt = obj_stmt.where(MusehubObject.object_id.notin_(have_objects)) if cursor: # Resume after the last seen object_id (keyset: strictly greater-than) obj_stmt = obj_stmt.where(MusehubObject.object_id > cursor) obj_rows = list((await session.execute(obj_stmt)).scalars().all()) has_more = len(obj_rows) > _PULL_OBJECTS_PAGE_SIZE if has_more: obj_rows = obj_rows[:_PULL_OBJECTS_PAGE_SIZE] import asyncio missing_objects = list(await asyncio.gather(*[_to_object_response(r) for r in obj_rows])) next_cursor = obj_rows[-1].object_id if has_more and obj_rows else None logger.info( "✅ Pull delta: %d commits, %d objects (has_more=%s) for repo=%s branch=%s", len(missing_commits), len(missing_objects), has_more, repo_id, branch, ) return PullResponse( commits=missing_commits, objects=missing_objects, remote_head=remote_head, has_more=has_more, next_cursor=next_cursor, ) # --------------------------------------------------------------------------- # Private helpers # --------------------------------------------------------------------------- async def _get_branch( session: AsyncSession, *, repo_id: str, branch: str ) -> MusehubBranch | None: stmt = select(MusehubBranch).where( MusehubBranch.repo_id == repo_id, MusehubBranch.name == branch, ) return (await session.execute(stmt)).scalar_one_or_none() async def _get_or_create_branch( session: AsyncSession, *, repo_id: str, branch: str ) -> MusehubBranch: existing = await _get_branch(session, repo_id=repo_id, branch=branch) if existing is not None: return existing new_branch = MusehubBranch(branch_id=compute_branch_id(repo_id, branch), repo_id=repo_id, name=branch) session.add(new_branch) await session.flush() logger.info("✅ Created branch '%s' for repo=%s", branch, repo_id) return new_branch async def _write_object( session: AsyncSession, *, repo_id: str, obj: ObjectInput, ) -> None: """Decode base64 content, write to the blob store and insert metadata row.""" from musehub.storage.backends import get_backend raw = base64.b64decode(obj.content_b64) uri = await get_backend().put(obj.object_id, raw) row = MusehubObject( object_id=obj.object_id, path=obj.path, size_bytes=len(raw), storage_uri=uri, ) session.add(row) session.add(MusehubObjectRef(repo_id=repo_id, object_id=obj.object_id)) logger.info( "✅ Stored object %s (%d bytes) for repo=%s", obj.object_id, len(raw), repo_id, ) async def commit_files_to_repo( session: AsyncSession, *, repo_id: str, branch: str, files: dict[str, bytes], message: str, author: str, delete_paths: list[str] | None = None, ) -> str: """Write files to a repo as a single commit and return the new commit_id. Creates MusehubObject rows, a MusehubSnapshot, a MusehubCommit, and advances (or creates) the branch head. Used for server-side commits that originate inside MuseHub rather than from a client push. Args: session: Active async SQLAlchemy session (caller owns transaction). repo_id: Target repo ID — must already exist. branch: Branch name to commit to (created if absent). files: ``{path: raw_bytes}`` mapping — paths to add/update. message: Commit message. author: Identity handle or ID of the author. delete_paths: Paths to remove from the HEAD manifest before committing. The current HEAD snapshot is read, the paths are excluded, and the resulting manifest is merged with ``files``. Returns: The new commit_id (``sha256:``). """ from musehub.muse_cli.snapshot import compute_snapshot_id from musehub.services.musehub_snapshot import upsert_snapshot_entries from musehub.storage.backends import get_backend repo_row = await session.get(MusehubRepo, repo_id) if repo_row is None: raise ValueError(f"commit_files_to_repo: repo not found: {repo_id}") backend = get_backend() # 1a. Build base manifest from HEAD when deletions are requested. base_manifest: dict[str, str] = {} if delete_paths: branch_row_pre = await _get_or_create_branch(session, repo_id=repo_id, branch=branch) if branch_row_pre.head_commit_id: head_commit = await session.get(MusehubCommit, branch_row_pre.head_commit_id) if head_commit and head_commit.snapshot_id: import msgpack as _mp head_snap = await session.get(MusehubSnapshot, head_commit.snapshot_id) if head_snap: base_manifest = _mp.unpackb(head_snap.manifest_blob, raw=False) for p in delete_paths: base_manifest.pop(p, None) # 1b. Write each new/updated file as an object. manifest: dict[str, str] = {**base_manifest} for path, content in files.items(): oid = blob_id(content) existing = await session.get(MusehubObject, oid) if existing is None: uri = await backend.put(oid, content) session.add(MusehubObject( object_id=oid, path=path, size_bytes=len(content), storage_uri=uri, )) session.add(MusehubObjectRef(repo_id=repo_id, object_id=oid)) manifest[path] = oid # 2. Upsert snapshot snap_id = compute_snapshot_id(manifest) await upsert_snapshot_entries(session, repo_id, snap_id, manifest) # 3. Resolve parent commit from existing branch head now = datetime.datetime.now(datetime.timezone.utc) branch_row = await _get_or_create_branch(session, repo_id=repo_id, branch=branch) parent_ids: list[str] = [] if branch_row.head_commit_id: parent_ids = [branch_row.head_commit_id] # 4. Derive a deterministic commit_id from content commit_seed = f"{repo_id}\x00{branch}\x00{snap_id}\x00{message}\x00{now.isoformat()}" commit_id = blob_id(commit_seed.encode()) session.add(MusehubCommit( commit_id=commit_id, branch=branch, parent_ids=parent_ids, message=message, author=author, timestamp=now, committed_at_raw=now.isoformat(), snapshot_id=snap_id, )) session.add(MusehubCommitRef(repo_id=repo_id, commit_id=commit_id)) # 5. Advance branch head branch_row.head_commit_id = commit_id await session.flush() logger.info( "✅ commit_files_to_repo: repo=%s branch=%s commit=%s files=%d", repo_id, branch, commit_id[:20], len(files), ) return commit_id