validate.py
python
sha256:a78e7e5a8740e03315f325d19edeb3aa1b306b3337d04abbaa9a9e0f3bbeb7a1
docs: MuseHub-first before ISR #74 — staging solidify NEXT
Human
13 hours ago
| 1 | """Ledger entry validation (§K9.7 / §K9.8 / §PE.3–§PE.4).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import re |
| 6 | from typing import Any |
| 7 | |
| 8 | from tools.honesty.provenance import validate_provenance |
| 9 | from tools.honesty.types import ( |
| 10 | ACTOR_ROLES, |
| 11 | BV_VERDICTS, |
| 12 | ENTRY_KINDS, |
| 13 | ISR_VERDICTS, |
| 14 | VERIFICATION_ARTIFACT_TYPES, |
| 15 | EntryValidationError, |
| 16 | ) |
| 17 | |
| 18 | _SHA256_RE = re.compile(r"^[0-9a-f]{64}$") |
| 19 | |
| 20 | |
| 21 | def _require_mapping(value: Any, field: str) -> dict[str, Any]: |
| 22 | if not isinstance(value, dict): |
| 23 | raise EntryValidationError(2, f"{field} must be an object") |
| 24 | return value |
| 25 | |
| 26 | |
| 27 | def _validate_sha256(value: Any, field: str) -> str: |
| 28 | if not isinstance(value, str) or not _SHA256_RE.match(value): |
| 29 | raise EntryValidationError(2, f"{field} must be lowercase 64-char hex sha256") |
| 30 | return value |
| 31 | |
| 32 | |
| 33 | def validate_verification_artifacts(artifacts: Any) -> list[dict[str, Any]]: |
| 34 | """Validate ``verification_evidence.artifacts`` per §PE.4.""" |
| 35 | if not isinstance(artifacts, list) or not artifacts: |
| 36 | raise EntryValidationError(24, "artifacts must be a non-empty list") |
| 37 | normalized: list[dict[str, Any]] = [] |
| 38 | for index, item in enumerate(artifacts): |
| 39 | obj = _require_mapping(item, f"artifacts[{index}]") |
| 40 | art_type = obj.get("type") |
| 41 | if art_type not in VERIFICATION_ARTIFACT_TYPES: |
| 42 | raise EntryValidationError( |
| 43 | 2, |
| 44 | f"artifacts[{index}].type must be test_output|deploy_health|screenshot", |
| 45 | ) |
| 46 | sha256 = _validate_sha256(obj.get("sha256"), f"artifacts[{index}].sha256") |
| 47 | ref = obj.get("ref") |
| 48 | if art_type in {"deploy_health", "screenshot"}: |
| 49 | _require_non_empty_str(ref, f"artifacts[{index}].ref") |
| 50 | elif ref is not None and not isinstance(ref, str): |
| 51 | raise EntryValidationError(2, f"artifacts[{index}].ref must be a string when present") |
| 52 | notes = obj.get("notes") |
| 53 | if notes is not None and not isinstance(notes, str): |
| 54 | raise EntryValidationError(2, f"artifacts[{index}].notes must be a string when present") |
| 55 | entry: dict[str, Any] = {"type": art_type, "sha256": sha256} |
| 56 | if ref is not None: |
| 57 | entry["ref"] = ref |
| 58 | if notes is not None: |
| 59 | entry["notes"] = notes |
| 60 | normalized.append(entry) |
| 61 | return normalized |
| 62 | |
| 63 | |
| 64 | def find_matching_verification_evidence( |
| 65 | entries: list[dict[str, Any]], |
| 66 | *, |
| 67 | phase_id: str, |
| 68 | frozen_spec: str | None, |
| 69 | ) -> dict[str, Any] | None: |
| 70 | """Return the last matching pass entry for Mode B (§PE.6.1).""" |
| 71 | matches: list[dict[str, Any]] = [] |
| 72 | for entry in entries: |
| 73 | if entry.get("kind") != "verification_evidence": |
| 74 | continue |
| 75 | if entry.get("actor_role") != "verifier": |
| 76 | continue |
| 77 | if entry.get("bv_verdict") != "pass": |
| 78 | continue |
| 79 | if entry.get("phase_id") != phase_id: |
| 80 | continue |
| 81 | if frozen_spec is not None and entry.get("frozen_spec") != frozen_spec: |
| 82 | continue |
| 83 | matches.append(entry) |
| 84 | return matches[-1] if matches else None |
| 85 | |
| 86 | |
| 87 | def find_matching_deploy_health( |
| 88 | entries: list[dict[str, Any]], |
| 89 | *, |
| 90 | phase_id: str, |
| 91 | frozen_spec: str | None, |
| 92 | ) -> dict[str, Any] | None: |
| 93 | """Return the last Mode C match: pass + ≥1 ``deploy_health`` artifact (§PD.3). |
| 94 | |
| 95 | Does not open network or treat ``ref`` as a URL. |
| 96 | """ |
| 97 | matches: list[dict[str, Any]] = [] |
| 98 | for entry in entries: |
| 99 | if entry.get("kind") != "verification_evidence": |
| 100 | continue |
| 101 | if entry.get("actor_role") != "verifier": |
| 102 | continue |
| 103 | if entry.get("bv_verdict") != "pass": |
| 104 | continue |
| 105 | if entry.get("phase_id") != phase_id: |
| 106 | continue |
| 107 | if frozen_spec is not None and entry.get("frozen_spec") != frozen_spec: |
| 108 | continue |
| 109 | artifacts = entry.get("artifacts") |
| 110 | if not isinstance(artifacts, list): |
| 111 | continue |
| 112 | if not any( |
| 113 | isinstance(item, dict) and item.get("type") == "deploy_health" for item in artifacts |
| 114 | ): |
| 115 | continue |
| 116 | matches.append(entry) |
| 117 | return matches[-1] if matches else None |
| 118 | |
| 119 | |
| 120 | def find_matching_independent_second_review( |
| 121 | entries: list[dict[str, Any]], |
| 122 | *, |
| 123 | phase_id: str, |
| 124 | frozen_spec: str | None, |
| 125 | producer_session: str | None, |
| 126 | ) -> dict[str, Any] | None: |
| 127 | """Return the last matching ISR pass entry for Mode D (§ISR.5.3). |
| 128 | |
| 129 | Does not open a network connection, call a model, or read IDE session ids. |
| 130 | """ |
| 131 | matches: list[dict[str, Any]] = [] |
| 132 | for entry in entries: |
| 133 | if entry.get("kind") != "independent_second_review": |
| 134 | continue |
| 135 | if entry.get("actor_role") != "verifier": |
| 136 | continue |
| 137 | if entry.get("isr_verdict") != "pass": |
| 138 | continue |
| 139 | if entry.get("phase_id") != phase_id: |
| 140 | continue |
| 141 | if frozen_spec is not None and entry.get("frozen_spec") != frozen_spec: |
| 142 | continue |
| 143 | actor_session = entry.get("actor_session_id") |
| 144 | producer_id = entry.get("producer_session_id") |
| 145 | if not isinstance(actor_session, str) or not isinstance(producer_id, str): |
| 146 | continue |
| 147 | if actor_session == producer_id: |
| 148 | continue |
| 149 | if producer_session is not None: |
| 150 | if producer_id != producer_session: |
| 151 | continue |
| 152 | if actor_session == producer_session: |
| 153 | continue |
| 154 | matches.append(entry) |
| 155 | return matches[-1] if matches else None |
| 156 | |
| 157 | |
| 158 | def _require_non_empty_str(value: Any, field: str) -> str: |
| 159 | if not isinstance(value, str) or not value.strip(): |
| 160 | raise EntryValidationError(2, f"{field} must be a non-empty string") |
| 161 | return value |
| 162 | |
| 163 | |
| 164 | def validate_append_body(*, kind: str, body: dict[str, Any]) -> dict[str, Any]: |
| 165 | """Validate and normalize an append body before hashing.""" |
| 166 | if kind not in ENTRY_KINDS: |
| 167 | raise EntryValidationError(2, f"unknown entry kind: {kind}") |
| 168 | |
| 169 | if "entry_hash" in body or "prev_hash" in body: |
| 170 | raise EntryValidationError(2, "client must not supply entry_hash or prev_hash") |
| 171 | |
| 172 | body_kind = body.get("kind") |
| 173 | if body_kind is not None and body_kind != kind: |
| 174 | raise EntryValidationError(2, "body kind must match --kind when present") |
| 175 | |
| 176 | merged = dict(body) |
| 177 | merged["kind"] = kind |
| 178 | |
| 179 | version = merged.get("v", 1) |
| 180 | if version != 1: |
| 181 | raise EntryValidationError(2, "v must be integer 1") |
| 182 | |
| 183 | merged["v"] = 1 |
| 184 | |
| 185 | if kind == "genesis": |
| 186 | if "actor_role" in merged or "actor_session_id" in merged: |
| 187 | raise EntryValidationError(2, "genesis must not carry actor fields") |
| 188 | for key in ( |
| 189 | "assignment", |
| 190 | "artifact_sha256", |
| 191 | "passed", |
| 192 | "evidence", |
| 193 | "subject", |
| 194 | "ruling", |
| 195 | "bound_verdict_hash", |
| 196 | "hook", |
| 197 | "ok", |
| 198 | "reason", |
| 199 | "provenance", |
| 200 | "phase_id", |
| 201 | "frozen_spec", |
| 202 | "round", |
| 203 | "bv_verdict", |
| 204 | "artifacts", |
| 205 | "subject_sha256", |
| 206 | "isr_verdict", |
| 207 | "producer_session_id", |
| 208 | "producer_agent_id", |
| 209 | "verifier_agent_id", |
| 210 | "bound_verification_evidence_hash", |
| 211 | ): |
| 212 | if key in merged: |
| 213 | raise EntryValidationError(2, f"genesis must not carry {key}") |
| 214 | return merged |
| 215 | |
| 216 | actor_role = merged.get("actor_role") |
| 217 | if actor_role not in ACTOR_ROLES: |
| 218 | raise EntryValidationError( |
| 219 | 23 |
| 220 | if kind in {"verdict", "verification_evidence", "independent_second_review"} |
| 221 | else 2, |
| 222 | "invalid or missing actor_role", |
| 223 | ) |
| 224 | |
| 225 | actor_session = merged.get("actor_session_id") |
| 226 | _require_non_empty_str(actor_session, "actor_session_id") |
| 227 | |
| 228 | if kind == "task_assigned": |
| 229 | if actor_role != "overseer": |
| 230 | raise EntryValidationError(23, "task_assigned requires actor_role=overseer") |
| 231 | _require_mapping(merged.get("assignment"), "assignment") |
| 232 | elif kind == "verdict": |
| 233 | if actor_role != "verifier": |
| 234 | raise EntryValidationError(23, "verdict requires actor_role=verifier") |
| 235 | _require_non_empty_str(merged.get("artifact_sha256"), "artifact_sha256") |
| 236 | passed = merged.get("passed") |
| 237 | if not isinstance(passed, bool): |
| 238 | raise EntryValidationError(2, "passed must be a boolean") |
| 239 | evidence = _require_mapping(merged.get("evidence"), "evidence") |
| 240 | reexecuted = evidence.get("reexecuted") |
| 241 | if not isinstance(reexecuted, list) or not reexecuted: |
| 242 | raise EntryValidationError(24, "evidence.reexecuted must be a non-empty list") |
| 243 | if not all(isinstance(item, str) for item in reexecuted): |
| 244 | raise EntryValidationError(2, "evidence.reexecuted entries must be strings") |
| 245 | elif kind == "dispute_opened": |
| 246 | _require_non_empty_str(merged.get("subject"), "subject") |
| 247 | elif kind == "overseer_ruling": |
| 248 | if actor_role != "overseer": |
| 249 | raise EntryValidationError(23, "overseer_ruling requires actor_role=overseer") |
| 250 | _require_non_empty_str(merged.get("ruling"), "ruling") |
| 251 | elif kind == "approval_recorded": |
| 252 | if actor_role != "owner": |
| 253 | raise EntryValidationError(23, "approval_recorded requires actor_role=owner") |
| 254 | _require_non_empty_str(merged.get("artifact_sha256"), "artifact_sha256") |
| 255 | _require_non_empty_str(merged.get("bound_verdict_hash"), "bound_verdict_hash") |
| 256 | elif kind == "board_advance": |
| 257 | _require_non_empty_str(merged.get("artifact_sha256"), "artifact_sha256") |
| 258 | _require_non_empty_str(merged.get("bound_verdict_hash"), "bound_verdict_hash") |
| 259 | elif kind == "hook_check": |
| 260 | hook = merged.get("hook") |
| 261 | if hook not in {"board_done", "handoff", "register"}: |
| 262 | raise EntryValidationError(2, "hook_check.hook must be board_done|handoff|register") |
| 263 | ok = merged.get("ok") |
| 264 | if not isinstance(ok, bool): |
| 265 | raise EntryValidationError(2, "ok must be a boolean") |
| 266 | reason = merged.get("reason") |
| 267 | if reason is not None and not isinstance(reason, str): |
| 268 | raise EntryValidationError(2, "reason must be a string when present") |
| 269 | elif kind == "verification_evidence": |
| 270 | if actor_role != "verifier": |
| 271 | raise EntryValidationError(23, "verification_evidence requires actor_role=verifier") |
| 272 | _require_non_empty_str(merged.get("phase_id"), "phase_id") |
| 273 | _require_non_empty_str(merged.get("frozen_spec"), "frozen_spec") |
| 274 | round_val = merged.get("round") |
| 275 | if not isinstance(round_val, int) or round_val < 1: |
| 276 | raise EntryValidationError(2, "round must be an integer >= 1") |
| 277 | bv_verdict = merged.get("bv_verdict") |
| 278 | if bv_verdict not in BV_VERDICTS: |
| 279 | raise EntryValidationError(2, "bv_verdict must be pass|findings|blocked") |
| 280 | merged["artifacts"] = validate_verification_artifacts(merged.get("artifacts")) |
| 281 | subject_sha256 = merged.get("subject_sha256") |
| 282 | if subject_sha256 is not None: |
| 283 | _validate_sha256(subject_sha256, "subject_sha256") |
| 284 | notes = merged.get("notes") |
| 285 | if notes is not None and not isinstance(notes, str): |
| 286 | raise EntryValidationError(2, "notes must be a string when present") |
| 287 | elif kind == "independent_second_review": |
| 288 | if actor_role != "verifier": |
| 289 | raise EntryValidationError( |
| 290 | 23, "independent_second_review requires actor_role=verifier" |
| 291 | ) |
| 292 | _require_non_empty_str(merged.get("phase_id"), "phase_id") |
| 293 | _require_non_empty_str(merged.get("frozen_spec"), "frozen_spec") |
| 294 | round_val = merged.get("round") |
| 295 | if not isinstance(round_val, int) or round_val < 1: |
| 296 | raise EntryValidationError(2, "round must be an integer >= 1") |
| 297 | isr_verdict = merged.get("isr_verdict") |
| 298 | if isr_verdict not in ISR_VERDICTS: |
| 299 | raise EntryValidationError(2, "isr_verdict must be pass|findings|blocked") |
| 300 | producer_session_id = _require_non_empty_str( |
| 301 | merged.get("producer_session_id"), "producer_session_id" |
| 302 | ) |
| 303 | if actor_session == producer_session_id: |
| 304 | raise EntryValidationError( |
| 305 | 2, "actor_session_id must differ from producer_session_id" |
| 306 | ) |
| 307 | producer_agent_id = merged.get("producer_agent_id") |
| 308 | verifier_agent_id = merged.get("verifier_agent_id") |
| 309 | if producer_agent_id is not None and not isinstance(producer_agent_id, str): |
| 310 | raise EntryValidationError(2, "producer_agent_id must be a string when present") |
| 311 | if verifier_agent_id is not None and not isinstance(verifier_agent_id, str): |
| 312 | raise EntryValidationError(2, "verifier_agent_id must be a string when present") |
| 313 | if ( |
| 314 | isinstance(producer_agent_id, str) |
| 315 | and isinstance(verifier_agent_id, str) |
| 316 | and producer_agent_id == verifier_agent_id |
| 317 | ): |
| 318 | raise EntryValidationError( |
| 319 | 2, "producer_agent_id must differ from verifier_agent_id when both present" |
| 320 | ) |
| 321 | bound_hash = merged.get("bound_verification_evidence_hash") |
| 322 | if bound_hash is not None: |
| 323 | _require_non_empty_str(bound_hash, "bound_verification_evidence_hash") |
| 324 | notes = merged.get("notes") |
| 325 | if notes is not None and not isinstance(notes, str): |
| 326 | raise EntryValidationError(2, "notes must be a string when present") |
| 327 | |
| 328 | if "provenance" in merged: |
| 329 | merged["provenance"] = validate_provenance(merged["provenance"]) |
| 330 | |
| 331 | return merged |
| 332 | |
| 333 | |
| 334 | def find_passing_verdict( |
| 335 | entries: list[dict[str, Any]], |
| 336 | *, |
| 337 | artifact_sha256: str, |
| 338 | bound_verdict_hash: str, |
| 339 | ) -> bool: |
| 340 | """Return True when a passing verifier verdict matches the bound hash.""" |
| 341 | for entry in entries: |
| 342 | if entry.get("kind") != "verdict": |
| 343 | continue |
| 344 | if entry.get("actor_role") != "verifier": |
| 345 | continue |
| 346 | if entry.get("passed") is not True: |
| 347 | continue |
| 348 | if entry.get("artifact_sha256") != artifact_sha256: |
| 349 | continue |
| 350 | if entry.get("entry_hash") == bound_verdict_hash: |
| 351 | return True |
| 352 | return False |
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