Merge pull request #9 from Agent-Field/santosh/docker-cleanup-callback

Merging after local Go/Python test run (CI bypass approved).

Author: santoshkumarradha

control-plane/internal/handlers/execute.go

@@ -201,6 +201,83 @@ func (c *executionController) handleSync(ctx *gin.Context) {
 	}
 
 	resultBody, elapsed, asyncAccepted, callErr := c.callAgent(reqCtx, plan)
+
+	// If agent returned HTTP 202 (async acknowledgment), wait for callback completion
+	if callErr == nil && asyncAccepted {
+		logger.Logger.Info().
+			Str("execution_id", plan.exec.ExecutionID).
+			Str("agent", plan.target.NodeID).
+			Str("reasoner", plan.target.TargetName).
+			Msg("agent returned async acknowledgment, waiting for completion")
+
+		// Wait for agent to call back and complete the execution
+		// Use 90 second timeout to match the HTTP client timeout
+		exec, waitErr := c.waitForExecutionCompletion(reqCtx, plan.exec.ExecutionID, 90*time.Second)
+		if waitErr != nil {
+			logger.Logger.Error().
+				Err(waitErr).
+				Str("execution_id", plan.exec.ExecutionID).
+				Msg("failed to wait for async execution completion")
+			writeExecutionError(ctx, waitErr)
+			return
+		}
+
+		// Build response from completed execution
+		var result interface{}
+		if exec.ResultPayload != nil {
+			result = decodeJSON(exec.ResultPayload)
+		}
+
+		var durationMS int64
+		if exec.DurationMS != nil {
+			durationMS = *exec.DurationMS
+		}
+
+		var finishedAt string
+		if exec.CompletedAt != nil {
+			finishedAt = exec.CompletedAt.UTC().Format(time.RFC3339)
+		} else {
+			finishedAt = time.Now().UTC().Format(time.RFC3339)
+		}
+
+		// Check if execution failed
+		if exec.Status == types.ExecutionStatusFailed {
+			errMsg := "execution failed"
+			if exec.ErrorMessage != nil {
+				errMsg = *exec.ErrorMessage
+			}
+			response := ExecuteResponse{
+				ExecutionID:       exec.ExecutionID,
+				RunID:             exec.RunID,
+				Status:            string(exec.Status),
+				ErrorMessage:      &errMsg,
+				DurationMS:        durationMS,
+				FinishedAt:        finishedAt,
+				WebhookRegistered: exec.WebhookRegistered,
+			}
+			ctx.Header("X-Execution-ID", exec.ExecutionID)
+			ctx.Header("X-Run-ID", exec.RunID)
+			ctx.JSON(http.StatusOK, response)
+			return
+		}
+
+		// Return successful execution result
+		response := ExecuteResponse{
+			ExecutionID:       exec.ExecutionID,
+			RunID:             exec.RunID,
+			Status:            string(exec.Status),
+			Result:            result,
+			DurationMS:        durationMS,
+			FinishedAt:        finishedAt,
+			WebhookRegistered: exec.WebhookRegistered,
+		}
+		ctx.Header("X-Execution-ID", exec.ExecutionID)
+		ctx.Header("X-Run-ID", exec.RunID)
+		ctx.JSON(http.StatusOK, response)
+		return
+	}
+
+	// Agent returned HTTP 200 (synchronous result), process completion normally
 	job := completionJob{
 		controller: c,
 		plan:       plan,
@@ -209,26 +286,6 @@ func (c *executionController) handleSync(ctx *gin.Context) {
 		callErr:    callErr,
 		done:       make(chan error, 1),
 	}
-	if callErr == nil && asyncAccepted {
-		response := AsyncExecuteResponse{
-			ExecutionID:       plan.exec.ExecutionID,
-			RunID:             plan.exec.RunID,
-			WorkflowID:        plan.exec.RunID,
-			Status:            string(types.ExecutionStatusRunning),
-			Target:            fmt.Sprintf("%s.%s", plan.target.NodeID, plan.target.TargetName),
-			Type:              plan.targetType,
-			CreatedAt:         plan.exec.CreatedAt.UTC().Format(time.RFC3339),
-			EnqueuedAt:        plan.exec.CreatedAt.UTC().Format(time.RFC3339),
-			WebhookRegistered: plan.webhookRegistered,
-		}
-		if plan.webhookError != nil {
-			response.WebhookError = plan.webhookError
-		}
-		ctx.Header("X-Execution-ID", plan.exec.ExecutionID)
-		ctx.Header("X-Run-ID", plan.exec.RunID)
-		ctx.JSON(http.StatusAccepted, response)
-		return
-	}
 	if err := enqueueCompletion(job); err != nil {
 		logger.Logger.Error().Err(err).Str("execution_id", plan.exec.ExecutionID).Msg("failed to enqueue completion job")
 		writeExecutionError(ctx, err)
@@ -505,6 +562,72 @@ func (c *executionController) publishExecutionEvent(exec *types.Execution, statu
 	c.eventBus.Publish(event)
 }
 
+// waitForExecutionCompletion waits for an execution to complete by subscribing to the event bus.
+// It returns the completed execution record or an error if the execution fails or times out.
+// This is used when agents return HTTP 202 (async acknowledgment) but the sync endpoint needs to wait for completion.
+func (c *executionController) waitForExecutionCompletion(ctx context.Context, executionID string, timeout time.Duration) (*types.Execution, error) {
+	if c.eventBus == nil {
+		return nil, fmt.Errorf("event bus not available")
+	}
+
+	// Create unique subscriber ID for this wait operation
+	subscriberID := fmt.Sprintf("sync-wait-%s", executionID)
+
+	// Subscribe to events
+	eventChan := c.eventBus.Subscribe(subscriberID)
+	defer c.eventBus.Unsubscribe(subscriberID)
+
+	// Create timeout timer
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+
+	logger.Logger.Debug().
+		Str("execution_id", executionID).
+		Dur("timeout", timeout).
+		Msg("waiting for execution completion via event bus")
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+
+		case <-timer.C:
+			logger.Logger.Warn().
+				Str("execution_id", executionID).
+				Dur("timeout", timeout).
+				Msg("execution completion timeout")
+			return nil, fmt.Errorf("execution timeout after %v", timeout)
+
+		case event := <-eventChan:
+			// Only process events for this specific execution
+			if event.ExecutionID != executionID {
+				continue
+			}
+
+			// Check if this is a terminal event
+			if event.Type == events.ExecutionCompleted || event.Type == events.ExecutionFailed {
+				logger.Logger.Debug().
+					Str("execution_id", executionID).
+					Str("event_type", string(event.Type)).
+					Msg("received terminal execution event")
+
+				// Fetch the updated execution record
+				exec, err := c.store.GetExecutionRecord(ctx, executionID)
+				if err != nil {
+					return nil, fmt.Errorf("failed to fetch execution after completion: %w", err)
+				}
+				if exec == nil {
+					return nil, fmt.Errorf("execution %s not found after completion event", executionID)
+				}
+
+				return exec, nil
+			}
+
+			// Continue waiting for other event types (ExecutionUpdated, etc.)
+		}
+	}
+}
+
 type preparedExecution struct {
 	exec              *types.Execution
 	requestBody       []byte

control-plane/internal/handlers/nodes.go

@@ -394,7 +394,28 @@ func RegisterNodeHandler(storageProvider storage.StorageProvider, uiService *ser
 		var normalizedCandidates []string
 		var probeResults []types.CallbackTestResult
 
-		if len(candidateList) > 0 {
+		// Determine if auto-discovery should be skipped
+		// Skip auto-discovery if:
+		// 1. An explicit BaseURL was provided by the agent AND
+		// 2. Either no discovery mode is set OR mode is explicitly "manual"/"explicit"
+		skipAutoDiscovery := false
+		if newNode.BaseURL != "" {
+			// If callback discovery mode is explicitly set to manual/explicit, respect it
+			if newNode.CallbackDiscovery != nil &&
+				(newNode.CallbackDiscovery.Mode == "manual" || newNode.CallbackDiscovery.Mode == "explicit") {
+				skipAutoDiscovery = true
+				logger.Logger.Info().Msgf("✅ Using explicit callback URL for %s (mode=%s): %s",
+					newNode.ID, newNode.CallbackDiscovery.Mode, newNode.BaseURL)
+			} else if newNode.CallbackDiscovery == nil || newNode.CallbackDiscovery.Mode == "" {
+				// No discovery info provided - treat BaseURL as explicit
+				skipAutoDiscovery = true
+				logger.Logger.Info().Msgf("✅ Using explicit callback URL for %s (no discovery mode): %s",
+					newNode.ID, newNode.BaseURL)
+			}
+		}
+
+		if len(candidateList) > 0 && !skipAutoDiscovery {
+			logger.Logger.Debug().Msgf("🔍 Auto-discovering callback URL for %s from %d candidates", newNode.ID, len(candidateList))
 			probeCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
 			resolvedBaseURL, normalizedCandidates, probeResults = resolveCallbackCandidates(probeCtx, candidateList, defaultPort)
 			cancel()
@@ -1165,6 +1186,7 @@ func RegisterServerlessAgentHandler(storageProvider storage.StorageProvider, uiS
 				Description  string                 `json:"description"`
 				InputSchema  map[string]interface{} `json:"input_schema"`
 				OutputSchema map[string]interface{} `json:"output_schema"`
+				Tags         []string               `json:"tags"`
 			} `json:"reasoners"`
 			Skills []struct {
 				ID           string                 `json:"id"`
@@ -1204,6 +1226,7 @@ func RegisterServerlessAgentHandler(storageProvider storage.StorageProvider, uiS
 				ID:           r.ID,
 				InputSchema:  json.RawMessage(inputSchemaBytes),
 				OutputSchema: json.RawMessage(outputSchemaBytes),
+				Tags:         r.Tags,
 			}
 		}
 

control-plane/internal/handlers/ui/packages.go

@@ -81,6 +81,7 @@ type ReasonerDefinition struct {
 	Description  string                 `json:"description"`
 	InputSchema  map[string]interface{} `json:"input_schema"`
 	OutputSchema map[string]interface{} `json:"output_schema,omitempty"`
+	Tags         []string               `json:"tags,omitempty"`
 }
 
 // SkillDefinition represents a skill definition

control-plane/internal/handlers/ui/reasoners.go

@@ -41,6 +41,7 @@ type ReasonerWithNode struct {
 	InputSchema  interface{}        `json:"input_schema"`
 	OutputSchema interface{}        `json:"output_schema"`
 	MemoryConfig types.MemoryConfig `json:"memory_config"`
+	Tags         []string           `json:"tags"`
 
 	// Performance metrics (placeholder for future implementation)
 	AvgResponseTime *int       `json:"avg_response_time_ms,omitempty"`
@@ -141,6 +142,7 @@ func (h *ReasonersHandler) GetAllReasonersHandler(c *gin.Context) {
 				InputSchema:  reasoner.InputSchema,
 				OutputSchema: reasoner.OutputSchema,
 				MemoryConfig: reasoner.MemoryConfig,
+				Tags:         reasoner.Tags,
 				LastUpdated:  node.LastHeartbeat,
 			}
 
@@ -262,6 +264,7 @@ func (h *ReasonersHandler) GetReasonerDetailsHandler(c *gin.Context) {
 		InputSchema:  foundReasoner.InputSchema,
 		OutputSchema: foundReasoner.OutputSchema,
 		MemoryConfig: foundReasoner.MemoryConfig,
+		Tags:         foundReasoner.Tags,
 		LastUpdated:  node.LastHeartbeat,
 	}
 

control-plane/pkg/types/types.go

@@ -183,6 +183,7 @@ type ReasonerDefinition struct {
 	InputSchema  json.RawMessage `json:"input_schema"`
 	OutputSchema json.RawMessage `json:"output_schema"`
 	MemoryConfig MemoryConfig    `json:"memory_config"`
+	Tags         []string        `json:"tags,omitempty"`
 }
 
 // SkillDefinition defines a skill provided by an agent node.

control-plane/web/client/src/components/ReasonersList.tsx

@@ -28,9 +28,37 @@ const ReasonersList: React.FC<ReasonersListProps> = ({ reasoners }) => {
       </div>
       <div className="flex flex-wrap gap-2">
         {reasoners.map((reasoner) => (
-          <Badge key={reasoner.id} variant="secondary" className="text-xs">
-            {reasoner.id}
-          </Badge>
+          <div
+            key={reasoner.id}
+            className="min-w-[140px] rounded-lg border border-border-secondary bg-card px-3 py-2"
+          >
+            <div className="text-xs font-medium text-text-primary">
+              {reasoner.id}
+            </div>
+            {reasoner.tags && reasoner.tags.length > 0 ? (
+              <div className="mt-1 flex flex-wrap gap-1">
+                {reasoner.tags.slice(0, 3).map((tag) => (
+                  <Badge
+                    key={`${reasoner.id}-${tag}`}
+                    variant="outline"
+                    className="text-[10px] bg-background text-text-tertiary border-border-secondary"
+                  >
+                    #{tag}
+                  </Badge>
+                ))}
+                {reasoner.tags.length > 3 && (
+                  <Badge
+                    variant="outline"
+                    className="text-[10px] bg-background text-text-quaternary border-border-secondary"
+                  >
+                    +{reasoner.tags.length - 3}
+                  </Badge>
+                )}
+              </div>
+            ) : (
+              <p className="mt-1 text-[11px] text-text-tertiary">No tags</p>
+            )}
+          </div>
         ))}
       </div>
     </div>

control-plane/web/client/src/components/ReasonersSkillsTable.tsx

@@ -89,6 +89,7 @@ export function ReasonersSkillsTable({
       exposure_level: reasonerDIDs[reasoner.id]?.exposure_level,
       capabilities: reasonerDIDs[reasoner.id]?.capabilities,
       memory_retention: reasoner.memory_config?.memory_retention,
+      tags: reasoner.tags,
     })),
     ...skills.map((skill): TableItem => ({
       id: skill.id,