"""Chess domain plugin — a chess game as an ordered sequence of moves. Built live in Build With Muse, Episode 22 ("Build Something Weird With Muse"), the season finale, starting from ``muse/plugins/scaffold`` and following the exact pattern established by ``muse/plugins/todo`` (Episode 06) and ``muse/plugins/midi`` (Episode 11). One tracked file, ``game.moves``: one SAN move per line, in play order. Moves are addressed *by position* (unlike Todo's by-content addressing) — order is the entire point of a chess game. Two branches that diverge after the same shared opening is not a metaphorical "conflict"; it is literally what chess players call a *variation* — two different continuations from the same position. Muse's merge conflict is the exact same concept the domain already has a word for. """ import pathlib from difflib import SequenceMatcher from muse._version import __version__ from muse.core.object_store import read_object, write_object from muse.core.schema import DomainSchema, SequenceSchema from muse.core.types import Manifest, blob_id from muse.domain import ( ConflictRecord, DeleteOp, DriftReport, InsertOp, LiveState, MergeResult, PatchOp, SnapshotManifest, StateDelta, StateSnapshot, StructuredDelta, ) _DOMAIN_NAME = "chess" _GAME_FILE = "game.moves" def _load_moves(repo_root: pathlib.Path | None, snap: StateSnapshot) -> list[str]: object_id = snap["files"].get(_GAME_FILE) if object_id is None or repo_root is None: return [] raw = read_object(repo_root, object_id) if raw is None: # Uncommitted working-tree state -- same fallback Todo (Episode 06) # needed: snapshot() hashes the live file but only `commit` ever # writes it to the object store. live_path = repo_root / _GAME_FILE if not live_path.is_file(): return [] raw = live_path.read_bytes() return [line for line in raw.decode("utf-8").splitlines() if line.strip()] def _write_moves(repo_root: pathlib.Path, moves: list[str]) -> str: body = "\n".join(moves) if body: body += "\n" raw = body.encode("utf-8") object_id = blob_id(raw) write_object(repo_root, object_id, raw) return object_id def _move_content_id(move: str, position: int) -> str: # Position is part of the hashed content -- the same SAN move ("Nf3") # played at two different points in the game is a different element, # exactly like two different note-on events at different beats in the # MIDI domain (Episode 11). return blob_id(f"{position}:{move}".encode("utf-8")) class ChessPlugin: """A chess game as a position-addressed, LCS-diffable move sequence.""" def snapshot(self, live_state: LiveState) -> StateSnapshot: if isinstance(live_state, pathlib.Path): workdir = live_state game_path = workdir / _GAME_FILE files: Manifest = {} if game_path.is_file(): raw = game_path.read_bytes() files[_GAME_FILE] = blob_id(raw) return SnapshotManifest(files=files, domain=_DOMAIN_NAME, directories=[]) return live_state def diff( self, base: StateSnapshot, target: StateSnapshot, *, repo_root: pathlib.Path | None = None, ) -> StateDelta: base_oid = base["files"].get(_GAME_FILE) target_oid = target["files"].get(_GAME_FILE) if base_oid == target_oid: return StructuredDelta(domain=_DOMAIN_NAME, ops=[], summary="no change") base_moves = _load_moves(repo_root, base) target_moves = _load_moves(repo_root, target) child_ops: list[InsertOp | DeleteOp] = [] matcher = SequenceMatcher(a=base_moves, b=target_moves, autojunk=False) for tag, i1, i2, j1, j2 in matcher.get_opcodes(): if tag == "equal": continue for i in range(i1, i2): child_ops.append(DeleteOp( op="delete", address=_GAME_FILE, position=i, content_id=_move_content_id(base_moves[i], i), content_summary=f"removed move {i + 1}: {base_moves[i]}", )) for j in range(j1, j2): child_ops.append(InsertOp( op="insert", address=_GAME_FILE, position=j, content_id=_move_content_id(target_moves[j], j), content_summary=f"played move {j + 1}: {target_moves[j]}", )) added = sum(1 for op in child_ops if op["op"] == "insert") removed = sum(1 for op in child_ops if op["op"] == "delete") summary = f"{added} move(s) played, {removed} move(s) removed" file_change = "added" if base_oid is None else "deleted" if target_oid is None else "modified" ops = [PatchOp( op="patch", address=_GAME_FILE, child_ops=child_ops, child_domain=_DOMAIN_NAME, child_summary=summary, file_change=file_change, )] return StructuredDelta(domain=_DOMAIN_NAME, ops=ops, summary=summary) def merge( self, base: StateSnapshot, left: StateSnapshot, right: StateSnapshot, *, repo_root: pathlib.Path | None = None, ) -> MergeResult: base_moves = _load_moves(repo_root, base) left_moves = _load_moves(repo_root, left) right_moves = _load_moves(repo_root, right) # The shared prefix is however much of `base` both sides still agree # on -- the actual moves played, not just base's own length, since # one side may have already diverged from base's tail. shared_len = 0 max_shared = min(len(left_moves), len(right_moves)) while shared_len < max_shared and left_moves[shared_len] == right_moves[shared_len]: shared_len += 1 left_tail = left_moves[shared_len:] right_tail = right_moves[shared_len:] if not left_tail: # Left added nothing past the shared prefix -- right's tail wins # cleanly (right is a superset continuation, or equal). merged_moves = left_moves[:shared_len] + right_tail conflicts: list[str] = [] elif not right_tail: merged_moves = left_moves[:shared_len] + left_tail conflicts = [] else: # Both sides played a *different* next move from the same # position -- a genuine variation. This is not resolvable by any # automatic rule (picking one side silently discards real # analysis) -- surface it as a real conflict, same discipline as # Episode 08's code-domain conflict and Episode 11's MIDI note # conflict. merged_moves = left_moves[:shared_len] conflicts = [_GAME_FILE] if repo_root is not None and not conflicts: object_id = _write_moves(repo_root, merged_moves) files: Manifest = {_GAME_FILE: object_id} if merged_moves else {} else: files = {} conflict_records: list[ConflictRecord] = [] if conflicts: conflict_records.append(ConflictRecord( path=_GAME_FILE, conflict_type="variation", ours_summary=f"ours continues: {' '.join(left_tail[:3])}...", theirs_summary=f"theirs continues: {' '.join(right_tail[:3])}...", addresses=[f"{_GAME_FILE}::{shared_len}"], ours_action="modify", theirs_action="modify", )) return MergeResult( merged=SnapshotManifest(files=files, domain=_DOMAIN_NAME, directories=[]), conflicts=conflicts, conflict_records=conflict_records, ) def drift(self, committed: StateSnapshot, live: LiveState) -> DriftReport: current = self.snapshot(live) repo_root = live if isinstance(live, pathlib.Path) else None delta = self.diff(committed, current, repo_root=repo_root) return DriftReport( has_drift=len(delta["ops"]) > 0, summary=delta["summary"], delta=delta, ) def apply(self, delta: StateDelta, live_state: LiveState) -> LiveState: # The core engine already restores game.moves' bytes from the object # store during checkout -- nothing domain-specific to do here. return live_state def schema(self) -> DomainSchema: return DomainSchema( domain=_DOMAIN_NAME, description=( "A chess game as an ordered sequence of moves (game.moves, " "one SAN move per line). Moves are position-addressed, " "LCS-diffable. Two branches diverging after a shared " "opening produce a real 'variation' conflict -- the exact " "term chess players already use for this situation." ), top_level=SequenceSchema( kind="sequence", element_type="move", identity="by_position", diff_algorithm="lcs", alphabet=None, ), dimensions=[], merge_mode="three_way", schema_version=__version__, )