feat(v2): add guarded execution state transitions

This commit is contained in:
2026-07-27 21:43:54 +02:00
parent 86d1ca168d
commit f3858ac43a
2 changed files with 51 additions and 0 deletions
+28
View File
@@ -0,0 +1,28 @@
from __future__ import annotations
from collections.abc import Mapping
ACTIVE_STATES = frozenset({"queued", "preparing", "running", "verifying", "cancelling"})
TERMINAL_STATES = frozenset({"committed", "cancelled", "failed"})
_ALLOWED: Mapping[str, frozenset[str]] = {
"queued": frozenset({"preparing", "cancelled", "failed"}),
"preparing": frozenset({"running", "cancelling", "failed"}),
"running": frozenset({"verifying", "cancelling", "failed"}),
"verifying": frozenset({"committed", "failed"}),
"cancelling": frozenset({"cancelled", "failed"}),
"committed": frozenset(),
"cancelled": frozenset(),
"failed": frozenset(),
}
class TransitionError(ValueError):
code = "invalid_transition"
def transition(current: str, target: str) -> str:
"""Validate and return one durable execution-state transition."""
if target not in _ALLOWED.get(current, frozenset()):
raise TransitionError(f"Cannot transition execution from {current!r} to {target!r}.")
return target