Add gate_decision/src/gate_decision/core.py

gate_decision/src/gate_decision/core.py

@@ -0,0 +1,93 @@
+import json
+import argparse
+import os
+import sys
+from dataclasses import dataclass
+from typing import Any, Dict
+
+
+@dataclass
+class GateDecision:
+    decision: str
+    explanation: str
+
+
+class InvalidInputError(Exception):
+    """Raised when summary_data input is invalid."""
+    pass
+
+
+def _validate_summary_data(summary_data: Dict[str, Any]) -> None:
+    required_fields = ["mischfenster_p95", "retry_free_in_window_rate", "max_value"]
+    for field in required_fields:
+        if field not in summary_data:
+            raise InvalidInputError(f"Missing required field: {field}")
+    try:
+        float(summary_data["mischfenster_p95"])
+        float(summary_data["retry_free_in_window_rate"])
+        float(summary_data["max_value"])
+    except (TypeError, ValueError) as e:
+        raise InvalidInputError(f"Invalid field type: {e}") from e
+
+
+def make_decision(summary_data: Dict[str, Any]) -> Dict[str, Any]:
+    """Erzeugt eine Gate-Decision nach der v0-Regel basierend auf den Zusammenfassungsdaten eines Runs."""
+    _validate_summary_data(summary_data)
+
+    p95 = float(summary_data["mischfenster_p95"])
+    retry_free = float(summary_data["retry_free_in_window_rate"])
+    threshold_p95 = 200.0
+    threshold_retry = 0.95
+
+    if p95 <= threshold_p95 and retry_free >= threshold_retry:
+        decision = GateDecision(decision="pass", explanation="Run stable: p95 <= 200ms and high retry-free rate.")
+    else:
+        reasons = []
+        if p95 > threshold_p95:
+            reasons.append(f"p95 too high ({p95:.2f} > {threshold_p95})")
+        if retry_free < threshold_retry:
+            reasons.append(f"retry-free rate low ({retry_free:.2f} < {threshold_retry})")
+        decision = GateDecision(decision="fail", explanation="; ".join(reasons))
+
+    return {"decision": decision.decision, "explanation": decision.explanation}
+
+
+def _load_json(path: str) -> Dict[str, Any]:
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def _save_json(data: Dict[str, Any], path: str) -> None:
+    os.makedirs(os.path.dirname(path), exist_ok=True)
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2, ensure_ascii=False)
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Gate Decision v0 CLI")
+    parser.add_argument("--input", required=True, help="Pfad zur JSON-Eingabedatei mit Run-Summary-Daten.")
+    parser.add_argument("--output", required=False, default="output/gate_decision.json", help="Pfad zur Ausgabe-Datei.")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = _parse_args()
+
+    try:
+        summary_data = _load_json(args.input)
+        result = make_decision(summary_data)
+        _save_json(result, args.output)
+        print(f"Gate decision written to {args.output}: {result['decision']}")
+    except InvalidInputError as e:
+        print(f"Invalid input: {e}", file=sys.stderr)
+        sys.exit(2)
+    except FileNotFoundError as e:
+        print(f"File error: {e}", file=sys.stderr)
+        sys.exit(1)
+    except Exception as e:
+        print(f"Unexpected error: {e}", file=sys.stderr)
+        sys.exit(3)
+
+
+if __name__ == "__main__":
+    main()