Coverage for src / lexigram / contracts / workflow / content_checkpoint.py: 76%

1"""Content-addressed checkpoint primitives for workflow sagas. 

2 

3Provides the structured key, entry, and store protocol needed for 

4content-addressed checkpointing where stage outputs are keyed by 

5``sha256(stage_id || tenant_id || inputs)``. 

6""" 

7 

8from __future__ import annotations 

9 

10from dataclasses import dataclass, field 

11from datetime import date, datetime 

12import hashlib 

13import json as _stdlib_json 

14from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

15 

16if TYPE_CHECKING: 

17 from collections.abc import Sequence 

18 

19__all__ = [ 

20 "ContentCheckpointEntry", 

21 "ContentCheckpointKey", 

22 "ContentCheckpointStoreProtocol", 

23] 

24 

25 

26def _canonical_json_bytes(obj: Any) -> bytes: 

27 """Serialize *obj* to canonical JSON bytes for content addressing.""" 

28 

29 def _default(value: Any) -> Any: 

30 if isinstance(value, (datetime, date)): 

31 return value.isoformat() 

32 raise TypeError( 

33 f"Object of type {type(value).__name__} is not JSON serializable" 

34 ) 

35 

36 return _stdlib_json.dumps( 

37 obj, 

38 sort_keys=True, 

39 separators=(",", ":"), 

40 ensure_ascii=False, 

41 default=_default, 

42 ).encode("utf-8") 

43 

44 

45@dataclass(frozen=True) 

46class ContentCheckpointKey: 

47 """Structured key for content-addressed stage outputs. 

48 

49 The string form is ``f"{stage_id}|{tenant_id or '_global'}|{input_hash.hex()}|{config_hash.hex()}"``, 

50 stable across processes and deploys. 

51 """ 

52 

53 stage_id: str 

54 tenant_id: str | None 

55 input_hash: bytes 

56 config_hash: bytes 

57 

58 def as_str(self) -> str: 

59 """Return the string form for use as a storage key.""" 

60 tenant = self.tenant_id or "_global" 

61 return ( 

62 f"{self.stage_id}|{tenant}|{self.input_hash.hex()}|{self.config_hash.hex()}" 

63 ) 

64 

65 @classmethod 

66 def compute( 

67 cls, 

68 stage_id: str, 

69 tenant_id: str | None, 

70 inputs: dict[str, Any], 

71 stage_handler_version: str, 

72 config_affecting_output: dict[str, Any], 

73 ) -> ContentCheckpointKey: 

74 """Compute a content-addressed key from stage parameters.""" 

75 input_bytes = _canonical_json_bytes(inputs) 

76 config_bytes = _canonical_json_bytes( 

77 { 

78 "handler_version": stage_handler_version, 

79 "config": config_affecting_output, 

80 } 

81 ) 

82 return cls( 

83 stage_id=stage_id, 

84 tenant_id=tenant_id, 

85 input_hash=hashlib.sha256(input_bytes).digest(), 

86 config_hash=hashlib.sha256(config_bytes).digest(), 

87 ) 

88 

89 

90@dataclass(frozen=True) 

91class ContentCheckpointEntry: 

92 """A stored checkpoint entry for a content-addressed stage output. 

93 

94 When the output exceeds the inline threshold, ``output`` is ``None`` and 

95 ``output_blob_ref`` holds the ``lexigram-storage`` blob reference. 

96 """ 

97 

98 output: Any 

99 output_blob_ref: str | None 

100 completed_at: datetime 

101 stage_handler_version: str 

102 output_size_bytes: int 

103 metadata: dict[str, Any] = field(default_factory=dict) 

104 

105 

106@runtime_checkable 

107class ContentCheckpointStoreProtocol(Protocol): 

108 """Persistence contract for content-addressed checkpoint entries.""" 

109 

110 async def get(self, key: ContentCheckpointKey) -> ContentCheckpointEntry | None: 

111 """Retrieve a cached checkpoint entry, or ``None``.""" 

112 

113 async def set( 

114 self, key: ContentCheckpointKey, entry: ContentCheckpointEntry 

115 ) -> None: 

116 """Persist a checkpoint entry.""" 

117 

118 async def evict(self, key: ContentCheckpointKey) -> None: 

119 """Remove a checkpoint entry.""" 

120 

121 async def list_by_stage( 

122 self, 

123 stage_id: str, 

124 tenant_id: str | None = None, 

125 limit: int = 100, 

126 ) -> Sequence[ContentCheckpointKey]: 

127 """List checkpoint keys for a given stage, optionally filtered by tenant."""