Coverage for integrations / agent_engine / app_installer.py: 91.5%

414 statements  

« prev     ^ index     » next       coverage.py v7.14.0, created at 2026-05-12 04:49 +0000

1""" 

2Unified App Installer — Cross-Platform Package Installation API. 

3 

4Handles installation from ANY platform through a single interface: 

5 - Linux: Nix packages, Flatpak, AppImage 

6 - Windows: .exe/.msi via Wine binfmt integration 

7 - Android: .apk via Android subsystem (binder/ashmem) 

8 - macOS: .app/.dmg via Darling (experimental) 

9 - HART OS: Extensions from extensions/ directory 

10 

11Detection chain: 

12 1. File extension → platform mapping 

13 2. Magic bytes (MZ for PE, PK for APK/ZIP, ELF header) 

14 3. URL pattern → package manager dispatch 

15 

16Each installer type registers in AppRegistry after successful install. 

17""" 

18 

19import hashlib 

20import json 

21import logging 

22import os 

23import shutil 

24import subprocess 

25import tempfile 

26import time 

27from dataclasses import dataclass, field 

28from enum import Enum 

29from typing import Dict, List, Optional 

30 

31logger = logging.getLogger('hevolve.installer') 

32 

33 

34class InstallerPlatform(Enum): 

35 """Platform classification for installers.""" 

36 NIX = 'nix' 

37 FLATPAK = 'flatpak' 

38 APPIMAGE = 'appimage' 

39 WINDOWS = 'windows' 

40 ANDROID = 'android' 

41 MACOS = 'macos' 

42 EXTENSION = 'extension' 

43 UNKNOWN = 'unknown' 

44 

45 

46class InstallStatus(Enum): 

47 """Installation lifecycle states.""" 

48 PENDING = 'pending' 

49 DOWNLOADING = 'downloading' 

50 VERIFYING = 'verifying' 

51 INSTALLING = 'installing' 

52 CONFIGURING = 'configuring' 

53 COMPLETED = 'completed' 

54 FAILED = 'failed' 

55 UNINSTALLING = 'uninstalling' 

56 

57 

58@dataclass 

59class InstallRequest: 

60 """A package installation request.""" 

61 source: str # File path, URL, or package name 

62 platform: InstallerPlatform = InstallerPlatform.UNKNOWN 

63 name: str = '' # Display name (auto-detected if empty) 

64 version: str = '' 

65 sha256: str = '' # Expected hash for verification 

66 options: Dict = field(default_factory=dict) # Platform-specific options 

67 

68 

69@dataclass 

70class InstallResult: 

71 """Result of an installation attempt.""" 

72 success: bool 

73 platform: str 

74 name: str 

75 version: str = '' 

76 install_path: str = '' 

77 app_id: str = '' 

78 error: str = '' 

79 duration_seconds: float = 0.0 

80 

81 

82# ─── Extension → Platform mapping ─────────────────────────── 

83 

84_EXT_PLATFORM_MAP = { 

85 # Windows 

86 '.exe': InstallerPlatform.WINDOWS, 

87 '.msi': InstallerPlatform.WINDOWS, 

88 '.bat': InstallerPlatform.WINDOWS, 

89 # Android 

90 '.apk': InstallerPlatform.ANDROID, 

91 '.xapk': InstallerPlatform.ANDROID, 

92 '.aab': InstallerPlatform.ANDROID, 

93 # macOS 

94 '.dmg': InstallerPlatform.MACOS, 

95 '.app': InstallerPlatform.MACOS, 

96 '.pkg': InstallerPlatform.MACOS, 

97 # Linux 

98 '.flatpakref': InstallerPlatform.FLATPAK, 

99 '.AppImage': InstallerPlatform.APPIMAGE, 

100 '.appimage': InstallerPlatform.APPIMAGE, 

101 # HART OS 

102 '.hartpkg': InstallerPlatform.EXTENSION, 

103} 

104 

105# ─── Magic bytes for binary detection ──────────────────────── 

106 

107_MAGIC_PLATFORM_MAP = { 

108 b'MZ': InstallerPlatform.WINDOWS, # PE executable 

109 b'\x7fELF': InstallerPlatform.APPIMAGE, # Could be AppImage (ELF) 

110 b'PK': InstallerPlatform.ANDROID, # ZIP/APK 

111} 

112 

113 

114def detect_platform(file_path: str) -> InstallerPlatform: 

115 """Detect the platform of an installer file. 

116 

117 Uses extension first, then magic bytes as fallback. 

118 """ 

119 _, ext = os.path.splitext(file_path) 

120 

121 # Extension-based detection 

122 if ext in _EXT_PLATFORM_MAP: 

123 return _EXT_PLATFORM_MAP[ext] 

124 

125 # Magic bytes detection 

126 if os.path.isfile(file_path): 

127 try: 

128 with open(file_path, 'rb') as f: 

129 header = f.read(4) 

130 for magic, platform in _MAGIC_PLATFORM_MAP.items(): 

131 if header[:len(magic)] == magic: 

132 # Distinguish APK (ZIP with AndroidManifest) from regular ZIP 

133 if magic == b'PK': 

134 import zipfile 

135 try: 

136 with zipfile.ZipFile(file_path) as zf: 

137 if 'AndroidManifest.xml' in zf.namelist(): 

138 return InstallerPlatform.ANDROID 

139 except zipfile.BadZipFile: 

140 pass 

141 continue 

142 return platform 

143 except (IOError, PermissionError): 

144 pass 

145 

146 return InstallerPlatform.UNKNOWN 

147 

148 

149def verify_checksum(file_path: str, expected_sha256: str) -> bool: 

150 """Verify SHA256 checksum of a file.""" 

151 if not expected_sha256: 

152 return True # No checksum to verify 

153 sha = hashlib.sha256() 

154 with open(file_path, 'rb') as f: 

155 for chunk in iter(lambda: f.read(8192), b''): 

156 sha.update(chunk) 

157 return sha.hexdigest() == expected_sha256 

158 

159 

160class AppInstaller: 

161 """Unified cross-platform application installer. 

162 

163 Dispatches installation to the appropriate platform handler 

164 based on file type detection. 

165 """ 

166 

167 def __init__(self): 

168 self._install_dir = os.environ.get( 

169 'HART_APP_DIR', '/var/lib/hart/apps') 

170 self._history: List[dict] = [] 

171 

172 def install(self, req: InstallRequest) -> InstallResult: 

173 """Install an application from any platform. 

174 

175 Args: 

176 req: InstallRequest with source path/URL and optional metadata. 

177 

178 Returns: 

179 InstallResult with success status and details. 

180 """ 

181 start = time.time() 

182 

183 # Auto-detect platform if not specified 

184 if req.platform == InstallerPlatform.UNKNOWN: 

185 if req.source.startswith('nixpkgs.') or req.source.startswith('nix:'): 

186 req.platform = InstallerPlatform.NIX 

187 elif req.source.startswith('flathub:') or req.source.startswith('flatpak:'): 

188 req.platform = InstallerPlatform.FLATPAK 

189 elif os.path.isfile(req.source): 

190 req.platform = detect_platform(req.source) 

191 else: 

192 # Try as nix package name 

193 req.platform = InstallerPlatform.NIX 

194 

195 # Verify checksum if file 

196 if os.path.isfile(req.source) and req.sha256: 

197 if not verify_checksum(req.source, req.sha256): 

198 return InstallResult( 

199 success=False, platform=req.platform.value, 

200 name=req.name or os.path.basename(req.source), 

201 error='Checksum verification failed', 

202 duration_seconds=time.time() - start) 

203 

204 # Dispatch to platform handler 

205 handlers = { 

206 InstallerPlatform.NIX: self._install_nix, 

207 InstallerPlatform.FLATPAK: self._install_flatpak, 

208 InstallerPlatform.APPIMAGE: self._install_appimage, 

209 InstallerPlatform.WINDOWS: self._install_windows, 

210 InstallerPlatform.ANDROID: self._install_android, 

211 InstallerPlatform.MACOS: self._install_macos, 

212 InstallerPlatform.EXTENSION: self._install_extension, 

213 } 

214 

215 handler = handlers.get(req.platform) 

216 if not handler: 

217 return InstallResult( 

218 success=False, platform=req.platform.value, 

219 name=req.name or req.source, 

220 error=f'No installer for platform: {req.platform.value}', 

221 duration_seconds=time.time() - start) 

222 

223 result = handler(req) 

224 result.duration_seconds = time.time() - start 

225 

226 # Record in history 

227 self._history.append({ 

228 'name': result.name, 

229 'platform': result.platform, 

230 'success': result.success, 

231 'timestamp': time.time(), 

232 'source': req.source, 

233 'error': result.error, 

234 }) 

235 

236 # Audit log successful installs 

237 if result.success: 

238 try: 

239 from security.immutable_audit_log import get_audit_log 

240 get_audit_log().log_event( 

241 'app_lifecycle', 'app_installer', 

242 f'Installed {result.name}', 

243 detail={ 

244 'platform': result.platform, 

245 'app_id': result.app_id, 

246 'source': req.source, 

247 'duration': round(result.duration_seconds, 2), 

248 }) 

249 except Exception: 

250 pass 

251 

252 # Auto-register in AppRegistry so the app appears in shell/spotlight 

253 self._auto_register_app(result, req) 

254 

255 return result 

256 

257 def uninstall(self, app_id: str, platform: str = '') -> InstallResult: 

258 """Uninstall an application.""" 

259 if platform == 'nix' or not platform: 

260 result = self._uninstall_nix(app_id) 

261 elif platform == 'flatpak': 

262 result = self._uninstall_flatpak(app_id) 

263 elif platform == 'appimage': 

264 result = self._uninstall_appimage(app_id) 

265 elif platform == 'windows': 

266 result = self._uninstall_windows(app_id) 

267 else: 

268 result = InstallResult( 

269 success=False, platform=platform, name=app_id, 

270 error=f'Uninstall not supported for: {platform}') 

271 

272 # Audit log successful uninstalls 

273 if result.success: 

274 try: 

275 from security.immutable_audit_log import get_audit_log 

276 get_audit_log().log_event( 

277 'app_lifecycle', 'app_installer', 

278 f'Uninstalled {app_id}', 

279 detail={ 

280 'platform': result.platform, 

281 'app_id': app_id, 

282 }) 

283 except Exception: 

284 pass 

285 

286 # Auto-unregister from AppRegistry 

287 self._auto_unregister_app(app_id) 

288 

289 return result 

290 

291 def _auto_register_app(self, result: InstallResult, req: InstallRequest): 

292 """Register successfully installed app in AppRegistry for shell/spotlight.""" 

293 try: 

294 from core.platform.registry import get_registry 

295 from core.platform.app_manifest import AppManifest, AppType 

296 

297 registry = get_registry() 

298 if not registry.has('apps'): 

299 return 

300 apps = registry.get('apps') 

301 

302 app_id = result.app_id or result.name.lower().replace(' ', '_') 

303 if apps.get(app_id): 

304 return # Already registered 

305 

306 # Map installer platform to app type 

307 platform_type_map = { 

308 'nix': AppType.DESKTOP_APP.value, 

309 'flatpak': AppType.DESKTOP_APP.value, 

310 'appimage': AppType.DESKTOP_APP.value, 

311 'windows': AppType.DESKTOP_APP.value, 

312 'android': AppType.DESKTOP_APP.value, 

313 'extension': AppType.EXTENSION.value, 

314 } 

315 app_type = platform_type_map.get(result.platform, AppType.DESKTOP_APP.value) 

316 

317 # Build entry dict with required keys per app type 

318 entry = {} 

319 if app_type == AppType.DESKTOP_APP.value: 

320 entry['exec'] = app_id 

321 if result.install_path: 

322 entry['install_path'] = result.install_path 

323 elif app_type == AppType.EXTENSION.value: 

324 entry['module'] = f'extensions.{app_id}' 

325 else: 

326 entry['exec'] = app_id 

327 

328 manifest = AppManifest( 

329 id=app_id, 

330 name=result.name, 

331 version=result.version or '1.0.0', 

332 type=app_type, 

333 icon='apps', 

334 entry=entry, 

335 group='Installed', 

336 tags=['installed', result.platform], 

337 ) 

338 apps.register(manifest) 

339 logger.info(f"Auto-registered app: {app_id} ({result.platform})") 

340 except Exception as e: 

341 logger.debug(f"App auto-register skipped: {e}") 

342 

343 def _auto_unregister_app(self, app_id: str): 

344 """Unregister app from AppRegistry on uninstall.""" 

345 try: 

346 from core.platform.registry import get_registry 

347 registry = get_registry() 

348 if not registry.has('apps'): 

349 return 

350 apps = registry.get('apps') 

351 if apps.get(app_id): 

352 apps.unregister(app_id) 

353 logger.info(f"Auto-unregistered app: {app_id}") 

354 except Exception as e: 

355 logger.debug(f"App auto-unregister skipped: {e}") 

356 

357 def list_installed(self) -> List[dict]: 

358 """List all installed applications across platforms.""" 

359 installed = [] 

360 

361 # Nix packages 

362 try: 

363 result = subprocess.run( 

364 ['nix-env', '-q', '--json'], 

365 capture_output=True, text=True, timeout=10) 

366 if result.returncode == 0: 

367 pkgs = json.loads(result.stdout) if result.stdout.strip() else {} 

368 for name, info in pkgs.items(): 

369 installed.append({ 

370 'name': name, 

371 'platform': 'nix', 

372 'version': info.get('version', ''), 

373 }) 

374 except (FileNotFoundError, subprocess.TimeoutExpired, json.JSONDecodeError): 

375 pass 

376 

377 # Flatpak 

378 try: 

379 result = subprocess.run( 

380 ['flatpak', 'list', '--app', '--columns=name,application,version'], 

381 capture_output=True, text=True, timeout=10) 

382 if result.returncode == 0: 

383 for line in result.stdout.strip().split('\n'): 

384 parts = line.split('\t') 

385 if len(parts) >= 2: 

386 installed.append({ 

387 'name': parts[0], 

388 'platform': 'flatpak', 

389 'app_id': parts[1] if len(parts) > 1 else '', 

390 'version': parts[2] if len(parts) > 2 else '', 

391 }) 

392 except (FileNotFoundError, subprocess.TimeoutExpired): 

393 pass 

394 

395 # AppImages 

396 appimage_dir = os.path.join(self._install_dir, 'appimages') 

397 if os.path.isdir(appimage_dir): 

398 for f in os.listdir(appimage_dir): 

399 if f.lower().endswith('.appimage'): 

400 installed.append({ 

401 'name': f.replace('.AppImage', '').replace('.appimage', ''), 

402 'platform': 'appimage', 

403 'path': os.path.join(appimage_dir, f), 

404 }) 

405 

406 # Wine apps 

407 wine_dir = os.path.join(self._install_dir, 'wine') 

408 if os.path.isdir(wine_dir): 

409 for f in os.listdir(wine_dir): 

410 if f.endswith('.desktop'): 

411 installed.append({ 

412 'name': f.replace('.desktop', ''), 

413 'platform': 'windows', 

414 }) 

415 

416 return installed 

417 

418 def search(self, query: str, platforms: Optional[List[str]] = None) -> List[dict]: 

419 """Search for available packages across platforms.""" 

420 results = [] 

421 

422 if not platforms or 'nix' in platforms: 

423 try: 

424 result = subprocess.run( 

425 ['nix', 'search', 'nixpkgs', query, '--json'], 

426 capture_output=True, text=True, timeout=30) 

427 if result.returncode == 0: 

428 pkgs = json.loads(result.stdout) if result.stdout.strip() else {} 

429 for attr, info in list(pkgs.items())[:20]: 

430 results.append({ 

431 'name': info.get('pname', attr), 

432 'platform': 'nix', 

433 'version': info.get('version', ''), 

434 'description': info.get('description', ''), 

435 'attr': attr, 

436 }) 

437 except (FileNotFoundError, subprocess.TimeoutExpired, json.JSONDecodeError): 

438 pass 

439 

440 if not platforms or 'flatpak' in platforms: 

441 try: 

442 result = subprocess.run( 

443 ['flatpak', 'search', query, '--columns=name,application,version,description'], 

444 capture_output=True, text=True, timeout=15) 

445 if result.returncode == 0: 

446 for line in result.stdout.strip().split('\n')[:20]: 

447 parts = line.split('\t') 

448 if parts and parts[0]: 

449 results.append({ 

450 'name': parts[0], 

451 'platform': 'flatpak', 

452 'app_id': parts[1] if len(parts) > 1 else '', 

453 'version': parts[2] if len(parts) > 2 else '', 

454 'description': parts[3] if len(parts) > 3 else '', 

455 }) 

456 except (FileNotFoundError, subprocess.TimeoutExpired): 

457 pass 

458 

459 return results 

460 

461 def history(self) -> List[dict]: 

462 """Get installation history.""" 

463 return list(self._history) 

464 

465 # ─── Platform Handlers ────────────────────────────────── 

466 

467 def _install_nix(self, req: InstallRequest) -> InstallResult: 

468 """Install a Nix package.""" 

469 pkg = req.source.replace('nixpkgs.', '').replace('nix:', '') 

470 name = req.name or pkg 

471 try: 

472 result = subprocess.run( 

473 ['nix-env', '-iA', f'nixpkgs.{pkg}'], 

474 capture_output=True, text=True, timeout=300) 

475 if result.returncode == 0: 

476 return InstallResult( 

477 success=True, platform='nix', name=name, 

478 app_id=pkg, install_path=f'/nix/store/.../{pkg}') 

479 return InstallResult( 

480 success=False, platform='nix', name=name, 

481 error=result.stderr.strip()[:500]) 

482 except FileNotFoundError: 

483 return InstallResult( 

484 success=False, platform='nix', name=name, 

485 error='nix-env not available') 

486 except subprocess.TimeoutExpired: 

487 return InstallResult( 

488 success=False, platform='nix', name=name, 

489 error='Installation timed out') 

490 

491 def _install_flatpak(self, req: InstallRequest) -> InstallResult: 

492 """Install a Flatpak package.""" 

493 ref = req.source.replace('flathub:', '').replace('flatpak:', '') 

494 name = req.name or ref 

495 try: 

496 result = subprocess.run( 

497 ['flatpak', 'install', '-y', 'flathub', ref], 

498 capture_output=True, text=True, timeout=300) 

499 if result.returncode == 0: 

500 return InstallResult( 

501 success=True, platform='flatpak', name=name, 

502 app_id=ref) 

503 return InstallResult( 

504 success=False, platform='flatpak', name=name, 

505 error=result.stderr.strip()[:500]) 

506 except FileNotFoundError: 

507 return InstallResult( 

508 success=False, platform='flatpak', name=name, 

509 error='flatpak not available') 

510 except subprocess.TimeoutExpired: 

511 return InstallResult( 

512 success=False, platform='flatpak', name=name, 

513 error='Installation timed out') 

514 

515 def _install_appimage(self, req: InstallRequest) -> InstallResult: 

516 """Install an AppImage (copy + make executable).""" 

517 if not os.path.isfile(req.source): 

518 return InstallResult( 

519 success=False, platform='appimage', 

520 name=req.name or req.source, error='File not found') 

521 

522 appimage_dir = os.path.join(self._install_dir, 'appimages') 

523 os.makedirs(appimage_dir, exist_ok=True) 

524 

525 filename = os.path.basename(req.source) 

526 name = req.name or filename.replace('.AppImage', '').replace('.appimage', '') 

527 dest = os.path.join(appimage_dir, filename) 

528 

529 try: 

530 shutil.copy2(req.source, dest) 

531 os.chmod(dest, 0o755) 

532 return InstallResult( 

533 success=True, platform='appimage', name=name, 

534 install_path=dest, app_id=name) 

535 except (IOError, PermissionError) as e: 

536 return InstallResult( 

537 success=False, platform='appimage', name=name, 

538 error=str(e)) 

539 

540 def _install_windows(self, req: InstallRequest) -> InstallResult: 

541 """Install a Windows executable via Wine.""" 

542 if not os.path.isfile(req.source): 

543 return InstallResult( 

544 success=False, platform='windows', 

545 name=req.name or req.source, error='File not found') 

546 

547 name = req.name or os.path.basename(req.source).replace('.exe', '').replace('.msi', '') 

548 

549 # Check Wine availability 

550 wine = shutil.which('wine64') or shutil.which('wine') 

551 if not wine: 

552 return InstallResult( 

553 success=False, platform='windows', name=name, 

554 error='Wine not installed. Enable Windows support in NixOS config: hart.kernel.windowsNative.enable = true') 

555 

556 try: 

557 ext = os.path.splitext(req.source)[1].lower() 

558 if ext == '.msi': 

559 cmd = [wine, 'msiexec', '/i', req.source, '/quiet'] 

560 else: 

561 cmd = [wine, req.source] 

562 

563 result = subprocess.run( 

564 cmd, capture_output=True, text=True, timeout=300, 

565 env={**os.environ, 'WINEPREFIX': os.path.join( 

566 self._install_dir, 'wine', 'prefix')}) 

567 

568 # Wine often returns 0 even for interactive installers 

569 return InstallResult( 

570 success=True, platform='windows', name=name, 

571 install_path=f'{self._install_dir}/wine/prefix', 

572 app_id=name) 

573 except subprocess.TimeoutExpired: 

574 return InstallResult( 

575 success=False, platform='windows', name=name, 

576 error='Installation timed out') 

577 except Exception as e: 

578 return InstallResult( 

579 success=False, platform='windows', name=name, 

580 error=str(e)) 

581 

582 def _install_android(self, req: InstallRequest) -> InstallResult: 

583 """Install an Android APK.""" 

584 if not os.path.isfile(req.source): 

585 return InstallResult( 

586 success=False, platform='android', 

587 name=req.name or req.source, error='File not found') 

588 

589 name = req.name or os.path.basename(req.source).replace('.apk', '') 

590 

591 # Check if Android subsystem is available 

592 if not os.path.exists('/dev/binder'): 

593 return InstallResult( 

594 success=False, platform='android', name=name, 

595 error='Android subsystem not enabled. Set hart.kernel.androidNative.enable = true in NixOS config') 

596 

597 # Try ADB-style install 

598 adb = shutil.which('adb') 

599 if adb: 

600 try: 

601 result = subprocess.run( 

602 [adb, 'install', '-r', req.source], 

603 capture_output=True, text=True, timeout=120) 

604 if result.returncode == 0: 

605 return InstallResult( 

606 success=True, platform='android', name=name, 

607 app_id=name) 

608 except (subprocess.TimeoutExpired, Exception): 

609 pass 

610 

611 # Fallback: copy to Android app directory 

612 android_dir = os.path.join(self._install_dir, 'android', 'apps') 

613 os.makedirs(android_dir, exist_ok=True) 

614 dest = os.path.join(android_dir, os.path.basename(req.source)) 

615 try: 

616 shutil.copy2(req.source, dest) 

617 return InstallResult( 

618 success=True, platform='android', name=name, 

619 install_path=dest, app_id=name) 

620 except (IOError, PermissionError) as e: 

621 return InstallResult( 

622 success=False, platform='android', name=name, 

623 error=str(e)) 

624 

625 def _install_macos(self, req: InstallRequest) -> InstallResult: 

626 """Install a macOS app via Darling (experimental).""" 

627 name = req.name or os.path.basename(req.source).replace('.dmg', '').replace('.app', '') 

628 

629 darling = shutil.which('darling') 

630 if not darling: 

631 return InstallResult( 

632 success=False, platform='macos', name=name, 

633 error='Darling not installed. macOS app support is experimental. ' 

634 'Consider using the app natively on macOS via remote desktop.') 

635 

636 return InstallResult( 

637 success=False, platform='macos', name=name, 

638 error='macOS app installation via Darling is not yet automated') 

639 

640 def _install_extension(self, req: InstallRequest) -> InstallResult: 

641 """Install a HART OS extension.""" 

642 name = req.name or os.path.basename(req.source) 

643 

644 try: 

645 from core.platform.registry import get_registry 

646 registry = get_registry() 

647 ext_reg = registry.get('extensions') 

648 if ext_reg: 

649 ext = ext_reg.load(req.source) 

650 return InstallResult( 

651 success=True, platform='extension', name=name, 

652 app_id=ext.manifest.id, version=ext.manifest.version) 

653 except Exception as e: 

654 return InstallResult( 

655 success=False, platform='extension', name=name, 

656 error=str(e)) 

657 

658 return InstallResult( 

659 success=False, platform='extension', name=name, 

660 error='Extension registry not available') 

661 

662 # ─── Uninstall handlers ───────────────────────────────── 

663 

664 def _uninstall_nix(self, pkg: str) -> InstallResult: 

665 try: 

666 result = subprocess.run( 

667 ['nix-env', '-e', pkg], 

668 capture_output=True, text=True, timeout=60) 

669 return InstallResult( 

670 success=result.returncode == 0, platform='nix', 

671 name=pkg, error=result.stderr.strip()[:500]) 

672 except (FileNotFoundError, subprocess.TimeoutExpired) as e: 

673 return InstallResult( 

674 success=False, platform='nix', name=pkg, error=str(e)) 

675 

676 def _uninstall_flatpak(self, app_id: str) -> InstallResult: 

677 try: 

678 result = subprocess.run( 

679 ['flatpak', 'uninstall', '-y', app_id], 

680 capture_output=True, text=True, timeout=60) 

681 return InstallResult( 

682 success=result.returncode == 0, platform='flatpak', 

683 name=app_id, error=result.stderr.strip()[:500]) 

684 except (FileNotFoundError, subprocess.TimeoutExpired) as e: 

685 return InstallResult( 

686 success=False, platform='flatpak', name=app_id, error=str(e)) 

687 

688 def _uninstall_appimage(self, name: str) -> InstallResult: 

689 appimage_dir = os.path.join(self._install_dir, 'appimages') 

690 for f in os.listdir(appimage_dir) if os.path.isdir(appimage_dir) else []: 

691 if name.lower() in f.lower(): 

692 os.remove(os.path.join(appimage_dir, f)) 

693 return InstallResult( 

694 success=True, platform='appimage', name=name) 

695 return InstallResult( 

696 success=False, platform='appimage', name=name, 

697 error='AppImage not found') 

698 

699 def _uninstall_windows(self, name: str) -> InstallResult: 

700 wine = shutil.which('wine64') or shutil.which('wine') 

701 if wine: 

702 try: 

703 subprocess.run( 

704 [wine, 'uninstaller'], 

705 capture_output=True, timeout=10) 

706 except Exception: 

707 pass 

708 return InstallResult( 

709 success=False, platform='windows', name=name, 

710 error='Wine uninstaller requires interactive session') 

711 

712 

713# ─── Singleton ────────────────────────────────────────────── 

714 

715_installer: Optional[AppInstaller] = None 

716 

717 

718def get_installer() -> AppInstaller: 

719 """Get the global AppInstaller instance.""" 

720 global _installer 

721 if _installer is None: 

722 _installer = AppInstaller() 

723 return _installer 

724 

725 

726# ─── Flask Route Registration ─────────────────────────────── 

727 

728def register_app_install_routes(app): 

729 """Register app installation API routes on a Flask app.""" 

730 from flask import jsonify, request 

731 

732 @app.route('/api/shell/apps/install', methods=['POST']) 

733 def shell_apps_install(): 

734 """Install an application (any platform). 

735 

736 Body: 

737 source: str — file path, URL, or package name 

738 platform: str — (optional) nix, flatpak, appimage, windows, android 

739 name: str — (optional) display name 

740 sha256: str — (optional) expected checksum 

741 """ 

742 data = request.get_json(force=True) 

743 source = data.get('source', '') 

744 if not source: 

745 return jsonify({'error': 'source required'}), 400 

746 

747 platform_str = data.get('platform', '') 

748 platform = InstallerPlatform.UNKNOWN 

749 for p in InstallerPlatform: 

750 if p.value == platform_str: 

751 platform = p 

752 break 

753 

754 req = InstallRequest( 

755 source=source, 

756 platform=platform, 

757 name=data.get('name', ''), 

758 version=data.get('version', ''), 

759 sha256=data.get('sha256', ''), 

760 options=data.get('options', {}), 

761 ) 

762 

763 installer = get_installer() 

764 result = installer.install(req) 

765 

766 return jsonify({ 

767 'success': result.success, 

768 'platform': result.platform, 

769 'name': result.name, 

770 'version': result.version, 

771 'app_id': result.app_id, 

772 'install_path': result.install_path, 

773 'error': result.error, 

774 'duration': round(result.duration_seconds, 2), 

775 }), 200 if result.success else 400 

776 

777 @app.route('/api/shell/apps/uninstall', methods=['POST']) 

778 def shell_apps_uninstall(): 

779 """Uninstall an application.""" 

780 data = request.get_json(force=True) 

781 app_id = data.get('app_id', '') 

782 platform = data.get('platform', '') 

783 if not app_id: 

784 return jsonify({'error': 'app_id required'}), 400 

785 

786 installer = get_installer() 

787 result = installer.uninstall(app_id, platform) 

788 

789 return jsonify({ 

790 'success': result.success, 

791 'name': result.name, 

792 'platform': result.platform, 

793 'error': result.error, 

794 }) 

795 

796 @app.route('/api/shell/apps/installed', methods=['GET']) 

797 def shell_apps_installed(): 

798 """List all installed applications across platforms.""" 

799 installer = get_installer() 

800 apps = installer.list_installed() 

801 return jsonify({ 

802 'apps': apps, 

803 'count': len(apps), 

804 }) 

805 

806 @app.route('/api/shell/apps/search', methods=['GET']) 

807 def shell_apps_search(): 

808 """Search for packages across platforms. 

809 

810 Query params: 

811 q: search query 

812 platforms: comma-separated list (nix,flatpak) 

813 """ 

814 query = request.args.get('q', '') 

815 if not query: 

816 return jsonify({'error': 'q parameter required'}), 400 

817 

818 platforms_str = request.args.get('platforms', '') 

819 platforms = platforms_str.split(',') if platforms_str else None 

820 

821 installer = get_installer() 

822 results = installer.search(query, platforms) 

823 return jsonify({ 

824 'query': query, 

825 'results': results, 

826 'count': len(results), 

827 }) 

828 

829 @app.route('/api/shell/apps/detect', methods=['POST']) 

830 def shell_apps_detect(): 

831 """Detect the platform of an installer file.""" 

832 data = request.get_json(force=True) 

833 file_path = data.get('path', '') 

834 if not file_path or not os.path.isfile(file_path): 

835 return jsonify({'error': 'Valid file path required'}), 400 

836 

837 platform = detect_platform(file_path) 

838 return jsonify({ 

839 'path': file_path, 

840 'platform': platform.value, 

841 'name': os.path.basename(file_path), 

842 'size': os.path.getsize(file_path), 

843 }) 

844 

845 @app.route('/api/shell/apps/history', methods=['GET']) 

846 def shell_apps_history(): 

847 """Get installation history.""" 

848 installer = get_installer() 

849 return jsonify({ 

850 'history': installer.history(), 

851 'count': len(installer.history()), 

852 }) 

853 

854 @app.route('/api/shell/apps/platforms', methods=['GET']) 

855 def shell_apps_platforms(): 

856 """List supported platforms and their availability.""" 

857 platforms = [] 

858 for p in InstallerPlatform: 

859 if p == InstallerPlatform.UNKNOWN: 

860 continue 

861 available = False 

862 tool = '' 

863 if p == InstallerPlatform.NIX: 

864 tool = 'nix-env' 

865 available = shutil.which('nix-env') is not None 

866 elif p == InstallerPlatform.FLATPAK: 

867 tool = 'flatpak' 

868 available = shutil.which('flatpak') is not None 

869 elif p == InstallerPlatform.APPIMAGE: 

870 available = True # Always available (just needs chmod +x) 

871 elif p == InstallerPlatform.WINDOWS: 

872 tool = 'wine64' 

873 available = shutil.which('wine64') is not None or \ 

874 shutil.which('wine') is not None 

875 elif p == InstallerPlatform.ANDROID: 

876 available = os.path.exists('/dev/binder') 

877 elif p == InstallerPlatform.MACOS: 

878 tool = 'darling' 

879 available = shutil.which('darling') is not None 

880 elif p == InstallerPlatform.EXTENSION: 

881 available = True 

882 

883 platforms.append({ 

884 'platform': p.value, 

885 'available': available, 

886 'tool': tool, 

887 }) 

888 

889 return jsonify({'platforms': platforms}) 

890 

891 logger.info("Registered app installation routes")