Coverage for src / lexigram / contracts / workflow / steps.py: 100%

1"""Pipeline step execution result types. 

2 

3Moved from ``lexigram.contracts.execution`` — step results are a workflow 

4concern, not a generic execution or Result concern. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass, field 

10from typing import TYPE_CHECKING, Any 

11 

12if TYPE_CHECKING: 

13 from collections.abc import Awaitable, Callable 

14 

15 from lexigram.contracts.core.result import Result 

16 

17__all__ = [ 

18 "SagaStep", 

19 "SagaStepError", 

20] 

21 

22 

23class SagaStepError(Exception): 

24 """Represents a recoverable failure from a single saga step action. 

25 

26 SagaProtocol step actions must return ``Result[T, SagaStepError]`` so that the 

27 orchestration layer can detect failures without catching exceptions. 

28 

29 Attributes: 

30 step_name: Identifier of the failing step (set by the orchestrator). 

31 detail: Human-readable description of what went wrong. 

32 """ 

33 

34 _code: str = "LEX_ERR_WF_002" 

35 

36 def __init__(self, detail: str, step_name: str = "") -> None: 

37 super().__init__(detail) 

38 self.detail = detail 

39 self.step_name = step_name 

40 

41 def __repr__(self) -> str: 

42 """Return developer-friendly repr.""" 

43 return f"{type(self).__name__}(step={self.step_name!r}, detail={self.detail!r})" 

44 

45 

46@dataclass(frozen=True) 

47class SagaStep: 

48 """Descriptor for a single saga orchestration step. 

49 

50 Shared across ``lexigram-workflow`` and ``lexigram-events`` via contracts. 

51 

52 The ``action`` callable **must** return ``Result[Any, SagaStepError]``. 

53 The ``compensation`` callable is called only when the saga is unwinding 

54 and may raise exceptions freely (infrastructure failures in compensation 

55 should propagate). 

56 

57 Attributes: 

58 name: Unique step identifier within the saga. 

59 action: Async callable returning ``Result[Any, SagaStepError]``. 

60 compensation: Async callable to undo the step on failure. 

61 max_retries: Maximum retry attempts if action returns ``Err``. 

62 retry_delay: Seconds to wait between retry attempts. 

63 idempotent: Whether the compensation is safe to call multiple times. 

64 """ 

65 

66 name: str 

67 action: Callable[..., Awaitable[Result[Any, SagaStepError]]] 

68 compensation: Callable[..., Awaitable[None]] | None = None 

69 max_retries: int = 3 

70 retry_delay: float = 1.0 

71 idempotent: bool = field(default=False)