src/ai_config_render/__init__.py (24327 bytes)
1 """Render one canonical AI-agent config into many agents' native formats. 2 3 It touches only what it owns: 4 - instruction files: blocks between <!-- aicr:NAME --> and <!-- /aicr:NAME --> 5 - MCP configs: the server names it renders now, and the ones it rendered last run 6 - skills and plain files: paths it wrote last run (tracked in a state file) 7 Everything it overwrites is backed up first (one rolling <file>.aicr-bak). 8 9 It never clobbers a local edit. The state file keeps a hash of what aicr last 10 wrote to each owned file and block. If a target no longer matches that hash, 11 someone edited it there: 12 - source unchanged -> the edit is imported into the source folder 13 - source changed too -> conflict: the edit is saved next to its source as 14 <source>.conflict-<machine>-<date>, then the source version is written 15 Text matching an [import] deny pattern is never imported. 16 17 `aicr watch` re-renders within seconds of a change to an owned file or the 18 source folder, and optionally runs a command after each render (commit, push). 19 """ 20 21 from __future__ import annotations 22 23 import argparse 24 import datetime as dt 25 import difflib 26 import hashlib 27 import json 28 import os 29 import re 30 import shutil 31 import subprocess 32 import sys 33 import time 34 import tomllib 35 from collections.abc import Callable 36 from dataclasses import dataclass, field 37 from pathlib import Path 38 from typing import Any 39 40 __version__ = "0.1.0" 41 42 CONFIG_NAME = "aicr.toml" 43 DEFAULT_MACHINE = Path("~/.config/aicr/machine.toml") 44 DEFAULT_STATE = "~/.config/aicr/state.json" 45 VAR = re.compile(r"\$\{(\w+)\}") 46 SKIP_NAMES = re.compile(r"__pycache__|\.aicr-(bak|tmp)|\.conflict-") 47 48 # A target is data. `mcp.format` is "json" (servers under `mcp.key`, default 49 # "mcpServers") or "toml" (servers under [mcp_servers.<name>]). Any field can be 50 # overridden, and new targets added, under [targets.<name>] in aicr.toml. 51 BUILTIN_TARGETS: dict[str, dict[str, Any]] = { 52 "claude": { 53 "instructions": "${home}/.claude/CLAUDE.md", 54 "skills": ["${home}/.claude/skills"], 55 "mcp": {"format": "json", "path": "${home}/.claude.json"}, 56 }, 57 "codex": { 58 "instructions": "${home}/.codex/AGENTS.md", 59 "skills": ["${home}/.codex/skills"], 60 "mcp": {"format": "toml", "path": "${home}/.codex/config.toml"}, 61 }, 62 "copilot": { 63 "instructions": "${home}/.copilot/copilot-instructions.md", 64 "skills": ["${home}/.copilot/skills"], 65 "mcp": {"format": "json", "path": "${home}/.copilot/mcp-config.json", "defaults": {"tools": ["*"]}}, 66 }, 67 "gemini": { 68 "instructions": "${home}/.gemini/GEMINI.md", 69 "skills": ["${home}/.gemini/skills"], 70 "mcp": {"format": "json", "path": "${home}/.gemini/settings.json"}, 71 }, 72 "antigravity": { 73 "instructions": "${home}/.gemini/GEMINI.md", 74 "skills": ["${home}/.gemini/antigravity/skills"], 75 "mcp": {"format": "json", "path": "${home}/.gemini/antigravity/mcp_config.json"}, 76 }, 77 "cursor": { 78 "mcp": {"format": "json", "path": "${home}/.cursor/mcp.json"}, 79 }, 80 "windsurf": { 81 "instructions": "${home}/.codeium/windsurf/memories/global_rules.md", 82 "mcp": {"format": "json", "path": "${home}/.codeium/windsurf/mcp_config.json"}, 83 }, 84 } 85 86 87 class Missing(Exception): 88 """A ${var} has no value.""" 89 90 91 def sub(value: Any, env: dict[str, str]) -> Any: 92 """Expand ${var} in every string of a nested value. An unknown var raises Missing.""" 93 if isinstance(value, str): 94 95 def rep(m: re.Match[str]) -> str: 96 if m.group(1) not in env: 97 raise Missing(m.group(1)) 98 return env[m.group(1)] 99 100 return VAR.sub(rep, value) 101 if isinstance(value, list): 102 return [sub(v, env) for v in value] 103 if isinstance(value, dict): 104 return {k: sub(v, env) for k, v in value.items()} 105 return value 106 107 108 def merge(a: dict[str, Any], b: dict[str, Any]) -> dict[str, Any]: 109 out = dict(a) 110 for k, v in b.items(): 111 out[k] = merge(out[k], v) if isinstance(v, dict) and isinstance(out.get(k), dict) else v 112 return out 113 114 115 def applies(entry: dict[str, Any], target: str, machine: str) -> bool: 116 """Filters: `targets` (default all), `machines` (default all).""" 117 if target and "targets" in entry and target not in entry["targets"]: 118 return False 119 return "machines" not in entry or machine in entry["machines"] 120 121 122 def sha(b: bytes) -> str: 123 return hashlib.sha256(b).hexdigest() 124 125 126 @dataclass 127 class Writer: 128 dry: bool 129 last: dict[str, str] = field(default_factory=dict) # key -> hash of what we wrote last run 130 host: str = "machine" 131 deny: list[re.Pattern[str]] = field(default_factory=list) # never import text that matches 132 changes: list[tuple[Path, str | None, str | None]] = field(default_factory=list) 133 pending: dict[Path, str] = field(default_factory=dict) # what a dry run would have written 134 rendered: dict[str, str] = field(default_factory=dict) 135 imported: list[tuple[str, Path]] = field(default_factory=list) 136 conflicts: list[tuple[str, Path]] = field(default_factory=list) 137 flags: list[str] = field(default_factory=list) 138 139 def guard(self, key: str, cur: bytes | None, new: bytes, src: Path | None, label: str) -> bytes: 140 """Return what the target should hold. A local edit is imported into src, or saved next to it as a conflict.""" 141 last = self.last.get(key) 142 if cur is None or cur == new or sha(cur) == last: 143 self.rendered[key] = sha(new) 144 return new 145 denied = any(rx.search(cur.decode("utf-8", "replace")) for rx in self.deny) 146 if last is not None and sha(new) == last: # only the target changed 147 if src is None or denied: 148 self.flags.append( 149 f"local edit to {label} not imported ({'denied text' if denied else 'templated source'}); target kept" 150 ) 151 self.rendered[key] = last # keep reporting it until someone resolves it 152 return cur 153 self.imported.append((label, src)) 154 if not self.dry: 155 src.write_bytes(cur) 156 self.rendered[key] = sha(cur) 157 return cur 158 # Both changed, or no record of what we wrote: keep the local copy next to the source, then write the source. 159 if src is None or denied: 160 self.flags.append(f"local edit to {label} overwritten, not saved ({'denied text' if denied else 'templated source'})") 161 else: 162 day = dt.datetime.now().astimezone().strftime("%Y%m%d") 163 dst = src.with_name(f"{src.name}.conflict-{self.host}-{day}") 164 self.conflicts.append((label, dst)) 165 if not self.dry: 166 dst.write_bytes(cur) 167 self.rendered[key] = sha(new) 168 return new 169 170 def read(self, p: Path) -> str | None: 171 return self.pending.get(p, p.read_text(encoding="utf-8") if p.exists() else None) 172 173 def backup(self, p: Path) -> None: 174 if p.exists() and not self.dry: 175 dst = p.with_name(f"{p.name}.aicr-bak") 176 if dst.is_dir(): 177 shutil.rmtree(dst) 178 if p.is_dir(): 179 shutil.copytree(p, dst) 180 else: 181 shutil.copy2(p, dst) 182 183 def text(self, p: Path, new: str) -> None: 184 old = self.read(p) 185 if old == new: 186 return 187 self.changes.append((p, old, new)) 188 self.pending[p] = new 189 if self.dry: 190 return 191 self.backup(p) 192 p.parent.mkdir(parents=True, exist_ok=True) 193 tmp = p.with_name(p.name + ".aicr-tmp") 194 tmp.write_text(new, encoding="utf-8", newline="") 195 os.replace(tmp, p) 196 197 def copy(self, src: Path, dst: Path, label: str) -> None: 198 have = dst.read_bytes() if dst.exists() else None 199 want = self.guard(str(dst), have, src.read_bytes(), src, label) 200 if have == want: 201 return 202 self.changes.append((dst, None, None)) 203 if not self.dry: 204 self.backup(dst) 205 dst.parent.mkdir(parents=True, exist_ok=True) 206 dst.write_bytes(want) 207 208 def remove(self, p: Path) -> None: 209 if not p.exists(): 210 return 211 self.changes.append((p, "", None)) 212 if not self.dry: 213 self.backup(p) 214 if p.is_dir(): 215 shutil.rmtree(p) 216 else: 217 p.unlink() 218 219 220 # ---------- instruction files ---------- 221 222 223 def block_re(name: str) -> re.Pattern[str]: 224 n = re.escape(name) 225 return re.compile(rf"<!-- aicr:{n} -->\n.*?<!-- /aicr:{n} -->\n?", re.DOTALL) 226 227 228 def block_body(text: str, name: str) -> str | None: 229 n = re.escape(name) 230 m = re.search(rf"<!-- aicr:{n} -->\n(.*?)<!-- /aicr:{n} -->", text, re.DOTALL) 231 return m.group(1).rstrip() + "\n" if m else None 232 233 234 def render_instructions(text: str, blocks: dict[str, str], dropped: set[str]) -> str: 235 """Replace or append each owned block; remove dropped ones; keep all other text.""" 236 for name in dropped: 237 text = block_re(name).sub("", text) 238 for name, body in blocks.items(): 239 new = f"<!-- aicr:{name} -->\n{body.rstrip()}\n<!-- /aicr:{name} -->\n" 240 rx = block_re(name) 241 first = rx.search(text) 242 if first: 243 # Keep the first position, drop duplicates. 244 text = text[: first.start()] + new + rx.sub("", text[first.end() :]) 245 else: 246 text = text.rstrip("\n") + ("\n\n" if text.strip() else "") + new 247 return text 248 249 250 # ---------- MCP ---------- 251 252 253 def toml_key(k: str) -> str: 254 return k if re.fullmatch(r"[A-Za-z0-9_-]+", k) else json.dumps(k) 255 256 257 def toml_value(v: Any) -> str: 258 if isinstance(v, bool): 259 return "true" if v else "false" 260 if isinstance(v, (int, float)): 261 return repr(v) 262 if isinstance(v, str): 263 return json.dumps(v, ensure_ascii=False).replace(chr(0x7F), "\\u007f") # TOML forbids a raw DEL 264 if isinstance(v, list): 265 return "[" + ", ".join(toml_value(x) for x in v) + "]" 266 if isinstance(v, dict): 267 return "{ " + ", ".join(f"{toml_key(k)} = {toml_value(x)}" for k, x in v.items()) + " }" 268 raise TypeError(f"cannot write {type(v).__name__} to TOML") 269 270 271 def render_toml(text: str, servers: dict[str, dict[str, Any]], owned: set[str]) -> str: 272 """Drop every [mcp_servers.<owned>...] table, then append ours in one marked block.""" 273 text = re.sub(r"# aicr:mcp\n.*?# /aicr:mcp\n?", "", text, flags=re.DOTALL) 274 out: list[str] = [] 275 skip = False 276 for line in text.splitlines(keepends=True): 277 m = re.match(r"\s*\[\[?\s*([^\]]+?)\s*\]\]?\s*(#.*)?$", line) 278 if m: 279 parts = [p.strip().strip("\"'") for p in re.split(r"\.(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)", m.group(1))] 280 skip = len(parts) >= 2 and parts[0] == "mcp_servers" and parts[1] in owned 281 if not skip: 282 out.append(line) 283 text = "".join(out).rstrip("\n") 284 if servers: 285 body = ["# aicr:mcp"] 286 for name, cfg in servers.items(): 287 body.append(f"[mcp_servers.{toml_key(name)}]") 288 body += [f"{toml_key(k)} = {toml_value(v)}" for k, v in cfg.items()] 289 body.append("") 290 text += ("\n\n" if text else "") + "\n".join(body) + "# /aicr:mcp\n" 291 tomllib.loads(text) # refuse to write a broken config 292 return text if not text or text.endswith("\n") else text + "\n" 293 294 295 def render_json(text: str, servers: dict[str, dict[str, Any]], owned: set[str], key: str) -> str: 296 data = json.loads(text) if text.strip() else {} 297 cur = data.setdefault(key, {}) 298 for name in owned - servers.keys(): 299 cur.pop(name, None) 300 cur.update(servers) 301 return json.dumps(data, indent=2, ensure_ascii=False) + "\n" 302 303 304 # ---------- render ---------- 305 306 307 def load(source: Path, machine_file: Path) -> tuple[dict[str, Any], dict[str, Any], dict[str, str]]: 308 cfg = tomllib.loads((source / CONFIG_NAME).read_text(encoding="utf-8")) 309 mach = tomllib.loads(machine_file.read_text(encoding="utf-8")) if machine_file.exists() else {} 310 env = { 311 "home": str(Path.home()), 312 "home_posix": Path.home().as_posix(), 313 "os": "windows" if os.name == "nt" else "macos" if sys.platform == "darwin" else "linux", 314 **{k: str(v) for k, v in cfg.get("vars", {}).items()}, 315 **{k: str(v) for k, v in mach.get("vars", {}).items()}, 316 } 317 env["machine"] = str(mach.get("name", env["os"])) 318 for k, v in env.items(): # one level of vars-in-vars, e.g. projects = "${home}/code" 319 env[k] = VAR.sub(lambda m: env.get(m.group(1), m.group(0)), v) 320 return cfg, mach, env 321 322 323 def targets(cfg: dict[str, Any]) -> dict[str, dict[str, Any]]: 324 """Built-ins merged with [targets.*]. Without a [targets] table, no target is enabled.""" 325 out = {} 326 for name, t in cfg.get("targets", {}).items(): 327 spec = merge(BUILTIN_TARGETS.get(name, {}), t) 328 if spec.pop("enabled", True): 329 out[name] = spec 330 return out 331 332 333 def render(source: Path, machine_file: Path, dry: bool) -> tuple[Writer, list[str]]: 334 cfg, mach, env = load(source, machine_file) 335 machine = env["machine"] 336 state_file = Path(sub(mach.get("state", DEFAULT_STATE), env)).expanduser() 337 state = json.loads(state_file.read_text(encoding="utf-8")) if state_file.exists() else {} 338 new_state: dict[str, Any] = {} 339 deny = [re.compile(p, re.IGNORECASE) for p in cfg.get("import", {}).get("deny", [])] 340 w = Writer(dry, last=state.get("_rendered", {}), host=machine, deny=deny) 341 notes: list[str] = [] 342 343 for tname, raw in targets(cfg).items(): 344 if tname in mach.get("skip_targets", []): 345 continue 346 try: 347 t = sub(raw, env) 348 except Missing as e: 349 notes.append(f"{tname}: skipped, no var {e}") 350 continue 351 prev = state.get(tname, {}) 352 cur = new_state.setdefault(tname, {}) 353 354 if "instructions" in t: 355 blocks: dict[str, str] = {} 356 p = Path(t["instructions"]).expanduser() 357 for name, r in cfg.get("rules", {}).items(): 358 if not applies(r, tname, machine): 359 continue 360 src = source / "rules" / f"{name}.{tname}.md" 361 src = src if src.exists() else source / "rules" / f"{name}.md" 362 try: 363 rule_text = src.read_text(encoding="utf-8") 364 body = sub(rule_text, env).rstrip() + "\n" 365 except Missing as e: 366 notes.append(f"{tname}: rule {name} skipped, no var {e}") 367 continue 368 have = block_body(w.read(p) or "", name) 369 blocks[name] = w.guard( 370 f"{p}#{name}", 371 have.encode("utf-8") if have is not None else None, 372 body.encode("utf-8"), 373 None if VAR.search(rule_text) else src, 374 f"{tname} rules/{name}", 375 ).decode("utf-8") 376 # Two targets can share one file (gemini, antigravity): merge their blocks. 377 file_state = new_state.setdefault("_instructions", {}).setdefault(str(p), []) 378 dropped = set(prev.get("rules", [])) - blocks.keys() - set(file_state) 379 file_state.extend(sorted(blocks)) 380 if blocks or dropped: 381 w.text(p, render_instructions(w.read(p) or "", blocks, dropped)) 382 cur["rules"] = sorted(blocks) 383 384 if "mcp" in t: 385 servers: dict[str, dict[str, Any]] = {} 386 for name, s in cfg.get("mcp", {}).items(): 387 if not applies(s, tname, machine): 388 continue 389 m = mach.get("mcp", {}).get(name, {}) 390 # Precedence, low to high: target defaults, server, server.target.<t>, machine, machine.target.<t> 391 spec: dict[str, Any] = t["mcp"].get("defaults", {}) 392 for layer in (s, s.get("target", {}).get(tname, {}), m, m.get("target", {}).get(tname, {})): 393 spec = merge(spec, layer) 394 spec = {k: v for k, v in spec.items() if k not in ("targets", "machines", "target")} 395 try: 396 servers[name] = sub(spec, env) 397 except Missing as e: 398 notes.append(f"{tname}: mcp {name} skipped, no var {e}") 399 p = Path(t["mcp"]["path"]).expanduser() 400 owned = set(servers) | set(prev.get("mcp", [])) 401 if servers or owned: 402 old = w.read(p) or "" 403 fmt = t["mcp"]["format"] 404 if fmt == "toml": 405 w.text(p, render_toml(old, servers, owned)) 406 elif fmt == "json": 407 w.text(p, render_json(old, servers, owned, t["mcp"].get("key", "mcpServers"))) 408 else: 409 raise SystemExit(f"{tname}: unknown mcp format {fmt!r} (use json or toml)") 410 cur["mcp"] = sorted(servers) 411 412 names = [n for n, s in cfg.get("skills", {}).items() if applies(s, tname, machine)] 413 for d in t.get("skills", []): 414 root = Path(d).expanduser() 415 for n in names: 416 skill = source / "skills" / n 417 for f in sorted(skill.rglob("*")) if skill.is_dir() else []: 418 if f.is_file() and not SKIP_NAMES.search(f.relative_to(skill).as_posix()): 419 w.copy(f, root / n / f.relative_to(skill), f"{tname} skills/{n}/{f.relative_to(skill).as_posix()}") 420 for n in set(prev.get("skills", [])) - set(names): 421 w.remove(root / n) 422 cur["skills"] = names 423 424 owned_files: list[str] = [] 425 for dest, f in cfg.get("files", {}).items(): 426 if not applies(f, "", machine): 427 continue 428 try: 429 dest_p, src = Path(sub(dest, env)).expanduser(), source / sub(f["src"], env) 430 data = src.read_text(encoding="utf-8") 431 templated = f.get("template", False) 432 want = (sub(data, env) if templated else data).encode("utf-8") 433 have = w.read(dest_p) 434 got = w.guard( 435 str(dest_p), 436 have.encode("utf-8") if have is not None else None, 437 want, 438 None if templated else src, 439 f"file {f['src']}", 440 ) 441 w.text(dest_p, got.decode("utf-8")) 442 except Missing as e: 443 notes.append(f"file {dest} skipped, no var {e}") 444 continue 445 owned_files.append(str(dest_p)) 446 for gone in set(state.get("_files", [])) - set(owned_files): 447 w.remove(Path(gone)) 448 new_state["_files"] = owned_files 449 new_state["_rendered"] = w.rendered 450 new_state.pop("_instructions", None) 451 452 if not dry: 453 state_file.parent.mkdir(parents=True, exist_ok=True) 454 state_file.write_text(json.dumps(new_state, indent=1), encoding="utf-8") 455 return w, notes 456 457 458 # ---------- watch ---------- 459 460 461 def watched(source: Path, machine_file: Path) -> list[Path]: 462 """Every file aicr owns on this machine, plus the source tree.""" 463 _, mach, env = load(source, machine_file) 464 state_file = Path(sub(mach.get("state", DEFAULT_STATE), env)).expanduser() 465 state = json.loads(state_file.read_text(encoding="utf-8")) if state_file.exists() else {} 466 files = {Path(k.split("#")[0]) for k in state.get("_rendered", {})} 467 files |= {p for p in source.rglob("*") if ".git" not in p.relative_to(source).parts} 468 return sorted(files) 469 470 471 def snapshot(files: list[Path]) -> dict[Path, tuple[int, int]]: 472 out = {} 473 for f in files: 474 try: 475 st = f.stat() 476 out[f] = (st.st_mtime_ns, st.st_size) 477 except OSError: 478 pass 479 return out 480 481 482 def git(source: Path, *args: str) -> str: 483 r = subprocess.run(["git", "-C", str(source), *args], capture_output=True, text=True, check=False) 484 return r.stdout.strip() if r.returncode == 0 else "" 485 486 487 def watch( 488 source: Path, 489 machine_file: Path, 490 run: Callable[[], None], 491 interval: float = 1.0, 492 settle: float = 2.0, 493 remote_every: float = 0.0, 494 rounds: int | None = None, 495 ) -> None: 496 """Call `run` once a burst of changes to owned files or the source tree has been quiet for `settle` seconds. 497 With remote_every > 0 and a git source, also pull and run when the upstream branch moves. 498 Our own writes do not loop: the snapshot is retaken after each run. 499 ponytail: polls stat() of the owned files every `interval` s, identical on Windows, macOS and Linux; 500 switch to inotify/FSEvents/ReadDirectoryChangesW if the owned set grows to many thousands of files.""" 501 files = watched(source, machine_file) 502 seen = snapshot(files) 503 dirty_since: float | None = None 504 next_remote = time.monotonic() 505 n = 0 506 while rounds is None or n < rounds: 507 n += 1 508 time.sleep(interval) 509 now = time.monotonic() 510 cur = snapshot(files) 511 if cur != seen: 512 seen, dirty_since = cur, now 513 trigger = dirty_since is not None and now - dirty_since >= settle 514 if not trigger and remote_every > 0 and now >= next_remote: 515 next_remote = now + remote_every 516 remote = git(source, "ls-remote", "origin", "HEAD").split() 517 if remote and remote[0] != git(source, "rev-parse", "HEAD"): 518 trigger = bool(git(source, "pull", "--ff-only", "-q") or True) 519 if trigger: 520 run() 521 files = watched(source, machine_file) 522 seen, dirty_since = snapshot(files), None 523 524 525 # ---------- CLI ---------- 526 527 528 def main(argv: list[str] | None = None) -> int: 529 ap = argparse.ArgumentParser(prog="aicr", description=(__doc__ or "").splitlines()[0]) 530 ap.add_argument("command", choices=["render", "diff", "check", "list-targets", "watch"]) 531 ap.add_argument("--source", type=Path, default=Path("."), help=f"folder with {CONFIG_NAME} (default: .)") 532 ap.add_argument("--machine", type=Path, default=DEFAULT_MACHINE, help="per-machine vars file") 533 ap.add_argument( 534 "--exec", dest="exec_cmd", help="watch: shell command to run after each render, e.g. a commit-and-push script" 535 ) 536 ap.add_argument("--remote-every", type=float, default=0.0, help="watch: seconds between upstream checks (0 = off)") 537 a = ap.parse_args(argv) 538 machine = a.machine.expanduser() 539 540 if a.command == "watch": 541 542 def run() -> None: 543 stamp = dt.datetime.now().astimezone().isoformat(timespec="seconds") 544 w, notes = render(a.source, machine, dry=False) 545 for line in report(w, notes): 546 print(f"{stamp} {line}", flush=True) 547 if a.exec_cmd: 548 subprocess.run(a.exec_cmd, shell=True, cwd=a.source, check=False) # noqa: S602 - the user's own command 549 550 watch(a.source, machine, run, remote_every=a.remote_every) 551 return 0 552 553 if a.command == "list-targets": 554 cfg = tomllib.loads((a.source / CONFIG_NAME).read_text(encoding="utf-8")) if (a.source / CONFIG_NAME).exists() else {} 555 enabled = targets(cfg) 556 for name in sorted(BUILTIN_TARGETS.keys() | enabled.keys()): 557 spec = enabled.get(name, BUILTIN_TARGETS.get(name, {})) 558 mark = "*" if name in enabled else " " 559 print(f"{mark} {name:12} {spec.get('instructions', '-')} mcp={spec.get('mcp', {}).get('path', '-')}") 560 return 0 561 562 w, notes = render(a.source, machine, dry=a.command != "render") 563 for p, old, new in w.changes: 564 if a.command == "diff" and new is not None: 565 sys.stdout.writelines(difflib.unified_diff((old or "").splitlines(True), new.splitlines(True), f"a/{p}", f"b/{p}")) 566 else: 567 print(f"{'removed' if old == '' and new is None else 'changed'} {p}") 568 for line in report(w, notes)[len(w.changes) :]: 569 print(line, file=sys.stderr) 570 return 1 if a.command == "check" and (w.changes or w.conflicts or w.flags) else 0 571 572 573 def report(w: Writer, notes: list[str]) -> list[str]: 574 return ( 575 [f"{'removed' if old == '' and new is None else 'changed'} {p}" for p, old, new in w.changes] 576 + [f"imported: {label} -> {p}" for label, p in w.imported] 577 + [f"CONFLICT: {label}: local edit saved as {p}" for label, p in w.conflicts] 578 + [f"FLAG: {f}" for f in w.flags] 579 + [f"note: {n}" for n in notes] 580 )