Domain Protocol

from muse.domain import (
    MuseDomainPlugin, LiveState, StateSnapshot,
    StateDelta, MergeResult, DriftReport, DomainSchema,
)
import pathlib

class MuseDomainPlugin(Protocol):

    def snapshot(self, live_state: LiveState) -> StateSnapshot:
        """Capture current state as a content-addressable SnapshotManifest.

        live_state is a Path (working tree) or an existing SnapshotManifest.
        Walk the domain's files, hash each one, return the manifest.
        Must honour .museignore — call ignore_filter(path) before adding.
        """

    def diff(
        self,
        base:   StateSnapshot,
        target: StateSnapshot,
        *,
        repo_root: pathlib.Path | None = None,
    ) -> StateDelta:
        """Compute the minimal StructuredDelta from base → target.

        StateDelta = {"domain": str, "ops": list[DomainOp], "summary": str,
                      "sem_ver_bump": "none"|"patch"|"minor"|"major",
                      "breaking_changes": list[str]}
        """

    def merge(
        self,
        base:  StateSnapshot,
        left:  StateSnapshot,
        right: StateSnapshot,
        *,
        repo_root: pathlib.Path | None = None,
    ) -> MergeResult:
        """Three-way merge. base is the common ancestor.

        MergeResult = {"merged": SnapshotManifest, "conflicts": list[str]}
        conflicts is a list of paths that could not be auto-merged.
        Consult .museattributes for per-path merge strategy overrides.
        """

    def drift(
        self,
        committed: StateSnapshot,
        live:      LiveState,
    ) -> DriftReport:
        """Detect uncommitted changes between HEAD snapshot and live working tree.

        DriftReport = dataclass(has_drift: bool, summary: str, delta: StateDelta | None)
        Called by `muse status` and `muse drift` commands.
        """

    def apply(self, delta: StateDelta, live_state: LiveState) -> LiveState:
        """Apply a delta to reconstruct historical state during checkout.

        For most file-based domains this is a no-op that returns live_state
        unchanged — the engine handles file restoration via the object store.
        Override when your domain needs post-processing (e.g. recompilation).
        """

    def schema(self) -> DomainSchema:
        """Declare the domain's data structure.

        Drives diff algorithm selection per dimension, merge mode,
        and the domain registry UI at /domains.
        """

{
    "domain":         "identity",
    "description":    "Cryptographic identity graph — humans, agents, orgs.",
    "schema_version": "1.0.0",
    "merge_mode":     "three_way",   # "three_way" | "crdt"
    "dimensions": [
        {
            "name":              "identities",
            "schema":            {"kind": "set"},   # "set" | "sequence" | "map" | "scalar"
            "independent_merge": True,        # this dim merges without blocking others
        },
        {
            "name":              "relationships",
            "schema":            {"kind": "set"},
            "independent_merge": True,
        },
    ],
}

# Ordered-sequence domains (MIDI note-level): position required
InsertOp  = {"op": "insert", "address": str, "content_id": str, "position": int}
DeleteOp  = {"op": "delete", "address": str, "content_id": str, "position": int}

# Name-addressed (unordered) domains: no position field — address IS the identity
AddressedInsertOp = {"op": "insert", "address": str, "content_id": str,
                      "content_summary": str}
AddressedDeleteOp = {"op": "delete", "address": str, "content_id": str,
                      "content_summary": str}

# Reposition within an ordered sequence — commutes with non-overlapping moves
MoveOp    = {"op": "move",   "address": str, "from_position": int, "to_position": int}

# Atomic leaf-level change; old_content_id enables conflict detection
ReplaceOp = {"op": "replace", "address": str,
              "old_content_id": str, "new_content_id": str}

# Entity modification: id + named field mutations (for structured records)
MutateOp  = {"op": "mutate", "address": str,
              "entity_id": str, "mutations": dict[str, Any]}

# Container modification: path + nested child ops (subtree patch)
PatchOp   = {"op": "patch",   "address": str, "child_ops": list[DomainOp]}

# Rename any first-class entity — directory, file, or symbol (address changed, content unchanged)
RenameOp          = {"op": "rename",            "address": str, "from_address": str}

{
  "domain":       "identity",
  "summary":      "Added 1 identity, 2 relationships",
  "sem_ver_bump": "minor",
  "ops": [
    {
      "op":         "insert",
      "address":    "identities/claude-code.json",
      "content_id": "sha256:a1b2c3...",
      "position":   null
    },
    {
      "op":             "replace",
      "address":        "identities/gabriel.json",
      "old_content_id": "sha256:d4e5f6...",
      "new_content_id": "sha256:7890ab..."
    }
  ]
}

left: InsertOp("identities/alice.json")   ─┐
right: InsertOp("identities/bob.json")    ─┘  → auto-merge (different addresses)

left: ReplaceOp("identities/gabriel.json", v1→v2)   ─┐
right: ReplaceOp("identities/gabriel.json", v1→v3)  ─┘  → CONFLICT (same addr, different targets)

left: ReplaceOp("identities/gabriel.json", v1→v2)   ─┐
right: ReplaceOp("identities/gabriel.json", v1→v2)  ─┘  → auto-merge (consensus: identical op)

{
  "MAX_CONNECTIONS": 10,
  "TIMEOUT_MS":      3000,
  "LOG_LEVEL":       "info"
}

{
  "domain": "json-docs",
  "ops": [
    {
      "op":             "replace",
      "address":        "config/app.json",
      "old_content_id": "sha256:base00…",
      "new_content_id": "sha256:left01…"    // MAX_CONNECTIONS: 10 → 20
    }
  ],
  "summary": "scale MAX_CONNECTIONS to 20"
}

{
  "domain": "json-docs",
  "ops": [
    {
      "op":             "replace",
      "address":        "config/app.json",
      "old_content_id": "sha256:base00…",
      "new_content_id": "sha256:right02…"   // MAX_CONNECTIONS: 10 → 50, TIMEOUT_MS: 3000 → 5000
    }
  ],
  "summary": "high-load tuning: MAX_CONNECTIONS 50, TIMEOUT_MS 5000"
}

muse merge feat/high-load

# Manually set MAX_CONNECTIONS: 50, TIMEOUT_MS: 5000 (take B's tuning, accept A's intent)
# ... edit config/app.json ...
muse code add config/app.json
muse commit -m "merge: resolve MAX_CONNECTIONS conflict — use 50 for high-load" \
  --agent-id claude-code --model-id claude-sonnet-4-6 --sign

class AddressedMergePlugin(MuseDomainPlugin, Protocol):
    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:
        """Address-keyed Map merge. Ops at different addresses commute; ops at the
        same address with incompatible intent produce a conflict entry."""

class HarmonyPlugin(MuseDomainPlugin, Protocol):
    def conflict_fingerprint(
        self,
        path:     str,
        ours_id:  str,   # "sha256:..." of ours blob
        theirs_id:str,   # "sha256:..." of theirs blob
        repo_root:pathlib.Path,
    ) -> str:            # 64-char hex semantic fingerprint
        """Produce a fingerprint that captures the conflict's structural shape
        independent of timestamps, signatures, or formatting noise.

        Two conflicts that represent the same logical change (e.g. the same
        edge being added with different timestamps) must return the same
        fingerprint — enabling Harmony Tier 3 to replay the resolution."""

    def similarity(self, fp_a: str, fp_b: str) -> float:
        """Score structural similarity between two conflict fingerprints.
        Return value in [0.0, 1.0]. Out-of-range values are clamped.
        DefaultPlugin returns 1.0 for identical fps, 0.0 otherwise."""

from __future__ import annotations
import hashlib, json, pathlib
from muse.domain import (
    DomainSchema, DriftReport, LiveState, MergeResult,
    MuseDomainPlugin, SnapshotManifest, StateDelta, StateSnapshot,
)

def _sha256(data: bytes) -> str:
    return "sha256:" + hashlib.sha256(data).hexdigest()

class JsonDocPlugin:
    """Version a flat directory of *.json documents."""

    def schema(self) -> DomainSchema:
        return {
            "domain":         "json-docs",
            "description":    "Flat directory of JSON documents.",
            "schema_version": "1.0.0",
            "merge_mode":     "three_way",
            "dimensions": [{
                "name": "documents",
                "schema": {"kind": "set"},
                "independent_merge": True,
            }],
        }

    def snapshot(self, live_state: LiveState) -> StateSnapshot:
        root = pathlib.Path(live_state) if isinstance(live_state, (str, pathlib.Path)) else None
        if root is None:
            return live_state   # already a SnapshotManifest
        files = {}
        for p in sorted(root.glob("**/*.json")):
            rel = p.relative_to(root).as_posix()
            files[rel] = _sha256(p.read_bytes())
        return SnapshotManifest(files=files, domain="json-docs", directories=[])

    def diff(self, base: StateSnapshot, target: StateSnapshot, **_) -> StateDelta:
        b, t = base["files"], target["files"]
        ops = []
        for path in t.keys() - b.keys():
            ops.append({"op": "insert", "address": path, "content_id": t[path], "position": None})
        for path in b.keys() - t.keys():
            ops.append({"op": "delete", "address": path, "content_id": b[path], "position": None})
        for path in b.keys() & t.keys():
            if b[path] != t[path]:
                ops.append({"op": "replace", "address": path,
                             "old_content_id": b[path], "new_content_id": t[path]})
        adds = sum(1 for o in ops if o["op"] == "insert")
        return {"domain": "json-docs", "ops": ops,
                "summary": f"{adds} added, {len(ops)-adds} changed",
                "sem_ver_bump": "minor" if adds else "patch"}

    def merge(self, base, left, right, **_) -> MergeResult:
        b, l, r = base["files"], left["files"], right["files"]
        merged, conflicts = dict(b), []
        for path in l.keys() | r.keys() | b.keys():
            lv, rv, bv = l.get(path), r.get(path), b.get(path)
            if lv == rv:                merged[path] = lv   # consensus / unchanged
            elif lv is None:           merged.pop(path, None)  # deleted on left
            elif rv is None:           merged.pop(path, None)  # deleted on right
            elif lv == bv:              merged[path] = rv   # only right changed
            elif rv == bv:              merged[path] = lv   # only left changed
            else:                       conflicts.append(path)  # both changed differently
        merged = {k: v for k, v in merged.items() if v is not None}
        return {"merged": SnapshotManifest(files=merged, domain="json-docs", directories=[]),
                "conflicts": conflicts}

    def drift(self, committed, live) -> DriftReport:
        current = self.snapshot(live)
        delta = self.diff(committed, current)
        return DriftReport(has_drift=bool(delta["ops"]), summary=delta["summary"], delta=delta)

    def apply(self, delta, live_state) -> LiveState:
        return live_state  # engine handles file restoration via object store

# Pattern            Strategy
*.lock               merge=ours
generated/**         merge=ours
tracks/master.mid    merge=theirs          # always take incoming master
config/schema.json   merge=union           # merge JSON keys additively
*.mid                merge=domain          # delegate to MidiPlugin (default)

# Query the resolved strategy for a path
muse check-attr tracks/drums.mid --json

# List all attribute rules
muse attributes list --json

# Validate the .museattributes file
muse attributes validate --json

Do you need Muse to understand the structure of your files?
├── No  → Use the code or json-docs domain. Add .museattributes
│         rules for files that need custom merge strategies. Done.
│
└── Yes → Do your files have multiple independent dimensions
          (e.g. notes vs. tempo in a MIDI file)?
          │
          ├── No  → Implement MuseDomainPlugin with merge_mode: three_way.
          │         Override merge() for your custom merge semantics.
          │         ~50 lines. See the JsonDocPlugin example.
          │
          └── Yes → Can edits to different dimensions always
                    merge without conflicting each other?
                    │
                    ├── Yes → Use independent_merge: True per dimension
                    │         in DomainSchema. The engine parallelises
                    │         dimension merges automatically.
                    │
                    └── No  → Implement AddressedMergePlugin.
                              Override merge_ops()
                              for address-keyed Map merge. Adds ~30 lines.
                              │
                              └── Also want Harmony Tier 3
                                  (semantic conflict replay)?
                                  └── Implement HarmonyPlugin too.
                                      Add conflict_fingerprint()
                                      + similarity(). ~20 more lines.

import hashlib

def domain_index(name: str) -> int:
    """Return the 31-bit HD path integer for a domain name string."""
    digest = hashlib.sha256(name.encode()).digest()
    return int.from_bytes(digest[:4], "big") & 0x7FFFFFFF

# Compute the HD path integer for any domain name
muse domain index muse/identity --json
# → {"name": "muse/identity", "domain_int": 1660078172, ...}

# Reverse-lookup: name from integer
muse domain lookup 1660078172 --json
# → {"domain_int": 1660078172, "name": "muse/identity", ...}

# List all known first-party domains
muse domain list --json

# Verify the integer matches the name (useful in CI)
muse domain check muse/identity 1660078172 --json

from muse.plugins.code.plugin     import CodePlugin
from muse.plugins.identity.plugin import IdentityPlugin
from muse.plugins.mist.plugin     import MistPlugin
from my_domain.plugin             import JsonDocPlugin   # your plugin

_REGISTRY: dict[str, MuseDomainPlugin] = {
    "code":      CodePlugin(),
    "identity":  IdentityPlugin(),
    "mist":      MistPlugin(),
    "json-docs": JsonDocPlugin(),   # registered here
}

# Init a repo with your domain
muse init --domain json-docs

# Inspect the active plugin
muse domain-info --json

# List all registered domains
muse domains --json

# See API surface of the active domain
muse api-surface --json

# Decode an HD path to human-readable labels
muse path annotate "m/1075233755'/1660078172'/0'/0'/0'/0'" --json