#!/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 argparse 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") exit_code = step.get("exit_code") return ( step.get("name") == "ci-postgres" and step.get("type") == "service" and step.get("state") == "failure" and type(exit_code) is int and not isinstance(exit_code, bool) and exit_code == 0 and isinstance(error, str) and POD_NOT_FOUND.fullmatch(error) is not None ) def verify(record: dict[str, Any], expected_commit: str) -> 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") actual_commit = record.get("commit") if actual_commit != expected_commit: anomalies.append( { "scope": "pipeline", "name": str(record.get("number", "unknown")), "state": pipeline_status, "reason": "pipeline commit does not equal the expected PR head", "expected_commit": expected_commit, "actual_commit": actual_commit, } ) 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": actual_commit, "expected_commit": expected_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 parse_arguments() -> argparse.Namespace: parser = argparse.ArgumentParser( description="verify the full Woodpecker terminal-green child-step contract" ) parser.add_argument( "--expect-commit", required=True, metavar="FULL_SHA", help="full 40-hex PR-head commit that the pipeline record must match", ) parser.add_argument("record", nargs="?", default="-", help="pipeline JSON file or -") arguments = parser.parse_args() if re.fullmatch(r"[0-9a-fA-F]{40}", arguments.expect_commit) is None: parser.error("--expect-commit must be a full 40-hex commit") arguments.expect_commit = arguments.expect_commit.lower() return arguments def main() -> int: arguments = parse_arguments() try: record = load_record(arguments.record) except (OSError, ValueError, json.JSONDecodeError) as error: return fail_usage(str(error)) code, result = verify(record, arguments.expect_commit) print(json.dumps(result, indent=2, sort_keys=True)) return code if __name__ == "__main__": raise SystemExit(main())