aaronrene / overseer-kit public
stamp.py python
116 lines 4.0 KB
Raw
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 13 hours ago
1 """Review stamp write path (§K5.7)."""
2
3 from __future__ import annotations
4
5 from datetime import datetime, timezone
6 from pathlib import Path
7
8 from cli.atomic import WriteFailure, atomic_write_text
9 from tools.freeze_reviewer.artifact import (
10 STAMP_MARKER,
11 FENCE_RE,
12 ParsedArtifact,
13 _operator_forced_md_prefix,
14 artifact_digest,
15 extract_existing_stamp,
16 pre_stamp_canonical_bytes,
17 )
18 from tools.freeze_reviewer.serializer import dump_freeze_mapping, parse_freeze_mapping
19 from tools.freeze_reviewer.types import ReviewStamp, ReviewerSettings
20
21
22 def utc_now_z() -> str:
23 """Return ISO-8601 UTC timestamp with trailing Z."""
24 return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
25
26
27 def build_stamp(
28 parsed: ParsedArtifact,
29 *,
30 reviewer: ReviewerSettings,
31 kit_version: str,
32 ) -> ReviewStamp:
33 """Build a review stamp payload for the current artifact."""
34 return ReviewStamp(
35 reviewed_at=utc_now_z(),
36 verdict="pass",
37 reviewer_mode=reviewer.mode,
38 reviewer_model=reviewer.model,
39 reviewer_provider=reviewer.provider,
40 kit_version=kit_version,
41 artifact_digest=artifact_digest(parsed),
42 )
43
44
45 def stamp_is_idempotent_noop(parsed: ParsedArtifact, new_stamp: ReviewStamp) -> bool:
46 """Return True when an existing pass stamp matches the recomputed digest."""
47 existing = extract_existing_stamp(parsed)
48 if not existing:
49 return False
50 if existing.get("verdict") != "pass":
51 return False
52 return existing.get("artifact_digest") == new_stamp.artifact_digest
53
54
55 def _insert_review_stamp(mapping: dict, stamp: ReviewStamp) -> dict:
56 updated = dict(mapping)
57 updated["review_stamp"] = stamp.to_mapping()
58 return updated
59
60
61 def render_stamped_text(parsed: ParsedArtifact, stamp: ReviewStamp) -> str:
62 """Render artifact text with the stamp applied."""
63 if parsed.kind == "markdown_fence" and parsed.freeze_mapping is not None and parsed.fence_match:
64 updated = _insert_review_stamp(parsed.freeze_mapping, stamp)
65 serialized = dump_freeze_mapping(updated)
66 start, end = parsed.fence_match.span()
67 fence_lang = parsed.fence_match.group(1)
68 return parsed.text[:start] + f"```{fence_lang}\n{serialized}```" + parsed.text[end:]
69
70 if parsed.kind == "yaml_whole" and parsed.freeze_mapping is not None:
71 updated = _insert_review_stamp(parsed.freeze_mapping, stamp)
72 return dump_freeze_mapping(updated)
73
74 if parsed.kind == "operator_forced_md":
75 stamp_yaml = dump_freeze_mapping({"review_stamp": stamp.to_mapping()})
76 marker_index = parsed.text.rfind(STAMP_MARKER)
77 if marker_index == -1:
78 base = parsed.text if parsed.text.endswith("\n") else parsed.text + "\n"
79 else:
80 base = _operator_forced_md_prefix(parsed.text)
81 return f"{base}\n{STAMP_MARKER}\n```yaml\n{stamp_yaml}```"
82
83 if parsed.kind == "operator_forced_yaml":
84 base = parsed.freeze_mapping if isinstance(parsed.freeze_mapping, dict) else {}
85 updated = _insert_review_stamp(base, stamp)
86 return dump_freeze_mapping(updated)
87
88 return parsed.text
89
90
91 def write_stamp(
92 path: Path,
93 parsed: ParsedArtifact,
94 stamp: ReviewStamp,
95 ) -> bool:
96 """Atomically write stamp; return False when idempotent no-op."""
97 if stamp_is_idempotent_noop(parsed, stamp):
98 return False
99 text = render_stamped_text(parsed, stamp)
100 atomic_write_text(path, text)
101 return True
102
103
104 def write_stamp_or_fail(path: Path, parsed: ParsedArtifact, stamp: ReviewStamp) -> tuple[bool, bool]:
105 """Write stamp; return (written, io_failed)."""
106 try:
107 written = write_stamp(path, parsed, stamp)
108 return written, False
109 except WriteFailure:
110 return False, True
111
112
113 def reference_digest(parsed: ParsedArtifact) -> str:
114 """Independently compute digest reference for data-integrity tests."""
115 digest_hex = __import__("hashlib").sha256(pre_stamp_canonical_bytes(parsed)).hexdigest()
116 return f"sha256:{digest_hex}"
File History 2 commits
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1 docs: MuseHub-first before ISR #74 — staging solidify NEXT Human 13 hours ago
sha256:4671b7f787ddbe63ced31c895b688c77ab495653b65a730b423329f26b3c1439 feat: K1-P1 complete — agent provenance, build-verification… Sonnet 4.6 patch 52 days ago