40-validation/validate-work-item-admission.py

#!/usr/bin/env python3
"""Validate controlled admission of mandatory roadmap work."""
from __future__ import annotations

import argparse
import json
import os
import re
from pathlib import Path

from evidence_lifecycle import seal_evidence

VALIDATOR_REF = "engineering-tools:40-validation/validate-work-item-admission.py"
DOC_REF = "engineering-tools:40-validation/WORK_ITEM_ADMISSION_CHANGE_CONTROL.md"
ROADMAP_REF = "gap-000:20-runtime-state/roadmap/ROADMAP.md"
GAP_MANIFEST_REF = "gap-000:manifest.json"
ALLOWED_NON_MANDATORY = {"optional-improvement", "discovery", "deferred"}


def atomic_json(path: Path, data: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
    os.replace(tmp, path)


def read_json(path: Path) -> dict:
    value = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(value, dict):
        raise ValueError(f"JSON object required: {path}")
    return value


def current_baseline(gap_root: Path) -> dict:
    manifest = read_json(gap_root / "manifest.json")
    roadmap = manifest.get("roadmap", {})
    return {
        "baseline": roadmap.get("baseline"),
        "baselineVersion": str(roadmap.get("baselineVersion")),
        "mandatoryTotal": roadmap.get("mandatoryTotal"),
    }


def optional_classifications(gap_root: Path) -> list[dict]:
    text = (gap_root / "20-runtime-state/roadmap/ROADMAP.md").read_text(encoding="utf-8")
    start = text.find("\n## Optionaler Backlog")
    if start < 0:
        return []
    rows = []
    pattern = r"^\|\s*(OPT-[^|]+?)\s*\|\s*([^|]+?)\s*\|\s*([^|]+?)\s*\|\s*([^|]+?)\s*\|\s*([^|]+?)\s*\|$"
    for row in re.finditer(pattern, text[start:], re.MULTILINE):
        if row.group(1).strip() == "ID":
            continue
        rows.append({
            "id": row.group(1).strip(),
            "classification": row.group(3).strip(),
            "roadmapImpact": row.group(4).strip(),
        })
    return rows


def validate_admission(previous: dict, current: dict, change: dict | None, optional: list[dict]) -> tuple[dict, list[str]]:
    errors: list[str] = []
    checks: list[str] = []
    changed = previous != current

    for key in ("baseline", "baselineVersion", "mandatoryTotal"):
        if key not in previous or key not in current:
            errors.append(f"missing baseline field: {key}")
    if not isinstance(current.get("mandatoryTotal"), int) or current.get("mandatoryTotal", 0) < 1:
        errors.append("current mandatoryTotal must be a positive integer")

    for entry in optional:
        if entry.get("classification") not in ALLOWED_NON_MANDATORY:
            errors.append(f"optional backlog entry has invalid classification: {entry.get('id')}")
        if entry.get("roadmapImpact") != "none":
            errors.append(f"optional backlog entry changes mandatory scope: {entry.get('id')}")
    if not errors:
        checks.append("non-mandatory findings remain outside mandatory scope")

    admitted = False
    if not changed:
        if change is not None:
            errors.append("change record supplied although baseline did not change")
        else:
            checks.append("unchanged baseline requires no admission")
    else:
        if change is None:
            errors.append("mandatory baseline changed without an explicit change record")
        else:
            required = ("changeId", "classification", "approved", "reason", "completionCriteriaImpact", "previous", "proposed")
            for key in required:
                if key not in change:
                    errors.append(f"change record missing field: {key}")
            if change.get("classification") != "mandatory-roadmap-change":
                errors.append("baseline change requires classification mandatory-roadmap-change")
            if change.get("approved") is not True:
                errors.append("mandatory roadmap change is not approved")
            if not str(change.get("reason", "")).strip():
                errors.append("mandatory roadmap change requires a reason")
            if not str(change.get("completionCriteriaImpact", "")).strip():
                errors.append("mandatory roadmap change requires completion-criteria impact")
            if change.get("previous") != previous:
                errors.append("change-record previous baseline does not match supplied snapshot")
            if change.get("proposed") != current:
                errors.append("change-record proposed baseline does not match current GAP baseline")
            version_changed = previous.get("baselineVersion") != current.get("baselineVersion")
            total_changed = previous.get("mandatoryTotal") != current.get("mandatoryTotal")
            if version_changed != total_changed or not (version_changed and total_changed):
                errors.append("baselineVersion and mandatoryTotal must change together")
            admitted = not errors
            if admitted:
                checks.append("approved mandatory roadmap change admitted")

    result = {
        "validationId": "work-item-admission-change-control",
        "status": "PASSED" if not errors else "FAILED",
        "previousBaseline": previous,
        "currentBaseline": current,
        "baselineChanged": changed,
        "changeRecordProvided": change is not None,
        "mandatoryWorkAdmitted": admitted,
        "optionalFindingsChecked": len(optional),
        "checksPassed": checks,
        "errors": errors,
        "roadmapChangedByValidator": False,
        "deploymentPerformed": False,
        "promotionPerformed": False,
    }
    return result, errors


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("gap_root", type=Path)
    parser.add_argument("repository_roots", type=Path)
    parser.add_argument("runtime_root", type=Path)
    parser.add_argument("--previous-baseline", required=True, type=Path)
    parser.add_argument("--change-record", type=Path)
    parser.add_argument("--work-item", default="KT-0019")
    args = parser.parse_args()

    gap_root = args.gap_root.resolve()
    runtime_root = args.runtime_root.resolve()
    previous = read_json(args.previous_baseline.resolve())
    current = current_baseline(gap_root)
    change = read_json(args.change_record.resolve()) if args.change_record else None
    optional = optional_classifications(gap_root)
    result, errors = validate_admission(previous, current, change, optional)

    materialization = runtime_root / "test/validations/KT-0019/admission-control-manifest.json"
    evidence = runtime_root / "test/reports/validation/work-item-admission-change-control.json"
    atomic_json(materialization, result)
    refs = [ROADMAP_REF, GAP_MANIFEST_REF, DOC_REF, "runtime:test/validations/KT-0019/admission-control-manifest.json"]
    sealed = seal_evidence(result, args.repository_roots.resolve(), VALIDATOR_REF, refs, work_item=args.work_item, chain="ROADMAP-MVP-001")
    atomic_json(evidence, sealed)

    if errors:
        for error in errors:
            print(f"ERROR: {error}")
        return 1
    print("Work-item admission control PASSED")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())