Merge branch 'nightly' of https://github.com/shuffle/shuffle into nightly

functions/onprem/orborus/go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/docker/docker v28.3.3+incompatible
 	github.com/docker/go-connections v0.5.0
 	github.com/satori/go.uuid v1.2.0
-	github.com/shuffle/shuffle-shared v0.9.79
+	github.com/shuffle/shuffle-shared v0.9.82
 	k8s.io/api v0.34.2
 	k8s.io/apimachinery v0.34.2
 )

functions/onprem/orborus/orborus.go

@@ -1159,6 +1159,7 @@ func fixk8sRoles() {
 	}
 }
 
+// TODO: Check if deployment or service already exist by labels and only create if not already exists
 func deployK8sWorker(image string, identifier string, env []string) error {
 	env = append(env, fmt.Sprintf("IS_KUBERNETES=true"))
 	env = append(env, fmt.Sprintf("KUBERNETES_NAMESPACE=%s", os.Getenv("KUBERNETES_NAMESPACE")))

functions/onprem/worker/README.md

@@ -1,6 +1,10 @@
 # Worker
 A worker implementation in Golang. This runs ALL Shuffle workflows onprem. In general receives jobs from Orborus.
 
+## Standalone run (testing)
+
+`go run worker.go standalone <executionid> <authorization> <optional:url>`
+
 ## Development
 The ideal way to test the Worker is with a single workflow execution, standalone. Here are some environment variables you can use:
 

functions/onprem/worker/go.mod

@@ -11,8 +11,8 @@ require (
 	github.com/docker/docker v28.3.3+incompatible
 	github.com/gorilla/mux v1.8.1
 	github.com/satori/go.uuid v1.2.0
-	github.com/shuffle/shuffle-shared v0.9.76
-	github.com/shuffle/singul v0.0.20
+	github.com/shuffle/shuffle-shared v0.9.82
+	github.com/shuffle/singul v0.0.24
 	k8s.io/api v0.34.2
 	k8s.io/apimachinery v0.34.2
 	k8s.io/client-go v0.34.2
@@ -60,7 +60,7 @@ require (
 	github.com/envoyproxy/protoc-gen-validate v1.2.1 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/frikky/kin-openapi v0.42.0 // indirect
-	github.com/frikky/schemaless v0.0.25 // indirect
+	github.com/frikky/schemaless v0.0.28 // indirect
 	github.com/fxamacker/cbor/v2 v2.9.0 // indirect
 	github.com/ghodss/yaml v1.0.0 // indirect
 	github.com/go-git/gcfg v1.5.1-0.20230307220236-3a3c6141e376 // indirect

functions/onprem/worker/worker.go

@@ -1498,11 +1498,77 @@ 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 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)
+
+			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?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,
+				"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)
+			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] Exiting execution handler - backend will requeue when agent completes", workflowExecution.ExecutionId)
+				return
+			}
+
+			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] 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)
 		if strings.Contains(imageName, " ") {