aaronrene / overseer-kit public
artifact.py python
175 lines 5.8 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 1 day ago
1 """Freeze artifact parsing and canonical digest bytes (§K5.4 / §K5.7)."""
2
3 from __future__ import annotations
4
5 import hashlib
6 import re
7 from dataclasses import dataclass
8 from pathlib import Path
9
10 import yaml
11
12 from tools.freeze_reviewer.serializer import dump_freeze_mapping, parse_freeze_mapping
13 from tools.freeze_reviewer.types import ArtifactKind, DeclarationStatus
14
15 UTF8_BOM = b"\xef\xbb\xbf"
16 FENCE_RE = re.compile(
17 r"```[ \t]*(yaml|yml)[ \t]*\n(.*?)```",
18 re.DOTALL | re.IGNORECASE,
19 )
20 STAMP_MARKER = "<!-- overseer:review-stamp -->"
21
22
23 @dataclass(frozen=True)
24 class ParsedArtifact:
25 """Parsed freeze artifact metadata."""
26
27 text: str
28 canonical_text: str
29 declaration: DeclarationStatus
30 kind: ArtifactKind
31 freeze_mapping: dict | None
32 fence_match: re.Match[str] | None
33 rel_path: str
34
35
36 def strip_bom_and_normalize_newlines(data: bytes) -> str:
37 """Apply §K4.7 canonical byte rules items 1 and 2."""
38 if data.startswith(UTF8_BOM):
39 data = data[UTF8_BOM.__len__() :]
40 try:
41 text = data.decode("utf-8")
42 except UnicodeDecodeError as exc:
43 raise ValueError("not-utf8") from exc
44 return text.replace("\r\n", "\n").replace("\r", "\n")
45
46
47 def read_artifact_bytes(path: Path) -> str:
48 """Read artifact bytes as canonical text or raise ``not-utf8``."""
49 return strip_bom_and_normalize_newlines(path.read_bytes())
50
51
52 def _is_valid_declaration(mapping: dict) -> bool:
53 if not isinstance(mapping.get("phase"), str):
54 return False
55 outputs = mapping.get("outputs")
56 if not isinstance(outputs, list) or not outputs:
57 return False
58 for item in outputs:
59 if not isinstance(item, dict):
60 return False
61 if item.get("frozen") is True:
62 return True
63 return False
64
65
66 def _detect_kind(path: Path, text: str) -> tuple[ArtifactKind, DeclarationStatus, dict | None, re.Match | None]:
67 suffix = path.suffix.lower()
68 if suffix in {".yaml", ".yml"}:
69 try:
70 mapping = parse_freeze_mapping(text)
71 except ValueError:
72 mapping = None
73 if mapping and _is_valid_declaration(mapping):
74 return "yaml_whole", "present", mapping, None
75 return "operator_forced_yaml", "absent", mapping if isinstance(mapping, dict) else None, None
76
77 match = FENCE_RE.search(text)
78 if match is not None:
79 body = match.group(2)
80 try:
81 mapping = parse_freeze_mapping(body)
82 except ValueError:
83 mapping = None
84 if mapping and _is_valid_declaration(mapping):
85 return "markdown_fence", "present", mapping, match
86 return "operator_forced_md", "absent", None, match
87
88
89 def parse_artifact(path: Path, *, rel_path: str) -> ParsedArtifact:
90 """Parse artifact text and detect declaration status."""
91 text = read_artifact_bytes(path)
92 kind, declaration, mapping, fence_match = _detect_kind(path, text)
93 return ParsedArtifact(
94 text=text,
95 canonical_text=text,
96 declaration=declaration,
97 kind=kind,
98 freeze_mapping=mapping,
99 fence_match=fence_match,
100 rel_path=rel_path,
101 )
102
103
104 def _operator_forced_md_prefix(text: str) -> str:
105 """Recover pre-stamp prose bytes for operator-forced Markdown."""
106 marker_index = text.rfind(STAMP_MARKER)
107 if marker_index == -1:
108 return text
109 prefix = text[:marker_index]
110 if prefix.endswith("\n\n"):
111 return prefix[:-1]
112 return prefix
113
114
115 def _remove_operator_forced_md_suffix(text: str) -> str:
116 return _operator_forced_md_prefix(text)
117
118
119 def pre_stamp_canonical_bytes(parsed: ParsedArtifact) -> bytes:
120 """Compute pre-stamp canonical form bytes for digest (§K5.7)."""
121 if parsed.kind == "markdown_fence" and parsed.freeze_mapping is not None and parsed.fence_match:
122 mapping = dict(parsed.freeze_mapping)
123 mapping.pop("review_stamp", None)
124 serialized = dump_freeze_mapping(mapping)
125 start, end = parsed.fence_match.span()
126 fence_lang = parsed.fence_match.group(1)
127 before = parsed.text[:start]
128 after = parsed.text[end:]
129 body = f"```{fence_lang}\n{serialized}```"
130 return (before + body + after).encode("utf-8")
131
132 if parsed.kind == "yaml_whole" and parsed.freeze_mapping is not None:
133 mapping = dict(parsed.freeze_mapping)
134 mapping.pop("review_stamp", None)
135 return dump_freeze_mapping(mapping).encode("utf-8")
136
137 if parsed.kind == "operator_forced_md":
138 return _remove_operator_forced_md_suffix(parsed.text).encode("utf-8")
139
140 if parsed.kind == "operator_forced_yaml":
141 if parsed.freeze_mapping is not None:
142 mapping = dict(parsed.freeze_mapping)
143 mapping.pop("review_stamp", None)
144 return dump_freeze_mapping(mapping).encode("utf-8")
145 return parsed.text.encode("utf-8")
146
147 return parsed.text.encode("utf-8")
148
149
150 def artifact_digest(parsed: ParsedArtifact) -> str:
151 """Return sha256 digest over pre-stamp canonical form."""
152 payload = pre_stamp_canonical_bytes(parsed)
153 digest_hex = hashlib.sha256(payload).hexdigest()
154 return f"sha256:{digest_hex}"
155
156
157 def extract_existing_stamp(parsed: ParsedArtifact) -> dict | None:
158 """Return existing review_stamp mapping if present."""
159 if parsed.freeze_mapping and isinstance(parsed.freeze_mapping.get("review_stamp"), dict):
160 return parsed.freeze_mapping["review_stamp"]
161 if parsed.kind == "operator_forced_md":
162 marker_index = parsed.text.rfind(STAMP_MARKER)
163 if marker_index == -1:
164 return None
165 tail = parsed.text[marker_index:]
166 match = FENCE_RE.search(tail)
167 if not match:
168 return None
169 try:
170 mapping = parse_freeze_mapping(match.group(2))
171 except ValueError:
172 return None
173 stamp = mapping.get("review_stamp")
174 return stamp if isinstance(stamp, dict) else None
175 return None
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 53 days ago