#!/usr/bin/env python3 from __future__ import annotations import json import sys from pathlib import Path ALLOWED_TYPES = { "PERFORMANCE", "CODE_STRUCTURE", "MODULE_STRUCTURE", "ARCHITECTURE", "DATA_ACCESS", "RELIABILITY", "CONCURRENCY", "TRANSACTION", "SECURITY", "OPERABILITY", "CONFIGURATION", "DEPENDENCY", "BUILD", "TESTABILITY", "CLEANUP", } ALLOWED_SCOPES = {"LOCAL", "MODULE", "CROSS_MODULE", "PROJECT"} ALLOWED_STATUSES = {"CANDIDATE", "READY", "BASELINING", "IN_PROGRESS", "VERIFYING", "WAITING_APPROVAL", "APPROVED", "MERGED", "REJECTED", "BLOCKED", "COMPLETE"} PERFORMANCE_BASELINE_STATUSES = {"IN_PROGRESS", "VERIFYING", "WAITING_APPROVAL", "APPROVED", "MERGED", "COMPLETE"} PERFORMANCE_COMPLETE_EVIDENCE_STATUSES = {"WAITING_APPROVAL", "APPROVED", "MERGED", "COMPLETE"} def _load(item_dir: Path) -> dict: path = item_dir / "work-item.json" if not path.exists(): raise FileNotFoundError(path) return json.loads(path.read_text(encoding="utf-8")) def _resolve(item_dir: Path, rel: str | None) -> Path | None: if not rel: return None return item_dir / rel def _comparison_fields(text: str) -> dict[str, str]: labels = ( "Same measurement command/procedure", "Same metric definitions", "Same dataset/load profile", "Environment materially equivalent", "Result", "Acceptance criteria satisfied", ) values: dict[str, str] = {} for raw in text.splitlines(): stripped = raw.strip().lstrip("- ") for label in labels: prefix = label + ":" if stripped.startswith(prefix): values[label] = stripped[len(prefix):].strip().upper() return values def verify_work_item(item_dir: Path) -> list[str]: item_dir = Path(item_dir) errors: list[str] = [] try: data = _load(item_dir) except (FileNotFoundError, json.JSONDecodeError) as exc: return [f"invalid work-item.json: {exc}"] required = ( "schemaVersion", "id", "project", "analysisRevision", "type", "scope", "target", "priority", "status", "problem", "goal", "acceptanceCriteria", "evidence", ) for key in required: if key not in data: errors.append(f"missing work item field: {key}") item_type = data.get("type") if item_type not in ALLOWED_TYPES: errors.append(f"invalid type: {item_type}") scope = data.get("scope") if scope not in ALLOWED_SCOPES: errors.append(f"invalid scope: {scope}") status = data.get("status") if status not in ALLOWED_STATUSES: errors.append(f"invalid status: {status}") if item_type == "PERFORMANCE": contract = data.get("measurementContract") if not isinstance(contract, dict): errors.append("performance baseline measurement contract is required before refactoring") else: for key in ("command", "cwd", "environment", "dataset", "metrics"): value = contract.get(key) if value in (None, "", []): errors.append(f"performance baseline measurement contract missing: {key}") environment = _resolve(item_dir, contract.get("environment")) if environment is not None and not environment.exists(): errors.append(f"performance environment evidence missing: {contract.get('environment')}") evidence = data.get("evidence") or {} baseline = evidence.get("baseline") or [] after = evidence.get("after") or [] comparison = evidence.get("comparison") def validate_metadata(phase: str, require: bool) -> dict | None: meta_path = item_dir / f"evidence/{phase}/metadata.json" if not meta_path.exists(): if require: errors.append(f"performance {phase} metadata is required") return None try: meta = json.loads(meta_path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: errors.append(f"performance {phase} metadata invalid: {exc}") return None for key in ("sourceRevision", "command", "cwd", "exitCode", "dataset", "metrics", "rawFiles"): if key not in meta or meta.get(key) in (None, "", []): if key == "exitCode" and meta.get(key) == 0: continue errors.append(f"performance {phase} metadata missing: {key}") if isinstance(contract, dict): for key in ("command", "cwd", "dataset", "metrics"): if meta.get(key) != contract.get(key): errors.append(f"performance {phase} metadata {key} differs from measurement contract") if meta.get("exitCode") not in (0,): errors.append(f"performance {phase} measurement exitCode is not zero") for rel in meta.get("rawFiles") or []: if not (meta_path.parent / rel).exists(): errors.append(f"performance {phase} metadata raw file missing: {rel}") return meta if status in PERFORMANCE_BASELINE_STATUSES: if not baseline: errors.append("performance baseline evidence is required before source changes") else: for rel in baseline: if not (item_dir / rel).exists(): errors.append(f"performance baseline evidence missing: {rel}") baseline_meta = validate_metadata("baseline", True) if baseline_meta is not None and baseline_meta.get("sourceRevision") != data.get("analysisRevision"): errors.append("performance baseline metadata sourceRevision differs from analysisRevision") if status in PERFORMANCE_COMPLETE_EVIDENCE_STATUSES: if not after: errors.append("performance after evidence is required") else: for rel in after: if not (item_dir / rel).exists(): errors.append(f"performance after evidence missing: {rel}") validate_metadata("after", True) if not comparison: errors.append("performance comparison evidence is required") elif not (item_dir / comparison).exists(): errors.append(f"performance comparison evidence missing: {comparison}") else: comparison_text = (item_dir / comparison).read_text(encoding="utf-8", errors="replace") fields = _comparison_fields(comparison_text) required_comparison_fields = ( "Same measurement command/procedure", "Same metric definitions", "Same dataset/load profile", "Environment materially equivalent", "Result", "Acceptance criteria satisfied", ) for field in required_comparison_fields: if not fields.get(field): errors.append(f"performance comparison missing field: {field}") for field in required_comparison_fields[:4]: value = fields.get(field) if value and value not in {"YES", "NO"}: errors.append(f"performance comparison invalid equivalence value for {field}: {value}") result = fields.get("Result") if result and result not in {"IMPROVED", "NEUTRAL", "REGRESSED", "INCOMPARABLE"}: errors.append(f"performance comparison invalid result: {result}") acceptance = fields.get("Acceptance criteria satisfied") if acceptance and acceptance not in {"YES", "NO"}: errors.append(f"performance comparison invalid acceptance value: {acceptance}") equivalent = all(fields.get(field) == "YES" for field in required_comparison_fields[:4]) if not equivalent and result and result != "INCOMPARABLE": errors.append("performance incomparable conditions cannot claim a comparable result") return errors def main(argv: list[str] | None = None) -> int: argv = sys.argv[1:] if argv is None else argv if len(argv) != 1: print("usage: verify_refactor_work_item.py ") return 2 errors = verify_work_item(Path(argv[0])) if errors: print("REFACTOR WORK ITEM VERIFICATION: FAIL") for error in errors: print(f"- {error}") return 1 print("REFACTOR WORK ITEM VERIFICATION: PASS") return 0 if __name__ == "__main__": raise SystemExit(main())