Coverage for core / config_cache.py: 78.8%
85 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Cached configuration loader.
4Replaces repeated `open("config.json")` calls across helper.py, create_recipe.py,
5reuse_recipe.py, and hart_intelligence with a single cached load.
7Before: config.json read 3+ times at module import (once per file).
8After: config.json read exactly once, cached in memory.
10Configuration Loading Priority (highest to lowest):
111. Environment variables — always checked first by get_secret()
122. Encrypted vault (SecretsManager) — if migrated to encrypted storage
133. config.json (standalone) — developer mode, repo root
144. langchain_config.json (bundled) — Nunba/cx_Freeze, next to executable
155. Empty dict fallback — env-vars-only mode
17Deployment mode detection:
18- Bundled (Nunba): sys.frozen == True → looks for langchain_config.json next to .exe
19- Standalone: looks for config.json in repo root (parent of core/)
20- HART OS: /etc/hart/hart.env loaded by systemd, no config.json needed
22Note: Nunba's AIKeyVault loads encrypted keys into env vars BEFORE
23config_cache runs. get_secret() checks env vars first, so vault keys
24always take precedence.
26See deploy/deployment-manifest.json for the full deployment mode matrix
27including tier definitions, service port assignments, and variant configs.
28"""
30import json
31import os
32import logging
33import threading
35logger = logging.getLogger('hevolve_core')
37_config = None
38_config_lock = threading.Lock()
41def get_config() -> dict:
42 """
43 Load config.json once and cache it.
44 Thread-safe singleton pattern.
45 """
46 global _config
47 if _config is not None:
48 return _config
50 with _config_lock:
51 # Double-check after acquiring lock
52 if _config is not None:
53 return _config
55 # Try encrypted vault first (security module)
56 try:
57 from security.secrets_manager import SecretsManager
58 mgr = SecretsManager()
59 # If secrets manager has been migrated, use it
60 if mgr._secrets:
61 _config = dict(mgr._secrets)
62 logger.info("Config loaded from encrypted vault")
63 return _config
64 except Exception:
65 pass
67 # Fall back to config.json (standalone) or langchain_config.json (bundled)
68 _base_dir = os.path.dirname(os.path.dirname(__file__))
69 # In frozen/bundled mode, config lives next to the exe as langchain_config.json
70 if getattr(__import__('sys'), 'frozen', False):
71 _base_dir = os.path.dirname(__import__('sys').executable)
72 _config_candidates = [
73 os.path.join(_base_dir, 'config.json'),
74 os.path.join(_base_dir, 'langchain_config.json'),
75 ]
76 for _cp in _config_candidates:
77 try:
78 with open(_cp, 'r') as f:
79 _config = json.load(f)
80 logger.info(f"Config loaded from {os.path.basename(_cp)}")
81 return _config
82 except FileNotFoundError:
83 continue
84 logger.warning("config.json not found, using environment variables only")
85 _config = {}
87 return _config
90def get_secret(name: str, default: str = '') -> str:
91 """
92 Get a configuration value by name.
93 Checks environment variable first, then cached config.
94 """
95 # Env vars take precedence
96 env_val = os.environ.get(name)
97 if env_val:
98 return env_val
100 config = get_config()
101 return config.get(name, default)
104def reload_config():
105 """Force reload of configuration (for testing or after migration)."""
106 global _config
107 with _config_lock:
108 _config = None
109 return get_config()
112# ── Endpoint Resolution ──
113# Single source of truth for API URLs.
114# In bundled Nunba mode (NUNBA_BUNDLED=1), all DB/action/prompt/vision
115# endpoints resolve to the local Flask server (localhost:5000).
116# In standalone/cloud mode, they resolve to cloud URLs from config.json.
118def _local_base() -> str:
119 """Local Nunba server base URL."""
120 return f"http://localhost:{os.environ.get('NUNBA_PORT', '5000')}"
123def is_bundled() -> bool:
124 """True when running inside Nunba (pip-installed, bundled, or frozen)."""
125 return bool(os.environ.get('NUNBA_BUNDLED') or getattr(__import__('sys'), 'frozen', False))
128def get_db_url() -> str:
129 """Database API base URL (replaces hardcoded mailer.hertzai.com).
131 Resolves to the LOCAL Nunba backend in bundled mode — every
132 db_routes.py endpoint (/create_action, /createpromptlist,
133 /getprompt, /getprompt_onlyuserid, /getprompt_all, …) is mounted
134 on the same Flask process, so the same DB_URL covers
135 write+read+update from inside the bundled app.
137 For cross-device CLOUD reads (e.g. /prompts wants to merge in
138 agents the user owns on hevolve.ai but never synced down to this
139 machine), use get_central_db_url() instead — that one always
140 points at the central cloud regardless of bundle mode.
141 """
142 if is_bundled():
143 return _local_base()
144 return get_secret('DB_URL', get_config().get('IP_ADDRESS', {}).get('database_url', ''))
147# Central cloud DB base URL. Pinned to the kong-routed central
148# instance; does NOT collapse to localhost in bundled mode the way
149# get_db_url() does. Override with HEVOLVE_CENTRAL_DB_URL when
150# running against a private Hevolve installation. Resolves to '' if
151# the env override is set to empty, which callers must treat as
152# "no central available — skip cross-device merge".
153_DEFAULT_CENTRAL_DB_URL = 'https://azurekong.hertzai.com:8443/db'
156def get_central_db_url() -> str:
157 """Central cloud DB base URL — always the central instance.
159 Used by readers that want cross-device data (e.g. /prompts merging
160 in agents created on another device, /prompts/public showing
161 cloud-only catalogue entries). Distinct from get_db_url() because
162 that one collapses to localhost in bundled mode and would silently
163 no-op the merge.
165 Default: kong-routed central instance. Override:
166 HEVOLVE_CENTRAL_DB_URL=https://my-private-cloud:8443/db
167 Empty override is honored ('' = disable cross-device reads).
168 """
169 override = os.environ.get('HEVOLVE_CENTRAL_DB_URL')
170 if override is not None:
171 return override
172 return _DEFAULT_CENTRAL_DB_URL
175def get_stop_api_url() -> str:
176 """VLM computer-use stop endpoint URL.
178 Resolves to HARTOS's local /api/vlm/stop in bundled mode (the
179 handler ported from OmniParser/omnitool/gradio/agentic_rpc.py
180 lives at hart_intelligence_entry.py:vlm_stop). When the user
181 clicks Stop in Nunba's indicator window, this URL is what gets
182 POST'd; the handler flips the stop flag in
183 integrations.vlm.local_loop._vlm_stop_flags and the next
184 iteration of run_local_agentic_loop exits cleanly with
185 exit_reason='stopped' before another action runs on the screen.
187 Standalone mode falls back to the env-var override (if set) or
188 empty string (callers skip the POST when falsy).
190 Override:
191 HEVOLVE_STOP_API_URL=http://my-cluster:5001/api/vlm/stop
192 Empty override is honored ('' = disable).
194 Used to live as a hardcoded `http://gcp_training2.hertzai.com:5001/stop`
195 literal in Nunba's main.py and app.py — pointed at a cloud
196 OmniParser instance that's not part of any current local
197 install. The /stop endpoint ITSELF was never ported when the
198 VLM execution loop moved into HARTOS, so the call timed out on
199 every shutdown of every install for months. Now the endpoint
200 is in HARTOS and this resolver wires Nunba to it correctly.
201 """
202 override = os.environ.get('HEVOLVE_STOP_API_URL')
203 if override is not None:
204 return override
205 if is_bundled():
206 return f"{_local_base()}/api/vlm/stop"
207 return ''
210def get_action_api() -> str:
211 """Action API URL for create/query actions."""
212 if is_bundled():
213 return f'{_local_base()}/create_action'
214 return get_secret('ACTION_API', get_config().get('IP_ADDRESS', {}).get('database_url', ''))
217def get_student_api() -> str:
218 """Student profile API URL."""
219 if is_bundled():
220 return f'{_local_base()}/getstudent_by_user_id'
221 return get_secret('STUDENT_API', '')
224def get_vision_api() -> str:
225 """Vision/image inference API URL."""
226 if is_bundled():
227 return f'{_local_base()}/upload/vision'
228 return get_secret('LLAVA_API', '')
231def get_book_parsing_api() -> str:
232 """PDF/book parsing API URL."""
233 if is_bundled():
234 return f'{_local_base()}/upload/parse_pdf'
235 return get_secret('BOOKPARSING_API', '')
238def get_visual_context_api(user_id, mins=5) -> str:
239 """Visual context query URL (recent actions by time window)."""
240 base = _local_base() if is_bundled() else os.environ.get('HEVOLVE_MAILER_URL', _local_base())
241 return f'{base}/get_visual_bymins?user_id={user_id}&mins={mins}'