import datetime
import re
import subprocess
import sys
from pathlib import Path
BEGIN_MARKER = (
"<!-- BEGIN merge-pr state: generated by scripts/update_wip.py on every `just merge-pr`."
" Anything between these markers is overwritten; write notes outside them. -->"
)
END_MARKER = "<!-- END merge-pr state -->"
_BEGIN_RE = re.compile(r"^<!-- BEGIN merge-pr state\b.*-->[ \t]*$", re.MULTILINE)
_END_RE = re.compile(r"^" + re.escape(END_MARKER) + r"[ \t]*$", re.MULTILINE)
def read_preserving_newlines(path):
raw = path.read_bytes()
crlf = raw.count(b"\r\n")
lf = raw.count(b"\n") - crlf
newline = "\r\n" if crlf > lf else "\n"
return raw.decode("utf-8").replace("\r\n", "\n"), newline
def write_preserving_newlines(path, text, newline):
path.write_bytes(text.replace("\n", newline).encode("utf-8"))
def run(cmd, required=True):
try:
return subprocess.run(
cmd, capture_output=True, text=True, encoding="utf-8", check=True
).stdout.strip()
except subprocess.CalledProcessError as e:
if not required:
return None
print(f"Command failed: {' '.join(cmd)}", file=sys.stderr)
if e.stdout:
print(e.stdout, file=sys.stderr)
if e.stderr:
print(e.stderr, file=sys.stderr)
sys.exit(e.returncode)
def read_cargo_version(cargo_file):
try:
for line in cargo_file.read_text(encoding="utf-8").splitlines():
m = re.match(r'version\s*=\s*"([^"]+)"', line)
if m:
return m.group(1)
except OSError:
pass
return None
def _cell(value):
return value.replace("|", "\\|").replace("\n", " ")
def render_state_block(state):
branch = state.get("branch")
if branch in (None, "main"):
active = "none — `main` is current"
else:
active = f"`{_cell(branch)}` — **not `main`**; check what is checked out"
versions = [f"`retch-cli` {_cell(state.get('cli_version') or 'unknown')}"]
if state.get("sysinfo_version"):
versions.append(f"`retch-sysinfo` {_cell(state['sysinfo_version'])}")
head = f"`{_cell(state.get('head_hash') or 'unknown')}`"
if state.get("head_subject"):
head += f" — {_cell(state['head_subject'])}"
tag = state.get("newest_tag")
rows = [
("Active branch", active),
("main HEAD", head),
("Version", " · ".join(versions)),
("Newest tag", f"`{_cell(tag)}`" if tag else "none"),
("Updated", f"{_cell(state.get('date') or 'unknown')}, by `just merge-pr`"),
]
lines = [
BEGIN_MARKER,
"## Repository state (generated)",
"",
"| | |",
"|---|---|",
*[f"| **{label}** | {value} |" for label, value in rows],
"",
END_MARKER,
]
return "\n".join(lines) + "\n"
def apply_state_block(text, block):
begins = list(_BEGIN_RE.finditer(text))
ends = list(_END_RE.finditer(text))
if not begins and not ends:
return block + "\n" + text, "inserted"
if len(begins) != 1 or len(ends) != 1:
raise ValueError(
f"expected exactly one BEGIN and one END merge-pr state marker, found "
f"{len(begins)} BEGIN and {len(ends)} END"
)
begin, end = begins[0], ends[0]
if end.start() < begin.end():
raise ValueError("the END merge-pr state marker comes before the BEGIN marker")
stop = end.end() + 1 if text[end.end():end.end() + 1] == "\n" else end.end()
return text[:begin.start()] + block + text[stop:], "replaced"
def update_file(path, state):
text, newline = read_preserving_newlines(path)
new_text, action = apply_state_block(text, render_state_block(state))
write_preserving_newlines(path, new_text, newline)
return action
def gather_state(root_dir):
return {
"branch": run(["git", "rev-parse", "--abbrev-ref", "HEAD"]),
"head_hash": run(["git", "rev-parse", "--short", "HEAD"]),
"head_subject": run(["git", "log", "-1", "--format=%s"]),
"cli_version": read_cargo_version(root_dir / "Cargo.toml"),
"sysinfo_version": read_cargo_version(root_dir / "crates" / "sysinfo" / "Cargo.toml"),
"newest_tag": run(["git", "describe", "--tags", "--abbrev=0"], required=False),
"date": datetime.date.today().isoformat(),
}
def main():
try:
sys.stdout.reconfigure(encoding="utf-8")
except (AttributeError, ValueError):
pass
root_dir = Path(__file__).resolve().parent.parent
wip_file = root_dir / "WIP.md"
if not wip_file.exists():
print("WIP.md not found. Skipping update.", file=sys.stderr)
return 0
state = gather_state(root_dir)
try:
action = update_file(wip_file, state)
except ValueError as e:
print(f"WIP.md NOT updated: {e}. Fix the markers by hand; nothing was written.",
file=sys.stderr)
return 1
where = "at the top of the file (first run)" if action == "inserted" else "in place"
print(f"WIP.md state block {action} {where}: main HEAD `{state['head_hash']}`, "
f"v{state['cli_version']}. Hand-written entries below it are unchanged; add one "
f"if this merge changed the open-task list.")
return 0
def _self_test():
import tempfile
failures = []
def check(name, cond, detail=""):
if not cond:
failures.append(f"{name}: {detail}")
state = {
"branch": "main", "head_hash": "abc1234", "head_subject": "Fix a | b (#1)",
"cli_version": "1.2.3", "sysinfo_version": "0.1.9", "newest_tag": "v1.2.0",
"date": "2026-01-02",
}
block = render_state_block(state)
check("block starts with BEGIN", block.startswith(BEGIN_MARKER + "\n"), repr(block[:40]))
check("block ends with END", block.endswith(END_MARKER + "\n"), repr(block[-40:]))
check("pipe in subject is escaped", "Fix a \\| b (#1)" in block, block)
check("main renders as none", "none — `main` is current" in block, block)
off_main = render_state_block({**state, "branch": "feature/x"})
check("off-main branch is flagged", "**not `main`**" in off_main, off_main)
no_tag = render_state_block({**state, "newest_tag": None, "sysinfo_version": None})
check("no tag renders as none", "| **Newest tag** | none |" in no_tag, no_tag)
check("absent sysinfo is omitted", "retch-sysinfo" not in no_tag, no_tag)
check("the block never contains the old header strings",
"### Active Branch:" not in block and "**main HEAD**:" not in block, block)
prose = ("## Old entry\n### Active Branch: feature/old\n"
"**main HEAD**: `0000000` - ancient\nnotes\n")
inserted, action = apply_state_block(prose, block)
check("absent markers -> inserted", action == "inserted", action)
check("inserted at the very top", inserted.startswith(BEGIN_MARKER), inserted[:60])
check("prose after an insert is byte-identical", inserted.endswith("\n" + prose),
repr(inserted[-80:]))
newer = render_state_block({**state, "head_hash": "def5678", "head_subject": "Next"})
replaced, action = apply_state_block(inserted, newer)
check("present markers -> replaced", action == "replaced", action)
check("replacement took effect", "`def5678`" in replaced and "`abc1234`" not in replaced,
replaced)
check("prose after a replace is byte-identical", replaced.endswith("\n" + prose),
repr(replaced[-80:]))
again, _ = apply_state_block(replaced, newer)
check("replacing is idempotent", again == replaced, "a second run changed the file")
middle = "# WIP\nintro\n" + block + "tail\n"
moved, _ = apply_state_block(middle, newer)
check("a block mid-file is replaced where it is",
moved.startswith("# WIP\nintro\n" + BEGIN_MARKER) and moved.endswith(END_MARKER
+ "\ntail\n"),
repr(moved))
mention = "Prose may mention " + END_MARKER + " mid-line.\n"
kept, action = apply_state_block(mention, block)
check("a mid-line mention is not a marker", action == "inserted" and kept.endswith(mention),
action)
for name, bad in [
("duplicate block", block + "x\n" + block),
("BEGIN without END", BEGIN_MARKER + "\nstuff\n"),
("END without BEGIN", "stuff\n" + END_MARKER + "\n"),
("END before BEGIN", END_MARKER + "\n" + BEGIN_MARKER + "\n"),
]:
try:
apply_state_block(bad, block)
check(f"{name} refuses", False, "no ValueError raised")
except ValueError:
pass
with tempfile.TemporaryDirectory() as d:
tmp = Path(d)
f = tmp / "crlf.md"
f.write_bytes(b"### Active Branch: feature/x\r\nbody\r\n**main HEAD**: `old`\r\n")
check("CRLF first run inserts", update_file(f, state) == "inserted")
check("CRLF second run replaces", update_file(f, state) == "replaced")
raw = f.read_bytes()
lone_lf = raw.count(b"\n") - raw.count(b"\r\n")
check("CRLF survives, with no lone LF", lone_lf == 0 and raw.count(b"\r\n") > 3,
f"crlf={raw.count(bytes([13, 10]))} lone_lf={lone_lf}")
check("old header lines untouched in the file",
b"### Active Branch: feature/x\r\n" in raw and b"**main HEAD**: `old`\r\n" in raw,
repr(raw[-80:]))
f2 = tmp / "lf.md"
f2.write_bytes(b"notes\n")
update_file(f2, state)
check("LF round-trips", b"\r" not in f2.read_bytes(), "a CR was introduced")
f3 = tmp / "bad.md"
before = (block + block).encode("utf-8")
f3.write_bytes(before)
try:
update_file(f3, state)
check("malformed file refuses", False, "no ValueError raised")
except ValueError:
check("a refusal leaves the file untouched", f3.read_bytes() == before, "file changed")
naive = re.sub(r"### Active Branch:.*", "x", "### Active Branch: y\r\nbody\r\n", count=1)
check("the .* trap is real", naive.count("\r\n") == 1,
"expected the substitution to eat one CR -- if this fails, re's dot changed")
if failures:
for f_ in failures:
print(f" FAIL {f_}", file=sys.stderr)
print(f"update_wip.py self-test FAILED ({len(failures)})", file=sys.stderr)
return 1
print("update_wip.py self-test passed")
return 0
if __name__ == "__main__":
if "--self-test" in sys.argv[1:]:
sys.exit(_self_test())
sys.exit(main())