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 Legal holds are stored in-memory. Persistence is out of scope for v1. 

20 TTL evaluation is a pure sync function — no I/O, Rust-portable. 

21 """ 

22 

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

24 self._config = config 

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

26 

27 # ── Legal holds (sync) ── 

28 

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

30 

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

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

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

34 hold = LegalHold( 

35 hold_id=hold_id, 

36 bank_id=bank_id, 

37 reason=reason, 

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

39 set_by=set_by, 

40 ) 

41 self._holds[key] = hold 

42 return hold 

43 

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

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

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

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

48 

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

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

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

52 return any(k.startswith(prefix) for k in self._holds) 

53 

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

55 """Get all active holds for a bank.""" 

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

57 return [h for k, h in self._holds.items() if k.startswith(prefix)] 

58 

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

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

61 holds = self.get_holds(bank_id) 

62 if holds: 

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

64 

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

66 

67 def evaluate_memory_ttl( 

68 self, 

69 memory_id: str, 

70 bank_id: str, 

71 created_at: datetime | None, 

72 last_recalled_at: datetime | None, 

73 tags: list[str] | None, 

74 fact_type: str | None, 

75 now: datetime, 

76 ) -> LifecycleAction: 

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

78 

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

80 """ 

81 if not self._config.enabled: 

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

83 

84 # Legal hold blocks all lifecycle actions 

85 if self.is_under_hold(bank_id): 

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

87 

88 ttl = self._config.ttl 

89 

90 # Check exempt tags 

91 if tags and ttl.exempt_tags: 

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

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

94 

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

96 archive_days = ttl.archive_after_days 

97 if fact_type and ttl.fact_type_overrides: 

98 override = ttl.fact_type_overrides.get(fact_type) 

99 if override is not None: 

100 archive_days = override 

101 

102 # Check delete threshold (based on creation date) 

103 if created_at: 

104 age_days = (now - created_at).days 

105 if age_days >= ttl.delete_after_days: 

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

107 

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

109 reference_date = last_recalled_at or created_at 

110 if reference_date: 

111 days_since = (now - reference_date).days 

112 if days_since >= archive_days: 

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

114 

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