stack/packages/mosaic/framework/tools/woodpecker/verify-terminal-green.py

#!/usr/bin/env python3
"""Verify Mosaic's full-step Woodpecker terminal-green contract.

RM-61 permits one named, signature-scoped exception for issue #1000. The
exception retires when #1000 is fixed; all other non-success states block.
This program consumes the JSON/API record emitted by pipeline-status.sh -f json.
It does not fetch, retry, or re-trigger pipelines.
"""

from __future__ import annotations

import json
import re
import sys
from collections import Counter
from pathlib import Path
from typing import Any

EXEMPTION_ID = "WP-K8S-1000-CI-POSTGRES-TEARDOWN"
EXEMPTION_ISSUE = "https://git.mosaicstack.dev/mosaicstack/stack/issues/1000"
POD_NOT_FOUND = re.compile(
    r'^pods "wp-svc-[0-9a-hjkmnp-tv-z]{26}-ci-postgres" not found$'
)


def fail_usage(message: str) -> int:
    print(f"terminal-green contract input error: {message}", file=sys.stderr)
    return 2


def load_record(argument: str | None) -> dict[str, Any]:
    if argument in (None, "-"):
        value = json.load(sys.stdin)
    else:
        with Path(argument).open(encoding="utf-8") as handle:
            value = json.load(handle)
    if not isinstance(value, dict):
        raise ValueError("pipeline record must be a JSON object")
    return value


def is_issue_1000_artifact(step: dict[str, Any]) -> bool:
    error = step.get("error")
    return (
        step.get("name") == "ci-postgres"
        and step.get("type") == "service"
        and step.get("state") == "failure"
        and step.get("exit_code") == 0
        and isinstance(error, str)
        and POD_NOT_FOUND.fullmatch(error) is not None
    )


def verify(record: dict[str, Any]) -> tuple[int, dict[str, Any]]:
    anomalies: list[dict[str, Any]] = []
    candidates: list[dict[str, Any]] = []
    steps: list[dict[str, Any]] = []

    pipeline_status = record.get("status")
    if pipeline_status != "success":
        anomalies.append(
            {
                "scope": "pipeline",
                "name": str(record.get("number", "unknown")),
                "state": pipeline_status,
                "reason": "pipeline status is not success",
            }
        )

    workflows = record.get("workflows")
    if not isinstance(workflows, list) or not workflows:
        anomalies.append(
            {
                "scope": "pipeline",
                "name": str(record.get("number", "unknown")),
                "state": pipeline_status,
                "reason": "workflows are missing or empty",
            }
        )
        workflows = []

    for workflow_index, workflow in enumerate(workflows):
        if not isinstance(workflow, dict):
            anomalies.append(
                {
                    "scope": "workflow",
                    "name": str(workflow_index),
                    "state": None,
                    "reason": "workflow is not an object",
                }
            )
            continue
        workflow_name = str(workflow.get("name", workflow_index))
        if workflow.get("state") != "success":
            anomalies.append(
                {
                    "scope": "workflow",
                    "name": workflow_name,
                    "state": workflow.get("state"),
                    "reason": "workflow state is not success",
                }
            )
        children = workflow.get("children")
        if not isinstance(children, list) or not children:
            anomalies.append(
                {
                    "scope": "workflow",
                    "name": workflow_name,
                    "state": workflow.get("state"),
                    "reason": "child-step list is missing or empty",
                }
            )
            continue
        for child_index, child in enumerate(children):
            if not isinstance(child, dict):
                anomalies.append(
                    {
                        "scope": "step",
                        "name": f"{workflow_name}[{child_index}]",
                        "state": None,
                        "reason": "step is not an object",
                    }
                )
                continue
            steps.append(child)
            if child.get("state") == "success":
                continue
            if is_issue_1000_artifact(child):
                candidates.append(child)
                continue
            anomalies.append(
                {
                    "scope": "step",
                    "name": child.get("name"),
                    "type": child.get("type"),
                    "state": child.get("state"),
                    "exit_code": child.get("exit_code"),
                    "error": child.get("error"),
                    "reason": "non-success step does not match the #1000 teardown signature",
                }
            )

    if len(candidates) > 1:
        anomalies.append(
            {
                "scope": "exemption",
                "name": EXEMPTION_ID,
                "state": "invalid",
                "reason": "the #1000 exemption may apply to exactly one step",
            }
        )

    exemption_applies = len(candidates) == 1 and not anomalies
    state_counts = Counter(str(step.get("state", "missing")) for step in steps)
    result: dict[str, Any] = {
        "schema_version": "mosaic-terminal-green/v1",
        "verdict": "terminal-green" if not anomalies else "not-terminal-green",
        "pipeline_number": record.get("number"),
        "commit": record.get("commit"),
        "pipeline_status": pipeline_status,
        "total_steps": len(steps),
        "state_counts": dict(sorted(state_counts.items())),
        "exempted_steps": 1 if exemption_applies else 0,
        "anomalies": anomalies,
    }
    if exemption_applies:
        candidate = candidates[0]
        result["exemptions"] = [
            {
                "exemption_id": EXEMPTION_ID,
                "step": candidate.get("name"),
                "signature": candidate.get("error"),
                "tracking_issue": EXEMPTION_ISSUE,
                "retires_when": "issue #1000 is fixed",
            }
        ]
    else:
        result["exemptions"] = []

    return (0 if not anomalies else 1), result


def main() -> int:
    if len(sys.argv) > 2 or (len(sys.argv) == 2 and sys.argv[1] in {"-h", "--help"}):
        print(f"usage: {Path(sys.argv[0]).name} [pipeline.json|-]", file=sys.stderr)
        return 0 if len(sys.argv) == 2 else 2
    try:
        record = load_record(sys.argv[1] if len(sys.argv) == 2 else None)
    except (OSError, ValueError, json.JSONDecodeError) as error:
        return fail_usage(str(error))
    code, result = verify(record)
    print(json.dumps(result, indent=2, sort_keys=True))
    return code


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