"""Code domain plugin — semantic version control for source code. This plugin implements :class:`~muse.domain.MuseDomainPlugin` and :class:`~muse.domain.AddressedMergePlugin` for software repositories. Philosophy ---------- Git models files as sequences of lines. The code plugin models them as **collections of named symbols** — functions, classes, methods, variables. Two commits that only reformat a Python file (no semantic change) produce identical symbol ``content_id`` values and therefore *no* structured delta. Two commits that rename a function produce a ``ReplaceOp`` annotated ``"renamed to bar"`` rather than a red/green line diff. Live State ---------- ``LiveState`` is either a ``pathlib.Path`` pointing to the repository root or a ``SnapshotManifest`` dict. The path form is used by the CLI; the dict form is used by in-memory merge and diff operations. Snapshot Format --------------- A code snapshot is a ``SnapshotManifest``: .. code-block:: json { "files": { "src/utils.py": "", "README.md": "" }, "domain": "code" } The ``files`` values are **raw-bytes SHA-256 hashes** (not AST hashes). This ensures the object store can correctly restore files verbatim on ``muse checkout``. Semantic identity (AST-based hashing) is used only inside ``diff()`` when constructing the structured delta. Delta Format ------------ ``diff()`` returns a ``StructuredDelta``. For Python files (and other languages with adapters) it produces ``PatchOp`` entries whose ``child_ops`` carry symbol-level operations: - ``AddressedInsertOp`` — a symbol was added (address ``"src/utils.py::my_func"``). - ``AddressedDeleteOp`` — a symbol was removed. - ``ReplaceOp`` — a symbol changed. The ``new_summary`` field describes the change: ``"renamed to bar"``, ``"implementation changed"``, etc. Non-Python files produce coarse ``AddressedInsertOp`` / ``AddressedDeleteOp`` / ``ReplaceOp`` at the file level. Merge Semantics --------------- The plugin implements :class:`~muse.domain.AddressedMergePlugin` — a Map CRDT with address-keyed operations and Harmony for conflict resolution: - Agent A modifies ``foo()`` and Agent B modifies ``bar()`` in the same file → **auto-merge** (ops on different addresses commute). - Both agents modify ``foo()`` → **symbol-level conflict** at address ``"src/utils.py::foo"`` rather than a coarse file conflict. Schema ------ The code domain schema declares five dimensions: ``structure`` The module/file tree — ``TreeSchema`` with GumTree diff. ``symbols`` The AST symbol tree — ``TreeSchema`` with GumTree diff. ``imports`` The import set — ``SetSchema`` with ``by_content`` identity. ``variables`` Top-level variable assignments — ``SetSchema``. ``metadata`` Configuration and non-code files — ``SetSchema``. """ import logging import os import pathlib import stat as _stat from muse._version import __version__ from muse.core.merge_debug import merge_debug_log from muse.core.attributes import load_attributes, resolve_strategy from muse.core.diff_algorithms import snapshot_diff from muse.core.ignore import is_ignored, load_ignore_config, resolve_patterns from muse.core.snapshot import ( _BUILTIN_SECRET_PATTERNS, detect_directory_renames, directories_from_manifest, load_ignore_patterns, ) from muse.core.cohen_transform import compute_regions from muse.core.object_store import read_object, write_object from muse.core.types import blob_id, hash_file from muse.core.op_merge import merge_op_lists, ops_commute from muse.core.stat_cache import load_cache from muse.core.schema import ( DimensionSpec, DomainSchema, SetSchema, TreeSchema, ) from muse.domain import ( AddressedDeleteOp, AddressedInsertOp, ConflictRecord, RenameOp, DomainOp, DirStatus, DriftReport, LiveState, MergeResult, PatchOp, SnapshotManifest, StagedEntry, StageStatus, StateDelta, StateSnapshot, StructuredDelta, ) from muse.plugins.code.stage import ( clear_stage, read_stage, read_stage_dir_renames, stage_path, write_stage, ) from muse.plugins.code.ast_parser import ( SymbolTree, parse_symbols, ) from muse.plugins.code.symbol_diff import ( build_diff_ops, delta_summary, ) from muse.core.types import Manifest logger = logging.getLogger(__name__) type FileStrMap = dict[str, str] # file_path → hash/content (generic manifest) type SymbolTreeMap = dict[str, SymbolTree] # file_path → symbol tree type PatchOpMap = dict[str, PatchOp] # address → patch op type AppliedStrategies = dict[str, str] # file_path → strategy name type StagedEntryMap = dict[str, StagedEntry] # file_path → staged entry _DOMAIN_NAME = "code" # Directories that are never versioned regardless of .museignore. # These are implicit ignores that apply to all code repositories. _ALWAYS_IGNORE_DIRS: frozenset[str] = frozenset({ ".git", ".muse", "__pycache__", ".mypy_cache", ".pytest_cache", ".ruff_cache", "node_modules", ".venv", "venv", ".tox", ".nox", ".coverage", "htmlcov", "dist", "build", ".eggs", ".DS_Store", }) def _head_manifest_for(root: pathlib.Path) -> Manifest: """Return the manifest from the current HEAD commit (empty dict if none). Resolves the branch through ``get_head_commit_id`` (the store abstraction) rather than reading the ref file directly. Used by ``snapshot()`` (to build the staged manifest) and ``stage_status()`` (to compute the unstaged diff). Kept outside the class so it can be called from module-level helpers without a plugin instance. """ from muse.core.refs import ( get_head_commit_id, read_current_branch, ) from muse.core.commits import read_commit from muse.core.snapshots import read_snapshot try: branch = read_current_branch(root) commit_id = get_head_commit_id(root, branch) if not commit_id: return {} commit = read_commit(root, commit_id) if commit is None: return {} snap = read_snapshot(root, commit.snapshot_id) return dict(snap.manifest) if snap else {} except Exception: return {} def _head_snapshot_dirs_for(root: pathlib.Path) -> list[str]: """Return the ``directories`` list from the current HEAD snapshot. Used by ``stage_status`` to detect explicitly-committed empty directories. Returns an empty list for repositories with no commits or a missing snapshot. """ from muse.core.refs import ( get_head_commit_id, read_current_branch, ) from muse.core.commits import read_commit from muse.core.snapshots import read_snapshot as _read_snapshot try: branch = read_current_branch(root) commit_id = get_head_commit_id(root, branch) if not commit_id: return [] commit = read_commit(root, commit_id) if commit is None: return [] snap = _read_snapshot(root, commit.snapshot_id) return list(snap.directories) if snap else [] except Exception: return [] class CodePlugin: """Muse domain plugin for software source code repositories. Implements all six core protocol methods plus the optional :class:`~muse.domain.AddressedMergePlugin` address-keyed map merge extension and the :class:`~muse.domain.StagePlugin` selective-commit extension. The plugin does not implement :class:`~muse.domain.CRDTPlugin` — source code is human-authored and benefits from explicit conflict resolution (via Harmony) rather than automatic convergence. The plugin is stateless. The module-level singleton :data:`plugin` is the standard entry point. """ # ------------------------------------------------------------------ # 1. snapshot # ------------------------------------------------------------------ def snapshot(self, live_state: LiveState) -> StateSnapshot: """Capture the current working tree as a snapshot dict. Walks all regular files under *live_state*, hashing each one with SHA-256 (raw bytes). Honours ``.museignore`` and always ignores known tool-generated directories (``__pycache__``, ``.git``, etc.). Uses ``os.walk`` with in-place ``dirnames`` pruning so that always-ignored and hidden directories (e.g. ``.venv/``, ``node_modules/``, ``.muse/``) are never descended into. The ``StatCache`` is consulted before hashing so that unchanged files are not re-read from disk. Args: live_state: A ``pathlib.Path`` pointing to the repository root, or an existing ``SnapshotManifest`` dict (returned as-is). Returns: A ``SnapshotManifest`` mapping workspace-relative POSIX paths to their SHA-256 raw-bytes digests. """ if not isinstance(live_state, pathlib.Path): return live_state workdir = live_state patterns = _BUILTIN_SECRET_PATTERNS + resolve_patterns(load_ignore_config(workdir), _DOMAIN_NAME) cache = load_cache(workdir) files: Manifest = {} musekeep_dirs: set[str] = set() root_str = str(workdir) prefix_len = len(root_str) + 1 for dirpath, dirnames, filenames in os.walk(root_str, followlinks=False): rel_dir = "" if dirpath != root_str: rel_dir = dirpath[prefix_len:] if os.sep != "/": rel_dir = rel_dir.replace(os.sep, "/") # Prune subdirectories: skip _ALWAYS_IGNORE_DIRS, directories whose # contents are entirely ignored by .museignore, and nested muse # repos. Nested repos are independent version-controlled units — # their files belong to their own snapshot, not the parent repo's. # Using a sentinel file path (e.g. ".hypothesis/_") lets is_ignored # detect both directory patterns (.hypothesis/) and glob patterns # (.hypothesis/**) without descending into the directory first. dirnames[:] = sorted( d for d in dirnames if d not in _ALWAYS_IGNORE_DIRS and not is_ignored( f"{rel_dir}/{d}/_" if rel_dir else f"{d}/_", patterns ) and not os.path.isdir(os.path.join(dirpath, d, ".muse")) ) for fname in sorted(filenames): abs_str = os.path.join(dirpath, fname) try: st = os.lstat(abs_str) except OSError: continue if not _stat.S_ISREG(st.st_mode): continue rel = abs_str[prefix_len:] if os.sep != "/": rel = rel.replace(os.sep, "/") if is_ignored(rel, patterns): continue files[rel] = cache.get_cached(rel, abs_str, st.st_mtime, st.st_size, st.st_ino) if fname == ".musekeep" and rel_dir: musekeep_dirs.add(rel_dir) cache.prune(set(files)) cache.save() # If a stage index is active, filter the manifest so that only staged # files (using their staged object IDs) and previously-committed files # (at their committed state) are included. This is the core of the # selective-commit model: unstaged working-tree changes are invisible # to ``muse commit``. stage = read_stage(workdir) if stage: committed = _head_manifest_for(workdir) staged_manifest: Manifest = {} # Strip any VCS-internal paths that may have leaked into a prior snapshot. staged_manifest.update( {k: v for k, v in committed.items() if not k.startswith(".muse/")} ) for rel_path, entry in stage.items(): if rel_path.startswith(".muse/"): continue # never allow VCS internals in snapshot if entry.get("kind") == "dir": continue # sentinel entries go into directories, not files if entry["mode"] == "D": staged_manifest.pop(rel_path, None) else: staged_manifest[rel_path] = entry["object_id"] # Include explicitly-staged empty directories (sentinel entries). staged_empty_dirs: set[str] = { rel for rel, entry in stage.items() if entry.get("kind") == "dir" and entry["mode"] == "A" } staged_deleted_dirs: set[str] = { rel for rel, entry in stage.items() if entry.get("kind") == "dir" and entry["mode"] == "D" } # Carry forward committed empty dirs from HEAD, minus any staged # for deletion. Without this, committing a second directory drops # the first from snapshot.directories. committed_empty_dirs = set(_head_snapshot_dirs_for(workdir)) staged_dirs = sorted( ( set(directories_from_manifest(staged_manifest)) | staged_empty_dirs | (committed_empty_dirs - staged_deleted_dirs) ) # Invariant: a path can never be tracked as both a file and # a directory at once. A path that has become a file this # commit (present in staged_manifest) must never also carry # forward a stale directory entry from HEAD, even if the # staging layer failed to explicitly record its removal. - set(staged_manifest) ) return SnapshotManifest(files=staged_manifest, domain=_DOMAIN_NAME, directories=staged_dirs) # Only include directories that contain tracked files. Empty # directories are tracked explicitly via staging (sentinel entries). dirs = sorted(directories_from_manifest(files)) return SnapshotManifest(files=files, domain=_DOMAIN_NAME, directories=dirs) def workdir_snapshot(self, root: pathlib.Path) -> SnapshotManifest: """Capture the raw working tree, bypassing any active stage. Identical to :meth:`snapshot` but skips the stage-overlay logic, so every on-disk file is reflected at its current content hash. Used by ``muse diff --working``. """ patterns = _BUILTIN_SECRET_PATTERNS + resolve_patterns(load_ignore_config(root), _DOMAIN_NAME) cache = load_cache(root) files: Manifest = {} root_str = str(root) prefix_len = len(root_str) + 1 on_disk_dirs: set[str] = set() for dirpath, dirnames, filenames in os.walk(root_str, followlinks=False): rel_dir = "" if dirpath != root_str: rel_dir = dirpath[prefix_len:] if os.sep != "/": rel_dir = rel_dir.replace(os.sep, "/") dirnames[:] = sorted( d for d in dirnames if d not in _ALWAYS_IGNORE_DIRS and not is_ignored( f"{rel_dir}/{d}/_" if rel_dir else f"{d}/_", patterns ) and not os.path.isdir(os.path.join(dirpath, d, ".muse")) ) # Track every non-root directory that exists on disk so that # committed empty dirs still present in the working tree are not # falsely reported as deleted by the diff engine. if rel_dir: on_disk_dirs.add(rel_dir) for fname in sorted(filenames): abs_str = os.path.join(dirpath, fname) try: st = os.lstat(abs_str) except OSError: continue if not _stat.S_ISREG(st.st_mode): continue rel = abs_str[prefix_len:] if os.sep != "/": rel = rel.replace(os.sep, "/") if is_ignored(rel, patterns): continue files[rel] = cache.get_cached(rel, abs_str, st.st_mtime, st.st_size, st.st_ino) # Only include dirs that are tracked (committed or staged) — untracked # dirs must not appear in the working tree snapshot so that muse diff # doesn't report them as changes (same as git: untracked is invisible to diff). committed_dirs = set(_head_snapshot_dirs_for(root)) staged = read_stage(root) staged_dirs: set[str] = { p for p, e in staged.items() if e.get("kind") == "dir" } tracked_on_disk_dirs = on_disk_dirs & (committed_dirs | staged_dirs) wdirs = sorted(set(directories_from_manifest(files)) | tracked_on_disk_dirs) return SnapshotManifest(files=files, domain=_DOMAIN_NAME, directories=wdirs) # ------------------------------------------------------------------ # StagePlugin implementation # ------------------------------------------------------------------ def stage_index_path(self, root: pathlib.Path) -> pathlib.Path: """Return the absolute path of ``.muse/code/stage.json``.""" return stage_path(root) def read_stage(self, root: pathlib.Path) -> StagedEntryMap: """Read the code-domain stage index.""" return read_stage(root) def write_stage( self, root: pathlib.Path, entries: StagedEntryMap ) -> None: """Persist *entries* as the code-domain stage index.""" write_stage(root, entries) def clear_stage(self, root: pathlib.Path) -> None: """Remove the code-domain stage index.""" clear_stage(root) def stage_status(self, root: pathlib.Path) -> StageStatus: """Return a three-bucket view of the working tree vs the stage. Compares: 1. The current stage against HEAD → **staged** bucket. 2. Working tree files against their HEAD/stage state → **unstaged**. 3. Files present on disk but neither tracked nor staged → **untracked**. """ stage = read_stage(root) committed = _head_manifest_for(root) # Staged bucket: all stage entries (files AND empty-dir sentinels). staged: StagedEntryMap = dict(stage) # Separate staged empty-dir entries (kind == "dir") from file entries # — they must not participate in file-level diff logic. staged_added_dirs: set[str] = { p for p, e in stage.items() if e.get("kind") == "dir" and e["mode"] == "A" } staged_deleted_dirs: set[str] = { p for p, e in stage.items() if e.get("kind") == "dir" and e["mode"] == "D" } # Exclude dir sentinels from the file-level staged bucket so the # staged/unstaged file passes below never trip on them. staged_files_only: StagedEntryMap = { p: e for p, e in stage.items() if e.get("kind") != "dir" } # Build the full working-tree manifest (reuse snapshot logic). patterns = _BUILTIN_SECRET_PATTERNS + resolve_patterns(load_ignore_config(root), _DOMAIN_NAME) cache = load_cache(root) workdir_files: Manifest = {} on_disk_dirs: set[str] = set() # muse#76: directories containing ANY file (ignored or not) — tracked # separately from workdir_files, which only holds non-ignored files. # A directory whose only content is an individually-ignored file is # not empty; "empty on-disk directory" must mean zero files at all. dirs_with_any_file: set[str] = set() root_str = str(root) prefix_len = len(root_str) + 1 for dirpath, dirnames, filenames in os.walk(root_str, followlinks=False): # Compute cur_rel first so it can be used in dirnames filtering. if dirpath != root_str: cur_rel = dirpath[prefix_len:] if os.sep != "/": cur_rel = cur_rel.replace(os.sep, "/") on_disk_dirs.add(cur_rel) else: cur_rel = "" # Match the same pruning as the snapshot walk: skip _ALWAYS_IGNORE_DIRS, # is_ignored directories, and nested Muse repos (dirs containing .muse/). dirnames[:] = sorted( d for d in dirnames if d not in _ALWAYS_IGNORE_DIRS and not is_ignored( f"{cur_rel}/{d}/_" if cur_rel else f"{d}/_", patterns ) and not os.path.isdir(os.path.join(dirpath, d, ".muse")) ) for fname in sorted(filenames): abs_str = os.path.join(dirpath, fname) try: st = os.lstat(abs_str) except OSError: continue if not _stat.S_ISREG(st.st_mode): continue rel = abs_str[prefix_len:] if os.sep != "/": rel = rel.replace(os.sep, "/") parts = rel.split("/") for i in range(1, len(parts)): dirs_with_any_file.add("/".join(parts[:i])) if is_ignored(rel, patterns): continue workdir_files[rel] = cache.get_cached( rel, abs_str, st.st_mtime, st.st_size, st.st_ino ) # Unstaged: files whose working-tree content diverges from what is staged # (or from HEAD for untracked-by-stage files). # # Two passes: # 1. Staged files: compare working tree against the staged object. # A file modified after staging must appear in the unstaged bucket. # 2. Committed files not in the stage: compare against HEAD. # Dir sentinels are excluded from both passes. unstaged: Manifest = {} for rel_path, staged_entry in staged_files_only.items(): if staged_entry["mode"] == "D": # Staged for deletion — if the file reappeared on disk, flag it. if rel_path in workdir_files: unstaged[rel_path] = "modified" continue staged_oid = staged_entry["object_id"] current_id = workdir_files.get(rel_path) if current_id is None: # Staged but deleted from disk since staging. unstaged[rel_path] = "deleted" elif current_id != staged_oid: # Modified on disk after staging — not yet re-staged. unstaged[rel_path] = "modified" for rel_path, committed_id in committed.items(): if rel_path in staged_files_only: continue # already covered in the stage pass above current_id = workdir_files.get(rel_path) if current_id is None: unstaged[rel_path] = "deleted" elif current_id != committed_id: unstaged[rel_path] = "modified" # Untracked: on disk, not committed, not staged (files only). all_known_files = set(committed) | set(staged_files_only) untracked_set: set[str] = { rel_path for rel_path in workdir_files if rel_path not in all_known_files } # Working-tree rename detection: a committed file deleted from disk # whose content_id matches an untracked file's content_id is a rename, # not a delete + new untracked file. Only unstaged deletes qualify — # files explicitly staged for deletion represent deliberate user action. renamed: dict[str, str] = {} unstaged_deleted = [p for p, lbl in unstaged.items() if lbl == "deleted"] if unstaged_deleted and untracked_set: # Build content_id → sorted list of untracked paths for O(1) lookup. untracked_by_cid: dict[str, list[str]] = {} for ut_path in sorted(untracked_set): cid = workdir_files[ut_path] untracked_by_cid.setdefault(cid, []).append(ut_path) for old_path in sorted(unstaged_deleted): old_cid = committed.get(old_path) if old_cid is None: continue candidates = untracked_by_cid.get(old_cid) if not candidates: continue new_path = candidates.pop(0) if not candidates: del untracked_by_cid[old_cid] renamed[old_path] = new_path del unstaged[old_path] untracked_set.discard(new_path) untracked: list[str] = sorted(untracked_set) # Directory tracking: compare on-disk dirs against committed dirs. # Only explicitly-empty dirs (not derivable from file paths) are surfaced. # Dirs that contain tracked files are implicitly covered by their files. committed_snap_dirs: list[str] = _head_snapshot_dirs_for(root) # Dirs derivable from committed file paths — implicitly tracked. dirs_with_committed_files: set[str] = set() for f in committed: parts = f.split("/") for i in range(1, len(parts)): dirs_with_committed_files.add("/".join(parts[:i])) # Explicitly-committed empty dirs = snapshot dirs minus implicit dirs. committed_empty_dirs: set[str] = set(committed_snap_dirs) - dirs_with_committed_files # Dirs with files in the current working tree (not empty). dirs_with_workdir_files: set[str] = set() for f in workdir_files: parts = f.split("/") for i in range(1, len(parts)): dirs_with_workdir_files.add("/".join(parts[:i])) # Empty dirs on disk = all dirs seen during walk minus those with files. empty_on_disk_dirs: set[str] = on_disk_dirs - dirs_with_workdir_files - dirs_with_any_file # Directory rename detection: read explicit renames recorded by muse mv. # Only renames where the old path is in staged_deleted_dirs AND the new # path is in staged_added_dirs are valid (guards against stale entries). raw_dir_renames = read_stage_dir_renames(root) valid_dir_renames: dict[str, str] = { old: new for old, new in raw_dir_renames.items() if old in staged_deleted_dirs and new in staged_added_dirs } renamed_old_dirs = set(valid_dir_renames.keys()) renamed_new_dirs = set(valid_dir_renames.values()) # Untracked dirs: empty on disk, not in committed, not staged. dir_added = sorted( empty_on_disk_dirs - committed_empty_dirs - staged_added_dirs ) # Deleted dirs: committed empty dirs no longer on disk and not already # staged for deletion (e.g. via muse mv or muse rm). dir_deleted = sorted(committed_empty_dirs - on_disk_dirs - staged_deleted_dirs) directories = DirStatus( added=dir_added, deleted=dir_deleted, staged_added=sorted(staged_added_dirs - committed_empty_dirs - renamed_new_dirs), staged_deleted=sorted(staged_deleted_dirs - renamed_old_dirs), staged_renamed=valid_dir_renames, ) return StageStatus( staged=staged, unstaged=unstaged, untracked=untracked, renamed=renamed, directories=directories, ) # ------------------------------------------------------------------ # 2. diff # ------------------------------------------------------------------ def diff( self, base: StateSnapshot, target: StateSnapshot, *, repo_root: pathlib.Path | None = None, ) -> StateDelta: """Compute the structured delta between two snapshots. Without ``repo_root`` Produces coarse file-level ops (``InsertOp`` / ``DeleteOp`` / ``ReplaceOp``). Used by ``muse checkout`` which only needs file paths. With ``repo_root`` Reads source bytes from the object store, parses AST for supported languages (Python), and produces ``PatchOp`` entries with symbol-level ``child_ops``. Used by ``muse commit`` (to store the structured delta) and ``muse read`` / ``muse diff``. Args: base: Base snapshot (older state). target: Target snapshot (newer state). repo_root: Repository root for object-store access and symbol extraction. ``None`` → file-level ops only. Returns: A ``StructuredDelta`` with ``domain="code"``. """ base_files = base["files"] target_files = target["files"] base_dirs = list(base.get("directories") or []) target_dirs = list(target.get("directories") or []) # ── Directory rename detection ──────────────────────────────────────── # Detect which added/deleted directories are actually renames and # suppress the corresponding file-level insert/delete ops so the delta # is clean: one RenameOp instead of N inserts + N deletes. added_dirs = set(target_dirs) - set(base_dirs) deleted_dirs = set(base_dirs) - set(target_dirs) dir_renames = detect_directory_renames( deleted_dirs, added_dirs, base_files, target_files ) # Build the set of file paths covered by directory renames so we can # filter them out of the file-level diff inputs. covered: set[str] = set() for old_dir, new_dir in dir_renames: old_prefix = f"{old_dir}/" new_prefix = f"{new_dir}/" covered |= {p for p in base_files if p.startswith(old_prefix)} covered |= {p for p in target_files if p.startswith(new_prefix)} renamed_old = {old for old, _ in dir_renames} renamed_new = {new for _, new in dir_renames} # Build directory-level ops. dir_ops: list[DomainOp] = [] for old_dir, new_dir in dir_renames: dir_ops.append( RenameOp( op="rename", address=new_dir + "/", from_address=old_dir + "/", ) ) for d in sorted(added_dirs - renamed_new): dir_ops.append(AddressedInsertOp(op="insert", address=d + "/", content_id="", content_summary=f"directory: {d}/")) for d in sorted(deleted_dirs - renamed_old): dir_ops.append(AddressedDeleteOp(op="delete", address=d + "/", content_id="", content_summary=f"directory: {d}/")) # ── File-level diff (excluding covered paths) ───────────────────────── if covered: filtered_base = SnapshotManifest( files={k: v for k, v in base_files.items() if k not in covered}, domain=base["domain"], directories=[], ) filtered_target = SnapshotManifest( files={k: v for k, v in target_files.items() if k not in covered}, domain=target["domain"], directories=[], ) else: filtered_base = base filtered_target = target if repo_root is None: file_delta = snapshot_diff(self.schema(), filtered_base, filtered_target) file_ops: list[DomainOp] = list(file_delta.get("ops", [])) else: file_ops = _semantic_ops( filtered_base["files"], filtered_target["files"], repo_root, workdir=repo_root, ) all_ops = dir_ops + file_ops summary = delta_summary(all_ops) return StructuredDelta(domain=_DOMAIN_NAME, ops=all_ops, summary=summary) # ------------------------------------------------------------------ # 3. merge # ------------------------------------------------------------------ def merge( self, base: StateSnapshot, left: StateSnapshot, right: StateSnapshot, *, repo_root: pathlib.Path | None = None, ) -> MergeResult: """Three-way merge at file granularity, respecting ``.museattributes``. Standard three-way logic, augmented by per-path strategy overrides declared in ``.museattributes``: - Both sides agree → consensus wins (including both deleted). - Only one side changed → take that side. - Both sides changed differently → consult ``.museattributes``: - ``ours`` — take left; remove from conflict list. - ``theirs`` — take right; remove from conflict list. - ``base`` — revert to the common ancestor; remove from conflicts. - ``union`` — keep all additions from both sides; prefer left for conflicting blobs; remove from conflict list. - ``manual`` — force into conflict list regardless of auto resolution. - ``auto`` — default three-way conflict. This is the fallback used by ``muse cherry-pick`` and contexts where structured deltas are unavailable. :meth:`merge_ops` provides symbol-level conflict detection when both sides have structured deltas. Args: base: Common ancestor snapshot. left: Our branch snapshot. right: Their branch snapshot. repo_root: Repository root; when provided, ``.museattributes`` is consulted for per-path strategy overrides. Returns: A ``MergeResult`` with the reconciled snapshot, any file-level conflicts, and ``applied_strategies`` recording which rules fired. """ attrs = load_attributes(repo_root, domain=_DOMAIN_NAME) if repo_root else [] base_files = base["files"] left_files = left["files"] right_files = right["files"] merged: Manifest = dict(base_files) conflicts: list[str] = [] conflict_records: list[ConflictRecord] = [] applied_strategies: AppliedStrategies = {} all_paths = set(base_files) | set(left_files) | set(right_files) merge_debug_log("plugin.merge.enter", { "all_paths_count": len(all_paths), "base_file_count": len(base_files), "left_file_count": len(left_files), "right_file_count": len(right_files), # Log paths where l==r but both changed from base (convergent edits — should NOT conflict) "convergent_same": [ p for p in all_paths if base_files.get(p) != left_files.get(p) and left_files.get(p) == right_files.get(p) ], # Log paths where l != r (potential conflicts) "divergent_paths": [ p for p in all_paths if left_files.get(p) != right_files.get(p) ], }) for path in sorted(all_paths): b = base_files.get(path) l = left_files.get(path) r = right_files.get(path) if l == r: # Both sides agree — or both deleted (including convergent same change). # "manual" must NOT fire here: when both branches agree (whether # nothing changed or both independently made the same edit), there # is no divergence to review. Firing manual on l == r produced # false conflicts for every unchanged file matching a manual rule. if l is None: merged.pop(path, None) else: merged[path] = l elif b == l: # Only right changed. if r is None: merged.pop(path, None) else: merged[path] = r if attrs and resolve_strategy(attrs, path) == "manual": conflicts.append(path) applied_strategies[path] = "manual" conflict_records.append(ConflictRecord( path=path, conflict_type="manual", theirs_action=_file_action(b, r), )) elif b == r: # Only left changed. if l is None: merged.pop(path, None) else: merged[path] = l if attrs and resolve_strategy(attrs, path) == "manual": conflicts.append(path) applied_strategies[path] = "manual" conflict_records.append(ConflictRecord( path=path, conflict_type="manual", ours_action=_file_action(b, l), )) else: # Both sides changed differently — consult attributes. strategy = resolve_strategy(attrs, path) if attrs else "auto" if strategy == "ours": if l is None: merged.pop(path, None) else: merged[path] = l applied_strategies[path] = "ours" elif strategy == "theirs": if r is None: merged.pop(path, None) else: merged[path] = r applied_strategies[path] = "theirs" elif strategy == "base": if b is None: merged.pop(path, None) else: merged[path] = b applied_strategies[path] = "base" elif strategy == "union": # Weave-based union: merge both sides' line-level changes. # When repo_root is available, read the blobs and run # _independence_merge_blob() so both sides' additions # survive (Manyana add-wins at the text level). This is # what makes docs/** and *.md union rules genuinely useful # for concurrent prose additions. # Binary blobs (contain null bytes) can't be text-merged — # fall through to prefer-ours. if repo_root is not None and b and l and r: b_blob = read_object(repo_root, b) l_blob = read_object(repo_root, l) r_blob = read_object(repo_root, r) if (b_blob is not None and l_blob is not None and r_blob is not None and b"\x00" not in b_blob and b"\x00" not in l_blob and b"\x00" not in r_blob): merged_bytes = _independence_merge_blob(b_blob, l_blob, r_blob) merged_id = blob_id(merged_bytes) write_object(repo_root, merged_id, merged_bytes) merged[path] = merged_id applied_strategies[path] = "union" continue # Fallback when blobs unavailable or binary: prefer ours. merged[path] = l or r or b or "" applied_strategies[path] = "union" elif strategy == "manual": conflicts.append(path) merged[path] = l or r or b or "" applied_strategies[path] = "manual" conflict_records.append(ConflictRecord( path=path, conflict_type="manual", ours_action=_file_action(b, l), theirs_action=_file_action(b, r), )) else: # "auto" — standard three-way conflict. conflicts.append(path) merged[path] = l or r or b or "" conflict_records.append(ConflictRecord( path=path, conflict_type="file_level", ours_action=_file_action(b, l), theirs_action=_file_action(b, r), )) merge_debug_log("plugin.merge.result", { "conflicts": conflicts, "conflict_records": [ {"path": cr.path, "type": cr.conflict_type} for cr in conflict_records ], "applied_strategies": applied_strategies, }) return MergeResult( merged=SnapshotManifest( files=merged, domain=_DOMAIN_NAME, directories=directories_from_manifest(merged), ), conflicts=conflicts, applied_strategies=applied_strategies, conflict_records=conflict_records, ) # ------------------------------------------------------------------ # 4. drift # ------------------------------------------------------------------ def drift(self, committed: StateSnapshot, live: LiveState) -> DriftReport: """Report how much the working tree has drifted from the last commit. Called by ``muse status``. Takes a snapshot of the current live state and diffs it against the committed snapshot. Args: committed: The last committed snapshot. live: Current live state (path or snapshot manifest). Returns: A ``DriftReport`` describing what has changed since the last commit. """ current = self.snapshot(live) delta = self.diff(committed, current) # Remove delete ops for files that are in .museignore but still on # disk. These were previously committed (exist in HEAD) but have # since been added to .museignore to stop tracking them. Surfacing # them as deleted in status or diff is misleading — they are present # and unchanged; the user just does not want to track them anymore. # Only file-level delete ops are considered (address has no "::"). if isinstance(live, pathlib.Path) and delta.get("ops"): ignore_patterns = load_ignore_patterns(live) filtered = [ op for op in delta["ops"] if not ( op.get("op") == "delete" and "::" not in op.get("address", "") and is_ignored(op["address"], ignore_patterns) and (live / op["address"]).exists() ) ] if len(filtered) != len(delta["ops"]): delta = StructuredDelta( domain=delta["domain"], ops=filtered, summary=delta_summary(filtered), ) return DriftReport( has_drift=len(delta["ops"]) > 0, summary=delta["summary"], delta=delta, ) # ------------------------------------------------------------------ # 5. apply # ------------------------------------------------------------------ def apply(self, delta: StateDelta, live_state: LiveState) -> LiveState: """Apply a delta to the working tree. Called by ``muse checkout`` after the core engine has already restored file-level objects from the object store. The code plugin has no domain-specific post-processing to perform, so this is a pass-through. Args: delta: The typed operation list (unused at post-checkout time). live_state: Current live state (returned unchanged). Returns: *live_state* unchanged. """ return live_state # ------------------------------------------------------------------ # 6. schema # ------------------------------------------------------------------ def schema(self) -> DomainSchema: """Declare the structural schema of the code domain. Returns: A ``DomainSchema`` with five semantic dimensions: ``structure``, ``symbols``, ``imports``, ``variables``, and ``metadata``. """ return DomainSchema( domain=_DOMAIN_NAME, description=( "Semantic version control for source code. " "Treats code as a structured system of named symbols " "(functions, classes, methods) rather than lines of text. " "Two commits that only reformat a file produce no delta. " "Renames and moves are detected via content-addressed " "symbol identity." ), top_level=TreeSchema( kind="tree", node_type="module", diff_algorithm="gumtree", ), dimensions=[ DimensionSpec( name="structure", description=( "Module / file tree. Tracks which files exist and " "how they relate to each other." ), schema=TreeSchema( kind="tree", node_type="file", diff_algorithm="gumtree", ), independent_merge=False, ), DimensionSpec( name="symbols", description=( "AST symbol tree. Functions, classes, methods, and " "variables — the primary unit of semantic change." ), schema=TreeSchema( kind="tree", node_type="symbol", diff_algorithm="gumtree", ), independent_merge=True, ), DimensionSpec( name="imports", description=( "Import set. Tracks added / removed import statements " "as an unordered set — order is semantically irrelevant." ), schema=SetSchema( kind="set", element_type="import", identity="by_content", ), independent_merge=True, ), DimensionSpec( name="variables", description=( "Top-level variable and constant assignments. " "Tracked as an unordered set." ), schema=SetSchema( kind="set", element_type="variable", identity="by_content", ), independent_merge=True, ), DimensionSpec( name="metadata", description=( "Non-code files: configuration, documentation, " "build scripts, etc. Tracked at file granularity." ), schema=SetSchema( kind="set", element_type="file", identity="by_content", ), independent_merge=True, ), ], merge_mode="three_way", schema_version=__version__, ) # ------------------------------------------------------------------ # AddressedMergePlugin — address-keyed map merge # ------------------------------------------------------------------ def merge_ops( self, base: StateSnapshot, ours_snap: StateSnapshot, theirs_snap: StateSnapshot, ours_ops: list[DomainOp], theirs_ops: list[DomainOp], *, repo_root: pathlib.Path | None = None, ) -> MergeResult: """Operation-level three-way merge using address-keyed map semantics. Uses :func:`~muse.core.op_merge.merge_op_lists` to determine which ``DomainOp`` pairs commute (auto-mergeable) and which conflict. For ``PatchOp`` entries at the same file address, the engine recurses into ``child_ops`` — so two agents modifying *different* functions in the same file produce a file-level conflict while concurrent modifications to the *same* function produce a symbol-level conflict address (e.g. ``"src/utils.py::calculate_total"``). **Conflict propagation** — the OT check operates at symbol granularity and can miss two classes of file-level conflict: 1. *Mixed op types*: one branch produced a ``ReplaceOp`` (no symbol tree available for that file type) and the other produced a ``PatchOp`` (symbol tree available). They never appear together in the OT conflict-detection loops. 2. *Commuting symbol changes*: two branches modify *different* symbols in the same file. The individual ``DomainOp``\\s commute so OT reports a clean merge, but Muse cannot reconstruct the merged file blob without a text-merge pass — the merged manifest would silently contain only the "ours" blob, discarding "theirs" changes. In both cases the file-level :meth:`merge` fallback correctly flags a conflict. ``merge_ops`` propagates those flags unless the path was already explicitly auto-resolved by a ``.museattributes`` strategy (``ours``, ``theirs``, ``base``, or ``union``). Args: base: Common ancestor snapshot. ours_snap: Our branch's final snapshot. theirs_snap: Their branch's final snapshot. ours_ops: Our branch's typed operation list. theirs_ops: Their branch's typed operation list. repo_root: Repository root for ``.museattributes`` lookup. Returns: A ``MergeResult`` where ``conflicts`` contains either symbol-level addresses (``"src/utils.py::calculate_total"``) or bare file paths (``"AGENTS.md"``) depending on the granularity at which the conflict was detected. """ # The core OT engine's _op_key for PatchOp hashes only the file path # and child_domain — not the child_ops themselves. This means two # PatchOps for the same file are treated as "consensus" regardless of # whether they touch the same or different symbols. We therefore # implement symbol-level conflict detection directly here. attrs = load_attributes(repo_root, domain=_DOMAIN_NAME) if repo_root else [] # ── Step 1: symbol-level conflict detection for PatchOps ────────── ours_patches: PatchOpMap = { op["address"]: op for op in ours_ops if op["op"] == "patch" } theirs_patches: PatchOpMap = { op["address"]: op for op in theirs_ops if op["op"] == "patch" } merge_debug_log("merge_ops.enter", { "ours_patches_files": sorted(ours_patches.keys()), "theirs_patches_files": sorted(theirs_patches.keys()), "shared_patch_files": sorted(set(ours_patches) & set(theirs_patches)), }) conflict_addresses: set[str] = set() for path in ours_patches: if path not in theirs_patches: continue for our_child in ours_patches[path]["child_ops"]: for their_child in theirs_patches[path]["child_ops"]: if not ops_commute(our_child, their_child): merge_debug_log("merge_ops.symbol_conflict", { "file": path, "ours_address": our_child["address"], "theirs_address": their_child["address"], }) conflict_addresses.add(our_child["address"]) merge_debug_log("merge_ops.step1_result", { "conflict_addresses_after_step1": sorted(conflict_addresses), }) # ── Step 1.5: independence-aware blob reconstruction ────────────── # For files where both sides changed the blob AND all symbol-level ops # commute, reconstruct the merged blob via _independence_merge_blob() # instead of letting the file-level fallback flag a manifest conflict. # This implements OR-Set semantics for imports and variables, and # independence-aware merge for functions and classes: concurrent # additions of non-overlapping symbols always merge cleanly regardless # of their line-level positions. # # We derive fresh PatchOps from the snapshots when the provided op # lists don't cover a changed path (e.g. when called with empty lists). # This keeps the test surface thin while ensuring correctness in all # callers. independence_resolved: dict[str, str] = {} # path → merged_content_id if repo_root is not None: conflict_file_paths: set[str] = { (addr.split("::")[0] if "::" in addr else addr) for addr in conflict_addresses } # Candidate paths: both sides changed the same file from base. base_files = base["files"] candidate_paths = sorted( p for p in set(ours_snap["files"]) | set(theirs_snap["files"]) if ( p not in conflict_file_paths and ours_snap["files"].get(p) != theirs_snap["files"].get(p) and base_files.get(p) != ours_snap["files"].get(p) and base_files.get(p) != theirs_snap["files"].get(p) and ours_snap["files"].get(p) is not None and theirs_snap["files"].get(p) is not None and base_files.get(p) is not None ) ) for path in candidate_paths: b_id = base_files[path] l_id = ours_snap["files"][path] r_id = theirs_snap["files"][path] # Check symbol independence — use provided PatchOps when # available, otherwise derive them from the snapshot diff. our_patch = ours_patches.get(path) their_patch = theirs_patches.get(path) if our_patch is None or their_patch is None: # Derive ops lazily for this path only. derived_ours = _semantic_ops( base_files, ours_snap["files"], repo_root ) derived_theirs = _semantic_ops( base_files, theirs_snap["files"], repo_root ) our_patch = next( (op for op in derived_ours if op["op"] == "patch" and op["address"] == path), None, ) their_patch = next( (op for op in derived_theirs if op["op"] == "patch" and op["address"] == path), None, ) if our_patch is None or their_patch is None: # No symbol-level info for this path on at least one side — # e.g. both sides' only change is a comment, which # content_id (computed from ast.unparse(), comment-blind by # design — see ast_parser.py) can't distinguish from "no # change", so build_diff_ops falls back to a whole-file # ReplaceOp with no child_ops instead of a symbol-addressed # PatchOp (muse#92). We don't have symbol knowledge to # confirm *which* symbols changed, but a clean three-way # LINE merge is unconditionally safe regardless of symbol # knowledge: if the base/ours/theirs diff regions don't # overlap at the text level, combining them is correct no # matter what kind of file this is or whether we understood # its structure. Only use the result when the merge is # genuinely clean (no conflict regions) -- any real overlap # still falls through to the existing file-level conflict. b_blob = read_object(repo_root, b_id) l_blob = read_object(repo_root, l_id) r_blob = read_object(repo_root, r_id) if b_blob is not None and l_blob is not None and r_blob is not None: from muse.core.cohen_transform import three_way_merge_lines b_lines = b_blob.decode("utf-8", errors="replace").splitlines(keepends=True) l_lines = l_blob.decode("utf-8", errors="replace").splitlines(keepends=True) r_lines = r_blob.decode("utf-8", errors="replace").splitlines(keepends=True) merged_lines, has_conflict = three_way_merge_lines(b_lines, l_lines, r_lines) if not has_conflict: merged_bytes = "".join(merged_lines).encode("utf-8") merged_id = blob_id(merged_bytes) write_object(repo_root, merged_id, merged_bytes) independence_resolved[path] = merged_id continue # Don't union-merge when neither side produced any symbols — the # file couldn't be parsed, so "no symbol conflicts" means "we have # no information", not "safe to merge". Letting _independence_merge_blob # fire here would silently write both divergent versions as union output # without surfacing a conflict (data corruption). if not our_patch["child_ops"] and not their_patch["child_ops"]: continue file_has_symbol_conflict = any( not ops_commute(oc, tc) for oc in our_patch["child_ops"] for tc in their_patch["child_ops"] ) if file_has_symbol_conflict: continue # genuine symbol conflict — let it surface normally # Independence merge is only safe when the two sides touch # DISJOINT symbol addresses (or share only convergent inserts — # the same symbol added with identical content on both sides, # which is idempotent). # # delete+delete on the SAME address "commutes" in the OT sense # (consensus delete), but the surrounding file bytes may still # diverge (e.g. both sides replaced an unparseable file and # produced "delete the base symbols" vacuously). Firing # independence_merge_blob in that case silently union-merges # genuinely conflicting content. our_addr_ops = {op["address"]: op for op in our_patch["child_ops"]} their_addr_ops = {op["address"]: op for op in their_patch["child_ops"]} shared_addrs = set(our_addr_ops) & set(their_addr_ops) if any( not ( our_addr_ops[addr]["op"] == "insert" and their_addr_ops[addr]["op"] == "insert" and our_addr_ops[addr].get("content_id") == their_addr_ops[addr].get("content_id") ) for addr in shared_addrs ): continue # non-convergent overlap — can't confirm independence b_blob = read_object(repo_root, b_id) l_blob = read_object(repo_root, l_id) r_blob = read_object(repo_root, r_id) if b_blob is None or l_blob is None or r_blob is None: continue # can't read blobs — fall through to file-level conflict merged_bytes = _independence_merge_blob(b_blob, l_blob, r_blob) merged_id = blob_id(merged_bytes) write_object(repo_root, merged_id, merged_bytes) independence_resolved[path] = merged_id # ── Step 2: coarse OT for non-PatchOp ops (file-level inserts/deletes) ── non_patch_ours: list[DomainOp] = [op for op in ours_ops if op["op"] != "patch"] non_patch_theirs: list[DomainOp] = [op for op in theirs_ops if op["op"] != "patch"] merge_debug_log("merge_ops.step2_input", { "non_patch_ours": non_patch_ours, "non_patch_theirs": non_patch_theirs, }) file_result = merge_op_lists( base_ops=[], ours_ops=non_patch_ours, theirs_ops=non_patch_theirs, ) for our_op, _ in file_result.conflict_ops: merge_debug_log("merge_ops.step2_conflict", {"address": our_op["address"]}) conflict_addresses.add(our_op["address"]) merge_debug_log("merge_ops.step2_result", { "conflict_addresses_after_step2": sorted(conflict_addresses), }) # Step 2's OT check operates on whole-file ops (e.g. two competing # file-level ReplaceOps) and knows nothing about Step 1.5's # independence-resolved merges -- it can re-flag a path as # conflicting even though independence_resolved already holds a # safe, clean merge for it (muse#92). Drop those paths here so # they never reach Step 3 / resolved_conflicts. if independence_resolved: conflict_addresses = { addr for addr in conflict_addresses if (addr.split("::")[0] if "::" in addr else addr) not in independence_resolved } # ── Step 3: apply .museattributes to symbol-level conflicts ────── # Symbol addresses are of the form "src/utils.py::function_name". # We resolve strategy against the file path portion so that a # path = "src/**/*.py" / strategy = "ours" rule suppresses symbol # conflicts in those files, not just file-level manifest conflicts. op_applied_strategies: AppliedStrategies = {} resolved_conflicts: list[str] = [] if attrs: for addr in sorted(conflict_addresses): file_path = addr.split("::")[0] if "::" in addr else addr strategy = resolve_strategy(attrs, file_path) if strategy in ("ours", "theirs", "base", "union"): op_applied_strategies[addr] = strategy elif strategy == "manual": resolved_conflicts.append(addr) op_applied_strategies[addr] = "manual" else: resolved_conflicts.append(addr) else: resolved_conflicts = sorted(conflict_addresses) merged_ops: list[DomainOp] = list(file_result.merged_ops) + list(ours_ops) # Fall back to file-level merge for the manifest (carries its own # applied_strategies from file-level attribute resolution). fallback = self.merge(base, ours_snap, theirs_snap, repo_root=repo_root) combined_strategies = {**fallback.applied_strategies, **op_applied_strategies} # Patch the manifest and conflict list with independence-resolved files. # These paths were auto-merged at the symbol level and must not appear # in the final conflict list even if the raw blob IDs differ. if independence_resolved: patched_files = dict(fallback.merged["files"]) patched_files.update(independence_resolved) fallback = MergeResult( merged=SnapshotManifest( files=patched_files, domain=_DOMAIN_NAME, directories=directories_from_manifest(patched_files), ), conflicts=[p for p in fallback.conflicts if p not in independence_resolved], applied_strategies=fallback.applied_strategies, conflict_records=fallback.conflict_records, ) # ── Step 4: propagate file-level conflicts missed by the OT check ── # Extract the file path from each OT-detected conflict address so we # can avoid duplicating conflicts that are already represented at the # symbol level (e.g. "src/utils.py::my_func" covers "src/utils.py"). auto_resolved_file_paths: set[str] = { (addr.split("::")[0] if "::" in addr else addr) for addr, strat in combined_strategies.items() if strat in ("ours", "theirs", "base", "union") } ot_conflict_file_paths: set[str] = { (addr.split("::")[0] if "::" in addr else addr) for addr in resolved_conflicts } propagated_file_conflicts: list[str] = [ p for p in fallback.conflicts if p not in auto_resolved_file_paths and p not in ot_conflict_file_paths ] all_conflicts: list[str] = resolved_conflicts + sorted(propagated_file_conflicts) merge_debug_log("merge_ops.result", { "fallback_conflicts": fallback.conflicts, "resolved_conflicts_from_OT": resolved_conflicts, "propagated_file_conflicts": sorted(propagated_file_conflicts), "auto_resolved_file_paths": sorted(auto_resolved_file_paths), "ot_conflict_file_paths": sorted(ot_conflict_file_paths), "all_conflicts_final": all_conflicts, "independence_resolved": sorted(independence_resolved.keys()), }) return MergeResult( merged=fallback.merged, conflicts=all_conflicts, applied_strategies=combined_strategies, dimension_reports=fallback.dimension_reports, op_log=merged_ops, ) # --------------------------------------------------------------------------- # Private helpers # --------------------------------------------------------------------------- def _independence_merge_blob( base_blob: bytes, ours_blob: bytes, theirs_blob: bytes, ) -> bytes: """Reconstruct a merged blob for a file whose symbol changes are independent. When ``merge_ops()`` confirms all child op pairs commute (no symbol-level conflict), this function produces the merged file content. Strategy -------- 1. Run ``three_way_merge_lines`` — if clean (no conflict regions), return it. 2. If the text merge has conflicts, those conflicts are spurious: they arise from concurrent insertions at the same line position (both sides added new content at the same anchor point in the file — e.g. two new functions at the end, or two new imports at the top). For independent symbols this is always safe to union-merge: include ours' additions then theirs' additions in document order. This is the Manyana add-wins principle applied at the text level. The union resolution is applied only here, gated by the caller having already verified symbol-level independence. Callers that detect a genuine symbol conflict take the normal conflict path and never reach this function. """ b_lines = base_blob.decode("utf-8", errors="replace").splitlines(keepends=True) l_lines = ours_blob.decode("utf-8", errors="replace").splitlines(keepends=True) r_lines = theirs_blob.decode("utf-8", errors="replace").splitlines(keepends=True) from muse.core.cohen_transform import three_way_merge_lines merged_lines, has_conflict = three_way_merge_lines(b_lines, l_lines, r_lines) if not has_conflict: return "".join(merged_lines).encode("utf-8") # Spurious conflict from concurrent insertions — union-resolve every conflict # region by including ours' lines then theirs' lines (Manyana add-wins). regions = compute_regions(b_lines, l_lines, r_lines) output: list[str] = [] for region in regions: if region.kind == "stable": output.extend(region.base_lines) elif region.kind == "ours_only": output.extend(region.ours_lines) elif region.kind == "theirs_only": output.extend(region.theirs_lines) elif region.kind == "both_same": output.extend(region.ours_lines) else: # conflict → union: ours additions then theirs additions output.extend(region.ours_lines) output.extend(region.theirs_lines) return "".join(output).encode("utf-8") def _file_action(base_id: str | None, other_id: str | None) -> str: """Return the Cohen-transform action label for one side of a file conflict. Compares the *other* side's object ID against the common ancestor: - ``"inserted"`` — the file was added from nothing (base absent, other present). - ``"deleted"`` — the file was removed (base present, other absent). - ``"modified"`` — the file existed on both sides but the content changed. """ if base_id is None and other_id is not None: return "inserted" if base_id is not None and other_id is None: return "deleted" return "modified" def _read_blob( repo_root: pathlib.Path, content_id: str, disk_fallback: pathlib.Path | None, ) -> bytes | None: """Read a blob from the object store; fall back to disk when not found. When ``disk_fallback`` is provided and the object store returns ``None`` (blob not yet committed — typical during ``muse diff`` on the working tree), we read the file directly from disk and verify its SHA-256 matches ``content_id`` before returning it. This guarantees we never parse stale content from a file whose hash has changed since the snapshot was taken. """ raw = read_object(repo_root, content_id) if raw is not None: return raw if disk_fallback is None or not disk_fallback.is_file(): return None try: candidate = disk_fallback.read_bytes() except OSError: return None if hash_file(disk_fallback) == content_id: return candidate return None def _semantic_ops( base_files: Manifest, target_files: Manifest, repo_root: pathlib.Path, workdir: pathlib.Path | None = None, ) -> list[DomainOp]: """Produce symbol-level ops by reading files from the object store. When *workdir* is supplied (working-tree diffs), blobs that are not yet in the object store are read directly from disk and verified against their content hash. This enables full semantic diffing for ``muse diff`` before a commit has been made. """ base_paths = set(base_files) target_paths = set(target_files) changed_paths = ( (target_paths - base_paths) # added | (base_paths - target_paths) # removed | { # modified p for p in base_paths & target_paths if base_files[p] != target_files[p] } ) base_trees: SymbolTreeMap = {} target_trees: SymbolTreeMap = {} for path in changed_paths: if path in base_files: raw = _read_blob(repo_root, base_files[path], None) if raw is not None: base_trees[path] = _parse_with_fallback(raw, path) if path in target_files: disk_path = (workdir / path) if workdir is not None else None raw = _read_blob(repo_root, target_files[path], disk_path) if raw is not None: target_trees[path] = _parse_with_fallback(raw, path) return build_diff_ops(base_files, target_files, base_trees, target_trees) def _parse_with_fallback(source: bytes, file_path: str) -> SymbolTree: """Parse symbols from *source*, returning an empty tree on any error.""" try: return parse_symbols(source, file_path) except Exception: logger.debug("Symbol parsing failed for %s — falling back to file-level.", file_path) return {} # --------------------------------------------------------------------------- # Module-level singleton # --------------------------------------------------------------------------- #: The singleton plugin instance registered in ``muse/plugins/registry.py``. plugin = CodePlugin()