Add comprehensive agent role integration from Bees-AgenticWorkers
- Add role-based configuration to AgentConfig with 15 predefined roles - Enhanced message types for role-based collaboration - Role-based topic subscription system - Agent initialization with automatic role assignment - Role announcements and collaboration settings - Support for expertise areas, reporting hierarchy, and deliverables 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
		
							
								
								
									
										109
									
								
								pubsub/pubsub.go
									
									
									
									
									
								
							
							
						
						
									
										109
									
								
								pubsub/pubsub.go
									
									
									
									
									
								
							| @@ -4,6 +4,7 @@ import ( | ||||
| 	"context" | ||||
| 	"encoding/json" | ||||
| 	"fmt" | ||||
| 	"strings" | ||||
| 	"sync" | ||||
| 	"time" | ||||
|  | ||||
| @@ -61,6 +62,18 @@ const ( | ||||
| 	CoordinationComplete MessageType = "coordination_complete"  // Coordination session completed | ||||
| 	DependencyAlert      MessageType = "dependency_alert"       // Dependency detected | ||||
| 	EscalationTrigger    MessageType = "escalation_trigger"     // Human escalation needed | ||||
| 	 | ||||
| 	// Role-based collaboration messages | ||||
| 	RoleAnnouncement     MessageType = "role_announcement"      // Agent announces its role and capabilities | ||||
| 	ExpertiseRequest     MessageType = "expertise_request"      // Request for specific expertise | ||||
| 	ExpertiseResponse    MessageType = "expertise_response"     // Response offering expertise | ||||
| 	StatusUpdate         MessageType = "status_update"          // Regular status updates from agents | ||||
| 	WorkAllocation       MessageType = "work_allocation"        // Allocation of work to specific roles | ||||
| 	RoleCollaboration    MessageType = "role_collaboration"     // Cross-role collaboration message | ||||
| 	MentorshipRequest    MessageType = "mentorship_request"     // Junior role requesting mentorship | ||||
| 	MentorshipResponse   MessageType = "mentorship_response"    // Senior role providing mentorship | ||||
| 	ProjectUpdate        MessageType = "project_update"         // Project-level status updates | ||||
| 	DeliverableReady     MessageType = "deliverable_ready"      // Notification that deliverable is complete | ||||
| ) | ||||
|  | ||||
| // Message represents a Bzzz/Antennae message | ||||
| @@ -70,6 +83,14 @@ type Message struct { | ||||
| 	Timestamp time.Time              `json:"timestamp"` | ||||
| 	Data      map[string]interface{} `json:"data"` | ||||
| 	HopCount  int                    `json:"hop_count,omitempty"` // For Antennae hop limiting | ||||
| 	 | ||||
| 	// Role-based collaboration fields | ||||
| 	FromRole        string   `json:"from_role,omitempty"`        // Role of sender | ||||
| 	ToRoles         []string `json:"to_roles,omitempty"`         // Target roles | ||||
| 	RequiredExpertise []string `json:"required_expertise,omitempty"` // Required expertise areas | ||||
| 	ProjectID       string   `json:"project_id,omitempty"`       // Associated project | ||||
| 	Priority        string   `json:"priority,omitempty"`         // Message priority (low, medium, high, urgent) | ||||
| 	ThreadID        string   `json:"thread_id,omitempty"`        // Conversation thread ID | ||||
| } | ||||
|  | ||||
| // NewPubSub creates a new PubSub instance for Bzzz coordination and Antennae meta-discussion | ||||
| @@ -188,6 +209,50 @@ func (p *PubSub) JoinDynamicTopic(topicName string) error { | ||||
| 	return nil | ||||
| } | ||||
|  | ||||
| // JoinRoleBasedTopics joins topics based on role and expertise | ||||
| func (p *PubSub) JoinRoleBasedTopics(role string, expertise []string, reportsTo []string) error { | ||||
| 	var topicsToJoin []string | ||||
| 	 | ||||
| 	// Join role-specific topic | ||||
| 	if role != "" { | ||||
| 		roleTopic := fmt.Sprintf("bzzz/roles/%s/v1", strings.ToLower(strings.ReplaceAll(role, " ", "_"))) | ||||
| 		topicsToJoin = append(topicsToJoin, roleTopic) | ||||
| 	} | ||||
| 	 | ||||
| 	// Join expertise-specific topics | ||||
| 	for _, exp := range expertise { | ||||
| 		expertiseTopic := fmt.Sprintf("bzzz/expertise/%s/v1", strings.ToLower(strings.ReplaceAll(exp, " ", "_"))) | ||||
| 		topicsToJoin = append(topicsToJoin, expertiseTopic) | ||||
| 	} | ||||
| 	 | ||||
| 	// Join reporting hierarchy topics | ||||
| 	for _, supervisor := range reportsTo { | ||||
| 		supervisorTopic := fmt.Sprintf("bzzz/hierarchy/%s/v1", strings.ToLower(strings.ReplaceAll(supervisor, " ", "_"))) | ||||
| 		topicsToJoin = append(topicsToJoin, supervisorTopic) | ||||
| 	} | ||||
| 	 | ||||
| 	// Join all identified topics | ||||
| 	for _, topicName := range topicsToJoin { | ||||
| 		if err := p.JoinDynamicTopic(topicName); err != nil { | ||||
| 			fmt.Printf("⚠️ Failed to join role-based topic %s: %v\n", topicName, err) | ||||
| 			continue | ||||
| 		} | ||||
| 	} | ||||
| 	 | ||||
| 	fmt.Printf("🎯 Joined %d role-based topics for role: %s\n", len(topicsToJoin), role) | ||||
| 	return nil | ||||
| } | ||||
|  | ||||
| // JoinProjectTopic joins a project-specific topic | ||||
| func (p *PubSub) JoinProjectTopic(projectID string) error { | ||||
| 	if projectID == "" { | ||||
| 		return fmt.Errorf("project ID cannot be empty") | ||||
| 	} | ||||
| 	 | ||||
| 	topicName := fmt.Sprintf("bzzz/projects/%s/coordination/v1", projectID) | ||||
| 	return p.JoinDynamicTopic(topicName) | ||||
| } | ||||
|  | ||||
| // LeaveDynamicTopic leaves a specific task topic | ||||
| func (p *PubSub) LeaveDynamicTopic(topicName string) { | ||||
| 	p.dynamicTopicsMux.Lock() | ||||
| @@ -267,6 +332,50 @@ func (p *PubSub) PublishAntennaeMessage(msgType MessageType, data map[string]int | ||||
| 	return p.antennaeTopic.Publish(p.ctx, msgBytes) | ||||
| } | ||||
|  | ||||
| // PublishRoleBasedMessage publishes a role-based collaboration message | ||||
| func (p *PubSub) PublishRoleBasedMessage(msgType MessageType, data map[string]interface{}, opts MessageOptions) error { | ||||
| 	msg := Message{ | ||||
| 		Type:              msgType, | ||||
| 		From:              p.host.ID().String(), | ||||
| 		Timestamp:         time.Now(), | ||||
| 		Data:              data, | ||||
| 		FromRole:          opts.FromRole, | ||||
| 		ToRoles:           opts.ToRoles, | ||||
| 		RequiredExpertise: opts.RequiredExpertise, | ||||
| 		ProjectID:         opts.ProjectID, | ||||
| 		Priority:          opts.Priority, | ||||
| 		ThreadID:          opts.ThreadID, | ||||
| 	} | ||||
|  | ||||
| 	msgBytes, err := json.Marshal(msg) | ||||
| 	if err != nil { | ||||
| 		return fmt.Errorf("failed to marshal role-based message: %w", err) | ||||
| 	} | ||||
|  | ||||
| 	// Determine which topic to use based on message type | ||||
| 	var topic *pubsub.Topic | ||||
| 	switch msgType { | ||||
| 	case RoleAnnouncement, ExpertiseRequest, ExpertiseResponse, StatusUpdate,  | ||||
| 		 WorkAllocation, RoleCollaboration, MentorshipRequest, MentorshipResponse, | ||||
| 		 ProjectUpdate, DeliverableReady: | ||||
| 		topic = p.antennaeTopic // Use Antennae topic for role-based messages | ||||
| 	default: | ||||
| 		topic = p.bzzzTopic // Default to Bzzz topic | ||||
| 	} | ||||
|  | ||||
| 	return topic.Publish(p.ctx, msgBytes) | ||||
| } | ||||
|  | ||||
| // MessageOptions holds options for role-based messages | ||||
| type MessageOptions struct { | ||||
| 	FromRole          string | ||||
| 	ToRoles           []string | ||||
| 	RequiredExpertise []string | ||||
| 	ProjectID         string | ||||
| 	Priority          string | ||||
| 	ThreadID          string | ||||
| } | ||||
|  | ||||
| // handleBzzzMessages processes incoming Bzzz coordination messages | ||||
| func (p *PubSub) handleBzzzMessages() { | ||||
| 	for { | ||||
|   | ||||
		Reference in New Issue
	
	Block a user
	 anthonyrawlins
					anthonyrawlins