"""Supervisor with explicit dispatch policy. Author: Sandeep Belgavi.""" from dataclasses import dataclass @dataclass(frozen=True) class Ticket: incident: str service: str symptom: str def dispatch(ticket, router, specialists, allowed_services): if not isinstance(ticket, Ticket) or not ticket.incident: raise ValueError("invalid ticket") if ticket.service not in allowed_services: return {"status": "denied", "route": "none"} try: route = router(ticket.symptom) except Exception: return {"status": "route_error", "route": "none"} if not isinstance(route, str) or route not in {"latency", "availability"}: return {"status": "escalate", "route": "none"} if route not in specialists: return {"status": "unavailable", "route": route} # Only this typed envelope crosses into a specialist, never raw credentials. envelope = {"incident": ticket.incident, "service": ticket.service} try: result = specialists[route](envelope) except Exception: return {"status": "worker_error", "route": route} if not isinstance(result, str) or not result.startswith("fixture:"): return {"status": "invalid_result", "route": route} return {"status": "review", "route": route, "evidence": result} if __name__ == "__main__": result = dispatch(Ticket("INC-42", "checkout", "slow"), lambda _: "latency", {"latency": lambda _: "fixture:L1"}, {"checkout"}) print(f"status={result['status']} route={result['route']} evidence={result['evidence']}")