40-validation/validate-mandatory-discovery-closure.py

#!/usr/bin/env python3
from __future__ import annotations
import argparse, json, os, re, sys
from pathlib import Path
from evidence_lifecycle import seal_evidence

VALIDATOR_REF = "engineering-tools:40-validation/validate-mandatory-discovery-closure.py"
TERMINAL_STATES = {"resolved", "rejected"}
NON_BLOCKING_CLASSES = {"optional-improvement", "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 parse_frontmatter(path: Path) -> dict:
    text = path.read_text(encoding="utf-8")
    if not text.startswith("---\n"):
        raise RuntimeError(f"missing front matter: {path}")
    block = text.split("---", 2)[1]
    out = {}
    for line in block.splitlines():
        if ":" not in line:
            continue
        key, value = line.split(":", 1)
        out[key.strip()] = value.strip().strip('"').strip("'")
    return out


def parse_optional_backlog(roadmap: Path) -> list[dict]:
    rows=[]
    in_section=False
    for line in roadmap.read_text(encoding="utf-8").splitlines():
        if line.strip()=="## Optionaler Backlog": in_section=True; continue
        if in_section and line.startswith("## "): break
        if in_section and line.startswith("|") and not line.startswith("|---") and "ID" not in line:
            cols=[c.strip() for c in line.strip().strip("|").split("|")]
            if len(cols)>=5:
                rows.append({"id":cols[0],"title":cols[1],"classification":cols[2],"roadmapImpact":cols[3],"status":cols[4]})
    return rows


def main() -> None:
    ap=argparse.ArgumentParser()
    ap.add_argument("gap_root", type=Path)
    ap.add_argument("repository_roots", type=Path)
    ap.add_argument("runtime_root", type=Path)
    ap.add_argument("--work-item", default="KT-0022")
    args=ap.parse_args()
    gap=args.gap_root.resolve(); roots=args.repository_roots.resolve(); runtime=args.runtime_root.resolve()
    manifests={}
    for rid in ("engineering-tools","runtime"):
        p=roots/rid/"manifest.json"
        if not p.is_file(): raise RuntimeError(f"missing repository manifest: {rid}")
        manifests[rid]=json.loads(p.read_text(encoding="utf-8"))
    discovery_dir=gap/"20-runtime-state/discoveries"
    discoveries=[]
    blockers=[]
    for path in sorted(discovery_dir.glob("DISC-*.md")):
        fm=parse_frontmatter(path)
        state=fm.get("status","").lower()
        rec={"id":fm.get("id"),"title":fm.get("title"),"status":state,"resolution":fm.get("resolution"),"followUpWorkItem":fm.get("followUpWorkItem"),"path":str(path.relative_to(gap))}
        rec["completionBlocking"] = state not in TERMINAL_STATES
        discoveries.append(rec)
        if rec["completionBlocking"]: blockers.append(rec)
    optional=parse_optional_backlog(gap/"20-runtime-state/roadmap/ROADMAP.md")
    invalid_optional=[x for x in optional if x["classification"] not in NON_BLOCKING_CLASSES or x["roadmapImpact"].lower() not in {"none","no","keine"}]
    blockers.extend(invalid_optional)
    coverage=runtime/"test/validations/KT-0017/contract-coverage-review-manifest.json"
    if not coverage.is_file(): raise RuntimeError("missing KT-0017 coverage review")
    coverage_data=json.loads(coverage.read_text(encoding="utf-8"))
    prior_blockers=coverage_data.get("completionBlockers",[])
    if prior_blockers: blockers.extend(prior_blockers)
    status="PASSED" if not blockers else "FAILED"
    material={
      "schemaVersion":"1.0",
      "reviewId":args.work_item,
      "workItem":args.work_item,
      "status":status,
      "recordedDiscoveries":discoveries,
      "discoveryCount":len(discoveries),
      "terminalDiscoveryCount":sum(1 for x in discoveries if not x["completionBlocking"]),
      "mandatoryOrCompletionBlockingDiscoveries":blockers,
      "optionalAndDeferredFindings":optional,
      "optionalBacklogCount":len(optional),
      "priorContractCoverageBlockers":prior_blockers,
      "mandatoryDiscoveryClosureMet":not blockers,
      "mandatoryRoadmapChangeRequired":False,
      "deploymentPerformed":False,
      "promotionPerformed":False
    }
    mp=runtime/"test/validations"/args.work_item/"mandatory-discovery-closure-manifest.json"
    atomic_json(mp,material)
    refs=[
      "runtime:test/validations/KT-0017/contract-coverage-review-manifest.json",
      f"runtime:test/validations/{args.work_item}/mandatory-discovery-closure-manifest.json",
      "engineering-tools:40-validation/MANDATORY_DISCOVERY_CLOSURE_REVIEW.md"
    ]
    ev=seal_evidence(material, roots, VALIDATOR_REF, refs, work_item=args.work_item, chain="DISCOVERY-CLOSURE")
    atomic_json(runtime/"test/reports/validation/mandatory-discovery-closure.json",ev)
    print(json.dumps({"status":status,"discoveries":len(discoveries),"completionBlockers":len(blockers),"optionalBacklog":len(optional)},indent=2))
    if blockers: raise SystemExit(1)

if __name__=="__main__":
    try: main()
    except Exception as exc:
        print("ERROR: "+str(exc), file=sys.stderr); raise SystemExit(1)