From 3a641709f702f7f0ce36aa4e5b4c543c06519c32 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Sat, 5 Sep 2026 15:25:05 +0530 Subject: [PATCH 01/30] fix workflow retrieval logic in checkExecutionStatus function for agents --- shared.go | 21 ++++++++++++++++----- 1 file changed, 16 insertions(+), 5 deletions(-) diff --git a/shared.go b/shared.go index 926be536..11bfcaf5 100644 --- a/shared.go +++ b/shared.go @@ -36092,11 +36092,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 From 4bec06245038255e6c826bad3eb0fd0d77b773e6 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Sat, 5 Sep 2026 15:58:28 +0530 Subject: [PATCH 02/30] onprem: fix workflow retrieval logic in PrepareSingleAction function --- shared.go | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/shared.go b/shared.go index 11bfcaf5..92186e3f 100644 --- a/shared.go +++ b/shared.go @@ -22787,22 +22787,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.") } From 8d9aba0964ed723c3055e469562370bde5dbf9a7 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Sat, 5 Sep 2026 16:37:08 +0530 Subject: [PATCH 03/30] onprem: fix app name matching logic and adjust base URL for cloud environment --- cloudSync.go | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/cloudSync.go b/cloudSync.go index 281a7a72..242e497b 100755 --- a/cloudSync.go +++ b/cloudSync.go @@ -2318,7 +2318,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 +2355,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 } @@ -3073,7 +3073,10 @@ 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" + baseUrl := "http://localhost:5001" + if project.Environment == "cloud" { + baseUrl = "https://shuffler.io" + } if os.Getenv("BASE_URL") != "" { baseUrl = os.Getenv("BASE_URL") } From a49198680175ad414e5bcba5e33e91b63e86aee5 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Sat, 5 Sep 2026 22:24:59 +0530 Subject: [PATCH 04/30] fix environment handling in sendAgentActionSelfRequest --- shared.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/shared.go b/shared.go index 92186e3f..008d85bb 100644 --- a/shared.go +++ b/shared.go @@ -18686,7 +18686,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 +18698,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) From a3feca28bc87807d55f41ede4b04fb3bb17736dd Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Mon, 7 Sep 2026 15:11:40 +0530 Subject: [PATCH 05/30] fetch and update workflowExecution if fresh execution is available in handleAgentDecisionStreamResult --- shared.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/shared.go b/shared.go index 008d85bb..d1708fa7 100644 --- a/shared.go +++ b/shared.go @@ -19101,6 +19101,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 } From 69d09193bef743f22f92a591c6f0f8bc2e66ff3b Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Mon, 7 Sep 2026 15:12:05 +0530 Subject: [PATCH 06/30] enhance app action retrieval by cleaning app name and searching in workflows --- shared.go | 33 ++++++++++++++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/shared.go b/shared.go index d1708fa7..f48d48f8 100644 --- a/shared.go +++ b/shared.go @@ -36819,7 +36819,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 == "" { + 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 { @@ -36827,6 +36851,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) From a8a8fd4a0976bfb6a1e91e320605e77752a09830 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Tue, 8 Sep 2026 18:24:16 +0530 Subject: [PATCH 07/30] refactor action handling in HandleAiAgentExecutionStart to improve app name and tool name validation --- ai.go | 48 +++++++++++++++++++++++++++++++++--------------- 1 file changed, 33 insertions(+), 15 deletions(-) diff --git a/ai.go b/ai.go index d493cbc7..b529e45c 100644 --- a/ai.go +++ b/ai.go @@ -8604,11 +8604,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 +8615,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 +8634,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 { From 0d95a38266c2b27ac519f350221ca57eb38e8ac0 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Mon, 7 Sep 2026 15:13:16 +0530 Subject: [PATCH 08/30] add standalone execution check before finalizing agent output status --- ai.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/ai.go b/ai.go index b529e45c..2ceb7351 100644 --- a/ai.go +++ b/ai.go @@ -10621,8 +10621,11 @@ 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" + execution.CompletedAt = agentOutput.CompletedAt + } execution.Results[foundResultIndex].Status = "SUCCESS" execution.Results[foundResultIndex].CompletedAt = agentOutput.CompletedAt SetWorkflowExecution(ctx, execution, true) From 297a7f1c07a4611de6bd909c0122f0cfae72895a Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Mon, 7 Sep 2026 18:02:29 +0530 Subject: [PATCH 09/30] add caching for agent output recovery in handleAgentDecisionStreamResult --- shared.go | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/shared.go b/shared.go index f48d48f8..cc9a7f53 100644 --- a/shared.go +++ b/shared.go @@ -18818,11 +18818,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") { From b9ac2ad39cce10cc6e986f044f42703edf3e3dd6 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Tue, 8 Sep 2026 16:55:01 +0530 Subject: [PATCH 10/30] add environment handling for agent decision execution --- cloudSync.go | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/cloudSync.go b/cloudSync.go index 242e497b..48b2c44d 100755 --- a/cloudSync.go +++ b/cloudSync.go @@ -2409,6 +2409,17 @@ 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, From 5cb51fd271bf6838262d6aa7aa5b44f00fb8e32e Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Tue, 8 Sep 2026 16:55:16 +0530 Subject: [PATCH 11/30] add environment handling for agent decision execution (2) --- cloudSync.go | 1 + 1 file changed, 1 insertion(+) diff --git a/cloudSync.go b/cloudSync.go index 48b2c44d..a17f1ef6 100755 --- a/cloudSync.go +++ b/cloudSync.go @@ -2426,6 +2426,7 @@ func runAgentDecisionDirectAppCall(execution WorkflowExecution, decision AgentDe Name: decision.Action, // overwritten below if schema match found //AuthenticationId: resolvedAuthId, Parameters: []WorkflowAppActionParameter{}, + Environment: foundEnv, ExecutionDelay: selectedDelay, SourceWorkflow: execution.Workflow.ID, From dde71d7dc6cc4a6ba0f9f60dc19942b776a05dc8 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Tue, 8 Sep 2026 16:55:57 +0530 Subject: [PATCH 12/30] onprem: increase agent tool timeout and adjust client timeout for better request handling --- cloudSync.go | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/cloudSync.go b/cloudSync.go index a17f1ef6..f6520bbd 100755 --- a/cloudSync.go +++ b/cloudSync.go @@ -2516,24 +2516,29 @@ func runAgentDecisionDirectAppCall(execution WorkflowExecution, decision AgentDe } } - //ExecutionDelay: selectedDelay, - timeout := time.Duration(30) * time.Second + toolTimeout := 120 + if v := os.Getenv("AGENT_TOOL_TIMEOUT"); len(v) > 0 { + if parsed, err := strconv.Atoi(v); err == nil && parsed > 0 && parsed <= 300 { + toolTimeout = parsed + } + } + 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 { requestUrl += fmt.Sprintf("parent_node=%s", parentNode) } - // Gives it time to return properly with +2 delay + // Gives it time to return properly with +5 delay client := GetExternalClientWithTimeout(requestUrl, 0) - client.Timeout = timeout + (1 * time.Second) + client.Timeout = timeout + (5 * time.Second) //if debug { //log.Printf("\n\n\n\nRequest timeout: %d", client.Timeout) From 32b584a8c41eedd38c501d64cb3bdfb450df9752 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Tue, 8 Sep 2026 16:57:04 +0530 Subject: [PATCH 13/30] onprem --- shared.go | 57 ++++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 56 insertions(+), 1 deletion(-) diff --git a/shared.go b/shared.go index cc9a7f53..e072e27d 100644 --- a/shared.go +++ b/shared.go @@ -19234,6 +19234,50 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut } } + // Special handler for AI Agent hybrid dispatch (where worker ran run_agent and backend dispatched to Cloud) + isAgentHybrid := (actionResult.Action.AppName == "shuffle-ai" || actionResult.Action.AppName == "AI Agent" || actionResult.Action.AppName == "Shuffle Agent" || actionResult.Action.Name == "run_agent") && + (strings.Contains(actionResult.Result, "\"mode\":\"hybrid\"") || strings.Contains(actionResult.Result, "\"mode\": \"hybrid\"")) + + if isAgentHybrid && actionResult.Status != "SKIPPED" { + log.Printf("[INFO][%s] AI Agent hybrid dispatch detected for action %s (%s). Setting node and workflow status to WAITING.", workflowExecution.ExecutionId, actionResult.Action.Label, actionResult.Action.ID) + + actionResult.Status = "WAITING" + actionResult.CompletedAt = time.Now().Unix() * 1000 + workflowExecution.Status = "WAITING" + + foundWaiting := false + for resultIndex, result := range workflowExecution.Results { + if result.Action.ID != actionResult.Action.ID { + continue + } + + workflowExecution.Results[resultIndex] = actionResult + actionResultBody, err := json.Marshal(actionResult) + if err == nil { + cacheId := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) + _ = SetCache(ctx, cacheId, actionResultBody, 600) + } + foundWaiting = true + break + } + + if !foundWaiting { + workflowExecution.Results = append(workflowExecution.Results, actionResult) + actionResultBody, err := json.Marshal(actionResult) + if err == nil { + cacheId := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) + _ = SetCache(ctx, cacheId, actionResultBody, 600) + } + } + + err = SetWorkflowExecution(ctx, workflowExecution, true) + if err != nil { + log.Printf("[ERROR][%s] Failed setting workflow execution during AI Agent hybrid return: %s", workflowExecution.ExecutionId, err) + } + + 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" { @@ -23796,7 +23840,18 @@ func ValidateNewWorkerExecution(ctx context.Context, body []byte, shouldReset bo } } - if execution.Status == "EXECUTING" { + hasWaiting := false + for _, result := range execution.Results { + if result.Status == "WAITING" { + hasWaiting = true + break + } + } + + if hasWaiting { + log.Printf("[INFO][%s] Workflow execution has action(s) in WAITING status. Preserving WAITING status.", execution.ExecutionId) + execution.Status = "WAITING" + } else if execution.Status == "EXECUTING" { //log.Printf("[INFO] Inside executing.") extra := 0 for _, trigger := range execution.Workflow.Triggers { From 90d80cc61728486bbb6d7cabdfb93a999f6b59d7 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Wed, 9 Sep 2026 11:22:18 +0530 Subject: [PATCH 14/30] fix: adjust CompletedAt timestamp handling for agent execution --- ai.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/ai.go b/ai.go index 2ceb7351..8bebc89d 100644 --- a/ai.go +++ b/ai.go @@ -10624,7 +10624,11 @@ data_filter: isStandalone := execution.ExecutionId == execution.WorkflowId || execution.ExecutionId == execution.Workflow.ID if isStandalone { execution.Status = "FINISHED" - execution.CompletedAt = agentOutput.CompletedAt + 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 From 4e3103e11bfd6a2e034d9195424728d12b8b3397 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Wed, 9 Sep 2026 13:34:05 +0530 Subject: [PATCH 15/30] fix: prevent potential nil pointer panic in HandleAiAgentExecutionStart for non-cloud environments --- executions.go | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/executions.go b/executions.go index 41632ae5..4cc477e1 100644 --- a/executions.go +++ b/executions.go @@ -379,9 +379,16 @@ 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) + // HandleAiAgentExecutionStart requires OpenSearch/Cloud DB infrastructure. + // The worker process runs without OpenSearch, so calling this there causes a nil pointer panic. + // On the worker, Cloud already handles re-invocation via the redeployment queue — skip here. + 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) } }() } From bb19d5d6457054f1c6ee0b1e0d43d4e87ce77c27 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Wed, 9 Sep 2026 15:46:28 +0530 Subject: [PATCH 16/30] Revert "onprem" This reverts commit 32b584a8c41eedd38c501d64cb3bdfb450df9752. --- shared.go | 57 +------------------------------------------------------ 1 file changed, 1 insertion(+), 56 deletions(-) diff --git a/shared.go b/shared.go index e072e27d..cc9a7f53 100644 --- a/shared.go +++ b/shared.go @@ -19234,50 +19234,6 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut } } - // Special handler for AI Agent hybrid dispatch (where worker ran run_agent and backend dispatched to Cloud) - isAgentHybrid := (actionResult.Action.AppName == "shuffle-ai" || actionResult.Action.AppName == "AI Agent" || actionResult.Action.AppName == "Shuffle Agent" || actionResult.Action.Name == "run_agent") && - (strings.Contains(actionResult.Result, "\"mode\":\"hybrid\"") || strings.Contains(actionResult.Result, "\"mode\": \"hybrid\"")) - - if isAgentHybrid && actionResult.Status != "SKIPPED" { - log.Printf("[INFO][%s] AI Agent hybrid dispatch detected for action %s (%s). Setting node and workflow status to WAITING.", workflowExecution.ExecutionId, actionResult.Action.Label, actionResult.Action.ID) - - actionResult.Status = "WAITING" - actionResult.CompletedAt = time.Now().Unix() * 1000 - workflowExecution.Status = "WAITING" - - foundWaiting := false - for resultIndex, result := range workflowExecution.Results { - if result.Action.ID != actionResult.Action.ID { - continue - } - - workflowExecution.Results[resultIndex] = actionResult - actionResultBody, err := json.Marshal(actionResult) - if err == nil { - cacheId := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) - _ = SetCache(ctx, cacheId, actionResultBody, 600) - } - foundWaiting = true - break - } - - if !foundWaiting { - workflowExecution.Results = append(workflowExecution.Results, actionResult) - actionResultBody, err := json.Marshal(actionResult) - if err == nil { - cacheId := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) - _ = SetCache(ctx, cacheId, actionResultBody, 600) - } - } - - err = SetWorkflowExecution(ctx, workflowExecution, true) - if err != nil { - log.Printf("[ERROR][%s] Failed setting workflow execution during AI Agent hybrid return: %s", workflowExecution.ExecutionId, err) - } - - 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" { @@ -23840,18 +23796,7 @@ func ValidateNewWorkerExecution(ctx context.Context, body []byte, shouldReset bo } } - hasWaiting := false - for _, result := range execution.Results { - if result.Status == "WAITING" { - hasWaiting = true - break - } - } - - if hasWaiting { - log.Printf("[INFO][%s] Workflow execution has action(s) in WAITING status. Preserving WAITING status.", execution.ExecutionId) - execution.Status = "WAITING" - } else if execution.Status == "EXECUTING" { + if execution.Status == "EXECUTING" { //log.Printf("[INFO] Inside executing.") extra := 0 for _, trigger := range execution.Workflow.Triggers { From 366818d44003d0329f6fa96ca3d1cadd53428c11 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Wed, 9 Sep 2026 16:35:03 +0530 Subject: [PATCH 17/30] feat: add special handling for AI Agent hybrid mode in workflow execution --- shared.go | 47 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/shared.go b/shared.go index cc9a7f53..6a01161d 100644 --- a/shared.go +++ b/shared.go @@ -19234,6 +19234,53 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut } } + // Specific handler for AI Agent hybrid thing (where worker ran run_agent and it reached to Cloud) + isAgentHybrid := (actionResult.Action.AppName == "shuffle-ai" || actionResult.Action.AppName == "AI Agent" || actionResult.Action.AppName == "Shuffle Agent" || actionResult.Action.Name == "run_agent") && + (strings.Contains(actionResult.Result, "\"mode\":\"hybrid\"") || strings.Contains(actionResult.Result, "\"mode\": \"hybrid\"")) + + if isAgentHybrid && actionResult.Status != "SKIPPED" { + log.Printf("[INFO][%s] AI Agent hybrid dispatch detected 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 strings.Contains(existingResult.Result, "decisions") && !strings.Contains(actionResult.Result, "decisions") { + actionResult.Result = existingResult.Result + } + + workflowExecution.Results[resultIndex] = actionResult + actionResultBody, marshalError := json.Marshal(actionResult) + if marshalError == nil { + cacheIdentifier := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) + _ = SetCache(ctx, cacheIdentifier, actionResultBody, 600) + } + foundWaiting = true + break + } + + if !foundWaiting { + workflowExecution.Results = append(workflowExecution.Results, actionResult) + actionResultBody, marshalError := json.Marshal(actionResult) + if marshalError == nil { + cacheIdentifier := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) + _ = SetCache(ctx, cacheIdentifier, actionResultBody, 600) + } + } + + saveError := SetWorkflowExecution(ctx, workflowExecution, true) + if saveError != nil { + log.Printf("[ERROR][%s] Failed setting workflow execution during AI Agent hybrid 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" { From 30316ce5d931da0d6bf81b8356c9ec9accfd3269 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Wed, 9 Sep 2026 17:07:50 +0530 Subject: [PATCH 18/30] added a silly debug log for testing --- shared.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/shared.go b/shared.go index 6a01161d..1a6cd004 100644 --- a/shared.go +++ b/shared.go @@ -19238,6 +19238,8 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut isAgentHybrid := (actionResult.Action.AppName == "shuffle-ai" || actionResult.Action.AppName == "AI Agent" || actionResult.Action.AppName == "Shuffle Agent" || actionResult.Action.Name == "run_agent") && (strings.Contains(actionResult.Result, "\"mode\":\"hybrid\"") || strings.Contains(actionResult.Result, "\"mode\": \"hybrid\"")) + log.Printf("[HEYOOO-PARSED] ParsedExecutionResult: Label='%s', AppName='%s', ActionName='%s', isAgentHybrid=%t, ResultPreview=%.150s", actionResult.Action.Label, actionResult.Action.AppName, actionResult.Action.Name, isAgentHybrid, actionResult.Result) + if isAgentHybrid && actionResult.Status != "SKIPPED" { log.Printf("[INFO][%s] AI Agent hybrid dispatch detected for action %s (%s). Setting node status to WAITING.", workflowExecution.ExecutionId, actionResult.Action.Label, actionResult.Action.ID) From 10598da30b7b1c5d629176e69444cedac38b6fd9 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Wed, 9 Sep 2026 18:52:35 +0530 Subject: [PATCH 19/30] fix: improve hybrid mode detection for AI Agent actions in execution result parsing --- shared.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/shared.go b/shared.go index 1a6cd004..7aec695d 100644 --- a/shared.go +++ b/shared.go @@ -19235,8 +19235,8 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut } // Specific handler for AI Agent hybrid thing (where worker ran run_agent and it reached to Cloud) - isAgentHybrid := (actionResult.Action.AppName == "shuffle-ai" || actionResult.Action.AppName == "AI Agent" || actionResult.Action.AppName == "Shuffle Agent" || actionResult.Action.Name == "run_agent") && - (strings.Contains(actionResult.Result, "\"mode\":\"hybrid\"") || strings.Contains(actionResult.Result, "\"mode\": \"hybrid\"")) + isAgentAction := actionResult.Action.AppName == "shuffle-ai" || actionResult.Action.AppName == "AI Agent" || actionResult.Action.AppName == "Shuffle Agent" || actionResult.Action.Name == "run_agent" + isAgentHybrid := isAgentAction && (strings.Contains(strings.ToLower(actionResult.Result), "hybrid") || actionResult.Action.Name == "run_agent") log.Printf("[HEYOOO-PARSED] ParsedExecutionResult: Label='%s', AppName='%s', ActionName='%s', isAgentHybrid=%t, ResultPreview=%.150s", actionResult.Action.Label, actionResult.Action.AppName, actionResult.Action.Name, isAgentHybrid, actionResult.Result) From 90584c5816df7b644dddc91f0bf37d1e3d21c614 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Wed, 9 Sep 2026 23:27:38 +0530 Subject: [PATCH 20/30] fix: caching for action results containing decisions in execution result parsing --- shared.go | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/shared.go b/shared.go index 7aec695d..5b344215 100644 --- a/shared.go +++ b/shared.go @@ -19257,10 +19257,9 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut } workflowExecution.Results[resultIndex] = actionResult - actionResultBody, marshalError := json.Marshal(actionResult) - if marshalError == nil { + if strings.Contains(actionResult.Result, "decisions") { cacheIdentifier := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) - _ = SetCache(ctx, cacheIdentifier, actionResultBody, 600) + _ = SetCache(ctx, cacheIdentifier, []byte(actionResult.Result), 600) } foundWaiting = true break @@ -19268,10 +19267,9 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut if !foundWaiting { workflowExecution.Results = append(workflowExecution.Results, actionResult) - actionResultBody, marshalError := json.Marshal(actionResult) - if marshalError == nil { + if strings.Contains(actionResult.Result, "decisions") { cacheIdentifier := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) - _ = SetCache(ctx, cacheIdentifier, actionResultBody, 600) + _ = SetCache(ctx, cacheIdentifier, []byte(actionResult.Result), 600) } } From 89d7a17fa0e53d94d2bff9e3a3c0974e2a9b6728 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Thu, 10 Sep 2026 18:35:41 +0530 Subject: [PATCH 21/30] fix: remove unnecessary comments --- executions.go | 3 --- 1 file changed, 3 deletions(-) diff --git a/executions.go b/executions.go index 4cc477e1..0ab88506 100644 --- a/executions.go +++ b/executions.go @@ -379,9 +379,6 @@ func Fixexecution(ctx context.Context, workflowExecution WorkflowExecution) (Wor time.Sleep(1 * time.Second) sendAgentActionSelfRequest("WAITING", capturedExec, capturedExec.Results[resultIndex]) time.Sleep(2 * time.Second) - // HandleAiAgentExecutionStart requires OpenSearch/Cloud DB infrastructure. - // The worker process runs without OpenSearch, so calling this there causes a nil pointer panic. - // On the worker, Cloud already handles re-invocation via the redeployment queue — skip here. if project.Environment == "cloud" { _, err := HandleAiAgentExecutionStart(capturedExec, capturedAction, true, "fixexecution_timeout_recovery") if err != nil { From 4814fb0be4df4413459a4f06531b55232263c4ac Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Fri, 11 Sep 2026 16:12:43 +0530 Subject: [PATCH 22/30] fix: made merging of prior node results in AI Agent execution handling --- ai.go | 45 +++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 41 insertions(+), 4 deletions(-) diff --git a/ai.go b/ai.go index 8bebc89d..69d6f48e 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) + } + } } } @@ -10287,6 +10303,7 @@ data_filter: execution.Status = "EXECUTING" agentOutput.Status = "RUNNING" + foundAgentIndex := -1 for resultIndex, result := range execution.Results { if result.Action.ID != startNode.ID { continue @@ -10309,6 +10326,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) From 896f6d93b9b46af883e9a9fcfcf850b816197dc9 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Fri, 11 Sep 2026 16:15:59 +0530 Subject: [PATCH 23/30] fix AI Agent node handling and logging in execution result parsing --- shared.go | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/shared.go b/shared.go index 5b344215..71d829ec 100644 --- a/shared.go +++ b/shared.go @@ -19234,14 +19234,9 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut } } - // Specific handler for AI Agent hybrid thing (where worker ran run_agent and it reached to Cloud) - isAgentAction := actionResult.Action.AppName == "shuffle-ai" || actionResult.Action.AppName == "AI Agent" || actionResult.Action.AppName == "Shuffle Agent" || actionResult.Action.Name == "run_agent" - isAgentHybrid := isAgentAction && (strings.Contains(strings.ToLower(actionResult.Result), "hybrid") || actionResult.Action.Name == "run_agent") - - log.Printf("[HEYOOO-PARSED] ParsedExecutionResult: Label='%s', AppName='%s', ActionName='%s', isAgentHybrid=%t, ResultPreview=%.150s", actionResult.Action.Label, actionResult.Action.AppName, actionResult.Action.Name, isAgentHybrid, actionResult.Result) - - if isAgentHybrid && actionResult.Status != "SKIPPED" { - log.Printf("[INFO][%s] AI Agent hybrid dispatch detected for action %s (%s). Setting node status to WAITING.", workflowExecution.ExecutionId, actionResult.Action.Label, actionResult.Action.ID) + 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 @@ -19252,6 +19247,11 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut 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 } @@ -19275,7 +19275,7 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut saveError := SetWorkflowExecution(ctx, workflowExecution, true) if saveError != nil { - log.Printf("[ERROR][%s] Failed setting workflow execution during AI Agent hybrid return: %s", workflowExecution.ExecutionId, saveError) + log.Printf("[ERROR][%s] Failed setting workflow execution during AI Agent return: %s", workflowExecution.ExecutionId, saveError) } return &workflowExecution, true, nil From baa33a21c40766104a8bba6797ff5695953ce132 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Fri, 11 Sep 2026 16:33:53 +0530 Subject: [PATCH 24/30] fix: refactor backend URL retrieval logic and adjust tool timeout settings --- cloudSync.go | 40 ++++++++++++++++++++++++++-------------- 1 file changed, 26 insertions(+), 14 deletions(-) diff --git a/cloudSync.go b/cloudSync.go index f6520bbd..294c20e9 100755 --- a/cloudSync.go +++ b/cloudSync.go @@ -2276,6 +2276,29 @@ func HandleSuborgScheduleRun(request *http.Request, workflow *Workflow) { } } +func getBackendBaseUrl() string { + baseUrl := os.Getenv("BASE_URL") + if len(baseUrl) > 0 { + return baseUrl + } + + cloudrunUrl := os.Getenv("SHUFFLE_CLOUDRUN_URL") + if len(cloudrunUrl) > 0 { + return cloudrunUrl + } + + 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() @@ -2503,21 +2526,10 @@ 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() - toolTimeout := 120 - if v := os.Getenv("AGENT_TOOL_TIMEOUT"); len(v) > 0 { + toolTimeout := 30 + timeoutOverride := os.Getenv("AGENT_TOOL_TIMEOUT"); len(v) > 0 { if parsed, err := strconv.Atoi(v); err == nil && parsed > 0 && parsed <= 300 { toolTimeout = parsed } From 56bcb4af5f76b084af0ba77ae3051a7b690b3d27 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Fri, 11 Sep 2026 16:36:00 +0530 Subject: [PATCH 25/30] fix: streamline backend URL retrieval and adjust tool timeout settings --- cloudSync.go | 32 ++++++++------------------------ 1 file changed, 8 insertions(+), 24 deletions(-) diff --git a/cloudSync.go b/cloudSync.go index 294c20e9..5f559bd9 100755 --- a/cloudSync.go +++ b/cloudSync.go @@ -2529,9 +2529,10 @@ func runAgentDecisionDirectAppCall(execution WorkflowExecution, decision AgentDe baseURL := getBackendBaseUrl() toolTimeout := 30 - timeoutOverride := os.Getenv("AGENT_TOOL_TIMEOUT"); len(v) > 0 { - if parsed, err := strconv.Atoi(v); err == nil && parsed > 0 && parsed <= 300 { - toolTimeout = parsed + timeoutOverride := os.Getenv("AGENT_TOOL_TIMEOUT") + if len(timeoutOverride) > 0 { + if parsedTimeout, err := strconv.Atoi(timeoutOverride); err == nil && parsedTimeout > 0 && parsedTimeout <= 300 { + toolTimeout = parsedTimeout } } timeout := time.Duration(toolTimeout) * time.Second @@ -2548,9 +2549,9 @@ func runAgentDecisionDirectAppCall(execution WorkflowExecution, decision AgentDe requestUrl += fmt.Sprintf("parent_node=%s", parentNode) } - // Gives it time to return properly with +5 delay + // Gives it time to return properly with +2 delay client := GetExternalClientWithTimeout(requestUrl, 0) - client.Timeout = timeout + (5 * time.Second) + client.Timeout = timeout + (2 * time.Second) //if debug { //log.Printf("\n\n\n\nRequest timeout: %d", client.Timeout) @@ -2645,14 +2646,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) @@ -3102,17 +3096,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 := "http://localhost:5001" - if project.Environment == "cloud" { - 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) From b0bb7fa608bf1ec763c2d6277a78cdeccd8b698d Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Fri, 11 Sep 2026 16:38:20 +0530 Subject: [PATCH 26/30] fix: improve tool name normalization for additional format handling --- cloudSync.go | 2 ++ shared.go | 8 ++++---- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/cloudSync.go b/cloudSync.go index 5f559bd9..83b37b17 100755 --- a/cloudSync.go +++ b/cloudSync.go @@ -2900,6 +2900,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) diff --git a/shared.go b/shared.go index 71d829ec..c1e87c93 100644 --- a/shared.go +++ b/shared.go @@ -36863,15 +36863,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) @@ -36891,7 +36891,7 @@ func getPrioritisedAppActions(ctx context.Context, inputApp string, maxAmount in if err == nil && len(foundApps) > 0 { foundApp = &foundApps[0] } - if foundApp.ID == "" { + if foundApp.ID == "" && project.Environment == "cloud" { algoliaApp, err := HandleAlgoliaAppSearch(ctx, cleanName) if err == nil && len(algoliaApp.ObjectID) > 0 { if len(foundApp.Actions) == 0 { From d3e3d54e34377178c33b7da822dcd3fe67848b3c Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Mon, 14 Sep 2026 10:13:33 +0000 Subject: [PATCH 27/30] check for both local ai creds and cloud sync for on-prem AI Agent --- ai.go | 30 +++++++++++++++++------------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/ai.go b/ai.go index 69d6f48e..8d1a74dc 100644 --- a/ai.go +++ b/ai.go @@ -8332,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) @@ -8341,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) } } @@ -8371,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 From 60b31bbc66ae596e2aaefaf74bb7537a5d524a6a Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Mon, 14 Sep 2026 10:14:16 +0000 Subject: [PATCH 28/30] fix: improve backend URL retrieval logic in getBackendBaseUrl function --- cloudSync.go | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/cloudSync.go b/cloudSync.go index 83b37b17..a9442aec 100755 --- a/cloudSync.go +++ b/cloudSync.go @@ -2277,14 +2277,17 @@ func HandleSuborgScheduleRun(request *http.Request, workflow *Workflow) { } func getBackendBaseUrl() string { - baseUrl := os.Getenv("BASE_URL") - if len(baseUrl) > 0 { - return baseUrl + backendUrl := "" + if len(os.Getenv("BASE_URL")) > 0 { + backendUrl = os.Getenv("BASE_URL") } - cloudrunUrl := os.Getenv("SHUFFLE_CLOUDRUN_URL") - if len(cloudrunUrl) > 0 { - return cloudrunUrl + if len(os.Getenv("SHUFFLE_CLOUDRUN_URL")) > 0 { + backendUrl = os.Getenv("SHUFFLE_CLOUDRUN_URL") + } + + if len(backendUrl) > 0 { + return backendUrl } if project.Environment == "cloud" { From 5ff3a55834bf4bd51c3586386befc45bfce63d01 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Mon, 14 Sep 2026 10:14:59 +0000 Subject: [PATCH 29/30] fix: adjust tool timeout settings based on environment --- cloudSync.go | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/cloudSync.go b/cloudSync.go index a9442aec..60270fd6 100755 --- a/cloudSync.go +++ b/cloudSync.go @@ -2532,12 +2532,10 @@ func runAgentDecisionDirectAppCall(execution WorkflowExecution, decision AgentDe baseURL := getBackendBaseUrl() toolTimeout := 30 - timeoutOverride := os.Getenv("AGENT_TOOL_TIMEOUT") - if len(timeoutOverride) > 0 { - if parsedTimeout, err := strconv.Atoi(timeoutOverride); err == nil && parsedTimeout > 0 && parsedTimeout <= 300 { - toolTimeout = parsedTimeout - } + 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 @@ -2554,7 +2552,7 @@ func runAgentDecisionDirectAppCall(execution WorkflowExecution, decision AgentDe // Gives it time to return properly with +2 delay client := GetExternalClientWithTimeout(requestUrl, 0) - client.Timeout = timeout + (2 * time.Second) + client.Timeout = timeout + (1 * time.Second) //if debug { //log.Printf("\n\n\n\nRequest timeout: %d", client.Timeout) From 45a8f9a0928577d521c086883ab2be54ee0fd337 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Mon, 14 Sep 2026 10:15:39 +0000 Subject: [PATCH 30/30] fix: refactor base URL retrieval to use getBackendBaseUrl function --- shared.go | 18 ++---------------- 1 file changed, 2 insertions(+), 16 deletions(-) diff --git a/shared.go b/shared.go index c1e87c93..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 { @@ -19493,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 {