Harmony

# idiomatic workflow
muse merge feature-x --json
# ✔ [harmony] auto-resolved: config.py::MAX_CONNECTIONS   ← Tier 2 replay
# CONFLICT: src/server.py::timeout                         ← new — no prior pattern

muse checkout --ours src/server.py
muse code add src/server.py
muse commit -m "merge: resolve timeout conflict" \
  --agent-id claude-code --model-id claude-sonnet-4-6 --sign
# ✅ harmony: recorded resolution for 'src/server.py::timeout'

# next merge — same conflict auto-resolved
muse merge feature-y --json
# ✔ [harmony] auto-resolved: src/server.py::timeout

muse merge feature-a --json

# Ours: MAX_CONNECTIONS = 20   Theirs: MAX_CONNECTIONS = 50
# The feature branch value is higher — accept it
muse checkout --theirs config.py
muse code add config.py

muse commit -m "merge: accept feature-a MAX_CONNECTIONS=50" \
  --agent-id claude-code --model-id claude-sonnet-4-6 --sign

muse harmony patterns --json | python3 -c "import sys,json; [print(p['pattern_id'], p['path']) for p in json.load(sys.stdin)['patterns']]"

muse harmony best --pattern-id a4f3c9d8e1b2... --json

muse merge feature-b --json

# blob fingerprint — commutative; order of ours/theirs doesn't matter
lo, hi = sorted((ours_id, theirs_id))
blob_fp = SHA256({"ids": [lo, hi]})

# pattern_id — keyed on semantic_fp when a plugin provides it
if semantic_fp != blob_fp:
    # cross-content replay: different bytes, same semantic shape → same pattern
    pattern_id = SHA256({"path": path, "semantic_fp": semantic_fp})
else:
    # no plugin — exact-replay mode only
    pattern_id = SHA256({"blob_fp": blob_fp, "path": path, "semantic_fp": semantic_fp})

# best_resolution() sort — descending
resolutions.sort(
    key=lambda r: (r.human_verified, r.confidence, r.applied_count),
    reverse=True
)
# human_verified=True always wins regardless of confidence score
# then highest confidence
# then most frequently applied (battle-tested)

muse harmony engine <pattern_id> --json
muse harmony engine <pattern_id> --auto-apply-threshold 0.95 --json
muse harmony engine <pattern_id> --auto-escalate --json

# always take ours for generated lock files
muse harmony policy-add \
  --id lock-files-ours \
  --action prefer-ours \
  --path-pattern "*.lock" \
  --confidence 1.0

# always prefer theirs for mist artifacts (immutable content-addressed blobs)
muse harmony policy-add \
  --id mist-theirs \
  --action prefer-theirs \
  --domain mist \
  --path-pattern "artifacts/**" \
  --confidence 0.9

# escalate conflicts in the auth module to a security specialist
muse harmony policy-add \
  --id auth-escalate \
  --action escalate \
  --path-pattern "src/auth/**" \
  --escalate-to security-agent

muse harmony policies --json
muse harmony policy-remove --id lock-files-ours --json

muse harmony escalations --json                       # list all escalations
muse harmony escalations --status open --json         # open only
muse harmony resolve-escalation <escalation_id> \
  --resolution-id <resolution_id> --json              # close with a resolution

# Harmony calls this internally when escalation fires:
muse -C ~/ecosystem/musehub hub issue create \
  --title "Harmony escalation: src/auth/tokens.py::validate_token" \
  --body  "**Pattern ID**: a4f3c9d8e1b2...\n**Reason**: no resolution found after 3 tiers\n**Conflict type**: concurrent_edit\n**Ours**: sha256:f3a7b9...\n**Theirs**: sha256:c2d8e1..." \
  --label "harmony-escalation" \
  --assignee security-agent \
  --hub http://staging.musehub.ai \
  --json

# Resolve the conflict manually, then commit
muse code add src/auth/tokens.py
muse commit -m "resolve: auth token validation conflict (escalation #42)" \
  --agent-id security-agent --model-id claude-sonnet-4-6 --sign
# ✅ harmony: recorded resolution for 'src/auth/tokens.py::validate_token'
# ✅ harmony: escalation d9b4c1... closed
# ✅ hub: issue #42 closed automatically

from typing import Protocol, runtime_checkable

@runtime_checkable
class HarmonyPlugin(Protocol):

    def conflict_fingerprint(
        self,
        path: str,
        ours_id: str,
        theirs_id: str,
        repo_root: Path,
    ) -> str:
        """Return a domain-semantic fingerprint for this conflict.

        If the fingerprint differs from the blob_fingerprint, conflicts with
        different byte content but identical semantic shape share one pattern
        and replay the same resolution (cross-content replay).

        Return blob_fingerprint to opt out of semantic keying.
        """
        ...

    def similarity(self, fp_a: str, fp_b: str) -> float:
        """Score semantic similarity between two fingerprints in [0.0, 1.0].

        Called by Tier 3. Scores above EngineConfig.semantic_threshold
        (default 0.70) are returned as proposals.

        Default plugin: returns 1.0 for identical, 0.0 for all others.
        """
        ...

# normalization rules applied before fingerprinting
# Python keywords     → preserved as-is
# Identifiers         → "ID"
# String literals     → "STR"
# Number literals     → "NUM"
# Operators/punct     → preserved
# Comments/whitespace → stripped

# Tanimoto similarity on multisets A, B
# sim(A, B) = Σ min(A[t], B[t]) / Σ max(A[t], B[t])
# Two functions with different variable names but same control flow
# score ~0.85 — above semantic_threshold; proposed but not auto-applied

import hashlib, json
from pathlib import Path

class JsonConfigPlugin:
    """HarmonyPlugin for JSON config files.

    Fingerprints on which keys differ — not the values themselves.
    Conflicts that change the same key for different reasons share
    one pattern and replay the same resolution strategy.
    """

    def conflict_fingerprint(
        self,
        path: str,
        ours_id: str,
        theirs_id: str,
        repo_root: Path,
    ) -> str:
        try:
            ours   = json.loads((repo_root / (path + ".ours")).read_text())
            theirs = json.loads((repo_root / (path + ".theirs")).read_text())
        except (FileNotFoundError, json.JSONDecodeError):
            return _blob_fp(ours_id, theirs_id)  # fall back to exact replay

        changed_keys = sorted(
            k for k in ours.keys() | theirs.keys()
            if ours.get(k) != theirs.get(k)
        )
        payload = json.dumps({"path": path, "keys": changed_keys},
                              sort_keys=True).encode()
        return hashlib.sha256(payload).hexdigest()

    def similarity(self, fp_a: str, fp_b: str) -> float:
        # Same changed-key set → same semantic pattern
        return 1.0 if fp_a == fp_b else 0.0


def _blob_fp(ours_id: str, theirs_id: str) -> str:
    lo, hi = sorted((ours_id, theirs_id))
    return hashlib.sha256(json.dumps({"ids": [lo, hi]}).encode()).hexdigest()

class JsonConfigDomain:
    domain_name = "json_config"

    def harmony_plugin(self) -> JsonConfigPlugin | None:
        return JsonConfigPlugin()

# disable harmony for one merge (e.g. intentional divergence)
muse merge feature-x --no-harmony-autoupdate --json

.muse/harmony/
  policies/
    <policy_id>.json            ← one per policy; URL-safe alphanumeric name

  patterns/
    sha256/
      <64-hex-pattern-id>/
        pattern.json             ← ConflictPattern record (max 32 KiB)
        resolutions/
          sha256/
            <64-hex>.json       ← one Resolution per file (max 16 KiB)

  audit/
    <YYYYMMDD>-<hex-slice>.json ← append-only; newest-first by filename sort

  escalations/
    sha256/
      <64-hex>.json             ← one EscalationRecord per file (max 16 KiB)

import hashlib, json

entry = {
    "event_type":    "resolution_applied",
    "pattern_id":    "a4f3c9d8e1b2...",
    "resolution_id": "7c1a9e3f4d82...",
    "policy_id":     None,
    "acted_by":      {"agent_id": "claude-code", "model_id": "claude-sonnet-4-6"},
    "occurred_at":   "2026-04-21T16:00:00Z",
    "metadata":      {"confidence": 1.0, "applied_count_after": 1},
}

canonical = json.dumps(entry, sort_keys=True, separators=(",", ":"))
audit_id  = hashlib.sha256(canonical.encode()).hexdigest()
# audit_id is the filename: .muse/harmony/audit/20260421-a4f3c9.json

muse harmony audit --json
muse harmony audit --limit 50 --json

# 1. Merge — Harmony auto-applies saved resolutions
muse merge feature-x --json
# ✔ [harmony] auto-resolved: config.py::MAX_CONNECTIONS
# CONFLICT: src/server.py::timeout

# 2. Inspect what Harmony knows
muse harmony patterns --json
muse harmony resolutions --pattern-id <id> --json

# 3. Resolve remaining conflicts manually
muse checkout --ours src/server.py
muse code add src/server.py

# 4. Commit — Harmony records the resolution (confidence=1.0, human_verified=true)
muse commit -m "merge: resolve timeout conflict" \
  --agent-id claude-code --model-id claude-sonnet-4-6 --sign

# 5. Next time the same conflict recurs → Tier 2 auto-resolved
muse merge feature-y --json
# ✔ [harmony] auto-resolved: src/server.py::timeout