"""Muse-only VCS adapter backend (§4).""" from __future__ import annotations from adapters.base import BaseAdapter from adapters.errors import ReadError, WriteError from adapters.runner import CommandResult from adapters.types import ( AnchorResult, CommitResult, HeadResult, MirrorResult, RealignResult, StatusResult, ) GIT_FORBIDDEN = "git forbidden in this regime" class MuseOnlyAdapter(BaseAdapter): """MuseHub backend — git and mirror are hard no-ops.""" def status(self) -> StatusResult | ReadError: branch_result = self._muse("branch", "--show-current") if isinstance(branch_result, ReadError): return branch_result dirty_result = self._muse_dirty() if isinstance(dirty_result, ReadError): return dirty_result return StatusResult( regime=self.regime, dirty=dirty_result, branch=branch_result.stdout, notes=["canonical=muse", "git-forbidden"], muse_dirty=dirty_result, git_dirty=None, ) def read_head(self, ref: str) -> HeadResult | ReadError: if ref.startswith("origin/") or ref.startswith(self.config.vcs.git.remote + "/"): return ReadError("read_head", GIT_FORBIDDEN) muse_ref = ref.removeprefix("muse:") return self._muse_rev_parse_sha(muse_ref) def read_canonical_anchor(self) -> AnchorResult | ReadError: main = self.config.vcs.muse.main_branch if not main: return ReadError("read_canonical_anchor", "muse.main_branch not configured") head = self.read_head(f"muse:{main}") if isinstance(head, ReadError): return head return AnchorResult(anchor_sha=head.sha, source=f"muse:{main}") def realign(self, *, dry_run: bool, max_commits: int) -> RealignResult: return RealignResult( would_import=0, applied=False, from_ref=None, to_ref=None, reason="single-history", ) def commit_feature( self, *, branch: str, message: str, paths: list[str], ) -> CommitResult | ReadError | WriteError: unsafe = self._validate_paths(paths) if unsafe: return unsafe if self._is_protected_branch(branch): return WriteError( "commit_feature", f"refused protected branch {branch!r}", ) # §GSW.6.1: skip checkout when Muse HEAD is already on the branch — # dirty handover/roadmap paths must never fail the commit here. current = self._muse("rev-parse", "--abbrev-ref", "HEAD") if isinstance(current, ReadError): return current if current.stdout.strip() != branch: # §GSW.6.2: bare checkout refuses dirty tracked files (Muse 0.2.x); # retry with --autoshelf to carry them. --force is forbidden. checkout = self._muse("checkout", branch) if isinstance(checkout, ReadError): checkout = self._muse("checkout", "--autoshelf", branch) if isinstance(checkout, ReadError): return checkout if paths: for path in paths: # Muse 0.2.x has no top-level `add`; staging is `muse code add` # (matches the KH2 remediation string). Found live in GSW land-b. add = self._muse("code", "add", path) if isinstance(add, ReadError): return add commit = self._muse("commit", "-m", message) if isinstance(commit, ReadError): return commit head = self._muse_rev_parse_sha("HEAD") if isinstance(head, ReadError): return head return CommitResult(committed=True, sha=head.sha) def mirror(self, *, dry_run: bool) -> MirrorResult: return MirrorResult( diff_summary="", pushed=False, reason=GIT_FORBIDDEN, ) def _git(self, *args: str) -> CommandResult | ReadError: return ReadError("git", GIT_FORBIDDEN)