40-validation/validate-execution-lifecycle.py

#!/usr/bin/env python3
"""Validate the complete evidence execution lifecycle against retained runtime state."""
from __future__ import annotations
import argparse, json, sys
from pathlib import Path
from evidence_lifecycle import seal_evidence, validate_evidence, write_evidence

ALLOWED_CLASSES={"active","superseded","failed"}

def load(path: Path) -> dict:
    return json.loads(path.read_text(encoding="utf-8"))

def validate_index(roots: Path, runtime: Path, required_work_item: str = "KT-0012") -> tuple[list[str], dict]:
    errors=[]
    index_path=runtime/'test/reports/evidence-retention-index.json'
    index=load(index_path)
    entries=index.get('entries')
    if not isinstance(entries,list):
        return ['retention index requires entries array'], {}
    active_by_id={}
    records={}
    for entry in entries:
        identity=entry.get('evidenceId'); cls=entry.get('retentionClass'); rel=entry.get('path')
        if not identity or cls not in ALLOWED_CLASSES or not rel:
            errors.append(f'invalid retention entry: {entry}'); continue
        target=(runtime/'test/reports'/rel).resolve()
        reports=(runtime/'test/reports').resolve()
        if reports not in target.parents:
            errors.append(f'unsafe retained path: {rel}'); continue
        if not target.is_file():
            errors.append(f'missing retained evidence: {rel}'); continue
        data=load(target)
        freshness=validate_evidence(roots,target)
        if cls=='active' and freshness:
            errors.extend(f'{rel}: {e}' for e in freshness)
        records[rel]=(entry,data)
        if cls=='active': active_by_id.setdefault(identity,[]).append(rel)
        if cls=='failed':
            if str(data.get('result','')).lower()!='failed': errors.append(f'{rel}: failed class requires failed result')
            rollback=data.get('rollback',{})
            if rollback.get('attempted') is not True or rollback.get('status')!='SUCCEEDED':
                errors.append(f'{rel}: controlled failure requires successful rollback')
        if cls=='superseded' and not entry.get('supersededBy'):
            errors.append(f'{rel}: superseded entry requires supersededBy')
    for identity, paths in active_by_id.items():
        if len(paths)!=1: errors.append(f'{identity}: expected exactly one active entry, got {len(paths)}')
    for rel,(entry,data) in records.items():
        if entry.get('retentionClass')=='superseded':
            successor=entry.get('supersededBy')
            if successor not in records: errors.append(f'{rel}: missing superseding evidence {successor}')
            elif records[successor][0].get('retentionClass')!='active': errors.append(f'{rel}: successor is not active')
    success=[(r,e,d) for r,(e,d) in records.items() if e.get('retentionClass')=='active' and str(d.get('result','')).lower() in {'passed','success','successful','ok'}]
    failed=[(r,e,d) for r,(e,d) in records.items() if e.get('retentionClass')=='failed' and str(d.get('result','')).lower()=='failed' and d.get('workItem')==required_work_item and not validate_evidence(roots,runtime/'test/reports'/r)]
    superseded=[r for r,(e,d) in records.items() if e.get('retentionClass')=='superseded']
    if not success: errors.append('no successful active lifecycle evidence')
    if not failed: errors.append('no controlled failed lifecycle evidence')
    if not superseded: errors.append('no active-to-superseded transition evidence')
    summary={
      'activeSuccessCount':len(success),'failedRollbackCount':len(failed),
      'supersededCount':len(superseded),'entryCount':len(entries),
      'successEvidence':[r for r,_,_ in success],
      'failureEvidence':[r for r,_,_ in failed],
      'supersededEvidence':superseded,
    }
    return errors,summary

def main() -> int:
    p=argparse.ArgumentParser()
    p.add_argument('repository_roots',type=Path)
    p.add_argument('runtime_repository',type=Path)
    p.add_argument('--evidence-output',type=Path)
    p.add_argument('--work-item',default='KT-0012')
    p.add_argument('--chain',default='RC-0001')
    a=p.parse_args()
    errors,summary=validate_index(a.repository_roots,a.runtime_repository,a.work_item)
    if errors:
        for error in errors: print('ERROR:',error,file=sys.stderr)
        return 1
    print('OK: execution lifecycle end-to-end validation passed')
    if a.evidence_output:
        index_ref='runtime:test/reports/evidence-retention-index.json'
        payload=seal_evidence({
          'evidenceId':'execution-lifecycle-end-to-end-validation',
          'evidenceType':'execution-lifecycle-end-to-end-validation',
          'result':'passed','exitCode':0,'summary':summary,
        },a.repository_roots,
          'engineering-tools:40-validation/validate-execution-lifecycle.py',
          [index_ref],work_item=a.work_item,chain=a.chain)
        write_evidence(a.evidence_output,payload)
    return 0
if __name__=='__main__': raise SystemExit(main())