|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Validate deterministic agent artifact bundle payloads.""" |
| 3 | + |
| 4 | +from __future__ import annotations |
| 5 | + |
| 6 | +import argparse |
| 7 | +import json |
| 8 | +import re |
| 9 | +import sys |
| 10 | +from pathlib import Path |
| 11 | +from typing import Any |
| 12 | + |
| 13 | +REPO_ROOT = Path(__file__).resolve().parents[1] |
| 14 | +DEFAULT_BUNDLE_PATH = REPO_ROOT / "artifacts" / "agent_artifact_bundle_example.json" |
| 15 | + |
| 16 | +REQUIRED_BUNDLE_FIELDS = ( |
| 17 | + "ok", |
| 18 | + "result", |
| 19 | + "branch", |
| 20 | + "changed_files", |
| 21 | + "safe_pr_gate", |
| 22 | + "validation_evidence", |
| 23 | +) |
| 24 | +REQUIRED_SAFE_GATE_FIELDS = ( |
| 25 | + "allow_dirty", |
| 26 | + "allowed_prefixes", |
| 27 | + "branch", |
| 28 | + "changed_paths", |
| 29 | + "ok", |
| 30 | + "problems", |
| 31 | + "result", |
| 32 | + "status_short", |
| 33 | +) |
| 34 | +DISALLOWED_TIME_KEYS = { |
| 35 | + "timestamp", |
| 36 | + "generated_at", |
| 37 | + "created_at", |
| 38 | + "updated_at", |
| 39 | + "completed_at", |
| 40 | + "requested_at", |
| 41 | +} |
| 42 | +DISALLOWED_RANDOM_ID_KEYS = { |
| 43 | + "generated_id", |
| 44 | + "random_id", |
| 45 | + "request_id", |
| 46 | + "run_id", |
| 47 | + "uuid", |
| 48 | +} |
| 49 | +UUID_PATTERN = re.compile( |
| 50 | + r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$" |
| 51 | +) |
| 52 | + |
| 53 | + |
| 54 | +def _relative(path: Path) -> str: |
| 55 | + try: |
| 56 | + return path.resolve().relative_to(REPO_ROOT).as_posix() |
| 57 | + except ValueError: |
| 58 | + return path.as_posix() |
| 59 | + |
| 60 | + |
| 61 | +def _load_json_object(path: Path) -> dict[str, Any]: |
| 62 | + try: |
| 63 | + payload = json.loads(path.read_text(encoding="utf-8")) |
| 64 | + except FileNotFoundError as exc: |
| 65 | + raise RuntimeError(f"missing bundle file: {_relative(path)}") from exc |
| 66 | + except json.JSONDecodeError as exc: |
| 67 | + raise RuntimeError(f"invalid JSON in bundle file: {_relative(path)}") from exc |
| 68 | + if not isinstance(payload, dict): |
| 69 | + raise RuntimeError(f"bundle file must contain a JSON object: {_relative(path)}") |
| 70 | + return payload |
| 71 | + |
| 72 | + |
| 73 | +def _is_string_list(value: object) -> bool: |
| 74 | + return isinstance(value, list) and all(isinstance(item, str) for item in value) |
| 75 | + |
| 76 | + |
| 77 | +def _expected_result(ok: bool) -> str: |
| 78 | + return "PASS" if ok else "FAIL" |
| 79 | + |
| 80 | + |
| 81 | +def _bundle_from_payload(payload: dict[str, Any]) -> tuple[dict[str, Any] | None, list[str]]: |
| 82 | + bundle = payload.get("bundle", payload) |
| 83 | + if isinstance(bundle, dict): |
| 84 | + return bundle, [] |
| 85 | + return None, ["bundle must be a JSON object"] |
| 86 | + |
| 87 | + |
| 88 | +def _scan_for_nondeterministic_fields(value: object, path: str = "$") -> list[str]: |
| 89 | + issues: list[str] = [] |
| 90 | + if isinstance(value, dict): |
| 91 | + for key, child in value.items(): |
| 92 | + key_path = f"{path}.{key}" |
| 93 | + normalized = key.lower() |
| 94 | + if normalized in DISALLOWED_TIME_KEYS: |
| 95 | + issues.append(f"{key_path}: timestamp-like field is not allowed") |
| 96 | + if normalized in DISALLOWED_RANDOM_ID_KEYS: |
| 97 | + issues.append(f"{key_path}: random-looking generated id field is not allowed") |
| 98 | + issues.extend(_scan_for_nondeterministic_fields(child, key_path)) |
| 99 | + elif isinstance(value, list): |
| 100 | + for index, child in enumerate(value): |
| 101 | + issues.extend(_scan_for_nondeterministic_fields(child, f"{path}[{index}]")) |
| 102 | + elif isinstance(value, str) and UUID_PATTERN.fullmatch(value): |
| 103 | + issues.append(f"{path}: UUID-like value is not allowed") |
| 104 | + return issues |
| 105 | + |
| 106 | + |
| 107 | +def validate_bundle_payload(payload: dict[str, Any]) -> dict[str, Any]: |
| 108 | + issues: list[str] = [] |
| 109 | + issues.extend(_scan_for_nondeterministic_fields(payload)) |
| 110 | + |
| 111 | + bundle, bundle_issues = _bundle_from_payload(payload) |
| 112 | + issues.extend(bundle_issues) |
| 113 | + if bundle is None: |
| 114 | + return {"issues": sorted(issues), "ok": False, "result": "FAIL"} |
| 115 | + |
| 116 | + for field in REQUIRED_BUNDLE_FIELDS: |
| 117 | + if field not in bundle: |
| 118 | + issues.append(f"bundle missing required field: {field}") |
| 119 | + |
| 120 | + ok = bundle.get("ok") |
| 121 | + result = bundle.get("result") |
| 122 | + branch = bundle.get("branch") |
| 123 | + changed_files = bundle.get("changed_files") |
| 124 | + safe_pr_gate = bundle.get("safe_pr_gate") |
| 125 | + validation_evidence = bundle.get("validation_evidence") |
| 126 | + |
| 127 | + if not isinstance(ok, bool): |
| 128 | + issues.append("bundle.ok must be a boolean") |
| 129 | + if not isinstance(result, str): |
| 130 | + issues.append("bundle.result must be a string") |
| 131 | + if isinstance(ok, bool) and isinstance(result, str) and result != _expected_result(ok): |
| 132 | + issues.append("bundle.result must match bundle.ok") |
| 133 | + if not isinstance(branch, str): |
| 134 | + issues.append("bundle.branch must be a string") |
| 135 | + if not _is_string_list(changed_files): |
| 136 | + issues.append("bundle.changed_files must be a list of strings") |
| 137 | + |
| 138 | + if not isinstance(safe_pr_gate, dict): |
| 139 | + issues.append("bundle.safe_pr_gate must be a JSON object") |
| 140 | + else: |
| 141 | + issues.extend(_validate_safe_pr_gate(safe_pr_gate, ok)) |
| 142 | + |
| 143 | + if not isinstance(validation_evidence, list): |
| 144 | + issues.append("bundle.validation_evidence must be a list") |
| 145 | + else: |
| 146 | + issues.extend(_validate_validation_evidence(validation_evidence)) |
| 147 | + |
| 148 | + return {"issues": sorted(issues), "ok": not issues, "result": "PASS" if not issues else "FAIL"} |
| 149 | + |
| 150 | + |
| 151 | +def _validate_safe_pr_gate(safe_pr_gate: dict[str, Any], bundle_ok: object) -> list[str]: |
| 152 | + issues: list[str] = [] |
| 153 | + for field in REQUIRED_SAFE_GATE_FIELDS: |
| 154 | + if field not in safe_pr_gate: |
| 155 | + issues.append(f"bundle.safe_pr_gate missing required field: {field}") |
| 156 | + |
| 157 | + gate_ok = safe_pr_gate.get("ok") |
| 158 | + gate_result = safe_pr_gate.get("result") |
| 159 | + if not isinstance(gate_ok, bool): |
| 160 | + issues.append("bundle.safe_pr_gate.ok must be a boolean") |
| 161 | + if not isinstance(gate_result, str): |
| 162 | + issues.append("bundle.safe_pr_gate.result must be a string") |
| 163 | + if isinstance(gate_ok, bool) and isinstance(gate_result, str) and gate_result != _expected_result(gate_ok): |
| 164 | + issues.append("bundle.safe_pr_gate.result must match bundle.safe_pr_gate.ok") |
| 165 | + if isinstance(bundle_ok, bool) and isinstance(gate_ok, bool) and bundle_ok != gate_ok: |
| 166 | + issues.append("bundle.ok must match bundle.safe_pr_gate.ok") |
| 167 | + if not isinstance(safe_pr_gate.get("allow_dirty"), bool): |
| 168 | + issues.append("bundle.safe_pr_gate.allow_dirty must be a boolean") |
| 169 | + if not isinstance(safe_pr_gate.get("branch"), str): |
| 170 | + issues.append("bundle.safe_pr_gate.branch must be a string") |
| 171 | + for field in ("allowed_prefixes", "changed_paths", "problems", "status_short"): |
| 172 | + if not _is_string_list(safe_pr_gate.get(field)): |
| 173 | + issues.append(f"bundle.safe_pr_gate.{field} must be a list of strings") |
| 174 | + return issues |
| 175 | + |
| 176 | + |
| 177 | +def _validate_validation_evidence(validation_evidence: list[object]) -> list[str]: |
| 178 | + issues: list[str] = [] |
| 179 | + for index, entry in enumerate(validation_evidence): |
| 180 | + if not isinstance(entry, dict): |
| 181 | + issues.append(f"bundle.validation_evidence[{index}] must be a JSON object") |
| 182 | + continue |
| 183 | + if not isinstance(entry.get("command"), str): |
| 184 | + issues.append(f"bundle.validation_evidence[{index}].command must be a string") |
| 185 | + if not isinstance(entry.get("result"), str): |
| 186 | + issues.append(f"bundle.validation_evidence[{index}].result must be a string") |
| 187 | + return issues |
| 188 | + |
| 189 | + |
| 190 | +def validate_bundle_file(path: Path) -> dict[str, Any]: |
| 191 | + payload = _load_json_object(path) |
| 192 | + result = validate_bundle_payload(payload) |
| 193 | + return { |
| 194 | + "bundle": _relative(path), |
| 195 | + "issues": result["issues"], |
| 196 | + "ok": result["ok"], |
| 197 | + "result": result["result"], |
| 198 | + } |
| 199 | + |
| 200 | + |
| 201 | +def _parse_args(argv: list[str]) -> argparse.Namespace: |
| 202 | + parser = argparse.ArgumentParser(description="Validate a deterministic agent artifact bundle.") |
| 203 | + parser.add_argument("--bundle", type=Path, default=DEFAULT_BUNDLE_PATH, help="Bundle JSON path.") |
| 204 | + return parser.parse_args(argv) |
| 205 | + |
| 206 | + |
| 207 | +def main(argv: list[str] | None = None) -> int: |
| 208 | + args = _parse_args(sys.argv[1:] if argv is None else argv) |
| 209 | + try: |
| 210 | + result = validate_bundle_file(args.bundle) |
| 211 | + except RuntimeError as exc: |
| 212 | + result = { |
| 213 | + "error": { |
| 214 | + "message": str(exc), |
| 215 | + "type": exc.__class__.__name__, |
| 216 | + }, |
| 217 | + "ok": False, |
| 218 | + "result": "ERROR", |
| 219 | + } |
| 220 | + sys.stdout.write(json.dumps(result, indent=2, sort_keys=True) + "\n") |
| 221 | + return 0 if result["ok"] else 1 |
| 222 | + |
| 223 | + |
| 224 | +if __name__ == "__main__": |
| 225 | + raise SystemExit(main()) |
0 commit comments