package hil import ( "context" "encoding/json" "fmt" "log" "sync" "time" "synapse/types" ) // ============================================================================= // HiL Manager // ============================================================================= // DecisionMaker is the interface for human decision makers type DecisionMaker interface { // RequestDecision sends a decision request to a human RequestDecision(ctx context.Context, request *HILRequest) (*HILResponse, error) // GetUserRole returns the user's role for authorization GetUserRole(userID string) string } // HILConfig holds HiL configuration type HILConfig struct { // Level 1: Agent suggests + human confirms (24h default) Level1Timeout time.Duration // Level 2: Must human decide (12h default, 2h urgent) Level2Timeout time.Duration Level2UrgentTimeout time.Duration // Level 3: Expert/management decision (on-demand) Level3Timeout time.Duration // Default timeouts DefaultTimeout time.Duration } func DefaultHILConfig() *HILConfig { return &HILConfig{ Level1Timeout: 24 * time.Hour, Level2Timeout: 12 * time.Hour, Level2UrgentTimeout: 2 * time.Hour, Level3Timeout: 72 * time.Hour, DefaultTimeout: 24 * time.Hour, } } // ============================================================================= // HILRequest represents a human-in-the-loop request // ============================================================================= type HILRequest struct { ID string TaskID string EscalationType string // P2P, GROUP, INTER_GROUP EscalationLevel int // 1, 2, 3 Reason HILReason Context HILContext Options []HILOption SuggestedValue any Deadline time.Time Status HILStatus CreatedAt time.Time UpdatedAt time.Time DecidedAt *time.Time Response *HILResponse } // HILReason describes why human intervention is needed type HILReason struct { Code string Message string } // HILContext provides context for the decision type HILContext struct { TaskSummary string CurrentProgress int BlockedReason string RelevantHistory []HILHistoryItem } // HILHistoryItem is a historical item for context type HILHistoryItem struct { Timestamp time.Time Event string Details string } // HILOption represents a decision option type HILOption struct { Field string Value any Label string ProsCons string Confidence int // 0-100 } // HILStatus represents the status of a HiL request type HILStatus string const ( HILStatusPending HILStatus = "pending" HILStatusWaiting HILStatus = "waiting" HILStatusDeciding HILStatus = "deciding" HILStatusApproved HILStatus = "approved" HILStatusRejected HILStatus = "rejected" HILStatusTimeout HILStatus = "timeout" HILStatusEscalated HILStatus = "escalated" ) // HILResponse represents a human's decision type HILResponse struct { DecisionID string UserID string UserName string UserRole string Action string // approve, reject, modify, select Decision HILDecision Notes string Timestamp time.Time } // HILDecision contains the actual decision type HILDecision struct { ApprovedFields map[string]interface{} ModifiedFields map[string]interface{} SelectedOption any RejectionReason string } // ============================================================================= // HILProtocol // ============================================================================= type HILProtocol struct { mu sync.RWMutex // Agent identity agentEndpoint types.AgentEndpoint // HIL requests requests map[string]*HILRequest // Decision maker decisionMaker DecisionMaker // Configuration config *HILConfig // Callbacks callbacks map[string]chan *HILResponse } // NewHILProtocol creates a new HiL protocol handler func NewHILProtocol(agentEndpoint types.AgentEndpoint, config *HILConfig) *HILProtocol { if config == nil { config = DefaultHILConfig() } return &HILProtocol{ agentEndpoint: agentEndpoint, requests: make(map[string]*HILRequest), config: config, callbacks: make(map[string]chan *HILResponse), } } // SetDecisionMaker sets the decision maker func (h *HILProtocol) SetDecisionMaker(m DecisionMaker) { h.decisionMaker = m } // ============================================================================= // P2P HiL Methods // ============================================================================= // CreateP2PHILRequest creates a HiL request for P2P task func (h *HILProtocol) CreateP2PHILRequest( taskID string, reasonCode, reasonMessage string, context HILContext, options []HILOption, suggestedValue any, escalationLevel int, ) *HILRequest { h.mu.Lock() defer h.mu.Unlock() req := &HILRequest{ ID: fmt.Sprintf("hil-%d", time.Now().UnixNano()), TaskID: taskID, EscalationType: "p2p", EscalationLevel: escalationLevel, Reason: HILReason{ Code: reasonCode, Message: reasonMessage, }, Context: context, Options: options, SuggestedValue: suggestedValue, Status: HILStatusPending, CreatedAt: time.Now(), UpdatedAt: time.Now(), } // Set deadline based on escalation level req.Deadline = h.getDeadline(escalationLevel) h.requests[req.ID] = req log.Printf("[HiL] P2P HiL request created: %s (level: %d, task: %s)", req.ID, escalationLevel, taskID) return req } // CreateP2PHumanDecisionMessage creates a HUMAN_DECISION message payload func (h *HILProtocol) CreateP2PHumanDecisionMessage(req *HILRequest, response *HILResponse) *types.UnifiedSynapseMessage { payload := types.HumanDecisionPayload{} payload.TaskID = req.TaskID payload.DecisionID = response.DecisionID payload.ReplyTo = req.ID payload.DecidedBy.UserID = response.UserID payload.DecidedBy.UserName = response.UserName payload.DecidedBy.Role = response.UserRole payload.Decision.Action = response.Action payload.Notes = response.Notes payload.Timestamp = types.NowTimestamp() switch response.Action { case "approve": if response.Decision.ApprovedFields != nil { payload.Decision.ApprovedFields = response.Decision.ApprovedFields } case "reject": payload.Decision.RejectionReason = response.Decision.RejectionReason case "select": payload.Decision.SelectedOption = response.Decision.SelectedOption case "modify": payload.Decision.ModifiedFields = response.Decision.ModifiedFields } payloadBytes, _ := json.Marshal(payload) return &types.UnifiedSynapseMessage{ ID: types.GenerateMessageID(), Version: "1.0.0", Type: string(types.P2PHumanDecision), Sender: h.agentEndpoint, Payload: payloadBytes, Timestamp: types.NowTimestamp(), } } // ============================================================================= // Group HiL Methods // ============================================================================= // CreateGroupConsensusEscalation creates a consensus escalation request func (h *HILProtocol) CreateGroupConsensusEscalation( groupID string, voteID string, question string, votes []struct { VoterID string VoterRole string Option string Reason string }, consensusStatus struct { RequiredThreshold float64 ActualThreshold float64 LeadingOption string OptionsTied bool }, escalateReason struct { Code string Message string }, escalationLevel int, context struct { TaskID string OriginalTaskRequirements map[string]interface{} AgentRecommendations []struct { AgentID string SuggestedOption string } }, ) *types.UnifiedSynapseMessage { decisionOptions := make([]struct { Option string SupportCount int SupportAgents []string }, 0) // Count votes per option optionCounts := make(map[string]int) optionAgents := make(map[string][]string) for _, v := range votes { optionCounts[v.Option]++ optionAgents[v.Option] = append(optionAgents[v.Option], v.VoterID) } for opt, count := range optionCounts { decisionOptions = append(decisionOptions, struct { Option string SupportCount int SupportAgents []string }{ Option: opt, SupportCount: count, SupportAgents: optionAgents[opt], }) } payload := types.GroupConsensusEscalatePayload{} payload.VoteID = voteID payload.EscalationID = fmt.Sprintf("hil-%d", time.Now().UnixNano()) payload.Question = question // Copy votes payload.Votes = make([]struct { VoterID string `json:"voter_id"` VoterRole string `json:"voter_role"` Option string `json:"option"` Reason string `json:"reason,omitempty"` }, len(votes)) for i, v := range votes { payload.Votes[i].VoterID = v.VoterID payload.Votes[i].VoterRole = v.VoterRole payload.Votes[i].Option = v.Option payload.Votes[i].Reason = v.Reason } payload.ConsensusStatus.RequiredThreshold = consensusStatus.RequiredThreshold payload.ConsensusStatus.ActualThreshold = consensusStatus.ActualThreshold payload.ConsensusStatus.LeadingOption = consensusStatus.LeadingOption payload.ConsensusStatus.OptionsTied = consensusStatus.OptionsTied payload.EscalateReason.Code = escalateReason.Code payload.EscalateReason.Message = escalateReason.Message payload.EscalationLevel = escalationLevel // Copy decision options payload.DecisionOptions = make([]struct { Option string `json:"option"` SupportCount int `json:"support_count"` SupportAgents []string `json:"support_agents"` }, len(decisionOptions)) for i, d := range decisionOptions { payload.DecisionOptions[i].Option = d.Option payload.DecisionOptions[i].SupportCount = d.SupportCount payload.DecisionOptions[i].SupportAgents = d.SupportAgents } payload.Deadline = types.Timestamp(h.getDeadline(escalationLevel).Format(time.RFC3339)) payload.Context.TaskID = context.TaskID payload.Context.OriginalTaskRequirements = context.OriginalTaskRequirements // Copy agent recommendations payload.Context.AgentRecommendations = make([]struct { AgentID string `json:"agent_id"` SuggestedOption string `json:"suggested_option"` }, len(context.AgentRecommendations)) for i, r := range context.AgentRecommendations { payload.Context.AgentRecommendations[i].AgentID = r.AgentID payload.Context.AgentRecommendations[i].SuggestedOption = r.SuggestedOption } payloadBytes, _ := json.Marshal(payload) return &types.UnifiedSynapseMessage{ ID: types.GenerateMessageID(), Version: "1.0.0", Type: string(types.GroupConsensusEscalate), Sender: h.agentEndpoint, GroupID: groupID, Payload: payloadBytes, Timestamp: types.NowTimestamp(), } } // CreateGroupSubtaskEscalation creates a subtask escalation request func (h *HILProtocol) CreateGroupSubtaskEscalation( groupID string, originalTask struct { ID string Type string Description string }, subtask struct { ID string Description string AssignedTo string BlockedAt string BlockedReason struct { Code string Message string } AttemptedSolutions []string }, requiredDecision struct { FieldsNeeded []struct { Field string CurrentValue any SuggestedValue any Alternatives []any ValidationRules string } DecisionType string AffectsOtherSubtasks bool RollbackPossible bool }, escalationLevel int, ) *types.UnifiedSynapseMessage { payload := types.GroupSubtaskEscalatePayload{} payload.OriginalTask.ID = originalTask.ID payload.OriginalTask.Type = originalTask.Type payload.OriginalTask.Description = originalTask.Description payload.Subtask.ID = subtask.ID payload.Subtask.Description = subtask.Description payload.Subtask.AssignedTo = subtask.AssignedTo payload.Subtask.BlockedAt = subtask.BlockedAt payload.Subtask.BlockedReason.Code = subtask.BlockedReason.Code payload.Subtask.BlockedReason.Message = subtask.BlockedReason.Message payload.Subtask.AttemptedSolutions = subtask.AttemptedSolutions // Copy fields needed payload.RequiredDecision.FieldsNeeded = make([]struct { Field string `json:"field"` CurrentValue any `json:"current_value,omitempty"` SuggestedValue any `json:"suggested_value,omitempty"` Alternatives []any `json:"alternatives,omitempty"` ValidationRules string `json:"validation_rules,omitempty"` }, len(requiredDecision.FieldsNeeded)) for i, f := range requiredDecision.FieldsNeeded { payload.RequiredDecision.FieldsNeeded[i].Field = f.Field payload.RequiredDecision.FieldsNeeded[i].CurrentValue = f.CurrentValue payload.RequiredDecision.FieldsNeeded[i].SuggestedValue = f.SuggestedValue payload.RequiredDecision.FieldsNeeded[i].Alternatives = f.Alternatives payload.RequiredDecision.FieldsNeeded[i].ValidationRules = f.ValidationRules } payload.RequiredDecision.DecisionType = requiredDecision.DecisionType payload.RequiredDecision.AffectsOtherSubtasks = requiredDecision.AffectsOtherSubtasks payload.RequiredDecision.RollbackPossible = requiredDecision.RollbackPossible payload.EscalationLevel = escalationLevel deadline := types.Timestamp(h.getDeadline(escalationLevel).Format(time.RFC3339)) payload.Deadline = &deadline payloadBytes, _ := json.Marshal(payload) return &types.UnifiedSynapseMessage{ ID: types.GenerateMessageID(), Version: "1.0.0", Type: string(types.GroupSubtaskEscalate), Sender: h.agentEndpoint, GroupID: groupID, Payload: payloadBytes, Timestamp: types.NowTimestamp(), } } // ============================================================================= // HiL Status Management // ============================================================================= // GetRequest returns a HiL request by ID func (h *HILProtocol) GetRequest(id string) (*HILRequest, bool) { h.mu.RLock() defer h.mu.RUnlock() req, ok := h.requests[id] return req, ok } // ListRequests returns all HiL requests func (h *HILProtocol) ListRequests() []*HILRequest { h.mu.RLock() defer h.mu.RUnlock() result := make([]*HILRequest, 0, len(h.requests)) for _, req := range h.requests { result = append(result, req) } return result } // ListPendingRequests returns all pending HiL requests func (h *HILProtocol) ListPendingRequests() []*HILRequest { h.mu.RLock() defer h.mu.RUnlock() result := make([]*HILRequest, 0) for _, req := range h.requests { if req.Status == HILStatusPending || req.Status == HILStatusWaiting { result = append(result, req) } } return result } // UpdateRequestStatus updates the status of a HiL request func (h *HILProtocol) UpdateRequestStatus(id string, status HILStatus) error { h.mu.Lock() defer h.mu.Unlock() req, ok := h.requests[id] if !ok { return fmt.Errorf("HiL request not found: %s", id) } req.Status = status req.UpdatedAt = time.Now() if status == HILStatusApproved || status == HILStatusRejected { now := time.Now() req.DecidedAt = &now } return nil } // RecordResponse records a human decision response func (h *HILProtocol) RecordResponse(id string, response *HILResponse) error { h.mu.Lock() defer h.mu.Unlock() req, ok := h.requests[id] if !ok { return fmt.Errorf("HiL request not found: %s", id) } req.Response = response req.Status = HILStatusApproved // or rejected based on action req.UpdatedAt = time.Now() now := time.Now() req.DecidedAt = &now log.Printf("[HiL] Response recorded for %s: %s by %s", id, response.Action, response.UserName) return nil } // ============================================================================= // Timeout Management // ============================================================================= func (h *HILProtocol) getDeadline(escalationLevel int) time.Time { switch escalationLevel { case 1: return time.Now().Add(h.config.Level1Timeout) case 2: return time.Now().Add(h.config.Level2Timeout) case 3: return time.Now().Add(h.config.Level3Timeout) default: return time.Now().Add(h.config.DefaultTimeout) } } // CheckTimeouts checks for and handles timed out requests func (h *HILProtocol) CheckTimeouts() []*HILRequest { h.mu.Lock() defer h.mu.Unlock() timedOut := make([]*HILRequest, 0) now := time.Now() for _, req := range h.requests { if (req.Status == HILStatusPending || req.Status == HILStatusWaiting) && now.After(req.Deadline) { req.Status = HILStatusTimeout req.UpdatedAt = now timedOut = append(timedOut, req) log.Printf("[HiL] Request timed out: %s (task: %s)", req.ID, req.TaskID) } } return timedOut } // ============================================================================= // Human Decision Panel (for frontend integration) // ============================================================================= // GetDecisionPanelData returns data needed to render a decision panel func (h *HILProtocol) GetDecisionPanelData(req *HILRequest) map[string]interface{} { return map[string]interface{}{ "request_id": req.ID, "task_id": req.TaskID, "escalation_type": req.EscalationType, "escalation_level": req.EscalationLevel, "reason": req.Reason, "context": req.Context, "options": req.Options, "suggested_value": req.SuggestedValue, "deadline": req.Deadline.Format(time.RFC3339), "status": req.Status, "created_at": req.CreatedAt.Format(time.RFC3339), } } // CreateHumanDecisionPayload creates a HUMAN_DECISION payload func (h *HILProtocol) CreateHumanDecisionPayload( taskID string, replyTo string, user struct { UserID string UserName string Role string }, decision struct { Action string ApprovedFields map[string]interface{} ModifiedFields map[string]interface{} SelectedOption any RejectionReason string }, notes string, ) *types.HumanDecisionPayload { payload := &types.HumanDecisionPayload{} payload.TaskID = taskID payload.DecisionID = fmt.Sprintf("decision-%d", time.Now().UnixNano()) payload.ReplyTo = replyTo payload.DecidedBy.UserID = user.UserID payload.DecidedBy.UserName = user.UserName payload.DecidedBy.Role = user.Role payload.Decision.Action = decision.Action payload.Notes = notes payload.Timestamp = types.NowTimestamp() switch decision.Action { case "approve": payload.Decision.ApprovedFields = decision.ApprovedFields case "reject": payload.Decision.RejectionReason = decision.RejectionReason case "select": payload.Decision.SelectedOption = decision.SelectedOption case "modify": payload.Decision.ModifiedFields = decision.ModifiedFields } return payload } // ============================================================================= // Utility Methods // ============================================================================= // IsHILRequired checks if a task requires human intervention func IsHILRequired(reasonCode string) bool { hilRequiredCodes := map[string]bool{ "BRAND_STYLE": true, "CREATIVE_DIRECTION": true, "COMPLIANCE": true, "CAPABILITY_LACKING": true, "USER_REQUESTED": true, "HIGH_COST_RISK": true, } return hilRequiredCodes[reasonCode] } // GetEscalationLevel determines escalation level based on reason code func GetEscalationLevel(reasonCode string) int { levelMap := map[string]int{ "INFO_MISSING": 1, "CREATIVE_DIRECTION": 1, "BRAND_STYLE": 2, "COMPLIANCE": 2, "CAPABILITY_LACKING": 2, "USER_REQUESTED": 1, "HIGH_COST_RISK": 2, "RESOURCE_DEADLOCK": 2, "STRATEGIC_DECISION": 3, } if level, ok := levelMap[reasonCode]; ok { return level } return 1 // Default to level 1 }