Coverage for astrocyte/lifecycle.py: 100%

1"""Memory lifecycle management — TTL, legal hold, compliance forget. 

2 

3See docs/_design/memory-lifecycle.md for the design specification. 

4All policy functions are sync (Rust migration candidates). 

5""" 

6 

7from __future__ import annotations 

8 

9from datetime import datetime, timezone 

10 

11from astrocyte.config import LifecycleConfig 

12from astrocyte.errors import LegalHoldActive 

13from astrocyte.types import LegalHold, LifecycleAction 

14 

15 

16class LifecycleManager: 

17 """Manages memory lifecycle: TTL evaluation and legal holds. 

18 

19 Holds come from two places: ones set in this process (``set_legal_hold``, 

20 kept in memory) and ones persisted in the store (``Astrocyte.place_legal_hold``), 

21 which the caller loads with :meth:`set_stored_holds` before each check so 

22 every process and replica sees the same holds. A bank is under hold if 

23 either has one. TTL evaluation is a pure sync function — no I/O, Rust-portable. 

24 """ 

25 

26 def __init__(self, config: LifecycleConfig) -> None: 

27 self._config = config 

28 self._holds: dict[str, LegalHold] = {} # key: "{bank_id}\x00{hold_id}" 

29 self._stored: dict[str, list[LegalHold]] = {} # bank_id → holds as last read from the store 

30 

31 # ── Legal holds (sync) ── 

32 

33 _SEP = "\x00" # Null byte separator — cannot appear in bank_id or hold_id strings 

34 

35 def set_legal_hold(self, bank_id: str, hold_id: str, reason: str, *, set_by: str = "user:api") -> LegalHold: 

36 """Place a bank under legal hold.""" 

37 key = f"{bank_id}{self._SEP}{hold_id}" 

38 hold = LegalHold( 

39 hold_id=hold_id, 

40 bank_id=bank_id, 

41 reason=reason, 

42 set_at=datetime.now(timezone.utc), 

43 set_by=set_by, 

44 ) 

45 self._holds[key] = hold 

46 return hold 

47 

48 def release_legal_hold(self, bank_id: str, hold_id: str) -> bool: 

49 """Release a legal hold. Returns True if hold existed.""" 

50 key = f"{bank_id}{self._SEP}{hold_id}" 

51 return self._holds.pop(key, None) is not None 

52 

53 def set_stored_holds(self, bank_id: str, holds: list[LegalHold]) -> None: 

54 """The bank's persisted holds, as just read from the store.""" 

55 self._stored[bank_id] = list(holds) 

56 

57 def is_under_hold(self, bank_id: str) -> bool: 

58 """Check if any legal hold is active on this bank.""" 

59 return bool(self.get_holds(bank_id)) 

60 

61 def get_holds(self, bank_id: str) -> list[LegalHold]: 

62 """Get all active holds for a bank: this process's and the persisted ones.""" 

63 prefix = f"{bank_id}{self._SEP}" 

64 local = [h for k, h in self._holds.items() if k.startswith(prefix)] 

65 ids = {h.hold_id for h in local} 

66 return local + [h for h in self._stored.get(bank_id, []) if h.hold_id not in ids] 

67 

68 def check_forget_allowed(self, bank_id: str) -> None: 

69 """Raise LegalHoldActive if bank is under hold.""" 

70 holds = self.get_holds(bank_id) 

71 if holds: 

72 raise LegalHoldActive(bank_id=bank_id, hold_id=holds[0].hold_id) 

73 

74 # ── TTL evaluation (sync, pure) ── 

75 

76 def evaluate_memory_ttl( 

77 self, 

78 memory_id: str, 

79 bank_id: str, 

80 created_at: datetime | None, 

81 last_recalled_at: datetime | None, 

82 tags: list[str] | None, 

83 fact_type: str | None, 

84 now: datetime, 

85 ) -> LifecycleAction: 

86 """Evaluate TTL policy for a single memory. Sync, pure function. 

87 

88 Returns LifecycleAction with action="archive"|"delete"|"keep". 

89 """ 

90 if not self._config.enabled: 

91 return LifecycleAction(memory_id=memory_id, action="keep", reason="lifecycle_disabled") 

92 

93 # Legal hold blocks all lifecycle actions 

94 if self.is_under_hold(bank_id): 

95 return LifecycleAction(memory_id=memory_id, action="keep", reason="legal_hold") 

96 

97 ttl = self._config.ttl 

98 

99 # Check exempt tags 

100 if tags and ttl.exempt_tags: 

101 if any(t in ttl.exempt_tags for t in tags): 

102 return LifecycleAction(memory_id=memory_id, action="keep", reason="exempt") 

103 

104 # Determine archive threshold (may be overridden by fact_type) 

105 archive_days = ttl.archive_after_days 

106 if fact_type and ttl.fact_type_overrides: 

107 override = ttl.fact_type_overrides.get(fact_type) 

108 if override is not None: 

109 archive_days = override 

110 

111 # Check delete threshold (based on creation date) 

112 if created_at: 

113 age_days = (now - created_at).days 

114 if age_days >= ttl.delete_after_days: 

115 return LifecycleAction(memory_id=memory_id, action="delete", reason="ttl_expired") 

116 

117 # Check archive threshold (based on last recall or creation) 

118 reference_date = last_recalled_at or created_at 

119 if reference_date: 

120 days_since = (now - reference_date).days 

121 if days_since >= archive_days: 

122 return LifecycleAction(memory_id=memory_id, action="archive", reason="ttl_unretrieved") 

123 

124 return LifecycleAction(memory_id=memory_id, action="keep", reason="recent")