stamp.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 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}" |