#!/usr/bin/env python3 """Muse Domain Registry — standalone HTML generator. Produces a self-contained, shareable page that explains the MuseDomainPlugin protocol, shows the registered plugin ecosystem, and guides developers through scaffolding and publishing their own domain plugin. Stand-alone usage ----------------- python tools/render_domain_registry.py python tools/render_domain_registry.py --out artifacts/domain_registry.html """ from __future__ import annotations import json import pathlib import subprocess import sys _ROOT = pathlib.Path(__file__).resolve().parent.parent # --------------------------------------------------------------------------- # Live domain data from the CLI # --------------------------------------------------------------------------- def _compute_crdt_demos() -> list[dict]: """Run the four CRDT primitives live and return formatted demo output.""" sys.path.insert(0, str(_ROOT)) try: from muse.core.crdts import GCounter, LWWRegister, ORSet, VectorClock # ORSet base, _ = ORSet().add("annotation-GO:0001234") a, _ = base.add("annotation-GO:0001234") b = base.remove("annotation-GO:0001234", base.tokens_for("annotation-GO:0001234")) merged = a.join(b) orset_out = "\n".join([ "ORSet — add-wins concurrent merge:", f" base elements: {sorted(base.elements())}", f" A re-adds → elements: {sorted(a.elements())}", f" B removes → elements: {sorted(b.elements())}", f" join(A, B) → elements: {sorted(merged.elements())}", " [A's new token is not tombstoned — add always wins]", ]) # LWWRegister ra = LWWRegister.from_dict({"value": "80 BPM", "timestamp": 1.0, "author": "agent-A"}) rb = LWWRegister.from_dict({"value": "120 BPM", "timestamp": 2.0, "author": "agent-B"}) rm = ra.join(rb) lww_out = "\n".join([ "LWWRegister — last-write-wins scalar:", f" Agent A writes: '{ra.read()}' at t=1.0", f" Agent B writes: '{rb.read()}' at t=2.0 (later)", f" join(A, B) → '{rm.read()}' [higher timestamp wins]", " join(B, A) → same result [commutativity]", ]) # GCounter ca = GCounter().increment("agent-A").increment("agent-A") cb = GCounter().increment("agent-B").increment("agent-B").increment("agent-B") cm = ca.join(cb) gc_out = "\n".join([ "GCounter — grow-only distributed counter:", f" Agent A x2 → A slot: {ca.value_for('agent-A')}", f" Agent B x3 → B slot: {cb.value_for('agent-B')}", f" join(A, B) global value: {cm.value()}", " [monotonically non-decreasing — joins never lose counts]", ]) # VectorClock va = VectorClock().increment("agent-A") vb = VectorClock().increment("agent-B") vm = va.merge(vb) vc_out = "\n".join([ "VectorClock — causal ordering:", f" Agent A: {va.to_dict()}", f" Agent B: {vb.to_dict()}", f" concurrent_with(A, B): {va.concurrent_with(vb)}", f" merge(A, B): {vm.to_dict()} [component-wise max]", ]) return [ {"type": "ORSet", "sub": "Observed-Remove Set", "color": "#bc8cff", "icon": "∪", "output": orset_out}, {"type": "LWWRegister", "sub": "Last-Write-Wins Register", "color": "#58a6ff", "icon": "✎", "output": lww_out}, {"type": "GCounter", "sub": "Grow-Only Distributed Counter", "color": "#3fb950", "icon": "↑", "output": gc_out}, {"type": "VectorClock", "sub": "Causal Ordering", "color": "#f9a825", "icon": "⊕", "output": vc_out}, ] except Exception as exc: print(f" ⚠ CRDT demo failed ({exc}); using static fallback") return [] def _load_domains() -> list[dict]: """Run `muse domains --json` and return parsed output.""" try: result = subprocess.run( [sys.executable, "-m", "muse", "domains", "--json"], capture_output=True, text=True, cwd=str(_ROOT), timeout=15, ) if result.returncode == 0: raw = result.stdout.strip() data: list[dict] = json.loads(raw) return data except Exception: pass # Fallback: static reference data return [ { "domain": "music", "active": "true", "capabilities": ["Typed Deltas", "Domain Schema", "OT Merge"], "schema": { "schema_version": "1", "merge_mode": "three_way", "description": "MIDI and audio file versioning with note-level diff and semantic merge", "dimensions": [ {"name": "melodic", "description": "Note pitches and durations over time"}, {"name": "harmonic", "description": "Chord progressions and key signatures"}, {"name": "dynamic", "description": "Velocity and expression curves"}, {"name": "structural", "description": "Track layout, time signatures, tempo map"}, ], }, } ] # --------------------------------------------------------------------------- # Scaffold template (shown in the "Build in 3 steps" section) # --------------------------------------------------------------------------- _TYPED_DELTA_EXAMPLE = """\ # muse show --json (any commit, any domain) { "commit_id": "b26f3c99", "message": "Resolve: integrate shared-state (A+B reconciled)", "operations": [ { "op_type": "ReplaceOp", "address": "shared-state.mid", "before_hash": "a1b2c3d4", "after_hash": "e5f6g7h8", "dimensions": ["structural"] }, { "op_type": "InsertOp", "address": "beta-a.mid", "after_hash": "09ab1234", "dimensions": ["rhythmic", "dynamic"] } ], "summary": { "inserted": 1, "replaced": 1, "deleted": 0 } }""" _OT_MERGE_EXAMPLE = """\ # Scenario A — independent InsertOps at different addresses → commute → clean merge left: InsertOp("ot-notes-a.mid") # tick=0, C4 E4 G4 right: InsertOp("ot-notes-b.mid") # tick=480, D4 F4 A4 transform(left, right) → no overlap → both applied result: both files present, zero conflicts ✓ # Scenario B — same address, different content → genuine conflict base: shared-melody.mid # C4 G4 left: ReplaceOp("shared-melody.mid") # C4 E4 G4 (major triad) right: ReplaceOp("shared-melody.mid") # C4 Eb4 G4 (minor triad) transform(left, right) → same address, non-commuting content result: ❌ Merge conflict in 1 file(s): CONFLICT (both modified): shared-melody.mid [musical intent differs — human must choose major or minor]""" _SCAFFOLD_SNIPPET = """\ from __future__ import annotations from muse.domain import ( MuseDomainPlugin, LiveState, StateSnapshot, StateDelta, DriftReport, MergeResult, DomainSchema, ) class GenomicsPlugin(MuseDomainPlugin): \"\"\"Version control for genomic sequences.\"\"\" def snapshot(self, live_state: LiveState) -> StateSnapshot: # Serialize current genome state to a content-addressable blob raise NotImplementedError def diff(self, base: StateSnapshot, target: StateSnapshot) -> StateDelta: # Compute minimal delta between two snapshots raise NotImplementedError def merge(self, base: StateSnapshot, left: StateSnapshot, right: StateSnapshot) -> MergeResult: # Three-way merge — surface conflicts per dimension raise NotImplementedError def drift(self, committed: StateSnapshot, live: LiveState) -> DriftReport: # Detect uncommitted changes in the working state raise NotImplementedError def apply(self, delta: StateDelta, live_state: LiveState) -> LiveState: # Reconstruct historical state from a delta raise NotImplementedError def schema(self) -> DomainSchema: # Declare dimensions — drives diff algorithm selection raise NotImplementedError """ # --------------------------------------------------------------------------- # Planned / aspirational domains # --------------------------------------------------------------------------- _PLANNED_DOMAINS = [ { "name": "Genomics", "icon": "🧬", "status": "planned", "tagline": "Version sequences, variants, and annotations", "dimensions": ["sequence", "variants", "annotations", "metadata"], "color": "#3fb950", }, { "name": "3D / Spatial", "icon": "🌐", "status": "planned", "tagline": "Merge spatial fields, meshes, and simulation frames", "dimensions": ["geometry", "materials", "physics", "temporal"], "color": "#58a6ff", }, { "name": "Financial", "icon": "📈", "status": "planned", "tagline": "Track model versions, alpha signals, and risk state", "dimensions": ["signals", "positions", "risk", "parameters"], "color": "#f9a825", }, { "name": "Scientific Simulation", "icon": "⚛️", "status": "planned", "tagline": "Snapshot simulation state across timesteps and parameter spaces", "dimensions": ["state", "parameters", "observables", "checkpoints"], "color": "#ab47bc", }, { "name": "Your Domain", "icon": "✦", "status": "yours", "tagline": "Six methods. Any multidimensional state. Full VCS for free.", "dimensions": ["your_dim_1", "your_dim_2", "..."], "color": "#4f8ef7", }, ] # --------------------------------------------------------------------------- # Distribution model description # --------------------------------------------------------------------------- _DISTRIBUTION_LEVELS = [ { "tier": "Local", "icon": "💻", "title": "Local plugin (right now)", "color": "#3fb950", "steps": [ "muse domains --new <name>", "Implement 6 methods in muse/plugins/<name>/plugin.py", "Register in muse/plugins/registry.py", "muse init --domain <name>", ], "desc": "Works today. Scaffold → implement → register. " "Your plugin lives alongside the core.", }, { "tier": "Shareable", "icon": "📦", "title": "pip-installable package (right now)", "color": "#58a6ff", "steps": [ "Package your plugin as a Python module", "pip install git+https://github.com/you/muse-plugin-genomics", "Register the entry-point in pyproject.toml", "muse init --domain genomics", ], "desc": "Share your plugin as a standard Python package. " "Anyone with pip can install and use it.", }, { "tier": "MuseHub", "icon": "🌐", "title": "Centralized registry (coming — MuseHub)", "color": "#bc8cff", "steps": [ "musehub publish muse-plugin-genomics", "musehub search genomics", "muse init --domain @musehub/genomics", "Browse plugins at musehub.io", ], "desc": "MuseHub is a planned centralized registry — npm for Muse plugins. " "Versioned, searchable, one-command install.", }, ] # --------------------------------------------------------------------------- # HTML template # --------------------------------------------------------------------------- def _render_capability_card(cap: dict) -> str: color = cap["color"] return f"""
{cap['icon']} {cap['type']} {cap['sub']} 
{cap['output']}
""" def _render_domain_card(d: dict) -> str: domain = d.get("domain", "unknown") active = d.get("active") == "true" schema = d.get("schema", {}) desc = schema.get("description", "") dims = schema.get("dimensions", []) caps = d.get("capabilities", []) cap_html = " ".join( f'{c}' for c in caps ) dim_html = " · ".join( f'{dim["name"]}' for dim in dims ) status_cls = "active-badge" if active else "reg-badge" status_text = "● active" if active else "○ registered" dot = '' if active else "" short_desc = desc[:150] + ("…" if len(desc) > 150 else "") return f"""
{status_text} {domain} {dot}

{short_desc}

{cap_html}
Dimensions: {dim_html}
""" def _render_planned_card(p: dict) -> str: dims = " · ".join(f'{d}' for d in p["dimensions"]) cls = "planned-card yours" if p["status"] == "yours" else "planned-card" return f"""
{p['icon']}
{p['name']}
{p['tagline']}
{dims}
{'Build it →' if p["status"] == "yours" else 'coming soon'}
""" def _render_dist_card(d: dict) -> str: steps = "".join( f'
  • {s}
    • ' for s in d["steps"] ) return f"""
    {d['icon']}
    {d['tier']}
    {d['title']}

    {d['desc']}

      {steps}
    """ def render(output_path: pathlib.Path) -> None: """Generate the domain registry HTML page.""" print(" Loading live domain data...") domains = _load_domains() print(f" Found {len(domains)} registered domain(s)") print(" Computing live CRDT demos...") crdt_demos = _compute_crdt_demos() active_domains_html = "\n".join(_render_domain_card(d) for d in domains) planned_html = "\n".join(_render_planned_card(p) for p in _PLANNED_DOMAINS) dist_html = "\n".join(_render_dist_card(d) for d in _DISTRIBUTION_LEVELS) crdt_cards_html = "\n".join(_render_capability_card(c) for c in crdt_demos) html = _HTML_TEMPLATE.replace("{{ACTIVE_DOMAINS}}", active_domains_html) html = html.replace("{{PLANNED_DOMAINS}}", planned_html) html = html.replace("{{DIST_CARDS}}", dist_html) html = html.replace("{{SCAFFOLD_SNIPPET}}", _SCAFFOLD_SNIPPET) html = html.replace("{{TYPED_DELTA_EXAMPLE}}", _TYPED_DELTA_EXAMPLE) html = html.replace("{{OT_MERGE_EXAMPLE}}", _OT_MERGE_EXAMPLE) html = html.replace("{{CRDT_CARDS}}", crdt_cards_html) output_path.write_text(html, encoding="utf-8") size_kb = output_path.stat().st_size // 1024 print(f" HTML written ({size_kb}KB) → {output_path}") # Also write as index.html so the domain registry IS the landing page. index_path = output_path.parent / "index.html" index_path.write_text(html, encoding="utf-8") print(f" Landing page mirrored → {index_path}") # --------------------------------------------------------------------------- # Large HTML template # --------------------------------------------------------------------------- _HTML_TEMPLATE = """\ Muse — Version Anything

    muse

    Version Anything

    One protocol. Any domain. Six methods between you and a complete version control system — branching, merging, conflict resolution, time-travel, and typed diffs — for free.

    Build a Domain Plugin Watch the Demo →
    🎵 music 🧬 genomics 🌐 3d-spatial 📈 financial ⚛️ simulation 🔬 proteomics 🏗️ cad 🎮 game-state ✦ your-domain 🎵 music 🧬 genomics 🌐 3d-spatial 📈 financial ⚛️ simulation 🔬 proteomics 🏗️ cad 🎮 game-state ✦ your-domain
    The Contract

    The MuseDomainPlugin Protocol

    Every domain — music, genomics, 3D spatial, financial models — implements the same six-method protocol. The core engine handles everything else: content-addressed storage, DAG, branches, log, merge base, cherry-pick, revert, stash, tags.

    6methods to implement
    14CLI commands, free
    ∞domains possible
    0core changes needed
    Method
    Signature
    Purpose
    snapshot
    snapshot(live) → StateSnapshot
    Capture current state as a content-addressable blob
    diff
    diff(base, target) → StateDelta
    Compute minimal change between two snapshots (added · removed · modified)
    merge
    merge(base, left, right) → MergeResult
    Three-way reconcile divergent state lines; surface conflicts per dimension
    drift
    drift(committed, live) → DriftReport
    Detect uncommitted changes between HEAD and working state
    apply
    apply(delta, live) → LiveState
    Apply a delta during checkout to reconstruct historical state
    schema
    schema() → DomainSchema
    Declare data structure — drives diff algorithm selection per dimension
    Engine Capabilities

    What Every Plugin Gets for Free

    The core engine provides four advanced capabilities that any domain plugin can opt into. Implement the protocol — the engine does the rest.

    🔬 Typed Delta Algebra StructuredDelta — every change is a typed operation

    Unlike Git's blob diffs, Muse deltas are typed objects: InsertOp, ReplaceOp, DeleteOp — each carrying the address, before/after hashes, and affected dimensions. Machine-readable with muse show --json. 

    {{TYPED_DELTA_EXAMPLE}}
    🗂️ Domain Schema Per-domain dimensions drive diff algorithm selection

    Each plugin's schema() method declares its dimensions and merge mode. The engine uses this to select the right diff algorithm per dimension and to surface only the dimensions that actually conflict.

    {{ACTIVE_DOMAINS}}
    ⚙️ OT Merge Operational transformation — independent ops commute automatically

    Plugins implementing StructuredMergePlugin get operational transformation. Operations at different addresses commute automatically — only operations on the same address with incompatible intent surface a conflict. 

    {{OT_MERGE_EXAMPLE}}
    🔮 CRDT Primitives Convergent merge — any two replicas always reach the same state

    Plugins implementing CRDTPlugin get four battle-tested convergent data structures. No coordination required between replicas.

    {{CRDT_CARDS}}
    Build

    Build in Three Steps

    One command scaffolds the entire plugin skeleton. You fill in six methods. The full VCS follows.

    Step 1 · Scaffold
    Generate the skeleton
    One command creates the plugin directory, class, and all six method stubs with full type annotations.
    muse domains --new genomics
    Step 2 · Implement
    Fill in the six methods
    Replace each raise NotImplementedError with your domain's snapshot, diff, merge, drift, apply, and schema logic.
    vim muse/plugins/genomics/plugin.py
    Step 3 · Use
    Full VCS, instantly
    Register in registry.py, then every Muse command works for your domain out of the box.
    muse init --domain genomics
    The Scaffold

    What muse domains --new genomics produces

    A fully typed, immediately runnable plugin skeleton. Every method has the correct signature. You replace the stubs — the protocol does the rest.

    muse/plugins/genomics/plugin.py
    {{SCAFFOLD_SNIPPET}}

    Full walkthrough → docs/guide/plugin-authoring-guide.md · CRDT extension → docs/guide/crdt-reference.md

    Registry

    Registered Domains

    Domains currently registered in this Muse instance. The active domain is the one used when you run muse commit, muse diff, and all other commands.

    {{ACTIVE_DOMAINS}}
    Ecosystem

    The Plugin Ecosystem

    Music is the reference implementation. These are the domains planned next — and the slot waiting for yours.

    {{PLANNED_DOMAINS}}
    Distribution

    How to Share Your Plugin

    Three tiers of distribution — from local prototype to globally searchable registry. Start local, publish when ready.

    {{DIST_CARDS}}

    MuseHub is coming

    A centralized, searchable registry for Muse domain plugins — think npm or crates.io, but for any multidimensional versioned state. One command to publish. One command to install.

    🔍
    Searchable
    Find plugins by domain, capability, or keyword
    📦
    Versioned
    Semantic versioning, pinned installs, changelogs
    🔒
    Private registries
    Self-host for enterprise or research teams
    ⚡
    One command
    muse init --domain @musehub/genomics
    MuseHub — planned · building in public at github.com/cgcardona/musehub
    """ # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- if __name__ == "__main__": import argparse parser = argparse.ArgumentParser( description="Generate the Muse domain registry HTML page" ) parser.add_argument( "--out", default=str(_ROOT / "artifacts" / "domain_registry.html"), help="Output HTML path", ) args = parser.parse_args() out_path = pathlib.Path(args.out) out_path.parent.mkdir(parents=True, exist_ok=True) print("Generating domain_registry.html...") render(out_path) print(f"Open: file://{out_path.resolve()}")