40-validation/validate-roadmap-progress.py

#!/usr/bin/env python3
"""Validate GAP roadmap progress, phase state and active-work-item consistency."""
from __future__ import annotations

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

from evidence_lifecycle import seal_evidence

VALIDATOR_REF = "engineering-tools:40-validation/validate-roadmap-progress.py"
DOC_REF = "engineering-tools:40-validation/ROADMAP_PROGRESS_VALIDATION.md"
ROADMAP_REF = "gap-000:20-runtime-state/roadmap/ROADMAP.md"
DASHBOARD_REF = "gap-000:20-runtime-state/status/dashboard.md"
GAP_MANIFEST_REF = "gap-000:manifest.json"


@dataclass(frozen=True)
class WorkItem:
    id: str
    title: str
    status: str
    phase: str


@dataclass(frozen=True)
class Phase:
    title: str
    declared_status: str
    declared_completed: int
    declared_total: int
    items: tuple[WorkItem, ...]


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


def front_matter(path: Path) -> dict[str, object]:
    text = path.read_text(encoding="utf-8")
    if not text.startswith("---\n"):
        raise ValueError(f"missing front matter: {path}")
    end = text.find("\n---\n", 4)
    if end < 0:
        raise ValueError(f"unterminated front matter: {path}")
    result: dict[str, object] = {}
    current_list: str | None = None
    for raw in text[4:end].splitlines():
        if not raw.strip() or raw.lstrip().startswith("#"):
            continue
        if raw.startswith("  - ") and current_list:
            result.setdefault(current_list, [])
            assert isinstance(result[current_list], list)
            result[current_list].append(raw[4:].strip())
            continue
        match = re.match(r"^([A-Za-z][A-Za-z0-9_]*):\s*(.*)$", raw)
        if not match:
            continue
        key, value = match.groups()
        current_list = key if value == "" else None
        if value == "":
            result[key] = []
        elif value.lower() in {"true", "false"}:
            result[key] = value.lower() == "true"
        else:
            result[key] = value.strip('"\'')
    return result


def parse_roadmap(path: Path) -> tuple[dict, list[Phase], list[dict]]:
    text = path.read_text(encoding="utf-8")
    summary_patterns = {
        "baseline": r"Baseline:\s*([^\n]+)",
        "mandatoryTotal": r"Mandatory work items:\s*(\d+)",
        "mandatoryCompleted": r"Completed:\s*(\d+)",
        "mandatoryRemaining": r"Remaining:\s*(\d+)",
        "progressPercent": r"Progress:\s*(\d+)\s*%",
        "currentPhase": r"Current phase:\s*([^\n]+)",
        "nextMandatoryWorkItem": r"Next mandatory work item:\s*([^\n]+)",
        "optionalBacklogCount": r"Optional backlog:\s*(\d+)",
    }
    summary: dict[str, object] = {}
    for key, pattern in summary_patterns.items():
        match = re.search(pattern, text)
        if not match:
            raise ValueError(f"roadmap summary field missing: {key}")
        value: object = match.group(1).strip()
        if key in {"mandatoryTotal", "mandatoryCompleted", "mandatoryRemaining", "progressPercent", "optionalBacklogCount"}:
            value = int(value)
        summary[key] = value

    phase_matches = list(re.finditer(r"^## (Phase \d+ — .+)$", text, re.MULTILINE))
    phases: list[Phase] = []
    for index, match in enumerate(phase_matches):
        start = match.end()
        end = phase_matches[index + 1].start() if index + 1 < len(phase_matches) else text.find("\n## Optionaler Backlog", start)
        if end < 0:
            end = len(text)
        block = text[start:end]
        status_match = re.search(r"\*\*Status:\*\*\s*([a-z-]+)\s*—\s*(\d+)\s*/\s*(\d+)", block)
        if not status_match:
            raise ValueError(f"phase status missing: {match.group(1)}")
        items: list[WorkItem] = []
        for row in re.finditer(r"^\|\s*(KT-\d{4})\s*\|\s*([^|]+?)\s*\|\s*([a-z-]+)\s*\|$", block, re.MULTILINE):
            items.append(WorkItem(row.group(1), row.group(2).strip(), row.group(3), match.group(1)))
        phases.append(Phase(match.group(1), status_match.group(1), int(status_match.group(2)), int(status_match.group(3)), tuple(items)))

    optional_block_start = text.find("\n## Optionaler Backlog")
    optional: list[dict] = []
    if optional_block_start >= 0:
        for row in re.finditer(r"^\|\s*(OPT-[^|]+?)\s*\|\s*([^|]+?)\s*\|\s*([^|]+?)\s*\|\s*([^|]+?)\s*\|\s*([^|]+?)\s*\|$", text[optional_block_start:], re.MULTILINE):
            if row.group(1).strip() == "ID":
                continue
            optional.append({"id": row.group(1).strip(), "title": row.group(2).strip(), "classification": row.group(3).strip(), "roadmapImpact": row.group(4).strip(), "status": row.group(5).strip()})
    return summary, phases, optional


def derive(phases: list[Phase], optional: list[dict]) -> dict:
    items = [item for phase in phases for item in phase.items]
    completed = sum(item.status == "complete" for item in items)
    total = len(items)
    remaining = total - completed
    progress = round(completed * 100 / total) if total else 0
    next_item = next((item for item in items if item.status != "complete"), None)
    current_phase = next_item.phase if next_item else "complete"
    phase_results = []
    first_incomplete_seen = False
    for phase in phases:
        phase_completed = sum(item.status == "complete" for item in phase.items)
        if phase_completed == len(phase.items):
            expected_status = "complete"
        elif not first_incomplete_seen:
            expected_status = "active"
            first_incomplete_seen = True
        else:
            expected_status = "planned"
        phase_results.append({
            "phase": phase.title,
            "completed": phase_completed,
            "total": len(phase.items),
            "expectedStatus": expected_status,
            "declaredStatus": phase.declared_status,
            "declaredCompleted": phase.declared_completed,
            "declaredTotal": phase.declared_total,
        })
    return {
        "mandatoryTotal": total,
        "mandatoryCompleted": completed,
        "mandatoryRemaining": remaining,
        "progressPercent": progress,
        "currentPhase": current_phase,
        "nextMandatoryWorkItem": next_item.id if next_item else "none",
        "optionalBacklogCount": len(optional),
        "phases": phase_results,
        "items": [item.__dict__ for item in items],
    }


def dashboard_values(path: Path) -> dict:
    text = path.read_text(encoding="utf-8")
    patterns = {
        "repositoryVersion": r"Repository version:\s*`([^`]+)`",
        "lastCompleted": r"Last completed work item:\s*`([^`]+)`",
        "activeWorkItem": r"Active work item:\s*`([^`]+)`",
        "mandatoryCompleted": r"Mandatory completed:\s*(\d+)\s*/\s*(\d+)",
        "mandatoryRemaining": r"Mandatory remaining:\s*(\d+)",
        "progressPercent": r"Progress:\s*(\d+)\s*%",
        "currentPhase": r"Current phase:\s*([^\n]+)",
        "nextMandatoryWorkItem": r"Next mandatory work item:\s*([^\n]+)",
        "optionalBacklogCount": r"Optional backlog:\s*(\d+)",
    }
    out: dict[str, object] = {}
    for key, pattern in patterns.items():
        match = re.search(pattern, text)
        if not match:
            raise ValueError(f"dashboard field missing: {key}")
        if key == "mandatoryCompleted":
            out[key], out["mandatoryTotal"] = int(match.group(1)), int(match.group(2))
        elif key in {"mandatoryRemaining", "progressPercent", "optionalBacklogCount"}:
            out[key] = int(match.group(1))
        else:
            out[key] = match.group(1).strip()
    return out


def validate(gap_root: Path) -> tuple[dict, list[str], list[str]]:
    roadmap_path = gap_root / "20-runtime-state/roadmap/ROADMAP.md"
    dashboard_path = gap_root / "20-runtime-state/status/dashboard.md"
    manifest_path = gap_root / "manifest.json"
    summary, phases, optional = parse_roadmap(roadmap_path)
    derived = derive(phases, optional)
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    dashboard = dashboard_values(dashboard_path)
    errors: list[str] = []
    checks: list[str] = []

    for key in ("mandatoryTotal", "mandatoryCompleted", "mandatoryRemaining", "progressPercent", "currentPhase", "nextMandatoryWorkItem", "optionalBacklogCount"):
        if summary[key] != derived[key]:
            errors.append(f"roadmap summary mismatch for {key}: declared={summary[key]!r}, derived={derived[key]!r}")
        else:
            checks.append(f"roadmap.{key}")
        manifest_value = manifest.get("roadmap", {}).get(key)
        if manifest_value != derived[key]:
            errors.append(f"manifest roadmap mismatch for {key}: declared={manifest_value!r}, derived={derived[key]!r}")
        else:
            checks.append(f"manifest.{key}")
        dashboard_value = dashboard.get(key)
        if dashboard_value != derived[key]:
            errors.append(f"dashboard mismatch for {key}: declared={dashboard_value!r}, derived={derived[key]!r}")
        else:
            checks.append(f"dashboard.{key}")

    for phase in derived["phases"]:
        if phase["declaredStatus"] != phase["expectedStatus"]:
            errors.append(f"phase status mismatch for {phase['phase']}: declared={phase['declaredStatus']}, expected={phase['expectedStatus']}")
        if phase["declaredCompleted"] != phase["completed"] or phase["declaredTotal"] != phase["total"]:
            errors.append(f"phase count mismatch for {phase['phase']}")

    work_items_dir = gap_root / "20-runtime-state/knowledge-transfers"
    for item in derived["items"]:
        matches = list(work_items_dir.glob(f"{item['id']}-*.md"))
        if item["status"] in {"complete", "ready"} and len(matches) != 1:
            errors.append(f"expected one work-item file for {item['id']}, found {len(matches)}")
        if len(matches) == 1:
            fm = front_matter(matches[0])
            if fm.get("status") != item["status"]:
                errors.append(f"work-item status mismatch for {item['id']}: file={fm.get('status')}, roadmap={item['status']}")
            if item["status"] == "ready" and fm.get("approved") is not True:
                errors.append(f"active work item is not approved: {item['id']}")

    next_id = derived["nextMandatoryWorkItem"]
    active = manifest.get("activeWorkItem", {})
    if active.get("id") != next_id or active.get("status") != "ready" or active.get("approved") is not True:
        errors.append("manifest activeWorkItem does not match the derived next mandatory work item")
    if manifest.get("runtimeState", {}).get("productiveKnowledgeTransfers", {}).get("active") != next_id:
        errors.append("runtimeState active work item mismatch")
    if manifest.get("stateMachine", {}).get("activeWorkItem") != next_id:
        errors.append("stateMachine active work item mismatch")
    if dashboard.get("activeWorkItem") != next_id:
        errors.append("dashboard active work item mismatch")

    complete_ids = [item["id"] for item in derived["items"] if item["status"] == "complete"]
    expected_last = complete_ids[-1] if complete_ids else "none"
    if dashboard.get("lastCompleted") != expected_last:
        errors.append(f"dashboard last completed mismatch: {dashboard.get('lastCompleted')} != {expected_last}")

    if any(entry.get("roadmapImpact") != "none" for entry in optional):
        errors.append("optional backlog contains an entry with mandatory roadmap impact")

    result = {
        "validationId": "roadmap-progress",
        "status": "PASSED" if not errors else "FAILED",
        "baseline": manifest.get("roadmap", {}).get("baseline"),
        "baselineVersion": manifest.get("roadmap", {}).get("baselineVersion"),
        "derived": derived,
        "declaredRoadmap": summary,
        "dashboard": dashboard,
        "checksPassed": len(checks),
        "errors": errors,
        "roadmapChanged": False,
        "mandatoryWorkAdmitted": False,
        "deploymentPerformed": False,
        "promotionPerformed": False,
    }
    return result, errors, [f"gap-000:20-runtime-state/knowledge-transfers/{next_id.lower()}" ]


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("--work-item", default="KT-0018")
    args = parser.parse_args()

    gap_root = args.gap_root.resolve()
    runtime_root = args.runtime_root.resolve()
    result, errors, _ = validate(gap_root)
    materialization = runtime_root / "test/validations/KT-0018/roadmap-progress-manifest.json"
    evidence = runtime_root / "test/reports/validation/roadmap-progress.json"
    atomic_json(materialization, result)

    active_path = json.loads((gap_root / "manifest.json").read_text(encoding="utf-8"))["activeWorkItem"]["path"]
    artifact_refs = [ROADMAP_REF, DASHBOARD_REF, GAP_MANIFEST_REF, f"gap-000:{active_path}", DOC_REF, "runtime:test/validations/KT-0018/roadmap-progress-manifest.json"]
    sealed = seal_evidence(result, args.repository_roots.resolve(), VALIDATOR_REF, artifact_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(f"Roadmap progress validation PASSED: {result['derived']['mandatoryCompleted']} / {result['derived']['mandatoryTotal']} ({result['derived']['progressPercent']} %)")
    return 0


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