Coverage for core / platform / boot_service.py: 100.0%

23 statements  

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

1""" 

2Platform Boot Service — Independent platform substrate initialization. 

3 

4Runs independently of both LiquidUI (desktop) and hart_intelligence (agent 

5backend). Both servers call `ensure_platform()` at startup; the first call 

6bootstraps, subsequent calls are no-ops. 

7 

8Can also run standalone: 

9 python -m core.platform.boot_service # foreground 

10 python -m core.platform.boot_service --daemon # background 

11 

12Architecture: 

13 ┌──────────────────────────────┐ 

14 │ Platform Boot Service │ ← owns bootstrap 

15 │ EventBus, AppRegistry, │ 

16 │ Extensions, CapabilityRouter│ 

17 ├──────────────────────────────┤ 

18 │ LiquidUI │ Agent API │ ← both consume via ensure_platform() 

19 │ (desktop shell) │ (port 6777)│ 

20 └──────────────────────────────┘ 

21""" 

22 

23import logging 

24import threading 

25 

26logger = logging.getLogger('hevolve.platform.boot') 

27 

28_boot_lock = threading.Lock() 

29_booted = False 

30 

31 

32def ensure_platform(extensions_dir=None): 

33 """Ensure the platform substrate is bootstrapped. Idempotent. 

34 

35 Called by both LiquidUI and hart_intelligence at startup. 

36 First call bootstraps; subsequent calls return immediately. 

37 

38 Returns: 

39 The global ServiceRegistry, or None on failure. 

40 """ 

41 global _booted 

42 if _booted: 

43 return _get_registry_safe() 

44 

45 with _boot_lock: 

46 # Double-check after acquiring lock 

47 if _booted: 

48 return _get_registry_safe() 

49 

50 try: 

51 from core.platform.bootstrap import bootstrap_platform 

52 registry = bootstrap_platform(extensions_dir) 

53 _booted = True 

54 logger.info("Platform substrate ready") 

55 return registry 

56 except Exception as e: 

57 logger.error("Platform bootstrap failed: %s", e) 

58 return None 

59 

60 

61def is_booted(): 

62 """Check if the platform substrate has been bootstrapped.""" 

63 return _booted 

64 

65 

66def _get_registry_safe(): 

67 """Get the global registry without re-bootstrapping.""" 

68 try: 

69 from core.platform.registry import get_registry 

70 return get_registry() 

71 except Exception: 

72 return None 

73 

74 

75if __name__ == '__main__': 

76 import argparse 

77 import signal 

78 import sys 

79 import time 

80 

81 logging.basicConfig( 

82 level=logging.INFO, 

83 format='%(asctime)s %(name)s %(levelname)s %(message)s') 

84 

85 parser = argparse.ArgumentParser(description='HART OS Platform Boot Service') 

86 parser.add_argument('--daemon', action='store_true', 

87 help='Run as background daemon (keep alive)') 

88 parser.add_argument('--extensions-dir', default=None, 

89 help='Path to extensions directory') 

90 args = parser.parse_args() 

91 

92 registry = ensure_platform(args.extensions_dir) 

93 if registry is None: 

94 logger.error("Bootstrap failed — exiting") 

95 sys.exit(1) 

96 

97 if args.daemon: 

98 logger.info("Boot service running (Ctrl+C to stop)") 

99 stop = threading.Event() 

100 signal.signal(signal.SIGINT, lambda *_: stop.set()) 

101 signal.signal(signal.SIGTERM, lambda *_: stop.set()) 

102 stop.wait() 

103 else: 

104 logger.info("Bootstrap complete — exiting")