aaronrene / overseer-kit public
stamp.py python
181 lines 6.7 KB
Raw
sha256:8461d44b77376fbf06fa7c3e085d309e3010fd8d5886d63c63e69ce118811ad4 docs: record AFF-b feature-tip SHAs after AFF-b-ISR commit. Human 3 days ago
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}"
File History 1 commit
sha256:8461d44b77376fbf06fa7c3e085d309e3010fd8d5886d63c63e69ce118811ad4 docs: record AFF-b feature-tip SHAs after AFF-b-ISR commit. Human 3 days ago