Coverage for src/local_deep_research/web/warning_checks/__init__.py: 88%
131 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +0000
1"""Warning calculation for the settings UI.
3Thin orchestrator: reads settings from a single DB session,
4delegates to pure check functions in hardware.py and context.py.
5"""
7from typing import List
9from flask import session
10from loguru import logger
12from ...database.session_context import get_user_db_session
13from ...utilities.db_utils import get_settings_manager
14from .context import (
15 check_context_below_history,
16 check_context_truncation_history,
17)
18from .backup import (
19 check_backup_disabled,
20 check_backup_healthy,
21 check_no_backups_exist,
22)
23from .hardware import (
24 LOCAL_PROVIDERS,
25 check_high_context,
26 check_legacy_server_config,
27 check_model_mismatch,
28)
29from ...security.egress.policy import DEFAULT_EGRESS_SCOPE
30from ...security.egress.warnings import (
31 check_cloud_embeddings_enabled,
32 check_cloud_llm_enabled,
33 check_effective_scope,
34 check_public_egress_enabled,
35 check_trusted_destinations,
36 check_unprotected_egress,
37)
38from ...constants import DEFAULT_SEARCH_TOOL
41def _safe_check(check_fn, *args, **kwargs):
42 """Run a single warning check, returning None on failure."""
43 try:
44 return check_fn(*args, **kwargs)
45 except Exception:
46 name = getattr(check_fn, "__name__", repr(check_fn))
47 logger.exception(f"Warning check {name} failed")
48 return None
51def calculate_warnings() -> List[dict]:
52 """Calculate current warning conditions based on settings.
54 Uses a single DB session for all setting reads and history queries.
55 """
56 warnings: List[dict] = []
58 try:
59 username = session.get("username")
60 with get_user_db_session(username) as db_session:
61 if not db_session:
62 return []
64 settings_manager = get_settings_manager(db_session, username)
66 # Read all needed settings in one session
67 provider = settings_manager.get_setting(
68 "llm.provider", "ollama"
69 ).lower()
70 local_context = settings_manager.get_setting(
71 "llm.local_context_window_size", 8192
72 )
73 current_model = settings_manager.get_setting("llm.model", "")
74 dismiss_high_context = settings_manager.get_setting(
75 "app.warnings.dismiss_high_context", False
76 )
77 dismiss_model_mismatch = settings_manager.get_setting(
78 "app.warnings.dismiss_model_mismatch", False
79 )
80 dismiss_context_below_history = settings_manager.get_setting(
81 "app.warnings.dismiss_context_below_history", False
82 )
83 dismiss_context_truncation_history = settings_manager.get_setting(
84 "app.warnings.dismiss_context_truncation_history", False
85 )
86 dismiss_legacy_config = settings_manager.get_setting(
87 "app.warnings.dismiss_legacy_config", False
88 )
89 backup_enabled = settings_manager.get_setting(
90 "backup.enabled", True
91 )
92 dismiss_backup_disabled = settings_manager.get_setting(
93 "app.warnings.dismiss_backup_disabled", False
94 )
95 dismiss_no_backups = settings_manager.get_setting(
96 "app.warnings.dismiss_no_backups", False
97 )
99 logger.debug(f"Starting warning calculation - provider={provider}")
101 is_local = provider in LOCAL_PROVIDERS
103 # --- Hardware / settings checks (pure functions) ---
104 w = _safe_check(
105 check_high_context,
106 provider,
107 local_context,
108 dismiss_high_context,
109 )
110 if w:
111 warnings.append(w)
113 w = _safe_check(
114 check_model_mismatch,
115 provider,
116 current_model,
117 local_context,
118 dismiss_model_mismatch,
119 )
120 if w:
121 warnings.append(w)
123 w = _safe_check(check_legacy_server_config, dismiss_legacy_config)
124 if w:
125 warnings.append(w)
127 # --- Egress policy checks ---
128 egress_scope = settings_manager.get_setting(
129 "policy.egress_scope", DEFAULT_EGRESS_SCOPE
130 )
131 require_local_endpoint = bool(
132 settings_manager.get_setting(
133 "llm.require_local_endpoint", False
134 )
135 )
136 embeddings_provider = settings_manager.get_setting(
137 "embeddings.provider", ""
138 )
139 embeddings_base_url = settings_manager.get_setting(
140 "embeddings.openai.base_url", ""
141 )
142 require_local_embeddings = bool(
143 settings_manager.get_setting("embeddings.require_local", False)
144 )
145 primary_engine = settings_manager.get_setting(
146 "search.tool", DEFAULT_SEARCH_TOOL
147 )
148 trusted_inference = settings_manager.get_setting(
149 "policy.trusted_inference_providers", []
150 )
151 trusted_search = settings_manager.get_setting(
152 "policy.trusted_search_engines", []
153 )
155 # Resolve the EFFECTIVE posture so the banners are accurate. For
156 # `adaptive`, this turns the opaque "follows the primary" into a
157 # concrete scope; it also applies the PRIVATE_ONLY -> force-local
158 # coupling, so a private-resolving run doesn't falsely show the
159 # "cloud LLM enabled" banner. Best-effort: any failure falls back
160 # to the raw values (the page must never break on this).
161 effective_scope = str(egress_scope).lower()
162 effective_require_local_endpoint = require_local_endpoint
163 effective_require_local_embeddings = require_local_embeddings
164 try:
165 from ...security.egress.policy import context_from_snapshot
167 _snap = settings_manager.get_settings_snapshot()
168 if isinstance(_snap, dict):
169 # allow_dns=False: this runs on the /api/warnings page-
170 # render hot path; skip the synchronous getaddrinfo that
171 # ADAPTIVE resolution would otherwise do for a URL-engine
172 # primary (could block the render up to _DNS_TIMEOUT_SEC).
173 # The banner is advisory and falls back to static
174 # classification — accuracy here is best-effort by design.
175 _eff_ctx = context_from_snapshot(
176 _snap,
177 primary_engine or DEFAULT_SEARCH_TOOL,
178 username=username,
179 allow_dns=False,
180 )
181 effective_scope = _eff_ctx.scope.value
182 effective_require_local_endpoint = (
183 _eff_ctx.require_local_llm
184 )
185 effective_require_local_embeddings = (
186 _eff_ctx.require_local_embeddings
187 )
188 except Exception:
189 logger.debug(
190 "could not resolve effective egress scope for warnings",
191 exc_info=True,
192 )
194 adaptive_info_dismissed = bool(
195 settings_manager.get_setting(
196 "app.warnings.dismiss_adaptive_scope_info", False
197 )
198 )
200 # Each egress banner has its OWN dismiss flag. Previously all
201 # three shared app.warnings.dismiss_egress_policy, so dismissing
202 # the fresh-install "public egress" notice ALSO permanently hid
203 # the critical cloud-LLM / cloud-embeddings warnings — a
204 # false-safety trap (switch to OpenAI later, never warned).
205 public_egress_dismissed = bool(
206 settings_manager.get_setting(
207 "app.warnings.dismiss_egress_policy", False
208 )
209 )
210 cloud_llm_dismissed = bool(
211 settings_manager.get_setting(
212 "app.warnings.dismiss_cloud_llm", False
213 )
214 )
215 cloud_embeddings_dismissed = bool(
216 settings_manager.get_setting(
217 "app.warnings.dismiss_cloud_embeddings", False
218 )
219 )
221 # Informational: state what ADAPTIVE actually resolves to.
222 w = _safe_check(
223 check_effective_scope,
224 egress_scope,
225 effective_scope,
226 primary_engine,
227 adaptive_info_dismissed,
228 )
229 if w:
230 warnings.append(w)
232 w = _safe_check(
233 check_public_egress_enabled,
234 effective_scope,
235 public_egress_dismissed,
236 )
237 if w:
238 warnings.append(w)
240 # Loud, non-dismissible banner when protection is turned off.
241 w = _safe_check(check_unprotected_egress, egress_scope)
242 if w: 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true
243 warnings.append(w)
245 # Trusted off-machine destinations (stage D) relax classification.
246 w = _safe_check(
247 check_trusted_destinations, trusted_inference, trusted_search
248 )
249 if w: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 warnings.append(w)
252 w = _safe_check(
253 check_cloud_llm_enabled,
254 provider,
255 effective_require_local_endpoint,
256 cloud_llm_dismissed,
257 )
258 if w:
259 warnings.append(w)
261 w = _safe_check(
262 check_cloud_embeddings_enabled,
263 embeddings_provider,
264 embeddings_base_url,
265 effective_require_local_embeddings,
266 cloud_embeddings_dismissed,
267 )
268 if w: 268 ↛ 269line 268 didn't jump to line 269 because the condition on line 268 was never true
269 warnings.append(w)
271 # --- Backup checks ---
272 w = _safe_check(
273 check_backup_disabled, backup_enabled, dismiss_backup_disabled
274 )
275 if w: 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true
276 warnings.append(w)
278 # Check backup file status (lightweight filesystem glob)
279 dismiss_backup_info = settings_manager.get_setting(
280 "app.warnings.dismiss_backup_info", False
281 )
282 try:
283 from ...config.paths import get_user_backup_directory
284 from ...utilities.formatting import human_size
285 from ...database.backup.backup_service import (
286 is_safe_glob_result,
287 )
289 username = session.get("username")
290 if username: 290 ↛ 328line 290 didn't jump to line 328 because the condition on line 290 was always true
291 backup_dir = get_user_backup_directory(username)
292 total_size = 0
293 backup_count = 0
294 for f in backup_dir.glob("ldr_backup_*.db"):
295 # Skip symlinks / entries resolving outside backup_dir
296 # so a planted symlink can't inflate the count/size
297 # shown in warnings — same hardening as BackupService.
298 if not is_safe_glob_result(f, backup_dir): 298 ↛ 300line 298 didn't jump to line 300 because the condition on line 298 was always true
299 continue
300 try:
301 total_size += f.stat().st_size
302 backup_count += 1
303 except FileNotFoundError:
304 continue
306 w = _safe_check(
307 check_no_backups_exist,
308 backup_enabled,
309 backup_count,
310 dismiss_no_backups,
311 )
312 if w:
313 warnings.append(w)
315 w = _safe_check(
316 check_backup_healthy,
317 backup_enabled,
318 backup_count,
319 human_size(total_size),
320 dismiss_backup_info,
321 )
322 if w: 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true
323 warnings.append(w)
324 except Exception:
325 logger.debug("Backup status check skipped")
327 # --- History-based checks (need DB queries) ---
328 if is_local and not dismiss_context_below_history:
329 w = _safe_check(
330 check_context_below_history, db_session, local_context
331 )
332 if w:
333 warnings.append(w)
335 if is_local and not dismiss_context_truncation_history:
336 w = _safe_check(
337 check_context_truncation_history, db_session, local_context
338 )
339 if w:
340 warnings.append(w)
342 except Exception:
343 logger.exception("Error calculating warnings")
345 return warnings