Coverage for integrations / coding_agent / backend_repair_tools.py: 71.4%
35 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""Backend Repair Tools — agent-callable wrappers over Nunba's
2per-backend TTS venv install/wipe layer.
4Why this exists
5---------------
6``error_advice.handle_exception(category='tts.probe',
7context={'backend': 'indic_parler'})`` creates ``self_heal`` AgentGoals
8that the coding daemon picks up. The autoresearch tool family in
9``autoevolve_code_tools.py`` only does source edits — useful when the
10fix is "patch a Python file", useless when the fix is "the venv at
11``~/Documents/Nunba/data/venvs/indic_parler/`` is missing the
12``parler_tts`` package". Source-edit alone leaves the live broken
13venv broken until the next rebuild.
15This module gives the agent a tool that actually repairs the venv:
16``repair_backend_venv(backend_name, wipe_first=False)`` calls Nunba's
17canonical ``install_backend_full()``, the same function the user-side
18"Set up TTS" UI uses.
20Wired in
21--------
22* ``mcp_http_bridge.py`` registers ``BACKEND_REPAIR_TOOLS`` for autogen
23 via the same module-list registration loop that exposes
24 ``AUTOEVOLVE_CODE_TOOLS`` / ``AUTO_EVOLVE_TOOLS`` /
25 ``THOUGHT_EXPERIMENT_TOOLS`` — see
26 ``mcp_http_bridge._load_tools`` around the registration tuple list.
27* ``goal_manager._build_self_heal_prompt`` mentions
28 ``repair_backend_venv`` when the goal's ``category`` is one of the
29 TTS / subprocess install-failure shapes AND ``context.backend`` is
30 set on the goal config.
32Bundled-mode-only contract
33--------------------------
34In source-mode HARTOS (a checkout without the Nunba freeze), Nunba's
35``tts.*`` modules are not on ``sys.path``. The tool returns a
36graceful ``{success: False, message: 'Backend repair tools require
37the Nunba bundled environment ...'}`` JSON so the agent can mark the
38goal failed instead of crashing the autogen group.
39"""
41import json
42import logging
43from pathlib import Path
44from typing import Optional
46logger = logging.getLogger('hevolve.backend_repair')
49def _get_known_backends() -> set:
50 """Return engine_ids known to HARTOS's ``ENGINE_REGISTRY``.
52 Single source of truth — never hand-maintain a second list. When
53 the registry can't be loaded (HARTOS-side circular import during
54 boot, missing module), returns an empty set so the caller fails
55 closed with a clear message rather than silently accepting any
56 string.
57 """
58 try:
59 from integrations.channels.media.tts_router import ENGINE_REGISTRY
60 return {spec.engine_id for spec in ENGINE_REGISTRY.values()}
61 except Exception as e:
62 logger.warning(
63 f"backend_repair_tools: ENGINE_REGISTRY unavailable for "
64 f"validation ({type(e).__name__}: {e})"
65 )
66 return set()
69def _resolve_log_path(backend_name: str) -> Optional[str]:
70 """Return the path Nunba writes pip output to for this backend.
72 Mirrors ``tts.backend_venv._venv_log_path`` so the agent can grep
73 a specific file when ``install_backend_full`` returns False.
74 Best-effort — returns None if the path cannot be derived (the
75 repair still runs; only the diagnostic hint is missing).
76 """
77 try:
78 from core.platform_paths import get_log_dir # type: ignore
79 log_dir = Path(get_log_dir())
80 except Exception:
81 log_dir = Path.home() / 'Documents' / 'Nunba' / 'logs'
82 try:
83 return str(log_dir / f'venv_{backend_name}.log')
84 except Exception:
85 return None
88def repair_backend_venv(backend_name: str, wipe_first: bool = False) -> str:
89 """Reinstall a TTS backend's per-engine venv.
91 Use this tool when a ``self_heal`` goal carries
92 ``category='tts.probe'``, ``category='tts.install'``,
93 ``category='tts.install.self_heal_exhausted'``, or
94 ``category='subprocess.tool_load'`` AND ``context.backend`` names
95 a broken backend.
97 The work is delegated to Nunba's existing
98 ``tts.package_installer.install_backend_full`` — the same function
99 the user-side "Set up TTS" UI calls. That function:
101 1. Creates the per-engine venv if missing (idempotent).
102 2. Routes the canonical pip_install_plan from
103 ``ENGINE_REGISTRY[backend].pip_install_plan`` into either the
104 backend's venv (when ``install_target='venv'``) or the main
105 interpreter (when ``install_target='main'``).
106 3. Applies post-install patches.
107 4. Downloads model weights via huggingface_hub.
109 Args:
110 backend_name: One of the engine_ids in
111 ``integrations.channels.media.tts_router.ENGINE_REGISTRY``.
112 Common names: ``'indic_parler'``, ``'kokoro'``,
113 ``'melotts'``, ``'f5_tts'``, ``'chatterbox_turbo'``,
114 ``'chatterbox'``, ``'omnivoice'``, ``'piper'``,
115 ``'pocket_tts'``, ``'neutts_air'``.
116 wipe_first: When True, delete the existing venv directory
117 before reinstall. Use after corruption, partial installs,
118 or when changing transitive-dep cages. Default False
119 (idempotent reinstall — pip skips already-satisfied specs).
121 Returns:
122 JSON string with:
123 ``success`` (bool): pip + model_weights both succeeded.
124 ``backend`` (str): the validated backend name.
125 ``message`` (str): human-readable status.
126 ``log_path`` (str | None): path to the venv install log
127 (``~/Documents/Nunba/logs/venv_<backend>.log``).
128 ``wiped`` (bool): whether ``wipe_venv`` was actually called.
129 """
130 log_path = _resolve_log_path(backend_name)
132 # Validate against ENGINE_REGISTRY — prevents path-traversal /
133 # arbitrary-name injection BEFORE any filesystem touch.
134 known = _get_known_backends()
135 if not known:
136 return json.dumps({
137 'success': False,
138 'backend': backend_name,
139 'message': (
140 'Cannot validate backend: HARTOS ENGINE_REGISTRY '
141 'unavailable in this process. Likely a non-bundled '
142 'HARTOS server without tts_router on sys.path.'
143 ),
144 'log_path': log_path,
145 'wiped': False,
146 })
147 if backend_name not in known:
148 return json.dumps({
149 'success': False,
150 'backend': backend_name,
151 'message': (
152 f"Unknown backend {backend_name!r}. "
153 f"Known backends: {sorted(known)}"
154 ),
155 'log_path': log_path,
156 'wiped': False,
157 })
159 # Lazy-import Nunba's venv layer. In source-mode HARTOS the
160 # imports fail — return a structured "not in bundled" message.
161 try:
162 from tts.backend_venv import wipe_venv as _wipe_venv # type: ignore
163 from tts.package_installer import install_backend_full # type: ignore
164 except ImportError as e:
165 return json.dumps({
166 'success': False,
167 'backend': backend_name,
168 'message': (
169 f'Backend repair requires the Nunba bundled environment '
170 f'(tts.backend_venv + tts.package_installer must be '
171 f'importable). ImportError: {e}. Source-mode HARTOS '
172 f'deploys cannot reinstall TTS venvs — the agent should '
173 f'mark this self_heal goal failed and the operator '
174 f'should run the repair from a Nunba install.'
175 ),
176 'log_path': log_path,
177 'wiped': False,
178 })
180 wiped = False
181 if wipe_first:
182 try:
183 _wipe_venv(backend_name)
184 wiped = True
185 logger.info(f"repair_backend_venv: wiped {backend_name!r}")
186 except Exception as e:
187 return json.dumps({
188 'success': False,
189 'backend': backend_name,
190 'message': (
191 f'wipe_venv({backend_name!r}) failed: '
192 f'{type(e).__name__}: {e}'
193 ),
194 'log_path': log_path,
195 'wiped': False,
196 })
198 try:
199 ok, msg = install_backend_full(backend_name)
200 except Exception as e:
201 return json.dumps({
202 'success': False,
203 'backend': backend_name,
204 'message': (
205 f'install_backend_full({backend_name!r}) raised: '
206 f'{type(e).__name__}: {e}'
207 ),
208 'log_path': log_path,
209 'wiped': wiped,
210 })
212 return json.dumps({
213 'success': bool(ok),
214 'backend': backend_name,
215 'message': str(msg),
216 'log_path': log_path,
217 'wiped': wiped,
218 })
221# Tool registration list — same shape as AUTOEVOLVE_CODE_TOOLS /
222# AUTO_EVOLVE_TOOLS / THOUGHT_EXPERIMENT_TOOLS. Consumed by
223# mcp_http_bridge._load_tools via the registration tuple list around
224# line 998. The 'tags' field is documentary — the bridge's dedup is
225# by name, and the goal_manager.tool_tags filter is per goal_type.
226BACKEND_REPAIR_TOOLS = [
227 {
228 'name': 'repair_backend_venv',
229 'func': repair_backend_venv,
230 'description': (
231 'Reinstall a broken TTS backend venv (pip install plan + '
232 'model weights). Use when a self_heal goal carries '
233 'category in {tts.probe, tts.install, '
234 'tts.install.self_heal_exhausted, subprocess.tool_load} '
235 'AND context.backend names the failed backend. Optional '
236 'wipe_first=True for full clean reinstall (use after '
237 'corruption / version-conflict situations).'
238 ),
239 'tags': ['coding', 'self_heal', 'tts'],
240 },
241]