#!/usr/bin/env python3 """ canonical_sync.py — operator-triggered tool to sync pinned canonical Corviduo specs from their canonical sources. Reads .corviduo-canonicals.toml at the repo root. For each pin: 1. Locates canonical at ~/development// 2. Reads canonical content 3. Computes SHA-256 (first 16 hex chars) 4. Writes content to consumer_path (if different from current) 5. Updates the pin's pinned_sha256_16 + pinned_at in the manifest Reports per-pin: OK | SYNCED | WOULD | ERROR. Manifest shape (.corviduo-canonicals.toml): [[pins]] id = "contract-format-v2" canonical_source = "corviduo-project-template" canonical_path = "docs/contracts/CONTRACT-FORMAT.md" consumer_path = "docs/contracts/CONTRACT-FORMAT.md" pinned_sha256_16 = "a1b2c3d4e5f67890" pinned_at = "2026-05-15T22:00:00+00:00" Optional per-pin: tolerate_drift = true (warn-only in canonical_drift.py when --allow-warn is set). Usage: python scripts/canonical_sync.py [--manifest PATH] [--dry-run] Exit codes: 0 — success (any number of files synced) 2 — manifest missing or invalid 3 — canonical source path missing for one or more pins """ from __future__ import annotations import argparse import datetime import hashlib import sys import tomllib from pathlib import Path DEFAULT_MANIFEST = ".corviduo-canonicals.toml" DEV_ROOT = Path.home() / "development" def sha256_16(data: bytes) -> str: """SHA-256 hash, first 16 hex chars.""" return hashlib.sha256(data).hexdigest()[:16] def _replace_value_preserve_format(line: str, new_value: str) -> str: """Replace the quoted value in a `key = "value"` line, preserving leading whitespace, key name, and trailing newline. Always emits double quotes.""" leading_ws = line[: len(line) - len(line.lstrip())] trailing_nl = "\n" if line.endswith("\n") else "" key_part, _, _ = line.lstrip().partition("=") key = key_part.strip() return f'{leading_ws}{key} = "{new_value}"{trailing_nl}' def update_pin_in_manifest_text( text: str, pin_id: str, new_sha: str, new_at: str, ) -> str: """Surgically update one pin's `pinned_sha256_16` + `pinned_at` lines in the manifest text, preserving all comments, blank lines, key order, and formatting elsewhere. If either line is absent in the block (e.g. first sync of a newly-added pin), insert it after the last content line of the block. Raises KeyError if the pin_id isn't found in any `[[pins]]` block. """ lines = text.splitlines(keepends=True) # Find [[pins]] block boundaries: (start_idx_inclusive, end_idx_exclusive). blocks: list[tuple[int, int]] = [] current_start: int | None = None for i, line in enumerate(lines): if line.strip() == "[[pins]]": if current_start is not None: blocks.append((current_start, i)) current_start = i if current_start is not None: blocks.append((current_start, len(lines))) # Locate the block whose `id = ""` matches pin_id. target_block: tuple[int, int] | None = None for start, end in blocks: for i in range(start, end): stripped = lines[i].strip() if stripped.startswith("id ") or stripped.startswith("id="): _, _, val = stripped.partition("=") val = val.strip().strip('"').strip("'") if val == pin_id: target_block = (start, end) break if target_block: break if target_block is None: raise KeyError(f"pin {pin_id!r} not found in manifest") start, end = target_block # Within the target block: find existing pinned_sha256_16 / pinned_at # lines (if any), and the last "content" line (non-blank, non-comment) for # insertion fallback. sha_idx: int | None = None at_idx: int | None = None last_content_idx = start # the [[pins]] header itself is content for i in range(start, end): stripped = lines[i].strip() if stripped.startswith("pinned_sha256_16"): sha_idx = i elif stripped.startswith("pinned_at"): at_idx = i if stripped and not stripped.startswith("#"): last_content_idx = i sha_line = f'pinned_sha256_16 = "{new_sha}"\n' at_line = f'pinned_at = "{new_at}"\n' # In-place replacement preserves position + leading whitespace. if sha_idx is not None: lines[sha_idx] = _replace_value_preserve_format(lines[sha_idx], new_sha) if at_idx is not None: lines[at_idx] = _replace_value_preserve_format(lines[at_idx], new_at) # Collect inserts for absent lines. When both go at the same index, we # want the final output to read pinned_sha256_16 BEFORE pinned_at # (conventional order). Since each insert shifts subsequent lines down, # insert pinned_at first then pinned_sha256_16 — the second insert # pushes pinned_at to idx+1 and lands pinned_sha256_16 at idx. inserts: list[tuple[int, str]] = [] if at_idx is None: inserts.append((last_content_idx + 1, at_line)) if sha_idx is None: inserts.append((last_content_idx + 1, sha_line)) # Reverse-index sort keeps earlier-position inserts from shifting later # ones; within a tie the original list order is preserved (Python sort # is stable). for idx, line in sorted(inserts, key=lambda x: x[0], reverse=True): lines.insert(idx, line) return "".join(lines) def main() -> int: ap = argparse.ArgumentParser( description="Sync pinned canonical Corviduo specs from their sources.", ) ap.add_argument("--manifest", type=Path, default=Path(DEFAULT_MANIFEST)) ap.add_argument("--dry-run", action="store_true", help="Show what would be synced without writing files") args = ap.parse_args() if not args.manifest.exists(): print(f"error: manifest not found at {args.manifest}", file=sys.stderr) return 2 with args.manifest.open("rb") as f: manifest = tomllib.load(f) pins = manifest.get("pins", []) if not pins: print("warning: no pins in manifest", file=sys.stderr) return 0 # Read manifest as text upfront for surgical pin-line updates that # preserve comments / blank lines / key order in the original file. manifest_text = args.manifest.read_text() now = datetime.datetime.now(datetime.timezone.utc).isoformat( timespec="seconds", ) any_change = False any_error = False for pin in pins: pin_id = pin.get("id", "") source = pin.get("canonical_source") canon_rel = pin.get("canonical_path") consumer_rel = pin.get("consumer_path") if not all((source, canon_rel, consumer_rel)): print(f"ERROR {pin_id} (manifest entry incomplete)", file=sys.stderr) any_error = True continue canon_path = DEV_ROOT / source / canon_rel if not canon_path.exists(): print(f"ERROR {pin_id} (canonical missing: {canon_path})", file=sys.stderr) any_error = True continue canon_bytes = canon_path.read_bytes() current_sha = sha256_16(canon_bytes) pinned_sha = pin.get("pinned_sha256_16", "") consumer_path = Path(consumer_rel) consumer_up_to_date = ( consumer_path.exists() and sha256_16(consumer_path.read_bytes()) == current_sha ) if current_sha == pinned_sha and consumer_up_to_date: print(f"OK {pin_id}") continue if args.dry_run: label = pinned_sha if pinned_sha else "" print(f"WOULD {pin_id} ({label} -> {current_sha})") continue consumer_path.parent.mkdir(parents=True, exist_ok=True) consumer_path.write_bytes(canon_bytes) manifest_text = update_pin_in_manifest_text( manifest_text, pin_id, current_sha, now, ) label = pinned_sha if pinned_sha else "" print(f"SYNCED {pin_id} ({label} -> {current_sha})") any_change = True if any_change and not args.dry_run: args.manifest.write_text(manifest_text) print(f"Manifest updated: {args.manifest}") if any_error: return 3 return 0 if __name__ == "__main__": sys.exit(main())