From efaa3e32bc7eb8971d511bfedefcbfe4b738c279 Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Mon, 9 Feb 2026 15:41:35 +0530 Subject: [PATCH 1/4] feat: made ai-agent app work on worker --- functions/onprem/worker/worker.go | 80 +++++++++++++++++++++++++++---- 1 file changed, 71 insertions(+), 9 deletions(-) diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index eb3db030..f0a95c76 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -1498,11 +1498,63 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { action, _ = singul.HandleSingulStartnode(workflowExecution, action, []string{}) parsedAppname := strings.Replace(strings.ToLower(action.AppName), " ", "-", -1) - //if strings.ToLower(parsedAppname) == "singul" { - // parsedAppname = "shuffle-ai" - // appversion = "1.0.0" - // appname = "shuffle-ai" - //} + if parsedAppname == "ai-agent" { + log.Printf("[INFO][%s] Running AI Agent action %s via backend API", workflowExecution.ExecutionId, action.ID) + + // Call the backend API - it will start the agent ASYNC and return immediately + fullUrl := fmt.Sprintf("%s/api/v1/agent/hybrid/execute?execution_id=%s&authorization=%s", + baseUrl, workflowExecution.ExecutionId, workflowExecution.Authorization) + + serverUrl := os.Getenv("SHUFFLE_BACKEND_URL") + if len(serverUrl) > 0 { + fullUrl = fmt.Sprintf("%s/api/v1/agent/hybrid/execute?execution_id=%s&authorization=%s", + serverUrl, workflowExecution.ExecutionId, workflowExecution.Authorization) + } + + requestBody := map[string]interface{}{ + "id": action.ID, + "name": action.Name, + "label": action.Label, + "app_name": action.AppName, + "app_id": action.AppID, + "app_version": action.AppVersion, + "environment": action.Environment, + "parameters": action.Parameters, + } + + requestBodyBytes, err := json.Marshal(requestBody) + if err != nil { + log.Printf("[ERROR][%s] Failed marshalling request body: %s", workflowExecution.ExecutionId, err) + continue + } + + req, err := http.NewRequest("POST", fullUrl, bytes.NewBuffer(requestBodyBytes)) + if err != nil { + log.Printf("[ERROR][%s] Failed creating AI Agent request: %s", workflowExecution.ExecutionId, err) + continue + } + + req.Header.Set("Content-Type", "application/json") + + client := &http.Client{Timeout: 10 * time.Second} + resp, err := client.Do(req) + if err != nil { + log.Printf("[ERROR][%s] Failed triggering AI Agent (timeout/error): %s", workflowExecution.ExecutionId, err) + log.Printf("[INFO][%s] Worker exiting (exit 0) - backend will requeue execution when agent completes", workflowExecution.ExecutionId) + os.Exit(0) + } + + defer resp.Body.Close() + body, err := ioutil.ReadAll(resp.Body) + if err != nil { + log.Printf("[ERROR][%s] Failed reading AI Agent response: %s", workflowExecution.ExecutionId, err) + } else { + log.Printf("[INFO][%s] AI Agent triggered: %s", workflowExecution.ExecutionId, string(body)) + } + + log.Printf("[INFO][%s] Worker exiting cleanly (exit 0) - backend will requeue when agent completes", workflowExecution.ExecutionId) + os.Exit(0) + } imageName := fmt.Sprintf("%s:%s_%s", baseimagename, parsedAppname, action.AppVersion) if strings.Contains(imageName, " ") { @@ -4473,11 +4525,21 @@ func checkStandaloneRun() { // Anything else here. } - workflowExecution.Results = newResults - workflowExecution.Status = "EXECUTING" - workflowExecution.CompletedAt = 0 + // workflowExecution.Results = newResults + // workflowExecution.Status = "EXECUTING" + // workflowExecution.CompletedAt = 0 - marshalledResult, err := json.Marshal(workflowExecution) + // Creating a minimal payload to force reset without sending the full execution + // This helps avoid validation issues and potential complexity limits + simpleReset := map[string]interface{}{ + "execution_id": executionId, + "authorization": authorization, + "status": "EXECUTING", + "results": []interface{}{}, + "completed_at": 0, + } + + marshalledResult, err := json.Marshal(simpleReset) if err != nil { log.Printf("[ERROR] Failed marshalling body: %s", err) os.Exit(1) From 6eb783c1b11f577051aec0889d2489bdbdbdc81b Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Mon, 9 Feb 2026 21:18:38 +0530 Subject: [PATCH 2/4] cleaned up the agent api calling code block --- functions/onprem/worker/worker.go | 23 ++++++++++++++--------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index f0a95c76..3ad67a33 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -1498,10 +1498,15 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { action, _ = singul.HandleSingulStartnode(workflowExecution, action, []string{}) parsedAppname := strings.Replace(strings.ToLower(action.AppName), " ", "-", -1) + // if strings.ToLower(parsedAppname) == "singul" { + // parsedAppname = "shuffle-ai" + // appversion = "1.0.0" + // appname = "shuffle-ai" + // } + if parsedAppname == "ai-agent" { log.Printf("[INFO][%s] Running AI Agent action %s via backend API", workflowExecution.ExecutionId, action.ID) - // Call the backend API - it will start the agent ASYNC and return immediately fullUrl := fmt.Sprintf("%s/api/v1/agent/hybrid/execute?execution_id=%s&authorization=%s", baseUrl, workflowExecution.ExecutionId, workflowExecution.Authorization) @@ -1545,14 +1550,14 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { } defer resp.Body.Close() - body, err := ioutil.ReadAll(resp.Body) - if err != nil { - log.Printf("[ERROR][%s] Failed reading AI Agent response: %s", workflowExecution.ExecutionId, err) - } else { - log.Printf("[INFO][%s] AI Agent triggered: %s", workflowExecution.ExecutionId, string(body)) - } - - log.Printf("[INFO][%s] Worker exiting cleanly (exit 0) - backend will requeue when agent completes", workflowExecution.ExecutionId) + // body, err := ioutil.ReadAll(resp.Body) + // if err != nil { + // log.Printf("[ERROR][%s] Failed reading AI Agent response: %s", workflowExecution.ExecutionId, err) + // } else { + // log.Printf("[INFO][%s] AI Agent triggered: %s", workflowExecution.ExecutionId, string(body)) + // } + + log.Printf("[INFO][%s] Worker exiting (exit 0) - backend will requeue when agent completes", workflowExecution.ExecutionId) os.Exit(0) } From af7b59e4bb2982ad7e79f234215a9674eef1fb6c Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Mon, 9 Feb 2026 21:26:04 +0530 Subject: [PATCH 3/4] uncommented the code block --- functions/onprem/worker/worker.go | 18 ++++-------------- 1 file changed, 4 insertions(+), 14 deletions(-) diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 3ad67a33..6026db96 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -4530,21 +4530,11 @@ func checkStandaloneRun() { // Anything else here. } - // workflowExecution.Results = newResults - // workflowExecution.Status = "EXECUTING" - // workflowExecution.CompletedAt = 0 + workflowExecution.Results = newResults + workflowExecution.Status = "EXECUTING" + workflowExecution.CompletedAt = 0 - // Creating a minimal payload to force reset without sending the full execution - // This helps avoid validation issues and potential complexity limits - simpleReset := map[string]interface{}{ - "execution_id": executionId, - "authorization": authorization, - "status": "EXECUTING", - "results": []interface{}{}, - "completed_at": 0, - } - - marshalledResult, err := json.Marshal(simpleReset) + marshalledResult, err := json.Marshal(workflowExecution) if err != nil { log.Printf("[ERROR] Failed marshalling body: %s", err) os.Exit(1) From b4e5d604e97e3abd881861273d0045e545f4a93c Mon Sep 17 00:00:00 2001 From: Hari Krishna Date: Tue, 10 Feb 2026 15:31:59 +0530 Subject: [PATCH 4/4] changed the backend api and removed the force exit --- functions/onprem/worker/worker.go | 41 +++++++++++++++++++------------ 1 file changed, 25 insertions(+), 16 deletions(-) diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 6026db96..707ca7b6 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -1507,24 +1507,33 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { if parsedAppname == "ai-agent" { log.Printf("[INFO][%s] Running AI Agent action %s via backend API", workflowExecution.ExecutionId, action.ID) - fullUrl := fmt.Sprintf("%s/api/v1/agent/hybrid/execute?execution_id=%s&authorization=%s", - baseUrl, workflowExecution.ExecutionId, workflowExecution.Authorization) + fullUrl := fmt.Sprintf("%s/api/v1/agent?execution_id=%s&authorization=%s&action_id=%s", + baseUrl, workflowExecution.ExecutionId, workflowExecution.Authorization, action.ID) serverUrl := os.Getenv("SHUFFLE_BACKEND_URL") if len(serverUrl) > 0 { - fullUrl = fmt.Sprintf("%s/api/v1/agent/hybrid/execute?execution_id=%s&authorization=%s", - serverUrl, workflowExecution.ExecutionId, workflowExecution.Authorization) + fullUrl = fmt.Sprintf("%s/api/v1/agent?execution_id=%s&authorization=%s&action_id=%s", + serverUrl, workflowExecution.ExecutionId, workflowExecution.Authorization, action.ID) + } + + inputParamValue := "" + for _, param := range action.Parameters { + if strings.ToLower(param.Name) == "input" { + inputParamValue = param.Value + break + } } requestBody := map[string]interface{}{ - "id": action.ID, - "name": action.Name, - "label": action.Label, - "app_name": action.AppName, - "app_id": action.AppID, - "app_version": action.AppVersion, - "environment": action.Environment, - "parameters": action.Parameters, + "id": action.ID, + "params": map[string]interface{}{ + "tool_name": action.AppName, + "tool_id": action.AppID, + "environment": action.Environment, + "input": map[string]interface{}{ + "text": inputParamValue, + }, + }, } requestBodyBytes, err := json.Marshal(requestBody) @@ -1545,8 +1554,8 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { resp, err := client.Do(req) if err != nil { log.Printf("[ERROR][%s] Failed triggering AI Agent (timeout/error): %s", workflowExecution.ExecutionId, err) - log.Printf("[INFO][%s] Worker exiting (exit 0) - backend will requeue execution when agent completes", workflowExecution.ExecutionId) - os.Exit(0) + log.Printf("[INFO][%s] Exiting execution handler - backend will requeue when agent completes", workflowExecution.ExecutionId) + return } defer resp.Body.Close() @@ -1557,8 +1566,8 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { // log.Printf("[INFO][%s] AI Agent triggered: %s", workflowExecution.ExecutionId, string(body)) // } - log.Printf("[INFO][%s] Worker exiting (exit 0) - backend will requeue when agent completes", workflowExecution.ExecutionId) - os.Exit(0) + log.Printf("[INFO][%s] AI Agent triggered successfully - exiting execution handler, backend will requeue when agent completes", workflowExecution.ExecutionId) + return } imageName := fmt.Sprintf("%s:%s_%s", baseimagename, parsedAppname, action.AppVersion)