Coverage for scripts/resolve_conflicts.py: 100%
104 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-31 06:18 +0000
1#!/usr/bin/env python3
2"""Resolve sync conflicts by taking the upstream side — behind `/rhiza:update` step 6.
4`scripts/sync.py` exits **1** when a template change collides with a local edit, leaving
5`<<<<<<< ======= >>>>>>>` markers. `/update`'s policy is to
6take the **upstream** side everywhere: a rhiza-managed file is the template's to own, and
7local divergence in one is drift to be undone, not work to preserve.
9That policy was prose, which is the wrong home for text surgery. Conflict resolution is
10the one step that **rewrites files the user did not author** — a marker left behind ships
11`<<<<<<<` into a repo, and a mis-parsed block silently discards upstream's change. Prose
12also cannot be tested: nothing executes it, so the documented procedure had no coverage
13at all beyond `scan_conflict_artifacts` unit tests.
15Scope, deliberately narrow:
17* **Conflict markers are resolved.** Each ``<<<<<<< … ======= … >>>>>>>`` block is
18 replaced by its *theirs* section. Nested or malformed blocks are refused rather than
19 guessed at.
20* **`*.rej` files are reported, never applied.** Re-deriving where an unplaceable hunk
21 belongs is exactly the judgement that corrupts files, so this exits non-zero and leaves
22 it to a human.
24 The sync no longer produces rejects at all: ``git apply --reject`` was the only thing
25 that ever created one, and `_rhiza_merge.py` replaced the whole ``git apply`` path. An
26 earlier version of this script *deleted* a reject sitting beside a file it had just
27 resolved, on the sound reasoning that `sync.py` then emitted both artifacts for one
28 collision — markers plus the identical hunk — so applying the reject too would apply
29 the change twice. That cause is gone, and with it the justification: a reject found
30 today came from an older sync or a hand-run ``git apply``, and deleting it because we
31 happened to resolve markers in the same file would be a guess about contents nobody
32 checked.
34Usage:
35 uv run --python 3.12 --no-project python \
36 scripts/resolve_conflicts.py [TARGET] [--dry-run] [--json]
38Exit codes:
39 0 no conflicts, or every marker resolved and no rejects remain
40 1 `*.rej` files remain — they need a human (the sync cannot create these)
41 2 a malformed conflict block was found and nothing was written
42"""
44from __future__ import annotations
46import argparse
47import json
48import sys
49from pathlib import Path
50from typing import Any
52OURS = "<<<<<<<"
53SEPARATOR = "======="
54THEIRS = ">>>>>>>"
56EXIT_OK = 0
57EXIT_REJECTS_REMAIN = 1
58EXIT_MALFORMED = 2
60# Binary and vendored trees are never conflict-resolved by hand.
61_SKIP_DIRS = frozenset({".git", "node_modules", "__pycache__", ".venv"})
64class MalformedConflict(Exception):
65 """A conflict block that cannot be resolved without guessing."""
68def take_theirs(text: str) -> tuple[str, int]:
69 """Replace every conflict block in *text* with its upstream side.
71 Returns ``(resolved_text, blocks_resolved)``. Raises :class:`MalformedConflict` when
72 a block is unterminated or the markers are out of order — the caller must see that
73 rather than receive a plausible-looking file.
74 """
75 out: list[str] = []
76 resolved = 0
77 theirs: list[str] | None = None
78 state = "copy" # copy -> ours -> theirs
80 for number, line in enumerate(text.splitlines(keepends=True), start=1):
81 marker = line.split(" ", 1)[0].rstrip("\r\n")
82 if state == "copy":
83 if marker == OURS:
84 state, theirs = "ours", []
85 elif marker in (SEPARATOR, THEIRS):
86 raise MalformedConflict(f"line {number}: {marker!r} outside a conflict block")
87 else:
88 out.append(line)
89 elif state == "ours":
90 if marker == SEPARATOR:
91 state = "theirs"
92 elif marker == OURS:
93 raise MalformedConflict(f"line {number}: nested conflict block")
94 # Anything else is our side, which is discarded.
95 else: # theirs
96 if marker == THEIRS:
97 assert theirs is not None
98 out.extend(theirs)
99 resolved += 1
100 state, theirs = "copy", None
101 elif marker in (OURS, SEPARATOR):
102 raise MalformedConflict(f"line {number}: {marker!r} inside the upstream side")
103 else:
104 assert theirs is not None
105 theirs.append(line)
107 if state != "copy":
108 raise MalformedConflict("unterminated conflict block at end of file")
109 return "".join(out), resolved
112def _walk(target: Path) -> list[Path]:
113 """Return the candidate files under *target*, skipping vendored trees."""
114 found = []
115 for path in sorted(target.rglob("*")):
116 if any(part in _SKIP_DIRS for part in path.parts):
117 continue
118 if path.is_file():
119 found.append(path)
120 return found
123def find_conflicts(target: Path) -> tuple[list[Path], list[Path]]:
124 """Return ``(files_with_markers, reject_files)`` under *target*."""
125 marked: list[Path] = []
126 rejects: list[Path] = []
127 for path in _walk(target):
128 if path.suffix == ".rej":
129 rejects.append(path)
130 continue
131 try:
132 text = path.read_text()
133 except (OSError, UnicodeDecodeError):
134 continue # binary or unreadable: never a text conflict
135 if any(line.startswith(OURS) for line in text.splitlines()):
136 marked.append(path)
137 return marked, rejects
140def resolve(target: Path, *, dry_run: bool = False) -> dict[str, Any]:
141 """Take the upstream side throughout *target*; return a summary dict."""
142 marked, rejects = find_conflicts(target)
143 resolved: list[dict[str, Any]] = []
145 for path in marked:
146 original = path.read_text()
147 try:
148 new_text, blocks = take_theirs(original)
149 except MalformedConflict as exc:
150 return {
151 "resolved": [],
152 "rejects": [str(p.relative_to(target)) for p in rejects],
153 "notes": [f"{path.relative_to(target)}: {exc} — nothing was written"],
154 "exit_code": EXIT_MALFORMED,
155 }
156 if not dry_run:
157 path.write_text(new_text)
158 resolved.append({"path": str(path.relative_to(target)), "blocks": blocks})
160 outstanding = [str(path.relative_to(target)) for path in rejects]
162 notes: list[str] = []
163 if dry_run and resolved:
164 notes.append("dry run — nothing was written")
165 if outstanding:
166 notes.append(
167 f"{len(outstanding)} .rej file(s) remain. The sync no longer creates these — "
168 "`git apply --reject` was the only thing that ever did, and it is gone — so one "
169 "here came from an older sync or a hand-run `git apply`, and its contents cannot "
170 "be assumed redundant. Apply them by hand, then delete the .rej."
171 )
172 if not marked and not rejects:
173 notes.append("no conflicts found")
175 return {
176 "resolved": resolved,
177 "rejects": outstanding,
178 "notes": notes,
179 "exit_code": EXIT_REJECTS_REMAIN if outstanding else EXIT_OK,
180 }
183def main(argv: list[str] | None = None) -> int:
184 """Entry point: resolve conflict markers and return an exit code."""
185 parser = argparse.ArgumentParser(
186 description="Resolve sync conflicts by taking the upstream side.",
187 )
188 parser.add_argument(
189 "target", nargs="?", default=".", help="Repository root (default: current directory)."
190 )
191 parser.add_argument(
192 "--dry-run", action="store_true", help="Report what would change, write nothing."
193 )
194 parser.add_argument(
195 "--json", dest="json_output", action="store_true", help="Emit the summary as JSON."
196 )
197 args = parser.parse_args(argv)
199 summary = resolve(Path(args.target).resolve(), dry_run=args.dry_run)
201 if args.json_output:
202 print(json.dumps(summary, indent=2))
203 else:
204 for entry in summary["resolved"]:
205 print(f"resolved {entry['path']}: {entry['blocks']} block(s) -> upstream")
206 for path in summary["rejects"]:
207 print(f"reject {path}", file=sys.stderr)
208 for note in summary["notes"]:
209 stream = sys.stdout if summary["exit_code"] == EXIT_OK else sys.stderr
210 print(f"note {note}", file=stream)
211 return int(summary["exit_code"])
214if __name__ == "__main__":
215 raise SystemExit(main())