diff --git a/ai.go b/ai.go index d493cbc7..8d1a74dc 100644 --- a/ai.go +++ b/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,8 +8332,8 @@ 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) @@ -8325,13 +8341,24 @@ func HandleAiAgentExecutionStart(execution WorkflowExecution, startNode Action, 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,11 +8624,10 @@ 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)) @@ -8616,22 +8635,18 @@ func HandleAiAgentExecutionStart(execution WorkflowExecution, startNode Action, // 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. @@ -8639,14 +8654,37 @@ func HandleAiAgentExecutionStart(execution WorkflowExecution, startNode Action, 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) diff --git a/cloudSync.go b/cloudSync.go index 281a7a72..60270fd6 100755 --- a/cloudSync.go +++ b/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) diff --git a/executions.go b/executions.go index 41632ae5..0ab88506 100644 --- a/executions.go +++ b/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) } }() } diff --git a/shared.go b/shared.go index 926be536..87ed2008 100644 --- a/shared.go +++ b/shared.go @@ -18629,14 +18629,7 @@ func sendAgentActionSelfRequest(status string, workflowExecution WorkflowExecuti } actionResult.CompletedAt = timenow - baseUrl := fmt.Sprintf("https://shuffler.io") - if len(os.Getenv("BASE_URL")) > 0 { - baseUrl = os.Getenv("BASE_URL") - } - - if len(os.Getenv("SHUFFLE_CLOUDRUN_URL")) > 0 { - baseUrl = os.Getenv("SHUFFLE_CLOUDRUN_URL") - } + baseUrl := getBackendBaseUrl() marshalledResult, err := json.Marshal(actionResult) if err != nil { @@ -18686,7 +18679,7 @@ func sendAgentActionSelfRequest(status string, workflowExecution WorkflowExecuti log.Printf("[DEBUG][%s] AI Agent finished. Detected environment: '%s'", workflowExecution.ExecutionId, agentEnvironment) - if strings.ToLower(agentEnvironment) != "cloud" && agentEnvironment != "" { + if fullExecution.Workflow.ID != fullExecution.ExecutionId && strings.ToLower(agentEnvironment) != "cloud" && agentEnvironment != "" { log.Printf("[INFO][%s] AI Agent finished (status: %s). Redeploying workflow to env '%s' with MAX priority", workflowExecution.ExecutionId, status, agentEnvironment) @@ -18698,7 +18691,10 @@ func sendAgentActionSelfRequest(status string, workflowExecution WorkflowExecuti Priority: 11, // I'm assuming 11 is the max priority } - parsedEnv := fmt.Sprintf("%s_%s", strings.ToLower(strings.ReplaceAll(strings.ReplaceAll(agentEnvironment, " ", "-"), "_", "-")), fullExecution.ExecutionOrg) + parsedEnv := strings.ToLower(strings.ReplaceAll(strings.ReplaceAll(agentEnvironment, " ", "-"), "_", "-")) + if project.Environment == "cloud" { + parsedEnv = fmt.Sprintf("%s_%s", parsedEnv, fullExecution.ExecutionOrg) + } log.Printf("[INFO][%s] Redeploying workflow to queue: %s with priority %d", fullExecution.ExecutionId, parsedEnv, executionRequest.Priority) // log.Printf("[DEBUG] AI Agent finished - REDEPLOYING workflow to Queue: '%s' (Priority: %d). Original Env: '%s', Org: '%s'", parsedEnv, executionRequest.Priority, agentEnvironment, fullExecution.ExecutionOrg) @@ -18815,11 +18811,25 @@ func handleAgentDecisionStreamResult(workflowExecution WorkflowExecution, action mappedResult := AgentOutput{} - //err := json.Unmarshal([]byte(actionResult.Result), &mappedResult) err := json.Unmarshal([]byte(workflowExecution.Results[foundActionResultIndex].Result), &mappedResult) if err != nil { - log.Printf("[ERROR][%s] Failed unmarshalling agent result: %s. Data: %s", workflowExecution.ExecutionId, err, actionResult.Result) - return &workflowExecution, false, err + actionCacheId := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) + recovered := false + if cachedData, cacheErr := GetCache(ctx, actionCacheId); cacheErr == nil && cachedData != nil { + if cachedBytes, ok := cachedData.([]uint8); ok { + if cacheErr := json.Unmarshal(cachedBytes, &mappedResult); cacheErr == nil { + log.Printf("[INFO][%s] Recovered agent output from cache %s for action %s", workflowExecution.ExecutionId, actionCacheId, actionResult.Action.ID) + workflowExecution.Results[foundActionResultIndex].Result = string(cachedBytes) + recovered = true + err = nil + } + } + } + + if !recovered { + log.Printf("[ERROR][%s] Failed unmarshalling agent result: %s. Data: %s", workflowExecution.ExecutionId, err, actionResult.Result) + return &workflowExecution, false, err + } } if mappedResult.Status == "ABORTED" || (mappedResult.Status == "FINISHED" && workflowExecution.Status != "EXECUTING") { @@ -19098,6 +19108,10 @@ func handleAgentDecisionStreamResult(workflowExecution WorkflowExecution, action _ = returnAction + if freshExec, fetchErr := GetWorkflowExecution(ctx, workflowExecution.ExecutionId); fetchErr == nil && freshExec != nil { + workflowExecution = *freshExec + } + //go sendAgentActionSelfRequest("SUCCESS", workflowExecution, workflowExecution.Results[foundActionResultIndex]) return &workflowExecution, false, nil } @@ -19213,6 +19227,53 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut } } + isAgentNode := actionResult.Action.AppName == "shuffle-ai" && actionResult.Action.Name == "run_agent" + if isAgentNode && actionResult.Status != "SKIPPED" && actionResult.Status != "FAILURE" && actionResult.Status != "ABORTED" { + log.Printf("[INFO][%s] AI Agent node executed for action %s (%s). Setting node status to WAITING.", workflowExecution.ExecutionId, actionResult.Action.Label, actionResult.Action.ID) + + actionResult.Status = "WAITING" + actionResult.CompletedAt = time.Now().Unix() * 1000 + + foundWaiting := false + for resultIndex, existingResult := range workflowExecution.Results { + if existingResult.Action.ID != actionResult.Action.ID { + continue + } + + // If the agent loop already finished or failed in the background, don't overwrite it back to WAITING + if existingResult.Status == "FINISHED" || existingResult.Status == "FAILURE" { + return &workflowExecution, false, nil + } + + if strings.Contains(existingResult.Result, "decisions") && !strings.Contains(actionResult.Result, "decisions") { + actionResult.Result = existingResult.Result + } + + workflowExecution.Results[resultIndex] = actionResult + if strings.Contains(actionResult.Result, "decisions") { + cacheIdentifier := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) + _ = SetCache(ctx, cacheIdentifier, []byte(actionResult.Result), 600) + } + foundWaiting = true + break + } + + if !foundWaiting { + workflowExecution.Results = append(workflowExecution.Results, actionResult) + if strings.Contains(actionResult.Result, "decisions") { + cacheIdentifier := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) + _ = SetCache(ctx, cacheIdentifier, []byte(actionResult.Result), 600) + } + } + + saveError := SetWorkflowExecution(ctx, workflowExecution, true) + if saveError != nil { + log.Printf("[ERROR][%s] Failed setting workflow execution during AI Agent return: %s", workflowExecution.ExecutionId, saveError) + } + + return &workflowExecution, true, nil + } + if actionResult.Action.AppName == "shuffle-subflow" { // Verifying if the userinput should be sent properly or not if actionResult.Action.Name == "run_userinput" && actionResult.Status != "SKIPPED" { @@ -19425,14 +19486,7 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut log.Printf("[ERROR][%s] Failed to marshal updated decision for delayed decision update: %s", workflowExecution.ExecutionId, err) } - 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/streams", baseUrl) if debug { @@ -22787,22 +22841,27 @@ func PrepareSingleAction(ctx context.Context, parentRequest *http.Request, user return workflowExecution, errors.New("No source_execution provided") } - workflow, err := GetWorkflow(ctx, action.SourceWorkflow, true) + oldExec, err := GetWorkflowExecution(ctx, action.SourceExecution) if err != nil { return workflowExecution, err } + workflow, err := GetWorkflow(ctx, action.SourceWorkflow, true) + if err != nil { + if action.SourceWorkflow == action.SourceExecution || (oldExec != nil && oldExec.Workflow.ID == action.SourceWorkflow) { + workflow = &oldExec.Workflow + err = nil + } else { + return workflowExecution, err + } + } + if workflow.OrgId != user.ActiveOrg.Id && len(workflow.ID) > 0 { return workflowExecution, errors.New(fmt.Sprintf("Workflow doesn't belong to the same organization (%s vs %s)", workflow.OrgId, user.ActiveOrg.Id)) } // Check if the execution exists workflowExecution.WorkflowId = workflow.ID - oldExec, err := GetWorkflowExecution(ctx, action.SourceExecution) - if err != nil { - return workflowExecution, err - } - if oldExec.Workflow.ID != action.SourceWorkflow { return workflowExecution, errors.New("Previous execution (source_execution) doesn't belong to the workflow. Please try again.") } @@ -36092,11 +36151,22 @@ func checkExecutionStatus(ctx context.Context, exec *WorkflowExecution) *Workflo return exec } - workflow, err := GetWorkflow(ctx, exec.Workflow.ID, true) - if err != nil { - log.Printf("[WARNING] Failed getting workflow '%s': %s (exec status)", exec.Workflow.ID, err) - //workflow = &exec.Workflow - //return exec + var workflow *Workflow + if exec.Workflow.ID == exec.ExecutionId && len(exec.Workflow.Actions) > 0 { + workflow = &exec.Workflow + } else { + workflow, err = GetWorkflow(ctx, exec.Workflow.ID, true) + if err != nil { + if len(exec.Workflow.Actions) > 0 { + workflow = &exec.Workflow + } else { + log.Printf("[WARNING] Failed getting workflow '%s': %s (exec status)", exec.Workflow.ID, err) + workflow = &exec.Workflow + } + } + } + if workflow == nil { + workflow = &exec.Workflow } // Make sure it only handles/keeps the relevant actions @@ -36779,15 +36849,15 @@ func getPrioritisedAppActions(ctx context.Context, inputApp string, maxAmount in log.Printf("[DEBUG] Getting prioritised app actions for '%s'", inputApp) } - if strings.Contains(inputApp, ":") || len(inputApp) == 32 { + if strings.Contains(inputApp, ":") || len(inputApp) == 32 || len(inputApp) == 36 { appnamesplit := strings.Split(inputApp, ":") appId = appnamesplit[0] - if len(appId) != 32 { + if len(appId) != 32 && len(appId) != 36 { appId = "" } } - if len(appId) == 32 { + if len(appId) == 32 || len(appId) == 36 { foundApp, err = GetApp(ctx, appId, User{}, false) if err != nil { log.Printf("[ERROR] Failed getting app %s for prioritised actions: %s", appId, err) @@ -36796,7 +36866,31 @@ func getPrioritisedAppActions(ctx context.Context, inputApp string, maxAmount in } if foundApp.ID == "" && len(appName) > 0 { - log.Printf("[ERROR] Should find app + actions based on name (not implemented): %#v", appName) + cleanName := strings.TrimPrefix(appName, "app:") + if strings.Contains(cleanName, ":") { + parts := strings.Split(cleanName, ":") + cleanName = parts[len(parts)-1] + } + cleanName = strings.TrimSpace(strings.ToLower(cleanName)) + + foundApps, err := FindWorkflowAppByName(ctx, cleanName) + if err == nil && len(foundApps) > 0 { + foundApp = &foundApps[0] + } + if foundApp.ID == "" && project.Environment == "cloud" { + algoliaApp, err := HandleAlgoliaAppSearch(ctx, cleanName) + if err == nil && len(algoliaApp.ObjectID) > 0 { + if len(foundApp.Actions) == 0 { + discoveredApp, err := GetApp(ctx, algoliaApp.ObjectID, User{}, false) + if err == nil && discoveredApp != nil && len(discoveredApp.Actions) > 0 { + foundApp = discoveredApp + } + } + if foundApp.ID == "" { + foundApp.ID = algoliaApp.ObjectID + } + } + } } for _, action := range foundApp.Actions { @@ -36804,6 +36898,13 @@ func getPrioritisedAppActions(ctx context.Context, inputApp string, maxAmount in continue } + if action.AppID == "" { + action.AppID = foundApp.ID + } + if action.AppName == "" { + action.AppName = foundApp.Name + } + if len(action.CategoryLabel) > 0 { //log.Printf("ACTION TAG: %#v => %#v", action.Label, action.CategoryLabel) returnActions = append(returnActions, action)