Many more AI Agent cleanup mechanisms

Author: frikky

ai.go

@@ -6675,9 +6675,20 @@ SINGUL ACTIONS:
 			oldAgentOutput = mappedResult
 			previousAnswers := ""
 			relevantDecisions := []AgentDecision{}
+
+			hasFailure := false
 			for _, mappedDecision := range mappedResult.Decisions {
+				if mappedDecision.RunDetails.Status == "FAILURE" {
+					// Overrides as to get the correct index
+					if lastFinishedIndex < mappedDecision.I {
+						lastFinishedIndex = mappedDecision.I
+					}
+
+					hasFailure = true 
+				}
+
 				if mappedDecision.RunDetails.Status != "FINISHED" && mappedDecision.RunDetails.Status != "SUCCESS" {
-					log.Printf("[DEBUG][%s] SKIPPING decision index %d with status %s", execution.ExecutionId, mappedDecision.I, mappedDecision.RunDetails.Status)
+					log.Printf("[DEBUG][%s] SKIPPING decision index %d (%s) with status %s", execution.ExecutionId, mappedDecision.I, mappedDecision.RunDetails.Id, mappedDecision.RunDetails.Status)
 					continue
 				}
 
@@ -6705,6 +6716,10 @@ SINGUL ACTIONS:
 				break
 			}
 
+			if hasFailure { 
+				userMessage += "\n\nSome of the previous decisions failed. Finalise the agent.\n\n"
+			}
+
 			userMessage += fmt.Sprintf("\n\nPrevious decision results:\n%s", string(marshalledDecisions))
 			if len(previousAnswers) > 0 {
 				userMessage += fmt.Sprintf("\n\nAnswers to questions:\n%s", previousAnswers)
@@ -6767,6 +6782,7 @@ RULES:
 * Do NOT add unnecessary fields; only include fields required for the action.
 * Fields can reference previous action outputs using {{action_name}}. Example: {"body": "{{previous_action.field}}"}.
 * If questions are absolutely required, combine all into one "ask" action with multiple "question" fields. Do NOT create multiple separate decisions.
+* If any decision has failed, add the finish decision.
 
 END RULES
 ---
@@ -7202,7 +7218,19 @@ FINALISING:
 				agentOutput.Status = "RUNNING"
 			}
 
-			log.Printf("Got %d NEW decision(s)", len(mappedDecisions))
+			if debug { 
+				log.Printf("[DEBUG] Got %d NEW decision(s)", len(mappedDecisions))
+			}
+
+			// Verbose error handling optimisations
+			for _, mappedDecision := range mappedDecisions {
+				if mappedDecision.I == lastFinishedIndex && mappedDecision.RunDetails.Status == "FAILURE" {
+					if debug { 
+						log.Printf("\n\n\n\n\nMAPPING TO FAILURE DUE TO DECISION INDEX AND STATUS!!! Decisions that aren't 'finalise' should be ignored\n\n\n\n\n\n\n")
+					}
+				}
+			}
+
 			additions := 0
 			for _, mappedDecision := range mappedDecisions {
 				if mappedDecision.I < lastFinishedIndex {
@@ -7257,8 +7285,9 @@ FINALISING:
 		}
 
 		if resultMapping.Status == "FAILURE" {
-			agentOutput.Status = "FAILURE"
-			agentOutput.CompletedAt = time.Now().Unix()
+			log.Printf("\n\n\n\n\nMAPPING TO FAILURE!!!\n\n\nn\n\n\n\n")
+			//agentOutput.Status = "FAILURE"
+			//agentOutput.CompletedAt = time.Now().Unix()
 		}
 
 		if !createNextActions {
@@ -7315,6 +7344,11 @@ FINALISING:
 
 
 				agentOutput.Output = decision.Reason
+				agentOutput.Status = "FINISHED"
+				agentOutput.CompletedAt = time.Now().Unix()
+
+				//workflowExecution.Results[resultIndex].Status = "SUCCESS"
+				//go sendAgentActionSelfRequest("SUCCESS", workflowExecution, workflowExecution.Results[resultIndex])
 
 			} else if decision.Action == "ask" || decision.Action == "question" { 
 				agentOutput.Decisions[decisionIndex].RunDetails.StartedAt = time.Now().Unix()
@@ -7426,10 +7460,26 @@ FINALISING:
 			}
 		}
 
+
+		if agentOutput.Status == "FINISHED" && agentOutput.CompletedAt > 0 && execution.Status == "EXECUTING" {
+			log.Printf("[INFO][%s] AI Agent action %s finished.", execution.ExecutionId, startNode.ID)
+			for resultIndex, result := range execution.Results {
+				if result.Action.ID != startNode.ID {
+					continue
+				}
+
+				execution.Results[resultIndex].Status = "SUCCESS"
+				execution.Results[resultIndex].CompletedAt = agentOutput.CompletedAt
+				go sendAgentActionSelfRequest("SUCCESS", execution, execution.Results[resultIndex])
+				break
+			}
+		}
+
 	} else {
 		log.Printf("[ERROR] No result found in AI agent response. Status: %d. Body: %s", newresp.StatusCode, string(body))
 	}
 
+
 	if memorizationEngine == "shuffle_db" {
 		requestKey := fmt.Sprintf("chat_%s_%s", execution.ExecutionId, startNode.ID)
 

cloudSync.go

@@ -2186,7 +2186,10 @@ func RunAgentDecisionSingulActionHandler(execution WorkflowExecution, decision A
 	appId := ""
 	_ = appId
 	for key, value := range resp.Header {
-		log.Printf("\n\n\n\n[DEBUG][%s] HEADER: key: %s, value: %s\n\n\n\n", execution.ExecutionId, key, value)
+		//if debug { 
+		//	log.Printf("\n\n\n\n[DEBUG][%s] HEADER: key: %s, value: %s\n\n\n\n", execution.ExecutionId, key, value)
+		//}
+
 		if key == "X-Appname" && len(value) > 0 {
 			appname = value[0]
 			continue
@@ -2335,7 +2338,7 @@ func RunAgentDecisionAction(execution WorkflowExecution, agentOutput AgentOutput
 
 	go SetCache(ctx, decisionId, marshalledDecision, 60)
 
-	if decision.Action == "user_input" || decision.Action == "ask" || decision.Action == "question" || decision.Action == "finish" || decision.Category == "standalone" { 
+	if decision.Action == "user_input" || decision.Action == "answer" || decision.Action == "ask" || decision.Action == "question" || decision.Action == "finish" || decision.Category == "standalone" { 
 	} else {
 		// Singul handler
 		rawResponse, debugUrl, appname, err := RunAgentDecisionSingulActionHandler(execution, decision)

db-connector.go

@@ -1263,6 +1263,7 @@ func Fixexecution(ctx context.Context, workflowExecution WorkflowExecution) (Wor
 						finishedDecisions = append(finishedDecisions, decision.RunDetails.Id)
 						continue
 					} else if decision.RunDetails.Status == "FAILURE" {
+						//finishedDecisions = append(finishedDecisions, decision.RunDetails.Id)
 						failedFound = true
 						continue
 					}
@@ -1285,16 +1286,23 @@ func Fixexecution(ctx context.Context, workflowExecution WorkflowExecution) (Wor
 					}
 				}
 
+				// FIXME: Is failure hadnling here necessary?
+				// Changed it to do failure handling better in the agent itself
+				// due to having a 'finish' action that should handle it properly
 				if failedFound {
 					decisionsUpdated = true
 
+					/*
 					mappedOutput.Status = "FAILURE"
 					mappedOutput.CompletedAt = time.Now().Unix()
 					workflowExecution.Results[resultIndex].Status = "ABORTED"
 
 					go sendAgentActionSelfRequest("FAILURE", workflowExecution, workflowExecution.Results[resultIndex])
+					*/
+
+				} 
 
-				} else if len(finishedDecisions) == len(mappedOutput.Decisions) && mappedOutput.Status != "FINISHED" && mappedOutput.Status != "FAILURE" && mappedOutput.Status != "ABORTED" {
+				if len(finishedDecisions) == len(mappedOutput.Decisions) && mappedOutput.Status != "FINISHED" && mappedOutput.Status != "FAILURE" && mappedOutput.Status != "ABORTED" {
 					decisionsUpdated = true
 					mappedOutput.Status = "FINISHED"
 					mappedOutput.CompletedAt = time.Now().Unix()

shared.go

@@ -16226,6 +16226,15 @@ func handleAgentDecisionStreamResult(workflowExecution WorkflowExecution, action
 		if err != nil {
 			log.Printf("[ERROR][%s] Failed unmarshalling agent decision result for decision ID '%s': %s. Data: %s", workflowExecution.ExecutionId, decisionId, err, actionResult.Result)
 		} else {
+			if newDecision.Action == "answer" {
+				if debug { 
+					log.Printf("[DEBUG] Auto-finishing 'answer' decision for decision ID '%s'", decisionId)
+				}
+
+				newDecision.RunDetails.Status = "FINISHED"
+				mappedResult.Decisions[decisionIdResultIndex] = newDecision
+			}
+
 			if newDecision.RunDetails.Status != "" && newDecision.RunDetails.Id != "" && (newDecision.RunDetails.Status != mappedResult.Decisions[decisionIdResultIndex].RunDetails.Status || newDecision.RunDetails.StartedAt != mappedResult.Decisions[decisionIdResultIndex].RunDetails.StartedAt || newDecision.RunDetails.CompletedAt != mappedResult.Decisions[decisionIdResultIndex].RunDetails.CompletedAt) {
 
 				log.Printf("[DEBUG][%s] Updating decision ID '%s' with new status '%s' (old: '%s')", workflowExecution.ExecutionId, decisionId, newDecision.RunDetails.Status, mappedResult.Decisions[decisionIdResultIndex].RunDetails.Status)
@@ -16250,9 +16259,12 @@ func handleAgentDecisionStreamResult(workflowExecution WorkflowExecution, action
 	}
 
 	if mappedResult.Decisions[decisionIdResultIndex].RunDetails.Status == "FAILURE" || mappedResult.Decisions[decisionIdResultIndex].RunDetails.Status == "ABORTED" {
+		if debug { 
+			log.Printf("[DEBUG] Auto-failing agent due to decision ID '%s' being in status '%s'", decisionId, mappedResult.Decisions[decisionIdResultIndex].RunDetails.Status)
+		}
 
-		go sendAgentActionSelfRequest("FAILURE", workflowExecution, workflowExecution.Results[foundActionResultIndex])
-		return &workflowExecution, false, nil
+		//go sendAgentActionSelfRequest("FAILURE", workflowExecution, workflowExecution.Results[foundActionResultIndex])
+		//return &workflowExecution, false, nil
 	}
 
 	//mappedResult.Decisions[decisionIdResultIndex] = actionResult.Result
@@ -16261,6 +16273,12 @@ func handleAgentDecisionStreamResult(workflowExecution WorkflowExecution, action
 	allFinishedDecisions := []string{}
 	for decisionId, curDecision := range mappedResult.Decisions {
 		if curDecision.RunDetails.Status == "FINISHED" {
+			allFinishedDecisions = append(allFinishedDecisions, curDecision.RunDetails.Id)
+		} else if curDecision.RunDetails.Status == "FAILURE" {
+			if debug { 
+				log.Printf("[DEBUG] Treating decision ID '%s' as finished due to FAILURE status", curDecision.RunDetails.Id)
+			}
+
 			allFinishedDecisions = append(allFinishedDecisions, curDecision.RunDetails.Id)
 		}
 
@@ -16298,8 +16316,8 @@ func handleAgentDecisionStreamResult(workflowExecution WorkflowExecution, action
 		if len(failedDecisions) > 0 {
 			log.Printf("[WARNING][%s] Failed decision found. Should exit out agent %s. It should have exited before this point.", workflowExecution.ExecutionId, decisionId)
 
-			go sendAgentActionSelfRequest("FAILURE", workflowExecution, workflowExecution.Results[foundActionResultIndex])
-			break
+			//go sendAgentActionSelfRequest("FAILURE", workflowExecution, workflowExecution.Results[foundActionResultIndex])
+			//break
 		}
 
 		if len(foundDecisions) == len(finishedDecisions) {
@@ -20639,7 +20657,7 @@ func PrepareSingleAction(ctx context.Context, user User, appId string, body []by
 		// 1. Find the decision & reset cache
 		// 2. Update the execution itself to not have the relevant data
 		if len(decisionId) > 0 {
-			log.Printf("[DEBUG][%s] Handling Single action rerun for AI Agent decision. DecisionID: %#v", oldExec.ExecutionId, decisionId)
+			log.Printf("[DEBUG][%s] Handling Single action RERUN for AI Agent decision. DecisionID: %#v", oldExec.ExecutionId, decisionId)
 
 			if foundResultIndex == -1 {
 				return workflowExecution, errors.New("Failed to find the action. Please try again or contact [email protected] if this persists.")
@@ -20653,26 +20671,69 @@ func PrepareSingleAction(ctx context.Context, user User, appId string, body []by
 
 			availableDecisions := []string{}
 			foundDecisionIndex := -1
+			decisionPosition := -1
+
+			newDecisions := []AgentDecision{}
 			for decisionIndex, decision := range mappedOutput.Decisions {
 				availableDecisions = append(availableDecisions, decision.RunDetails.Id)
 				if decision.RunDetails.Id != decisionId {
+					if decision.Action == "finish" {
+						if debug { 
+							log.Printf("[DEBUG] Removing the 'finish' action due to rerun")
+						}
+
+						continue
+					}
+
+					newDecisions = append(newDecisions, decision)
+
 					continue
 				}
 
-				foundDecisionIndex = decisionIndex
-
+				// The position in the hierarchy
+				decisionPosition = mappedOutput.Decisions[decisionIndex].I
+				foundDecisionIndex = decisionIndex 
 				mappedOutput.CompletedAt = 0
 				mappedOutput.Decisions[decisionIndex].RunDetails.Status = "RUNNING"
 				mappedOutput.Decisions[decisionIndex].RunDetails.CompletedAt = 0
 				mappedOutput.Decisions[decisionIndex].RunDetails.RawResponse = ""
 				mappedOutput.Decisions[decisionIndex].RunDetails.DebugUrl = ""
-				break
+
+				newDecisions = append(newDecisions, mappedOutput.Decisions[decisionIndex])
 			}
 
 			if foundDecisionIndex == -1 {
 				return workflowExecution, errors.New(fmt.Sprintf("Failed to find and rerun decision '%s' out of '%s' in execution %s. Please try again or contact [email protected] if the error persists.", decisionId, strings.Join(availableDecisions, ","), oldExec.ExecutionId))
 			}
 
+			// Removing everything AFTER the one we are currently on
+			// Has to be done in a wonky way due to not having ordered arrays
+			newNewDecisions := []AgentDecision{}
+			for _, newDecision := range newDecisions {
+				if decisionPosition != -1 && newDecision.I > decisionPosition && newDecision.RunDetails.Id != decisionId {
+					if debug {
+						log.Printf("[DEBUG] SKIPPING decision %s as it's after the rerun position", newDecision.RunDetails.Id)
+					}
+
+					continue
+				}
+
+				if newDecision.RunDetails.Status == "" {
+					continue
+				}
+
+				newNewDecisions = append(newNewDecisions, newDecision)
+			}
+
+
+			for newDecisionIndex, newDecision := range newNewDecisions {
+				if newDecision.RunDetails.Id == decisionId {
+					foundDecisionIndex = newDecisionIndex
+				}
+			}
+
+			mappedOutput.Decisions = newNewDecisions 
+
 			mappedOutput.Status = "WAITING"
 			marshalledResult, err := json.Marshal(mappedOutput)
 			if err == nil {