40-validation/evidence_retention.py

#!/usr/bin/env python3
"""Freshness-gated evidence promotion and append-only retention history."""
from __future__ import annotations
import json, shutil
from datetime import datetime, timezone
from pathlib import Path
from evidence_lifecycle import validate_evidence, write_evidence


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _load_index(path: Path) -> dict:
    if not path.exists():
        return {"schemaVersion": "1", "entries": []}
    data=json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(data.get("entries"), list):
        raise ValueError("retention index requires entries array")
    return data


def _record_path(path: Path, base: Path) -> str:
    try:
        return str(path.resolve().relative_to(base.resolve()))
    except ValueError:
        return path.name


def _successful(data: dict) -> bool:
    result=str(data.get("result", data.get("overall", ""))).lower()
    exit_code=data.get("exitCode", 0)
    return result in {"passed","success","successful","ok"} and exit_code in {0,None}


def evidence_identity(data: dict) -> str:
    value=data.get("evidenceId") or data.get("evidenceType")
    if not isinstance(value,str) or not value:
        raise ValueError("evidence requires evidenceId or evidenceType")
    return value


def promote_evidence(roots: Path, source: Path, retained_root: Path, index_path: Path) -> dict:
    data=json.loads(source.read_text(encoding="utf-8"))
    if data.get("freshness") != "fresh" or not isinstance(data.get("provenance"),dict):
        raise ValueError("promotion requires automatically sealed evidence")
    errors=validate_evidence(roots,source)
    if errors:
        raise ValueError("evidence is not fresh: " + "; ".join(errors))
    if not _successful(data):
        raise ValueError("failed evidence cannot be promoted")
    identity=evidence_identity(data)
    index=_load_index(index_path)
    stamp=datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    target=retained_root/identity/'active'/f'{stamp}.json'
    target.parent.mkdir(parents=True,exist_ok=True)
    temp=target.with_suffix('.json.new'); shutil.copy2(source,temp); temp.replace(target)
    now=_now()
    previous=None
    for entry in index['entries']:
        if entry.get('evidenceId')==identity and entry.get('retentionClass')=='active':
            entry['retentionClass']='superseded'; entry['supersededAt']=now; entry['supersededBy']=str(target.relative_to(index_path.parent))
            previous=entry.get('path')
    record={"evidenceId":identity,"retentionClass":"active","path":str(target.relative_to(index_path.parent)),"promotedAt":now,"source":_record_path(source,index_path.parent),"workItem":data.get("workItem"),"chain":data.get("chain"),"replaces":previous}
    index['entries'].append(record); write_evidence(index_path,index)
    return record


def retain_failed(source: Path, retained_root: Path, index_path: Path) -> dict:
    data=json.loads(source.read_text(encoding="utf-8")); identity=evidence_identity(data)
    if _successful(data): raise ValueError("successful evidence must use promotion flow")
    stamp=datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    target=retained_root/identity/'failed'/f'{stamp}.json'; target.parent.mkdir(parents=True,exist_ok=True)
    temp=target.with_suffix('.json.new'); shutil.copy2(source,temp); temp.replace(target)
    index=_load_index(index_path)
    record={"evidenceId":identity,"retentionClass":"failed","path":str(target.relative_to(index_path.parent)),"retainedAt":_now(),"source":_record_path(source,index_path.parent),"workItem":data.get("workItem"),"chain":data.get("chain"),"failurePhase":data.get("failurePhase"),"rollback":data.get("rollback")}
    index['entries'].append(record); write_evidence(index_path,index); return record