Coverage for integrations / agent_engine / provision_tools.py: 64.0%

1""" 

2HART OS Provisioning AutoGen Tools — Agent tools for network provisioning. 

3 

45 tools registered with GoalManager under 'provision' tag: 

5 - provision_network_machine 

6 - scan_network_for_machines 

7 - check_provisioned_node 

8 - update_provisioned_node 

9 - list_provisioned_nodes 

10""" 

11 

12import json 

13import logging 

14from typing import Annotated 

15 

16logger = logging.getLogger('hevolve_provision_tools') 

17 

18 

19def provision_network_machine( 

20 target_host: Annotated[str, "IP address or hostname of the target machine"], 

21 ssh_user: Annotated[str, "SSH username (default: root)"] = "root", 

22 ssh_key_path: Annotated[str, "Path to SSH private key (optional)"] = "", 

23 join_peer: Annotated[str, "URL of existing hive node to join (optional)"] = "", 

24) -> str: 

25 """Install HART OS on a remote network machine via SSH. 

26 

27 This tool SSHs into the target machine, runs system checks, 

28 transfers the HART OS install bundle, and executes the installer. 

29 The new node is automatically registered with the hive. 

30 

31 Returns a JSON string with the provisioning result including 

32 the new node's ID and capability tier. 

33 """ 

34 from integrations.agent_engine.network_provisioner import NetworkProvisioner 

35 

36 result = NetworkProvisioner.provision_remote( 

37 target_host=target_host, 

38 ssh_user=ssh_user, 

39 ssh_key_path=ssh_key_path if ssh_key_path else None, 

40 join_peer=join_peer if join_peer else None, 

41 ) 

42 return json.dumps(result, indent=2) 

43 

44 

45def scan_network_for_machines( 

46 subnet: Annotated[str, "CIDR subnet to scan (e.g., '192.168.1.0/24'). Auto-detect if empty."] = "", 

47) -> str: 

48 """Scan the local network for machines available for HART OS provisioning. 

49 

50 Discovers machines with open SSH port (22) on the specified subnet. 

51 If no subnet is provided, auto-detects the local network. 

52 

53 Returns a JSON list of discovered machines with their IP, hostname, 

54 and SSH accessibility status. 

55 """ 

56 from integrations.agent_engine.network_provisioner import NetworkProvisioner 

57 

58 targets = NetworkProvisioner.discover_network_targets( 

59 subnet=subnet if subnet else None) 

60 

61 return json.dumps({ 

62 'count': len(targets), 

63 'targets': targets, 

64 'subnet': subnet or 'auto-detected', 

65 }, indent=2) 

66 

67 

68def check_provisioned_node( 

69 target_host: Annotated[str, "IP address or hostname of the provisioned node"], 

70 ssh_user: Annotated[str, "SSH username (default: root)"] = "root", 

71) -> str: 

72 """Check the health and status of a provisioned HART OS node. 

73 

74 SSHs into the node and checks: 

75 - Service status (backend, discovery, agent daemon, vision, LLM) 

76 - Backend HTTP health 

77 - Node identity 

78 - System uptime 

79 

80 Returns a JSON health report. 

81 """ 

82 from integrations.agent_engine.network_provisioner import NetworkProvisioner 

83 

84 health = NetworkProvisioner.check_remote_health( 

85 target_host=target_host, 

86 ssh_user=ssh_user, 

87 ) 

88 return json.dumps(health, indent=2) 

89 

90 

91def update_provisioned_node( 

92 target_host: Annotated[str, "IP address or hostname of the node to update"], 

93 ssh_user: Annotated[str, "SSH username (default: root)"] = "root", 

94) -> str: 

95 """Update HART OS on a remote provisioned node to the latest version. 

96 

97 SSHs into the node, pulls latest code (via git or re-transfer), 

98 restarts services, and verifies the backend comes back up. 

99 

100 Returns a JSON result with success status. 

101 """ 

102 from integrations.agent_engine.network_provisioner import NetworkProvisioner 

103 

104 result = NetworkProvisioner.update_remote( 

105 target_host=target_host, 

106 ssh_user=ssh_user, 

107 ) 

108 return json.dumps(result, indent=2) 

109 

110 

111def list_provisioned_nodes() -> str: 

112 """List all HART OS nodes provisioned by this hive. 

113 

114 Returns a JSON list of all provisioned nodes with their status, 

115 capability tier, IP address, and last health check time. 

116 """ 

117 from integrations.agent_engine.network_provisioner import NetworkProvisioner 

118 

119 nodes = NetworkProvisioner.list_provisioned() 

120 return json.dumps({ 

121 'count': len(nodes), 

122 'nodes': nodes, 

123 }, indent=2) 

124 

125 

126# Tool metadata for GoalManager registration 

127PROVISION_TOOLS = [ 

128 provision_network_machine, 

129 scan_network_for_machines, 

130 check_provisioned_node, 

131 update_provisioned_node, 

132 list_provisioned_nodes, 

133]