40-validation/validate-operational-recovery.py

#!/usr/bin/env python3
"""Validate operational restart, recovery and operator handover for GAP."""
from __future__ import annotations
import argparse, hashlib, json, os, re
from pathlib import Path
from evidence_lifecycle import seal_evidence

VALIDATOR_REF = "engineering-tools:40-validation/validate-operational-recovery.py"
DOC_REF = "engineering-tools:40-validation/OPERATIONAL_RUNBOOK_RECOVERY.md"
REQUIRED_SECTIONS = [
    "## Kanonische Eingaben", "## Operator-Ablauf", "## Checkpoint-Zustände",
    "## Unterbrechung und Wiederanlauf", "## Recovery-Regeln", "## Operator-Handover",
]
ALLOWED_STATES = {"prepared", "executing", "validated", "handoff-ready", "failed"}
RECOVERABLE_STATES = {"prepared", "executing", "validated", "handoff-ready"}

def sha256_file(path: Path) -> str:
    h=hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda:f.read(1024*1024), b""): h.update(chunk)
    return h.hexdigest()

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 load_json(path: Path) -> dict:
    return json.loads(path.read_text(encoding="utf-8"))

def validate_runbook(path: Path) -> list[str]:
    errors=[]
    if not path.is_file(): return [f"runbook missing: {path}"]
    text=path.read_text(encoding="utf-8")
    for section in REQUIRED_SECTIONS:
        if section not in text: errors.append(f"runbook section missing: {section}")
    for token in ("gap-000.zip", "source_bundle.zip", "target_bundle.zip", "handoff-ready", "Source Bundle"):
        if token not in text: errors.append(f"runbook contract token missing: {token}")
    return errors

def recovery_decision(checkpoint: dict, source_sha256: str, gap_version: str, active_work_item: str) -> tuple[str,list[str]]:
    errors=[]
    state=checkpoint.get("state")
    if state not in ALLOWED_STATES: errors.append(f"invalid checkpoint state: {state}")
    if checkpoint.get("sourceBundleSha256") != source_sha256: errors.append("Source Bundle SHA-256 differs from checkpoint")
    if checkpoint.get("gapVersion") != gap_version: errors.append("GAP version differs from checkpoint")
    if checkpoint.get("workItem") != active_work_item: errors.append("active work item differs from checkpoint")
    if errors: return "BLOCK",errors
    if state == "failed": return "NEW_ATTEMPT",[]
    if state in {"prepared","executing"}: return "RESTART_FROM_INPUTS",[]
    if state == "validated": return "REVALIDATE_AND_PACKAGE",[]
    if state == "handoff-ready": return "VERIFY_DELIVERABLE_HASHES",[]
    return "BLOCK",["unhandled checkpoint state"]

def phase4_complete(roadmap_text: str) -> tuple[bool,list[str]]:
    errors=[]
    required={"KT-0018":"complete","KT-0019":"complete","KT-0020":"complete","KT-0021":"complete"}
    for wid,status in required.items():
        pattern=rf"\|\s*{wid}\s*\|[^\n]*\|\s*{status}\s*\|"
        if not re.search(pattern,roadmap_text): errors.append(f"Phase 4 status not complete: {wid}")
    if "**Status:** complete — 4 / 4" not in roadmap_text: errors.append("Phase 4 aggregate status is not complete — 4 / 4")
    return not errors,errors

def validate(gap_root:Path,runtime_root:Path,source_bundle:Path,checkpoint_path:Path,runbook_path:Path,work_item:str="KT-0021")->tuple[dict,list[str]]:
    errors=validate_runbook(runbook_path)
    manifest=load_json(gap_root/"manifest.json")
    active=manifest.get("activeWorkItem",{}).get("id")
    gap_version=str(manifest.get("repository",{}).get("version"))
    source_sha=sha256_file(source_bundle)
    checkpoint=load_json(checkpoint_path)
    decision,decision_errors=recovery_decision(checkpoint,source_sha,str(checkpoint.get("gapVersion")),work_item)
    errors.extend(decision_errors)
    complete,phase_errors=phase4_complete((gap_root/"20-runtime-state/roadmap/ROADMAP.md").read_text(encoding="utf-8"))
    errors.extend(phase_errors)
    result={
      "validationId":"operational-runbook-recovery",
      "status":"PASSED" if not errors else "FAILED",
      "gapVersion":gap_version,
      "activeWorkItem":active,
      "validatedWorkItem":work_item,
      "sourceBundleSha256":source_sha,
      "checkpoint":checkpoint,
      "recoveryDecision":decision,
      "runbookComplete":not validate_runbook(runbook_path),
      "phase4CompletionCriteriaMet":complete,
      "chatMemoryRequired":False,
      "deploymentPerformed":False,
      "promotionPerformed":False,
      "errors":errors,
    }
    return result,errors

def main()->int:
    p=argparse.ArgumentParser()
    p.add_argument("gap_root",type=Path); p.add_argument("repository_roots",type=Path); p.add_argument("runtime_root",type=Path)
    p.add_argument("--source-bundle",type=Path,required=True); p.add_argument("--checkpoint",type=Path,required=True)
    p.add_argument("--runbook",type=Path); p.add_argument("--work-item",default="KT-0021")
    a=p.parse_args()
    runbook=a.runbook or Path(__file__).with_name("OPERATIONAL_RUNBOOK_RECOVERY.md")
    result,errors=validate(a.gap_root.resolve(),a.runtime_root.resolve(),a.source_bundle.resolve(),a.checkpoint.resolve(),runbook.resolve(),a.work_item)
    mat=a.runtime_root.resolve()/"test/validations/KT-0021/operational-recovery-manifest.json"
    ev=a.runtime_root.resolve()/"test/reports/validation/operational-runbook-recovery.json"
    atomic_json(mat,result)
    sealed=seal_evidence(result,a.repository_roots.resolve(),VALIDATOR_REF,[DOC_REF,"runtime:test/validations/KT-0021/operational-recovery-manifest.json"],work_item=a.work_item,chain="OPERATIONAL-RECOVERY")
    atomic_json(ev,sealed)
    if errors:
        for e in errors: print("ERROR:",e)
        return 1
    print(f"Operational recovery PASSED: decision={result['recoveryDecision']} phase4=complete")
    return 0
if __name__=="__main__": raise SystemExit(main())