support AI Agent execution in on-prem and hybrid environments and

ai.go

@@ -8256,10 +8256,26 @@ func HandleAiAgentExecutionStart(execution WorkflowExecution, startNode Action,
 	var err error
 	aiStarttime := time.Now().UnixMilli()
 
-	// Only fetch from DB if the passed execution has no results somehow
-	if len(execution.Results) == 0 {
-		if replacedExecution, fetchErr := GetWorkflowExecution(ctx, execution.ExecutionId); fetchErr == nil && replacedExecution != nil && len(replacedExecution.Results) > 0 {
-			execution = *replacedExecution
+	// Ensure we merge all prior node results from DB so earlier nodes (e.g. Shuffle Tools) are never clobbered or lost
+	if freshExecution, fetchErr := GetWorkflowExecution(ctx, execution.ExecutionId); fetchErr == nil && freshExecution != nil && len(freshExecution.Results) > 0 {
+		if len(execution.Results) == 0 {
+			execution.Results = freshExecution.Results
+		} else {
+			for _, dbRes := range freshExecution.Results {
+				if dbRes.Action.ID == startNode.ID {
+					continue
+				}
+				found := false
+				for _, curRes := range execution.Results {
+					if curRes.Action.ID == dbRes.Action.ID {
+						found = true
+						break
+					}
+				}
+				if !found {
+					execution.Results = append(execution.Results, dbRes)
+				}
+			}
 		}
 	}
 
@@ -8316,22 +8332,33 @@ func HandleAiAgentExecutionStart(execution WorkflowExecution, startNode Action,
 
 	// Validate On-Prem Configuration immediately
 	if project.Environment == "onprem" {
-			cloudSyncConfigured := false
-			if len(execution.Workflow.OrgId) > 0 {
+		cloudSyncConfigured := false
+		if len(execution.Workflow.OrgId) > 0 {
 			if validationOrg, orgErr := GetOrg(ctx, execution.Workflow.OrgId); orgErr == nil {
 				if len(validationOrg.CreatorOrg) > 0 {
 					validationOrg, orgErr = GetOrg(ctx, validationOrg.CreatorOrg)
 				}
 				if orgErr == nil && len(validationOrg.SyncConfig.Apikey) > 0 && validationOrg.CloudSyncActive && validationOrg.SyncConfig.AiCloudSync {
 					cloudSyncConfigured = true
 				}
+			}
+		}
+
+		hasLocalAi := false
+		if len(execution.Workflow.OrgId) > 0 {
+			if auths, err := GetAllWorkflowAppAuth(ctx, execution.Workflow.OrgId); err == nil {
+				for _, auth := range auths {
+					if strings.ToLower(auth.App.Name) == "openai" && (auth.Defined || auth.Validation.Valid || len(auth.Id) > 0) {
+						hasLocalAi = true
+						break
+					}
 				}
 			}
+		}
 
-			if !cloudSyncConfigured {
-			onpremAiConfigErr := "AI_MODEL or OPENAI_MODEL environment variable must be set for On-Premise AI Agent execution. Alternatively, enable Cloud Sync and turn on \"Shuffle Cloud AI\" to run AI requests through Shuffle Cloud without any additional configuration"
+		if !cloudSyncConfigured && !hasLocalAi {
+			onpremAiConfigErr := "To use the AI Agent on-premise, configure your LLM credentials by connecting the OpenAI app in Shuffle App Auth (supports OpenAI and any compatible provider/proxy), or enable Cloud Sync with \"Shuffle Cloud AI\" to run requests through Shuffle Cloud."
 			log.Printf("[ERROR] AI Configuration Error: %s", onpremAiConfigErr)
-
 			return abortAgentExecution(ctx, execution, startNode, "missing_onprem_ai_config", onpremAiConfigErr)
 		}
 	}
@@ -8355,14 +8382,7 @@ func HandleAiAgentExecutionStart(execution WorkflowExecution, startNode Action,
 	executionMode := ""
 
 	// Self-request starts here!
-	backendUrl := "https://shuffler.io"
-	if len(os.Getenv("BASE_URL")) > 0 {
-		backendUrl = os.Getenv("BASE_URL")
-	}
-
-	if len(os.Getenv("SHUFFLE_CLOUDRUN_URL")) > 0 {
-		backendUrl = os.Getenv("SHUFFLE_CLOUDRUN_URL")
-	}
+	backendUrl := getBackendBaseUrl()
 
 	// This is a part of making sure variables work properly, no matter where
 	// in Shuffle we are
@@ -8604,49 +8624,67 @@ func HandleAiAgentExecutionStart(execution WorkflowExecution, startNode Action,
 			executionMode = strings.ToLower(strings.TrimSpace(param.Value))
 		}
 
-		if param.Name == "action" {
+		if (param.Name == "action" || param.Name == "app_name" || param.Name == "tool_name") && len(param.Value) > 0 && param.Value != "openai" && param.Value != "AI Agent" && param.Value != "Shuffle Agent" {
 			param.Value = strings.ReplaceAll(param.Value, "app:undefined:api,", "")
 			param.Value = strings.ReplaceAll(param.Value, "app:undefined:api", "")
 
-			allowedActionString = param.Value
 			for _, actionStr := range strings.Split(param.Value, ",") {
 				actionStr = strings.ToLower(strings.TrimSpace(actionStr))
 
 				//if debug {
 				//	log.Printf("[DEBUG] STRING: %s", actionStr)
 				//}
 
-				if actionStr == "" || actionStr == "nothing" || actionStr == "shuffle ai" || actionStr == "api" {
+				if actionStr == "" || actionStr == "nothing" || actionStr == "shuffle ai" || actionStr == "api" || actionStr == "openai" || actionStr == "ai agent" || actionStr == "shuffle agent" {
 					if debug {
 						log.Printf("[DEBUG][%s] Skipping action '%s' as it is not a valid action.", execution.ExecutionId, actionStr)
 					}
 					continue
 				}
 
-				if !strings.HasPrefix(actionStr, "app:") {
-					if debug {
-						log.Printf("[DEBUG][%s] Skipping action '%s' as it is not a valid action.", execution.ExecutionId, actionStr)
-					}
-
+				trimmedActionStr := strings.TrimPrefix(actionStr, "app:")
+				if trimmedActionStr == "" || trimmedActionStr == "openai" {
 					continue
 				}
 
-				trimmedActionStr := strings.TrimPrefix(actionStr, "app:")
 				sortedAppActions := getPrioritisedAppActions(ctx, trimmedActionStr, 15)
 
 				// Sort alphabetically so the action list is byte-for-byte identical across every LLM loop, keeping the prompt cache prefix stable.
 				sort.Slice(sortedAppActions, func(i, j int) bool {
 					return sortedAppActions[i].Name < sortedAppActions[j].Name
 				})
 
-				if len(sortedAppActions) > 0 {
-					// Cuts off the potential md5:appname prefix
-					if len(trimmedActionStr) > 33 && string(trimmedActionStr[32]) == ":" {
-						trimmedActionStr = trimmedActionStr[33:]
+				// Cuts off the potential md5:appname or uuid:appname prefix
+				baseToolName := trimmedActionStr
+				if len(baseToolName) > 33 && string(baseToolName[32]) == ":" {
+					baseToolName = baseToolName[33:]
+				} else if len(baseToolName) > 37 && string(baseToolName[36]) == ":" {
+					baseToolName = baseToolName[37:]
+				}
+
+				if !ArrayContains(decidedApps, baseToolName) {
+					decidedApps = append(decidedApps, baseToolName)
+				}
+
+				allowedEntry := actionStr
+				if !strings.Contains(trimmedActionStr, ":") {
+					if len(sortedAppActions) > 0 && len(sortedAppActions[0].AppID) > 0 {
+						allowedEntry = fmt.Sprintf("app:%s:%s", sortedAppActions[0].AppID, baseToolName)
+					} else {
+						allowedEntry = fmt.Sprintf("app:%s", baseToolName)
 					}
+				}
 
-					decidedApps = append(decidedApps, trimmedActionStr)
-					specificAppMetadata += fmt.Sprintf("\n\n**Available actions and fields for Tool '%s'**:\n", trimmedActionStr)
+				if len(allowedActionString) > 0 {
+					if !strings.Contains(allowedActionString, allowedEntry) && !strings.Contains(allowedActionString, baseToolName) {
+						allowedActionString += "," + allowedEntry
+					}
+				} else {
+					allowedActionString = allowedEntry
+				}
+
+				if len(sortedAppActions) > 0 {
+					specificAppMetadata += fmt.Sprintf("\n\n**Available actions and fields for Tool '%s'**:\n", baseToolName)
 
 					previousDesc := ""
 					for counter, sortedAppAction := range sortedAppActions {
@@ -10269,6 +10307,7 @@ data_filter:
 				execution.Status = "EXECUTING"
 				agentOutput.Status = "RUNNING"
 
+				foundAgentIndex := -1
 				for resultIndex, result := range execution.Results {
 					if result.Action.ID != startNode.ID {
 						continue
@@ -10291,6 +10330,26 @@ data_filter:
 					if err != nil {
 						log.Printf("[ERROR] AI Agent: Failed setting cache for action result %s: %s", actionCacheId, err)
 					}
+					foundAgentIndex = resultIndex
+					break
+				}
+
+				if foundAgentIndex < 0 {
+					agentOutputMarshalled, err := json.Marshal(agentOutput)
+					initialResult := string(agentOutputMarshalled)
+					if err != nil {
+						initialResult = "{}"
+					}
+					agentResult := ActionResult{
+						Action:      startNode,
+						ExecutionId: execution.ExecutionId,
+						Result:      initialResult,
+						Status:      "WAITING",
+						StartedAt:   time.Now().UnixMilli(),
+					}
+					execution.Results = append(execution.Results, agentResult)
+					actionCacheId := fmt.Sprintf("%s_%s_result", execution.ExecutionId, startNode.ID)
+					_ = SetCache(ctx, actionCacheId, []byte(initialResult), 600)
 				}
 
 				SetWorkflowExecution(ctx, execution, true)
@@ -10603,8 +10662,15 @@ data_filter:
 		}
 
 		if agentOutput.Status == "FINISHED" && agentOutput.CompletedAt > 0 && execution.Status != "ABORTED" && execution.Status != "FAILURE" {
-			execution.Status = "FINISHED"
-			execution.CompletedAt = agentOutput.CompletedAt
+			isStandalone := execution.ExecutionId == execution.WorkflowId || execution.ExecutionId == execution.Workflow.ID
+			if isStandalone {
+				execution.Status = "FINISHED"
+				if agentOutput.CompletedAt > 100000000000 {
+					execution.CompletedAt = agentOutput.CompletedAt / 1000
+				} else {
+					execution.CompletedAt = agentOutput.CompletedAt
+				}
+			}
 			execution.Results[foundResultIndex].Status = "SUCCESS"
 			execution.Results[foundResultIndex].CompletedAt = agentOutput.CompletedAt
 			SetWorkflowExecution(ctx, execution, true)

cloudSync.go

@@ -2276,6 +2276,32 @@ func HandleSuborgScheduleRun(request *http.Request, workflow *Workflow) {
 	}
 }
 
+func getBackendBaseUrl() string {
+	backendUrl := ""
+	if len(os.Getenv("BASE_URL")) > 0 {
+		backendUrl = os.Getenv("BASE_URL")
+	}
+
+	if len(os.Getenv("SHUFFLE_CLOUDRUN_URL")) > 0 {
+		backendUrl = os.Getenv("SHUFFLE_CLOUDRUN_URL")
+	}
+
+	if len(backendUrl) > 0 {
+		return backendUrl
+	}
+
+	if project.Environment == "cloud" {
+		return "https://uk.shuffler.io"
+	}
+
+	port := os.Getenv("PORT")
+	if len(port) == 0 {
+		port = "5001"
+	}
+
+	return fmt.Sprintf("http://localhost:%s", port)
+}
+
 // runAgentDecisionDirectAppCall bypasses Singul and runs the app directly.
 func runAgentDecisionDirectAppCall(execution WorkflowExecution, decision AgentDecision) (rawResult []byte, debugUrl string, appName string, categoryLabels []string, actionName string, err error) {
 	ctx := context.Background()
@@ -2318,7 +2344,7 @@ func runAgentDecisionDirectAppCall(execution WorkflowExecution, decision AgentDe
 		} else {
 			toolLower := strings.ToLower(decision.Tool)
 			for _, app := range allApps {
-				if strings.ToLower(app.Name) != toolLower && strings.ToLower(app.ID) != toolLower && strings.ReplaceAll(strings.ToLower(app.Name), " ", "") != toolLower {
+				if strings.ToLower(app.Name) != toolLower && strings.ToLower(app.ID) != toolLower && strings.ReplaceAll(strings.ToLower(app.Name), " ", "") != strings.ReplaceAll(toolLower, "_", "") {
 					continue
 				} 
 
@@ -2355,7 +2381,7 @@ func runAgentDecisionDirectAppCall(execution WorkflowExecution, decision AgentDe
 		if err == nil {
 			toolLower := strings.ToLower(decision.Tool)
 			for _, app := range foundApps {
-				if strings.ToLower(app.Name) != toolLower && strings.ToLower(app.ID) != toolLower && strings.ReplaceAll(strings.ToLower(app.Name), " ", "") != toolLower {
+				if strings.ToLower(app.Name) != toolLower && strings.ToLower(app.ID) != toolLower && strings.ReplaceAll(strings.ToLower(app.Name), " ", "") != strings.ReplaceAll(toolLower, "_", "") {
 					continue
 				} 
 
@@ -2409,12 +2435,24 @@ func runAgentDecisionDirectAppCall(execution WorkflowExecution, decision AgentDe
 		log.Printf("[ERROR][%s] AI_AGENT_LLM_FAILURE: Failed to parse Agent decision.Delay '%s' as int: %s", execution.ExecutionId, decision.Delay, err)
 	}
 
+	foundEnv := ""
+	for _, act := range execution.Workflow.Actions {
+		if (act.ID == execution.ExecutionSourceNode || act.ID == execution.Start || act.AppName == "AI Agent" || act.AppID == "agent") && len(act.Environment) > 0 {
+			foundEnv = act.Environment
+			break
+		}
+	}
+	if len(foundEnv) == 0 && len(execution.Workflow.Actions) > 0 {
+		foundEnv = execution.Workflow.Actions[0].Environment
+	}
+
 	action := Action{
 		AppID:            resolvedAppId,
 		AppName:          resolvedAppName,
 		Name:             decision.Action, // overwritten below if schema match found
 		//AuthenticationId: resolvedAuthId,
 		Parameters:       []WorkflowAppActionParameter{},
+		Environment:      foundEnv,
 
 		ExecutionDelay: selectedDelay,
 		SourceWorkflow: execution.Workflow.ID,
@@ -2491,28 +2529,21 @@ func runAgentDecisionDirectAppCall(execution WorkflowExecution, decision AgentDe
 	}
 
 
-	baseURL := os.Getenv("BASE_URL")
-	if len(baseURL) == 0 {
-		if v := os.Getenv("SHUFFLE_CLOUDRUN_URL"); len(v) > 0 {
-			baseURL = v
-		} else {
-			port := os.Getenv("PORT")
-			if len(port) == 0 {
-				port = "5001"
-			}
-			baseURL = fmt.Sprintf("http://localhost:%s", port)
-		}
-	}
+	baseURL := getBackendBaseUrl()
 
-	//ExecutionDelay: selectedDelay,
-	timeout := time.Duration(30) * time.Second
+	toolTimeout := 30
+	if project.Environment != "cloud" || (len(foundEnv) > 0 && strings.ToLower(foundEnv) != "cloud") {
+		toolTimeout = 120
+	}
+
+	timeout := time.Duration(toolTimeout) * time.Second
 
 	// Immediate exits. 3 seconds due to body transfer worst case
 	if selectedDelay > 0 { 
 		timeout = time.Duration(2) * time.Second
 	}
 
-	requestUrl := fmt.Sprintf("%s/api/v1/apps/%s/run?delete=false&execution_id=%s&authorization=%s&org_id=%s&timeout=%d&delay=%d&decision_id=%s", baseURL, resolvedAppId, execution.ExecutionId, execution.Authorization, execution.ExecutionOrg, (timeout/1000000000)-1, selectedDelay, decision.RunDetails.Id)
+	requestUrl := fmt.Sprintf("%s/api/v1/apps/%s/run?delete=false&execution_id=%s&authorization=%s&org_id=%s&timeout=%d&delay=%d&decision_id=%s", baseURL, resolvedAppId, execution.ExecutionId, execution.Authorization, execution.ExecutionOrg, int(timeout.Seconds())-1, selectedDelay, decision.RunDetails.Id)
 
 	parentNode := ""
 	if len(parentNode) > 0 { 
@@ -2616,14 +2647,7 @@ func RunAgentDecisionSingulActionHandler(execution WorkflowExecution, decision A
 
 	_ = debugUrl
 
-	baseUrl := "https://shuffler.io"
-	if os.Getenv("BASE_URL") != "" {
-		baseUrl = os.Getenv("BASE_URL")
-	}
-
-	if os.Getenv("SHUFFLE_CLOUDRUN_URL") != "" {
-		baseUrl = os.Getenv("SHUFFLE_CLOUDRUN_URL")
-	}
+	baseUrl := getBackendBaseUrl()
 
 	requestUrl := fmt.Sprintf("%s/api/v1/apps/categories/run?authorization=%s&execution_id=%s", baseUrl, execution.Authorization, execution.ExecutionId)
 
@@ -2877,6 +2901,8 @@ func normalizeAgentToolName(tool string) string {
 
 	if len(tool) > 33 && tool[32] == ':' {
 		tool = tool[33:]
+	} else if len(tool) > 37 && tool[36] == ':' {
+		tool = tool[37:]
 	}
 
 	tool = strings.TrimSpace(tool)
@@ -3073,14 +3099,7 @@ func RunAgentDecisionAction(execution WorkflowExecution, agentOutput AgentOutput
 	//     2. Send the result through AI again to check if it changes (?). Should there be a verdict here?
 	//     3: Start the next steps of decisions after updates
 
-	baseUrl := "https://shuffler.io"
-	if os.Getenv("BASE_URL") != "" {
-		baseUrl = os.Getenv("BASE_URL")
-	}
-
-	if os.Getenv("SHUFFLE_CLOUDRUN_URL") != "" {
-		baseUrl = os.Getenv("SHUFFLE_CLOUDRUN_URL")
-	}
+	baseUrl := getBackendBaseUrl()
 
 	//url := fmt.Sprintf("%s/api/v1/apps/categories/run?authorization=%s&execution_id=%s", baseUrl, execution.Authorization, execution.ExecutionId)
 	url := fmt.Sprintf("%s/api/v1/streams", baseUrl)

executions.go

@@ -379,9 +379,13 @@ func Fixexecution(ctx context.Context, workflowExecution WorkflowExecution) (Wor
 								time.Sleep(1 * time.Second)
 								sendAgentActionSelfRequest("WAITING", capturedExec, capturedExec.Results[resultIndex])
 								time.Sleep(2 * time.Second)
-								_, err := HandleAiAgentExecutionStart(capturedExec, capturedAction, true, "fixexecution_timeout_recovery")
-								if err != nil {
-									log.Printf("[ERROR][%s] Failed re-invoking agent after decisions completed for action %s: %s", capturedExec.ExecutionId, capturedAction.ID, err)
+								if project.Environment == "cloud" {
+									_, err := HandleAiAgentExecutionStart(capturedExec, capturedAction, true, "fixexecution_timeout_recovery")
+									if err != nil {
+										log.Printf("[ERROR][%s] Failed re-invoking agent after decisions completed for action %s: %s", capturedExec.ExecutionId, capturedAction.ID, err)
+									}
+								} else {
+									log.Printf("[DEBUG][%s] Skipping HandleAiAgentExecutionStart in non-cloud environment (fixexecution_timeout_recovery) — Cloud handles redeployment via queue.", capturedExec.ExecutionId)
 								}
 							}()
 						}