Add rollout engine

90-execution/distributed-inference/model-rollouts/rollout.py

@@ -0,0 +1,63 @@
+"""rollout.py - async canary/progressive rollout state machine.
+
+States: DRAFT -> APPROVED -> CANARY_PULL -> CANARY_VERIFY -> CANARY_SERVING
+        -> PROGRESSIVE_ROLLOUT -> COMPLETE   (failures -> PAUSED/ROLLBACK)
+Persist steps + audit to PostgreSQL; use Redis for exclusive transition locks.
+"""
+from __future__ import annotations
+import enum
+from dataclasses import dataclass, field
+
+
+class State(str, enum.Enum):
+    DRAFT = "DRAFT"
+    APPROVED = "APPROVED"
+    CANARY_PULL = "CANARY_PULL"
+    CANARY_VERIFY = "CANARY_VERIFY"
+    CANARY_SERVING = "CANARY_SERVING"
+    PROGRESSIVE_ROLLOUT = "PROGRESSIVE_ROLLOUT"
+    COMPLETE = "COMPLETE"
+    PAUSED = "PAUSED"
+    ROLLBACK = "ROLLBACK"
+
+
+@dataclass
+class Rollout:
+    id: str
+    model_id: str
+    environment: str
+    plan: dict
+    state: State = State.DRAFT
+    history: list[str] = field(default_factory=list)
+
+    def transition(self, new: State, why: str = ""):
+        valid = {
+            State.DRAFT: {State.APPROVED},
+            State.APPROVED: {State.CANARY_PULL},
+            State.CANARY_PULL: {State.CANARY_VERIFY},
+            State.CANARY_VERIFY: {State.CANARY_SERVING, State.ROLLBACK},
+            State.CANARY_SERVING: {State.PROGRESSIVE_ROLLOUT, State.PAUSED, State.ROLLBACK},
+            State.PROGRESSIVE_ROLLOUT: {State.COMPLETE, State.PAUSED, State.ROLLBACK},
+            State.PAUSED: {State.PROGRESSIVE_ROLLOUT, State.ROLLBACK},
+        }.get(self.state, set())
+        if new not in valid:
+            raise ValueError(f"invalid transition {self.state} -> {new}")
+        self.state = new
+        self.history.append(f"{new.value}:{why}")
+
+    def gates_pass(self, metrics: dict) -> bool:
+        g = self.plan.get("quality_gates", {})
+        return all([
+            metrics.get("smoke_success_rate", 0) >= g.get("smoke_test_success_rate_min", 1.0),
+            metrics.get("error_rate", 1) <= g.get("error_rate_max", 0.01),
+            metrics.get("p95_ttft_ms", 1e9) <= g.get("p95_first_token_latency_ms_max", 3000),
+            metrics.get("p95_itl_ms", 1e9) <= g.get("p95_inter_token_latency_ms_max", 250),
+        ])
+
+
+def approve(r: Rollout, role: str, ticket: str) -> Rollout:
+    need = r.plan.get("approval", {}).get("required_role", "model_operator")
+    if role != need or not ticket:
+        raise PermissionError(f"requires {need} + change ticket")
+    r.transition(State.APPROVED, f"approved_by={role} ticket={ticket}")
+    return r