aaronrene / overseer-kit public
version_lock.py python
274 lines 9.2 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago
1 """``.overseer/version.lock`` reader/writer per §K4.6 + §K6.4 ``origin``."""
2
3 from __future__ import annotations
4
5 from dataclasses import dataclass
6 from datetime import datetime, timezone
7 from pathlib import Path
8 from typing import Any
9
10 import yaml
11
12 from cli.digest import FootprintRecord, compute_footprint_digest, sha256_hex
13
14 SUPPORTED_LOCK_VERSION = 1
15 LOCK_FILENAME = "version.lock"
16 ORIGIN_KIT = "kit"
17 ORIGIN_PRESERVED = "preserved"
18 SUPPORTED_ORIGINS = frozenset({ORIGIN_KIT, ORIGIN_PRESERVED})
19
20
21 @dataclass(frozen=True)
22 class FootprintEntry:
23 """One per-file manifest row in ``version.lock``."""
24
25 path: str
26 source: str
27 sha256: str
28 origin: str = ORIGIN_KIT
29
30
31 @dataclass(frozen=True)
32 class VersionLock:
33 """Parsed ``version.lock`` with full §K4.6 shape + optional ``origin``."""
34
35 lock_version: int
36 kit_version: str
37 config_version: int
38 installed_at: str
39 synced_at: str
40 footprint_digest: str
41 footprint: tuple[FootprintEntry, ...]
42
43 def to_dict(self) -> dict[str, Any]:
44 """Serialize to a YAML-compatible mapping."""
45 footprint_rows: list[dict[str, Any]] = []
46 for entry in self.footprint:
47 row: dict[str, Any] = {
48 "path": entry.path,
49 "source": entry.source,
50 "sha256": entry.sha256,
51 }
52 # Omit default kit origin for greenfield lock compatibility; always
53 # write preserved (and kit when any preserved exists for clarity).
54 if entry.origin != ORIGIN_KIT or any(
55 e.origin == ORIGIN_PRESERVED for e in self.footprint
56 ):
57 row["origin"] = entry.origin
58 footprint_rows.append(row)
59 return {
60 "lock_version": self.lock_version,
61 "kit_version": self.kit_version,
62 "config_version": self.config_version,
63 "installed_at": self.installed_at,
64 "synced_at": self.synced_at,
65 "footprint_digest": self.footprint_digest,
66 "footprint": footprint_rows,
67 }
68
69
70 class LockError(Exception):
71 """Raised when ``version.lock`` is missing, corrupt, or unsupported."""
72
73
74 def utc_now_iso() -> str:
75 """Return current UTC time as ISO-8601 with trailing ``Z``."""
76 return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
77
78
79 def lock_path(repo_root: Path) -> Path:
80 """Return the default lock file path."""
81 return repo_root / ".overseer" / LOCK_FILENAME
82
83
84 def entry_origin(entry: FootprintEntry | dict[str, Any]) -> str:
85 """Return origin for a lock entry; omitted defaults to ``kit`` (§K6.4)."""
86 if isinstance(entry, FootprintEntry):
87 return entry.origin
88 origin = entry.get("origin", ORIGIN_KIT)
89 return origin if isinstance(origin, str) else ORIGIN_KIT
90
91
92 def kit_only_records(entries: list[FootprintEntry] | tuple[FootprintEntry, ...]) -> list[FootprintRecord]:
93 """Build digest records over ``origin: kit`` (and omitted-default kit) only."""
94 records: list[FootprintRecord] = []
95 for entry in entries:
96 if entry_origin(entry) == ORIGIN_PRESERVED:
97 continue
98 records.append(FootprintRecord(path=entry.path, sha256_hex=entry.sha256))
99 return records
100
101
102 def compute_lock_digest(entries: list[FootprintEntry] | tuple[FootprintEntry, ...]) -> str:
103 """Compute ``footprint_digest`` per §K6.4 kit-only rule when preserved exist."""
104 has_preserved = any(entry_origin(e) == ORIGIN_PRESERVED for e in entries)
105 if has_preserved:
106 records = kit_only_records(entries)
107 else:
108 records = [FootprintRecord(path=e.path, sha256_hex=e.sha256) for e in entries]
109 return compute_footprint_digest(records)
110
111
112 def read_version_lock(path: Path) -> VersionLock:
113 """Parse and validate ``version.lock``; raise ``LockError`` on violation."""
114 if not path.is_file():
115 raise LockError("version.lock missing")
116
117 try:
118 raw_text = path.read_text(encoding="utf-8")
119 except OSError as exc:
120 raise LockError(f"cannot read version.lock: {exc}") from exc
121
122 try:
123 raw = yaml.safe_load(raw_text)
124 except yaml.YAMLError as exc:
125 raise LockError(f"unparseable version.lock: {exc}") from exc
126
127 if not isinstance(raw, dict):
128 raise LockError("version.lock root must be a mapping")
129
130 required_keys = (
131 "lock_version",
132 "kit_version",
133 "config_version",
134 "installed_at",
135 "synced_at",
136 "footprint_digest",
137 "footprint",
138 )
139 missing = [key for key in required_keys if key not in raw]
140 if missing:
141 raise LockError(f"version.lock missing required key(s): {', '.join(missing)}")
142
143 lock_version = raw["lock_version"]
144 if not isinstance(lock_version, int):
145 raise LockError("lock_version must be an integer")
146 if lock_version != SUPPORTED_LOCK_VERSION:
147 raise LockError(
148 f"unsupported lock_version {lock_version} (supported: {SUPPORTED_LOCK_VERSION})"
149 )
150
151 kit_version = raw["kit_version"]
152 if not isinstance(kit_version, str) or not kit_version.strip():
153 raise LockError("kit_version must be a non-empty string")
154
155 config_version = raw["config_version"]
156 if not isinstance(config_version, int):
157 raise LockError("config_version must be an integer")
158
159 installed_at = raw["installed_at"]
160 synced_at = raw["synced_at"]
161 if not isinstance(installed_at, str) or not isinstance(synced_at, str):
162 raise LockError("installed_at and synced_at must be strings")
163
164 footprint_digest = raw["footprint_digest"]
165 if not isinstance(footprint_digest, str) or not footprint_digest.startswith("sha256:"):
166 raise LockError("footprint_digest must be a sha256: prefixed string")
167
168 footprint_raw = raw["footprint"]
169 if not isinstance(footprint_raw, list):
170 raise LockError("footprint must be a list")
171
172 entries: list[FootprintEntry] = []
173 for item in footprint_raw:
174 if not isinstance(item, dict):
175 raise LockError("each footprint entry must be a mapping")
176 for field in ("path", "source", "sha256"):
177 if field not in item or not isinstance(item[field], str):
178 raise LockError(f"footprint entry missing or invalid {field}")
179 origin = item.get("origin", ORIGIN_KIT)
180 if not isinstance(origin, str) or origin not in SUPPORTED_ORIGINS:
181 raise LockError(f"footprint entry origin must be kit|preserved, got {origin!r}")
182 entries.append(
183 FootprintEntry(
184 path=item["path"],
185 source=item["source"],
186 sha256=item["sha256"],
187 origin=origin,
188 )
189 )
190
191 entries.sort(key=lambda e: e.path)
192 return VersionLock(
193 lock_version=lock_version,
194 kit_version=kit_version,
195 config_version=config_version,
196 installed_at=installed_at,
197 synced_at=synced_at,
198 footprint_digest=footprint_digest,
199 footprint=tuple(entries),
200 )
201
202
203 def build_version_lock(
204 *,
205 kit_version: str,
206 config_version: int,
207 footprint: list[tuple[str, str, bytes]],
208 installed_at: str | None = None,
209 synced_at: str | None = None,
210 prior_installed_at: str | None = None,
211 origins: dict[str, str] | None = None,
212 ) -> VersionLock:
213 """Build a new lock from rendered footprint ``(dest_path, source, bytes)`` tuples.
214
215 ``origins`` maps destination path → ``kit|preserved`` (default ``kit``).
216 """
217 now = utc_now_iso()
218 origin_map = origins or {}
219 sorted_footprint = sorted(footprint, key=lambda item: item[0])
220 entries: list[FootprintEntry] = []
221 for dest, source, content in sorted_footprint:
222 hex_digest = sha256_hex(content)
223 origin = origin_map.get(dest, ORIGIN_KIT)
224 if origin not in SUPPORTED_ORIGINS:
225 origin = ORIGIN_KIT
226 entries.append(
227 FootprintEntry(path=dest, source=source, sha256=hex_digest, origin=origin)
228 )
229 entries_tuple = tuple(entries)
230 digest = compute_lock_digest(entries_tuple)
231 return VersionLock(
232 lock_version=SUPPORTED_LOCK_VERSION,
233 kit_version=kit_version,
234 config_version=config_version,
235 installed_at=prior_installed_at or installed_at or now,
236 synced_at=synced_at or now,
237 footprint_digest=digest,
238 footprint=entries_tuple,
239 )
240
241
242 def build_version_lock_from_entries(
243 *,
244 kit_version: str,
245 config_version: int,
246 entries: list[FootprintEntry],
247 installed_at: str,
248 synced_at: str | None = None,
249 ) -> VersionLock:
250 """Build a lock from explicit manifest entries (for partial ``--only`` sync)."""
251 sorted_entries = sorted(entries, key=lambda e: e.path)
252 digest = compute_lock_digest(sorted_entries)
253 return VersionLock(
254 lock_version=SUPPORTED_LOCK_VERSION,
255 kit_version=kit_version,
256 config_version=config_version,
257 installed_at=installed_at,
258 synced_at=synced_at or utc_now_iso(),
259 footprint_digest=digest,
260 footprint=tuple(sorted_entries),
261 )
262
263
264 def write_version_lock(path: Path, lock: VersionLock) -> None:
265 """Write ``version.lock`` as YAML."""
266 from cli.atomic import atomic_write_text
267
268 text = yaml.safe_dump(
269 lock.to_dict(),
270 sort_keys=False,
271 allow_unicode=True,
272 default_flow_style=False,
273 )
274 atomic_write_text(path, text)
File History 2 commits
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439 feat: K1-P1 complete — agent provenance, build-verification… Sonnet 4.6 patch 52 days ago