Coverage for integrations / agent_engine / self_build_tools.py: 44.3%
183 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Unified Agent Goal Engine - Self-Build AutoGen Tools
4Tools for the coding agent to modify the OS at runtime via NixOS
5self-build. **Sandbox-first**: every change must pass a dry-run
6build before it can be applied to the live system.
8Pipeline: stage change -> dry-run (sandbox) -> apply (switch)
9 |
10 v
11 fail? -> rollback staged changes
13Same registration pattern as finance_tools.py / upgrade_tools.py.
14"""
15import json
16import logging
17import os
18import re
19import subprocess
20from typing import Annotated
22logger = logging.getLogger('hevolve_social')
24# Allowed package name pattern — prevents injection
25_PKG_RE = re.compile(r'^[a-zA-Z][a-zA-Z0-9_.-]{0,127}$')
26_RUNTIME_NIX = '/etc/hart/runtime.nix'
27_BUILDS_LOG = '/var/lib/hart/ota/history/builds.jsonl'
30def _is_nixos() -> bool:
31 """Check if we're running on NixOS."""
32 try:
33 result = subprocess.run(
34 ['nixos-version'], capture_output=True, text=True, timeout=5)
35 return result.returncode == 0
36 except Exception:
37 return False
40def _read_runtime_packages() -> list:
41 """Parse package list from runtime.nix."""
42 packages = []
43 if not os.path.isfile(_RUNTIME_NIX):
44 return packages
45 try:
46 with open(_RUNTIME_NIX) as f:
47 in_packages = False
48 for line in f:
49 stripped = line.strip()
50 if 'systemPackages' in stripped:
51 in_packages = True
52 continue
53 if in_packages and stripped == '];':
54 break
55 if in_packages and stripped and not stripped.startswith('#'):
56 packages.append(stripped)
57 except Exception:
58 pass
59 return packages
62def _stage_package(package: str) -> dict:
63 """Add a package to runtime.nix (staged, not yet built)."""
64 if not os.path.isfile(_RUNTIME_NIX):
65 return {'success': False, 'error': 'runtime.nix not found'}
66 try:
67 with open(_RUNTIME_NIX) as f:
68 content = f.read()
69 if package in content:
70 return {'success': True, 'status': 'already_present', 'package': package}
71 content = content.replace(
72 '# Packages added at runtime appear here',
73 f'# Packages added at runtime appear here\n {package}')
74 with open(_RUNTIME_NIX, 'w') as f:
75 f.write(content)
76 return {'success': True, 'status': 'staged', 'package': package}
77 except PermissionError:
78 return {'success': False, 'error': 'Permission denied writing runtime.nix'}
81def _unstage_package(package: str) -> dict:
82 """Remove a package from runtime.nix."""
83 if not os.path.isfile(_RUNTIME_NIX):
84 return {'success': False, 'error': 'runtime.nix not found'}
85 try:
86 with open(_RUNTIME_NIX) as f:
87 lines = f.readlines()
88 new_lines = [l for l in lines
89 if package not in l.strip() or l.strip().startswith('#')]
90 if len(new_lines) == len(lines):
91 return {'success': True, 'status': 'not_found', 'package': package}
92 with open(_RUNTIME_NIX, 'w') as f:
93 f.writelines(new_lines)
94 return {'success': True, 'status': 'removed', 'package': package}
95 except PermissionError:
96 return {'success': False, 'error': 'Permission denied writing runtime.nix'}
99def _run_build(mode: str, timeout: int = 600) -> dict:
100 """Run hart-self-build with the given mode."""
101 try:
102 result = subprocess.run(
103 ['hart-self-build', mode],
104 capture_output=True, text=True, timeout=timeout)
105 return {
106 'success': result.returncode == 0,
107 'mode': mode,
108 'returncode': result.returncode,
109 'output': result.stdout[-3000:] if result.stdout else '',
110 'errors': result.stderr[-1500:] if result.stderr else '',
111 }
112 except subprocess.TimeoutExpired:
113 return {'success': False, 'error': f'Build timed out ({timeout}s)'}
114 except FileNotFoundError:
115 return {'success': False, 'error': 'hart-self-build not available'}
118def register_self_build_tools(helper, assistant, user_id: str):
119 """Register self-build tools with an AutoGen agent (Tier 2).
121 All mutating tools enforce sandbox-first: dry-run must pass
122 before switch is allowed.
123 """
125 def get_self_build_status() -> str:
126 """Get current OS self-build status: NixOS version, generation,
127 runtime packages, and recent build history."""
128 info = {'self_build_available': _is_nixos()}
129 try:
130 result = subprocess.run(
131 ['nixos-version'], capture_output=True, text=True, timeout=5)
132 if result.returncode == 0:
133 info['nixos_version'] = result.stdout.strip()
134 except Exception:
135 pass
137 gen_link = '/nix/var/nix/profiles/system'
138 if os.path.islink(gen_link):
139 info['current_generation'] = os.readlink(gen_link)
141 info['runtime_config_exists'] = os.path.isfile(_RUNTIME_NIX)
142 info['runtime_packages'] = _read_runtime_packages()
144 if os.path.isfile(_BUILDS_LOG):
145 try:
146 with open(_BUILDS_LOG) as f:
147 lines = f.readlines()
148 info['recent_builds'] = [
149 json.loads(l) for l in lines[-5:] if l.strip()]
150 except Exception:
151 pass
153 # Check allowAgentBuilds flag
154 info['agent_builds_allowed'] = os.environ.get(
155 'HART_ALLOW_AGENT_BUILDS', 'false').lower() == 'true'
157 return json.dumps(info)
159 def install_package(
160 package: Annotated[str, "NixOS package name (e.g. 'htop', 'nodejs_20')"],
161 ) -> str:
162 """Stage a package for installation in the OS runtime config.
164 This ONLY stages the change in runtime.nix. You MUST call
165 sandbox_test_build() to verify it builds, then apply_build()
166 to make it live. Never skip the sandbox step.
167 """
168 if not _PKG_RE.match(package):
169 return json.dumps({'error': f'Invalid package name: {package}'})
170 if not _is_nixos():
171 return json.dumps({'error': 'Not running on NixOS'})
173 result = _stage_package(package)
174 if result['success']:
175 result['next_step'] = 'Call sandbox_test_build() to verify this builds correctly'
176 return json.dumps(result)
178 def remove_package(
179 package: Annotated[str, "NixOS package name to remove"],
180 ) -> str:
181 """Stage a package removal from the OS runtime config.
183 This ONLY removes it from runtime.nix. You MUST call
184 sandbox_test_build() to verify, then apply_build() to make live.
185 """
186 if not package or not package.strip():
187 return json.dumps({'error': 'Package name required'})
188 if not _is_nixos():
189 return json.dumps({'error': 'Not running on NixOS'})
191 result = _unstage_package(package.strip())
192 if result['success']:
193 result['next_step'] = 'Call sandbox_test_build() to verify this builds correctly'
194 return json.dumps(result)
196 def sandbox_test_build() -> str:
197 """Run a dry-run build to test staged changes in a sandbox.
199 This is MANDATORY before apply_build(). It evaluates the full
200 NixOS configuration without actually switching, catching any
201 errors (missing packages, syntax errors, dependency conflicts).
203 Returns success=true only if the build would succeed.
204 """
205 if not _is_nixos():
206 return json.dumps({'error': 'Not running on NixOS'})
208 result = _run_build('dry-run', timeout=300)
209 if result['success']:
210 result['sandbox_passed'] = True
211 result['next_step'] = 'Safe to call apply_build() to make changes live'
212 else:
213 result['sandbox_passed'] = False
214 result['next_step'] = (
215 'Fix the errors, then call sandbox_test_build() again. '
216 'Do NOT call apply_build() until sandbox passes.'
217 )
218 return json.dumps(result)
220 def apply_build() -> str:
221 """Apply staged changes by rebuilding the OS (nixos-rebuild switch).
223 REQUIRES: sandbox_test_build() must have passed first.
224 This tool checks the build log to verify a recent successful
225 dry-run before proceeding.
227 Creates a new NixOS generation with instant rollback available.
228 """
229 if not _is_nixos():
230 return json.dumps({'error': 'Not running on NixOS'})
232 # Check agent builds are allowed
233 if os.environ.get('HART_ALLOW_AGENT_BUILDS', 'false').lower() != 'true':
234 return json.dumps({
235 'error': 'Agent-triggered builds are disabled. '
236 'Set HART_ALLOW_AGENT_BUILDS=true or enable '
237 'selfBuild.allowAgentBuilds in NixOS config.'
238 })
240 # Verify a recent dry-run passed (within last 10 minutes)
241 dry_run_verified = False
242 if os.path.isfile(_BUILDS_LOG):
243 try:
244 import time
245 with open(_BUILDS_LOG) as f:
246 lines = f.readlines()
247 for line in reversed(lines):
248 if not line.strip():
249 continue
250 entry = json.loads(line)
251 if entry.get('mode') == 'dry-run' and entry.get('success'):
252 # Check timestamp (within 600 seconds)
253 ts = entry.get('timestamp', '')
254 if ts:
255 from datetime import datetime
256 try:
257 build_time = datetime.fromisoformat(ts)
258 age = (datetime.now() - build_time).total_seconds()
259 if age < 600:
260 dry_run_verified = True
261 except Exception:
262 pass
263 break
264 except Exception:
265 pass
267 if not dry_run_verified:
268 return json.dumps({
269 'error': 'No recent successful dry-run found. '
270 'You MUST call sandbox_test_build() first and it must pass. '
271 'This is a safety requirement — never skip the sandbox.',
272 'sandbox_passed': False,
273 })
275 result = _run_build('switch', timeout=600)
276 if result['success']:
277 result['message'] = (
278 'OS rebuilt successfully. New generation active. '
279 'Previous generation available for instant rollback.'
280 )
281 else:
282 result['message'] = (
283 'Build failed during apply. The system is unchanged — '
284 'NixOS builds are atomic. Check errors and try again.'
285 )
286 return json.dumps(result)
288 def show_build_diff() -> str:
289 """Show what would change between current system and staged config.
291 Runs `nixos-rebuild build --dry-run` diff output to show exactly
292 which packages/services would be added, removed, or changed.
293 """
294 if not _is_nixos():
295 return json.dumps({'error': 'Not running on NixOS'})
296 return json.dumps(_run_build('diff', timeout=120))
298 def list_generations() -> str:
299 """List available NixOS generations for rollback.
301 Each generation is a complete, bootable system snapshot.
302 Rollback is instant and risk-free.
303 """
304 generations = []
305 profile_dir = '/nix/var/nix/profiles'
306 if os.path.isdir(profile_dir):
307 try:
308 for entry in sorted(os.listdir(profile_dir), reverse=True):
309 if entry.startswith('system-') and entry.endswith('-link'):
310 gen_num = entry.replace('system-', '').replace('-link', '')
311 target = os.readlink(os.path.join(profile_dir, entry))
312 generations.append({
313 'generation': gen_num, 'path': target})
314 except Exception:
315 pass
316 current = ''
317 if os.path.islink(os.path.join(profile_dir, 'system')):
318 current = os.readlink(os.path.join(profile_dir, 'system'))
319 return json.dumps({
320 'current': current,
321 'generations': generations[:20],
322 'rollback_available': len(generations) > 1,
323 })
325 def rollback_build(
326 reason: Annotated[str, "Why rollback is needed"] = '',
327 ) -> str:
328 """Rollback to the previous NixOS generation.
330 Instant and risk-free — the previous generation is a complete
331 system that was already proven to work.
332 """
333 if not _is_nixos():
334 return json.dumps({'error': 'Not running on NixOS'})
336 logger.warning(f"Agent-triggered rollback: {reason}")
338 try:
339 result = subprocess.run(
340 ['sudo', 'nixos-rebuild', 'switch', '--rollback'],
341 capture_output=True, text=True, timeout=300)
342 return json.dumps({
343 'success': result.returncode == 0,
344 'status': 'rolled_back' if result.returncode == 0 else 'failed',
345 'reason': reason,
346 'output': result.stdout[-2000:] if result.stdout else '',
347 })
348 except FileNotFoundError:
349 return json.dumps({'error': 'nixos-rebuild not available'})
351 tools = [
352 ('get_self_build_status',
353 'Get OS self-build status: version, generation, packages, build history',
354 get_self_build_status),
355 ('install_package',
356 'Stage a NixOS package for installation (must sandbox_test_build before applying)',
357 install_package),
358 ('remove_package',
359 'Stage a NixOS package removal (must sandbox_test_build before applying)',
360 remove_package),
361 ('sandbox_test_build',
362 'MANDATORY: Test staged changes in sandbox (dry-run build) before applying',
363 sandbox_test_build),
364 ('apply_build',
365 'Apply staged changes to live OS (requires prior sandbox_test_build pass)',
366 apply_build),
367 ('show_build_diff',
368 'Show what would change between current system and staged config',
369 show_build_diff),
370 ('list_generations',
371 'List available NixOS generations for rollback',
372 list_generations),
373 ('rollback_build',
374 'Rollback to previous NixOS generation (instant, risk-free)',
375 rollback_build),
376 ]
378 for name, desc, func in tools:
379 helper.register_for_llm(name=name, description=desc)(func)
380 assistant.register_for_execution(name=name)(func)
382 logger.info(f"Registered {len(tools)} self-build tools for user {user_id}")
385# Tool descriptors for non-AutoGen registration (e.g. upgrade_tools.py pattern)
386SELF_BUILD_TOOLS = [
387 {'name': 'get_self_build_status',
388 'description': 'Get OS self-build status: version, generation, packages, build history.',
389 'function': lambda: get_self_build_status_standalone()},
390 {'name': 'sandbox_test_build',
391 'description': 'Test staged changes in sandbox before applying.',
392 'function': lambda: sandbox_test_build_standalone()},
393]
396def get_self_build_status_standalone() -> dict:
397 """Standalone version for non-AutoGen contexts."""
398 info = {'self_build_available': _is_nixos()}
399 info['runtime_packages'] = _read_runtime_packages()
400 info['agent_builds_allowed'] = os.environ.get(
401 'HART_ALLOW_AGENT_BUILDS', 'false').lower() == 'true'
402 return info
405def sandbox_test_build_standalone() -> dict:
406 """Standalone version for non-AutoGen contexts."""
407 if not _is_nixos():
408 return {'success': False, 'error': 'Not running on NixOS'}
409 return _run_build('dry-run', timeout=300)