Coverage for src / kemi / lifecycle.py: 100%

1from datetime import datetime, timezone 

2 

3from kemi.models import LifecycleState, MemoryObject 

4 

5 

6def evaluate_lifecycle( 

7 memory: MemoryObject, 

8 decay_threshold_hours: float = 720.0, 

9) -> LifecycleState: 

10 """Evaluate what lifecycle state a memory should be in. 

11 

12 Based on last_accessed_at compared to decay_threshold_hours. 

13 Default: 720 hours = 30 days. 

14 

15 If last_accessed_at is in the future (clock skew), returns ACTIVE. 

16 """ 

17 if memory.lifecycle_state == LifecycleState.DELETED: 

18 return LifecycleState.DELETED 

19 

20 if memory.lifecycle_state == LifecycleState.ARCHIVED: 

21 return LifecycleState.ARCHIVED 

22 

23 now = datetime.now(timezone.utc) 

24 hours_since_access = (now - memory.last_accessed_at).total_seconds() / 3600.0 

25 

26 if hours_since_access < 0: 

27 return LifecycleState.ACTIVE 

28 

29 if hours_since_access > decay_threshold_hours: 

30 return LifecycleState.DECAYING 

31 

32 return LifecycleState.ACTIVE 

33 

34 

35def transition(memory: MemoryObject, new_state: LifecycleState) -> MemoryObject: 

36 """Create a new MemoryObject with updated lifecycle state. 

37 

38 Does not mutate the original. Returns a new instance. 

39 

40 Raises ValueError if the transition is not allowed. 

41 """ 

42 validate_transition(memory.lifecycle_state, new_state) 

43 

44 return MemoryObject( 

45 memory_id=memory.memory_id, 

46 user_id=memory.user_id, 

47 content=memory.content, 

48 embedding=memory.embedding, 

49 score=memory.score, 

50 created_at=memory.created_at, 

51 last_accessed_at=memory.last_accessed_at, 

52 source=memory.source, 

53 importance=memory.importance, 

54 lifecycle_state=new_state, 

55 metadata=memory.metadata.copy() if memory.metadata else {}, 

56 embedding_dim=memory.embedding_dim, 

57 tags=memory.tags, 

58 confidence=memory.confidence, 

59 memory_type=memory.memory_type, 

60 session_id=memory.session_id, 

61 namespace=memory.namespace, 

62 version=memory.version, 

63 ) 

64 

65 

66def get_recall_filter() -> list[LifecycleState]: 

67 """Return the list of lifecycle states that should be included in recall results. 

68 

69 Excludes ARCHIVED and DELETED. Includes ACTIVE and DECAYING. 

70 """ 

71 return [LifecycleState.ACTIVE, LifecycleState.DECAYING] 

72 

73 

74_VALID_TRANSITIONS = { 

75 LifecycleState.ACTIVE: { 

76 LifecycleState.DECAYING, 

77 LifecycleState.DELETED, 

78 LifecycleState.ARCHIVED, 

79 }, 

80 LifecycleState.DECAYING: { 

81 LifecycleState.ACTIVE, 

82 LifecycleState.DELETED, 

83 LifecycleState.ARCHIVED, 

84 }, 

85 LifecycleState.ARCHIVED: set(), 

86 LifecycleState.DELETED: set(), 

87} 

88 

89 

90def validate_transition(from_state: LifecycleState, to_state: LifecycleState) -> None: 

91 """Validate a state transition. 

92 

93 Raises ValueError if the transition is not allowed. 

94 Valid transitions: 

95 - ACTIVE → DECAYING 

96 - ACTIVE → DELETED 

97 - ACTIVE → ARCHIVED 

98 - DECAYING → ACTIVE 

99 - DECAYING → DELETED 

100 - DECAYING → ARCHIVED 

101 

102 ARCHIVED and DELETED are terminal. 

103 """ 

104 if to_state not in _VALID_TRANSITIONS.get(from_state, set()): 

105 raise ValueError( 

106 f"Invalid transition from {from_state.value} to {to_state.value}. " 

107 f"Valid transitions from {from_state.value}: " 

108 f"{[s.value for s in _VALID_TRANSITIONS.get(from_state, set())]}" 

109 )