stamp.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Review stamp write path (§K5.7 / §FRV.3–§FRV.5).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from datetime import datetime, timezone |
| 6 | from pathlib import Path |
| 7 | from typing import Any |
| 8 | |
| 9 | from cli.atomic import WriteFailure, atomic_write_text |
| 10 | from tools.freeze_authorization.resolve import resolve_stamp_record |
| 11 | from tools.freeze_reviewer.artifact import ( |
| 12 | STAMP_MARKER, |
| 13 | ParsedArtifact, |
| 14 | _operator_forced_md_prefix, |
| 15 | artifact_digest, |
| 16 | extract_existing_stamp, |
| 17 | pre_stamp_canonical_bytes, |
| 18 | ) |
| 19 | from tools.freeze_reviewer.serializer import dump_freeze_mapping |
| 20 | from tools.freeze_reviewer.types import ( |
| 21 | CLI_OWNED_STAMP_KEYS, |
| 22 | STAMP_KEY_ORDER, |
| 23 | ReviewStamp, |
| 24 | ReviewerSettings, |
| 25 | ) |
| 26 | |
| 27 | |
| 28 | def utc_now_z() -> str: |
| 29 | """Return ISO-8601 UTC timestamp with trailing Z.""" |
| 30 | return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") |
| 31 | |
| 32 | |
| 33 | def build_stamp( |
| 34 | parsed: ParsedArtifact, |
| 35 | *, |
| 36 | reviewer: ReviewerSettings, |
| 37 | kit_version: str, |
| 38 | produced_by: str, |
| 39 | provider_kind: str, |
| 40 | checklist_ids: list[str], |
| 41 | checklist_source: str, |
| 42 | findings_count: int, |
| 43 | override_applied: bool = False, |
| 44 | ) -> ReviewStamp: |
| 45 | """Build a fourteen-key mechanical stamp; never emit ``verdict`` (§FRV.3–§FRV.4).""" |
| 46 | model = None if provider_kind == "rule_engine" else reviewer.model |
| 47 | return ReviewStamp( |
| 48 | reviewed_at=utc_now_z(), |
| 49 | mechanical_verdict="pass", |
| 50 | reviewer_mode=reviewer.mode, |
| 51 | reviewer_model=model, |
| 52 | reviewer_provider=reviewer.provider, |
| 53 | kit_version=kit_version, |
| 54 | artifact_digest=artifact_digest(parsed), |
| 55 | gate="mechanical", |
| 56 | produced_by=produced_by, |
| 57 | provider_kind=provider_kind, |
| 58 | checklist_ids=list(checklist_ids), |
| 59 | checklist_source=checklist_source, |
| 60 | findings_count=findings_count, |
| 61 | override_applied=override_applied, |
| 62 | ) |
| 63 | |
| 64 | |
| 65 | def merge_stamp_mapping(existing: dict[str, Any] | None, new: dict[str, Any]) -> dict[str, Any]: |
| 66 | """Merge per §FRV.5.1: fourteen CLI keys from N, then unknown keys from E; drop legacy verdict.""" |
| 67 | e = existing if isinstance(existing, dict) else {} |
| 68 | merged: dict[str, Any] = {} |
| 69 | for key in STAMP_KEY_ORDER: |
| 70 | if key in new: |
| 71 | merged[key] = new[key] |
| 72 | for key, value in e.items(): |
| 73 | if key in CLI_OWNED_STAMP_KEYS: |
| 74 | continue |
| 75 | if key not in merged: |
| 76 | merged[key] = value |
| 77 | return merged |
| 78 | |
| 79 | |
| 80 | def stamp_is_idempotent_noop(parsed: ParsedArtifact, new_stamp: ReviewStamp) -> bool: |
| 81 | """True when re-run is a no-op per §FRV.5.3 four conditions.""" |
| 82 | existing = extract_existing_stamp(parsed) |
| 83 | if not existing: |
| 84 | return False |
| 85 | resolved = resolve_stamp_record(existing) |
| 86 | if resolved.kind != "mechanical": |
| 87 | return False |
| 88 | if resolved.verdict != new_stamp.mechanical_verdict: |
| 89 | return False |
| 90 | if existing.get("artifact_digest") != new_stamp.artifact_digest: |
| 91 | return False |
| 92 | # Condition 4: preserve existing reviewed_at so a true no-op does not refresh it. |
| 93 | candidate = new_stamp.to_mapping() |
| 94 | if "reviewed_at" in existing and existing["reviewed_at"] is not None: |
| 95 | candidate["reviewed_at"] = existing["reviewed_at"] |
| 96 | from dataclasses import replace |
| 97 | |
| 98 | compare_stamp = replace( |
| 99 | new_stamp, |
| 100 | reviewed_at=str(candidate["reviewed_at"]), |
| 101 | override_applied=bool(candidate.get("override_applied", False)), |
| 102 | ) |
| 103 | # Align compare_stamp fields with candidate for merge fidelity |
| 104 | compare_stamp = replace( |
| 105 | compare_stamp, |
| 106 | gate=str(candidate.get("gate", "mechanical")), |
| 107 | produced_by=str(candidate.get("produced_by", compare_stamp.produced_by)), |
| 108 | provider_kind=str(candidate.get("provider_kind", compare_stamp.provider_kind)), |
| 109 | checklist_ids=list(candidate.get("checklist_ids") or []), |
| 110 | checklist_source=str(candidate.get("checklist_source", compare_stamp.checklist_source)), |
| 111 | findings_count=int(candidate.get("findings_count") or 0), |
| 112 | ) |
| 113 | rendered = render_stamped_text(parsed, compare_stamp) |
| 114 | return rendered == parsed.text |
| 115 | |
| 116 | |
| 117 | def _insert_review_stamp(mapping: dict, stamp: ReviewStamp) -> dict: |
| 118 | updated = dict(mapping) |
| 119 | existing = mapping.get("review_stamp") if isinstance(mapping.get("review_stamp"), dict) else {} |
| 120 | updated["review_stamp"] = merge_stamp_mapping(existing, stamp.to_mapping()) |
| 121 | return updated |
| 122 | |
| 123 | |
| 124 | def render_stamped_text(parsed: ParsedArtifact, stamp: ReviewStamp) -> str: |
| 125 | """Render artifact text with the stamp applied.""" |
| 126 | if parsed.kind == "markdown_fence" and parsed.freeze_mapping is not None and parsed.fence_match: |
| 127 | updated = _insert_review_stamp(parsed.freeze_mapping, stamp) |
| 128 | serialized = dump_freeze_mapping(updated) |
| 129 | start, end = parsed.fence_match.span() |
| 130 | fence_lang = parsed.fence_match.group(1) |
| 131 | return parsed.text[:start] + f"```{fence_lang}\n{serialized}```" + parsed.text[end:] |
| 132 | |
| 133 | if parsed.kind == "yaml_whole" and parsed.freeze_mapping is not None: |
| 134 | updated = _insert_review_stamp(parsed.freeze_mapping, stamp) |
| 135 | return dump_freeze_mapping(updated) |
| 136 | |
| 137 | if parsed.kind == "operator_forced_md": |
| 138 | existing = extract_existing_stamp(parsed) or {} |
| 139 | merged = merge_stamp_mapping(existing, stamp.to_mapping()) |
| 140 | stamp_yaml = dump_freeze_mapping({"review_stamp": merged}) |
| 141 | marker_index = parsed.text.rfind(STAMP_MARKER) |
| 142 | if marker_index == -1: |
| 143 | base = parsed.text if parsed.text.endswith("\n") else parsed.text + "\n" |
| 144 | else: |
| 145 | base = _operator_forced_md_prefix(parsed.text) |
| 146 | return f"{base}\n{STAMP_MARKER}\n```yaml\n{stamp_yaml}```" |
| 147 | |
| 148 | if parsed.kind == "operator_forced_yaml": |
| 149 | base = parsed.freeze_mapping if isinstance(parsed.freeze_mapping, dict) else {} |
| 150 | updated = _insert_review_stamp(base, stamp) |
| 151 | return dump_freeze_mapping(updated) |
| 152 | |
| 153 | return parsed.text |
| 154 | |
| 155 | |
| 156 | def write_stamp( |
| 157 | path: Path, |
| 158 | parsed: ParsedArtifact, |
| 159 | stamp: ReviewStamp, |
| 160 | ) -> bool: |
| 161 | """Atomically write stamp; return False when idempotent no-op.""" |
| 162 | if stamp_is_idempotent_noop(parsed, stamp): |
| 163 | return False |
| 164 | text = render_stamped_text(parsed, stamp) |
| 165 | atomic_write_text(path, text) |
| 166 | return True |
| 167 | |
| 168 | |
| 169 | def write_stamp_or_fail(path: Path, parsed: ParsedArtifact, stamp: ReviewStamp) -> tuple[bool, bool]: |
| 170 | """Write stamp; return (written, io_failed).""" |
| 171 | try: |
| 172 | written = write_stamp(path, parsed, stamp) |
| 173 | return written, False |
| 174 | except WriteFailure: |
| 175 | return False, True |
| 176 | |
| 177 | |
| 178 | def reference_digest(parsed: ParsedArtifact) -> str: |
| 179 | """Independently compute digest reference for data-integrity tests.""" |
| 180 | digest_hex = __import__("hashlib").sha256(pre_stamp_canonical_bytes(parsed)).hexdigest() |
| 181 | return f"sha256:{digest_hex}" |