sync_classify.py
python
sha256:0e9549ec7b463911bc08b7d586dc320b1ac9b1f5c943ee7e3865dcc6cb0f6f83
chore(governance): sync handover+roadmap to 84db8c8 (drift:…
Human
8 hours ago
| 1 | """Footprint file classification for ``overseer sync`` (§K4.3).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import fnmatch |
| 6 | from dataclasses import dataclass |
| 7 | from enum import Enum |
| 8 | from pathlib import Path |
| 9 | |
| 10 | from cli.digest import sha256_hex |
| 11 | from cli.footprint import FootprintFile |
| 12 | from cli.version_lock import FootprintEntry, VersionLock |
| 13 | |
| 14 | |
| 15 | class Classification(str, Enum): |
| 16 | """Three-way sync classification per §K4.3.""" |
| 17 | |
| 18 | UNCHANGED = "unchanged" |
| 19 | KIT_UPDATED = "kit-updated" |
| 20 | CONSUMER_MODIFIED = "consumer-modified" |
| 21 | BOTH_CHANGED = "both-changed" |
| 22 | MISSING = "missing" |
| 23 | |
| 24 | |
| 25 | @dataclass(frozen=True) |
| 26 | class ClassifiedFile: |
| 27 | """One classified footprint file with rendered and on-disk state.""" |
| 28 | |
| 29 | destination: str |
| 30 | source: str |
| 31 | classification: Classification |
| 32 | new_content: bytes |
| 33 | baseline_sha: str | None |
| 34 | current_sha: str | None |
| 35 | |
| 36 | @property |
| 37 | def needs_write(self) -> bool: |
| 38 | return self.classification in { |
| 39 | Classification.KIT_UPDATED, |
| 40 | Classification.MISSING, |
| 41 | } |
| 42 | |
| 43 | @property |
| 44 | def is_conflict(self) -> bool: |
| 45 | return self.classification in { |
| 46 | Classification.CONSUMER_MODIFIED, |
| 47 | Classification.BOTH_CHANGED, |
| 48 | } |
| 49 | |
| 50 | |
| 51 | def matches_glob(destination: str, globs: list[str]) -> bool: |
| 52 | """Return True when ``destination`` matches any ``--only`` glob.""" |
| 53 | return any(fnmatch.fnmatch(destination, pattern) for pattern in globs) |
| 54 | |
| 55 | |
| 56 | def classify_footprint( |
| 57 | rendered: list[FootprintFile], |
| 58 | lock: VersionLock | None, |
| 59 | repo_root: Path, |
| 60 | ) -> list[ClassifiedFile]: |
| 61 | """Classify every footprint file using lock baseline and on-disk bytes.""" |
| 62 | baseline = {entry.path: entry for entry in lock.footprint} if lock else {} |
| 63 | classified: list[ClassifiedFile] = [] |
| 64 | |
| 65 | for item in rendered: |
| 66 | entry = baseline.get(item.destination) |
| 67 | baseline_sha = entry.sha256 if entry else None |
| 68 | dest_path = repo_root / item.destination |
| 69 | if dest_path.is_file(): |
| 70 | current_bytes = dest_path.read_bytes() |
| 71 | current_sha = sha256_hex(current_bytes) |
| 72 | else: |
| 73 | current_bytes = b"" |
| 74 | current_sha = None |
| 75 | |
| 76 | new_sha = sha256_hex(item.content) |
| 77 | if current_sha is None: |
| 78 | classification = Classification.MISSING |
| 79 | elif current_sha == baseline_sha: |
| 80 | if new_sha == baseline_sha: |
| 81 | classification = Classification.UNCHANGED |
| 82 | else: |
| 83 | classification = Classification.KIT_UPDATED |
| 84 | elif new_sha == baseline_sha: |
| 85 | classification = Classification.CONSUMER_MODIFIED |
| 86 | else: |
| 87 | classification = Classification.BOTH_CHANGED |
| 88 | |
| 89 | classified.append( |
| 90 | ClassifiedFile( |
| 91 | destination=item.destination, |
| 92 | source=item.source, |
| 93 | classification=classification, |
| 94 | new_content=item.content, |
| 95 | baseline_sha=baseline_sha, |
| 96 | current_sha=current_sha, |
| 97 | ) |
| 98 | ) |
| 99 | |
| 100 | classified.sort(key=lambda row: row.destination) |
| 101 | return classified |
File History
1 commit
sha256:6abcf1fa82a7a621ccbc945f19acdba5bc0db54569599404a1452fb4a096a199
fix(ISR): default require_independent_second_reviewer to require
Human
minor
⚠
7 hours ago