aaronrene / scooling-lab public
state_machine.py python
53 lines 1.6 KB
Raw
sha256:2313b6a6353294a0ca08ff579624198da250b99163a86215aed31a905912e360 Land Scooling Lab T0/T2 training contract and seven-tier tests Human minor ⚠ breaking 43 days ago
1 """Deterministic T2 training job state machine."""
2
3 from __future__ import annotations
4
5 from types import MappingProxyType
6
7 from scooling_lab.contracts import TrainingJobStatus
8 from scooling_lab.errors import ApiError, ErrorCode
9
10
11 TRANSITION_TABLE: MappingProxyType[TrainingJobStatus, frozenset[TrainingJobStatus]] = (
12 MappingProxyType(
13 {
14 TrainingJobStatus.QUEUED: frozenset(
15 {
16 TrainingJobStatus.RUNNING,
17 TrainingJobStatus.FAILED,
18 TrainingJobStatus.CANCELLED,
19 }
20 ),
21 TrainingJobStatus.RUNNING: frozenset(
22 {
23 TrainingJobStatus.SUCCEEDED,
24 TrainingJobStatus.FAILED,
25 TrainingJobStatus.CANCELLED,
26 }
27 ),
28 TrainingJobStatus.SUCCEEDED: frozenset(),
29 TrainingJobStatus.FAILED: frozenset(),
30 TrainingJobStatus.CANCELLED: frozenset(),
31 }
32 )
33 )
34
35
36 def transition(
37 current: TrainingJobStatus, target: TrainingJobStatus
38 ) -> TrainingJobStatus:
39 """Return the next state or raise for invalid non-idempotent transitions."""
40
41 if current == target:
42 return current
43 if target in TRANSITION_TABLE[current]:
44 return target
45 raise ApiError(ErrorCode.INVALID_TRANSITION, 409)
46
47
48 def cancel(current: TrainingJobStatus) -> TrainingJobStatus:
49 """Cancel a queued or running job; terminal replay is idempotent."""
50
51 if current == TrainingJobStatus.CANCELLED:
52 return current
53 return transition(current, TrainingJobStatus.CANCELLED)
File History 1 commit
sha256:2313b6a6353294a0ca08ff579624198da250b99163a86215aed31a905912e360 Land Scooling Lab T0/T2 training contract and seven-tier tests Human minor ⚠ 43 days ago