Coverage for core / prompts_backup.py: 88.5%

1""" 

2core.prompts_backup - periodic local snapshot of the prompts/ directory. 

3 

4Closes the "user accidentally wipes data dir / corruption / bad 

5shutdown" recovery gap. Cloud sync (core.recipe_sync) handles 

6cross-device propagation, but it only helps when the central is 

7reachable AND the user previously pushed. Local snapshots are the 

8zero-network fallback: the user's last N agent states sit in a 

9sibling directory and can be restored in one cp -r. 

10 

11When called: 

12 Once per HARTOS boot (best-effort, non-blocking) - cheap because 

13 the prompts/ dir is typically <10MB even with hundreds of agents. 

14 No file watcher / continuous backup - the boot snapshot bounds 

15 data loss to "since last reboot" which matches user mental 

16 model for most desktop applications. 

17 

18Where stored: 

19 ``<data>/prompts_snapshots/YYYYMMDD_HHMMSS/`` - one dir per snapshot, 

20 each containing a flat copy of all .json files from prompts/ at 

21 that moment. See core.platform_paths.get_data_dir for the data 

22 root resolution. 

23 

24Retention: 

25 Keep the latest ``MAX_DAILY_SNAPSHOTS`` (default 7) - one per 

26 day if HARTOS reboots daily; old snapshots beyond that count 

27 get pruned automatically on each new snapshot. 

28 

29Restore: 

30 Manual for now - user copies files back. A future feature could 

31 expose a /prompts/restore endpoint. 

32 

33Tests in tests/unit/test_prompts_backup.py. 

34""" 

35 

36import logging 

37import os 

38import shutil 

39import time 

40from typing import List, Optional 

41 

42logger = logging.getLogger('hevolve.prompts_backup') 

43 

44#: How many recent snapshots to keep. Override via env. 

45MAX_DAILY_SNAPSHOTS: int = int(os.environ.get( 

46 'HEVOLVE_PROMPTS_SNAPSHOT_KEEP', '7')) 

47 

48#: Subdirectory under data/ where snapshot dirs live. Sibling of 

49#: prompts/ so the user can find it without a hunt. 

50SNAPSHOTS_SUBDIR: str = 'prompts_snapshots' 

51 

52 

53def _snapshots_root(prompts_dir: str) -> str: 

54 """Sibling directory of prompts_dir for snapshot storage.""" 

55 return os.path.join(os.path.dirname(prompts_dir), SNAPSHOTS_SUBDIR) 

56 

57 

58def list_snapshots(prompts_dir: str) -> List[str]: 

59 """Return existing snapshot directory names, oldest-first. 

60 

61 Snapshot names are YYYYMMDD_HHMMSS so lexical sort = chronological. 

62 Returns ``[]`` when the snapshots root doesn't exist. 

63 """ 

64 root = _snapshots_root(prompts_dir) 

65 if not os.path.isdir(root): 

66 return [] 

67 names = [] 

68 for name in os.listdir(root): 

69 full = os.path.join(root, name) 

70 if os.path.isdir(full) and name[:1].isdigit(): 

71 names.append(name) 

72 return sorted(names) 

73 

74 

75def snapshot_prompts(prompts_dir: str, 

76 max_keep: Optional[int] = None) -> Optional[str]: 

77 """Take one snapshot of prompts_dir. Returns the snapshot dir 

78 name on success, None on failure or no-op. 

79 

80 Best-effort: any IOError / permission issue is logged at debug 

81 and returns None so the boot path never crashes on a backup 

82 failure. 

83 

84 Idempotent within the same second: if a snapshot for the current 

85 YYYYMMDD_HHMMSS already exists, returns that name without copying 

86 again. (Multiple HARTOS boots within one second is rare but the 

87 guard avoids a partial-overwrite race.) 

88 """ 

89 if not os.path.isdir(prompts_dir): 

90 logger.debug(f'snapshot_prompts: source missing: {prompts_dir}') 

91 return None 

92 # Don't snapshot empty prompts/. Avoids creating misleading 

93 # "everything was wiped" snapshots when the user starts a fresh 

94 # install before any agent exists. 

95 try: 

96 entries = [f for f in os.listdir(prompts_dir) if f.endswith('.json')] 

97 except OSError as e: 

98 logger.debug(f'snapshot_prompts: cannot list {prompts_dir}: {e}') 

99 return None 

100 if not entries: 

101 logger.debug('snapshot_prompts: prompts dir empty, skipping') 

102 return None 

103 

104 snap_name = time.strftime('%Y%m%d_%H%M%S') 

105 root = _snapshots_root(prompts_dir) 

106 snap_path = os.path.join(root, snap_name) 

107 if os.path.isdir(snap_path): 

108 logger.debug(f'snapshot_prompts: {snap_name} already exists') 

109 return snap_name 

110 

111 try: 

112 os.makedirs(snap_path, exist_ok=True) 

113 except OSError as e: 

114 logger.debug(f'snapshot_prompts: cannot create {snap_path}: {e}') 

115 return None 

116 

117 # Belt + suspenders for M3 (post-shipment review): even though 

118 # the canonical recipe writers were converted to atomic temp+rename, 

119 # third-party writers / older HARTOS versions / hand-edits could 

120 # still leave torn JSON. Validate each file's JSON before copying; 

121 # skip + log on parse failure so the snapshot only contains valid 

122 # restorable state. 

123 import json as _json 

124 copied = 0 

125 skipped_corrupt = 0 

126 for fname in entries: 

127 src = os.path.join(prompts_dir, fname) 

128 dst = os.path.join(snap_path, fname) 

129 try: 

130 with open(src, 'r', encoding='utf-8') as _src_f: 

131 _content = _src_f.read() 

132 try: 

133 _json.loads(_content) 

134 except _json.JSONDecodeError as je: 

135 logger.warning( 

136 f'snapshot_prompts: skipping torn/corrupt {fname} ' 

137 f'({je}) - snapshot stays restore-safe') 

138 skipped_corrupt += 1 

139 continue 

140 shutil.copy2(src, dst) 

141 copied += 1 

142 except (IOError, OSError) as e: 

143 logger.debug(f'snapshot_prompts: copy {fname} failed: {e}') 

144 if skipped_corrupt: 

145 logger.warning( 

146 f'snapshot_prompts: {skipped_corrupt} torn/corrupt files ' 

147 f'skipped in snapshot {snap_name}') 

148 if copied == 0: 

149 # Empty snapshot is worse than no snapshot. 

150 try: 

151 os.rmdir(snap_path) 

152 except OSError: 

153 pass 

154 return None 

155 

156 logger.info( 

157 f'prompts_backup: snapshot {snap_name} saved ' 

158 f'({copied}/{len(entries)} files)') 

159 

160 # Retention: prune old snapshots beyond max_keep. 

161 keep = max_keep if max_keep is not None else MAX_DAILY_SNAPSHOTS 

162 _prune_old_snapshots(prompts_dir, keep=keep) 

163 return snap_name 

164 

165 

166def _prune_old_snapshots(prompts_dir: str, keep: int) -> int: 

167 """Delete snapshots beyond the most-recent *keep*. 

168 

169 Returns count of pruned snapshots. 

170 """ 

171 if keep < 1: 

172 return 0 

173 snapshots = list_snapshots(prompts_dir) 

174 excess = len(snapshots) - keep 

175 if excess <= 0: 

176 return 0 

177 pruned = 0 

178 root = _snapshots_root(prompts_dir) 

179 for name in snapshots[:excess]: # oldest first 

180 full = os.path.join(root, name) 

181 try: 

182 shutil.rmtree(full) 

183 pruned += 1 

184 logger.info(f'prompts_backup: pruned old snapshot {name}') 

185 except (IOError, OSError) as e: 

186 logger.debug(f'prompts_backup: prune {name} failed: {e}') 

187 return pruned 

188 

189 

190def snapshot_at_boot() -> Optional[str]: 

191 """Convenience entry-point for HARTOS boot. Resolves the 

192 prompts directory via core.platform_paths and snapshots it. 

193 

194 Returns the snapshot name on success, None on no-op or failure. 

195 """ 

196 try: 

197 from core.platform_paths import get_prompts_dir 

198 prompts_dir = get_prompts_dir() 

199 except Exception as e: 

200 logger.debug(f'snapshot_at_boot: cannot resolve prompts_dir: {e}') 

201 return None 

202 return snapshot_prompts(prompts_dir)