Coverage for integrations / blueprint_registry.py: 0.0%

1""" 

2Blueprint Registry — Single entry point for registering all HARTOS blueprints. 

3 

4Both Nunba (main.py, direct mode) and standalone (hart_intelligence_entry.py) 

5call register_all_blueprints(app) to get every endpoint wired. 

6 

7Each registration is try/except guarded — missing dependencies are logged 

8and skipped. The app starts with whatever is available. 

9""" 

10import logging 

11 

12logger = logging.getLogger('hevolve.blueprints') 

13 

14 

15def register_all_blueprints(app) -> dict: 

16 """Register all HARTOS blueprints on a Flask app. 

17 

18 Returns: 

19 Dict with 'registered' (list of names) and 'skipped' (list of names). 

20 """ 

21 registered = [] 

22 skipped = [] 

23 

24 # Collect already-registered blueprint names to avoid collisions 

25 existing_bp_names = {bp.name for bp in app.blueprints.values()} if hasattr(app, 'blueprints') else set() 

26 

27 def _try_register(name: str, import_fn): 

28 try: 

29 bp = import_fn() 

30 if bp is None: 

31 skipped.append(name) 

32 logger.debug("Blueprint returned None: %s", name) 

33 return 

34 # Skip if already registered (avoids duplicate name errors) 

35 bp_name = getattr(bp, 'name', name) 

36 if bp_name in existing_bp_names: 

37 skipped.append(name) 

38 logger.debug("Blueprint already registered, skipping: %s", bp_name) 

39 return 

40 app.register_blueprint(bp) 

41 existing_bp_names.add(bp_name) 

42 registered.append(name) 

43 logger.info("Registered blueprint: %s", name) 

44 except ImportError as e: 

45 skipped.append(name) 

46 logger.debug("Blueprint import failed (%s): %s", name, e) 

47 except Exception as e: 

48 skipped.append(name) 

49 logger.warning("Blueprint init failed (%s): %s", name, e) 

50 

51 # ── Hive Session ── 

52 _try_register('hive_session', lambda: ( 

53 __import__('integrations.coding_agent.claude_hive_session', 

54 fromlist=['create_hive_session_blueprint']) 

55 .create_hive_session_blueprint() 

56 )) 

57 

58 # ── Hive Signal Bridge ── 

59 _try_register('hive_signals', lambda: ( 

60 __import__('integrations.channels.hive_signal_bridge', 

61 fromlist=['create_signal_blueprint']) 

62 .create_signal_blueprint() 

63 )) 

64 

65 # ── Benchmark Prover ── 

66 _try_register('benchmark_prover', lambda: ( 

67 __import__('integrations.agent_engine.hive_benchmark_prover', 

68 fromlist=['create_benchmark_blueprint']) 

69 .create_benchmark_blueprint() 

70 )) 

71 

72 # ── App Marketplace ── 

73 _try_register('marketplace', lambda: ( 

74 __import__('integrations.agent_engine.app_marketplace', 

75 fromlist=['marketplace_bp']) 

76 .marketplace_bp 

77 )) 

78 

79 # ── Robotics Hardware Bridge ── 

80 _try_register('robotics', lambda: ( 

81 __import__('integrations.robotics.hardware_bridge', 

82 fromlist=['create_robotics_blueprint']) 

83 .create_robotics_blueprint() 

84 )) 

85 

86 # ── Robot Intelligence API ── 

87 # Canonical routes at /api/robotics/ai/* via robot_intelligence_bp 

88 # (registered directly in hart_intelligence_entry.py). The previous 

89 # create_intelligence_blueprint() entry was a duplicate /api/robotics/ 

90 # intelligence/{think,robots} subset — removed 2026-04-15. 

91 

92 # ── Compute Optimizer ── 

93 def _register_optimizer(): 

94 mod = __import__('core.compute_optimizer', 

95 fromlist=['create_optimizer_blueprint', 'get_optimizer']) 

96 bp = mod.create_optimizer_blueprint() 

97 if bp: 

98 # Start optimizer in background (event-driven, zero overhead when idle) 

99 try: 

100 mod.get_optimizer().start() 

101 except Exception: 

102 pass 

103 return bp 

104 _try_register('compute_optimizer', _register_optimizer) 

105 

106 logger.info( 

107 "Blueprint registry: %d registered, %d skipped", 

108 len(registered), len(skipped), 

109 ) 

110 return {'registered': registered, 'skipped': skipped}