Agent stability fixes

ai.go

@@ -7647,12 +7647,28 @@ func ReduceAgentResponseData(rawResponse []byte, dataFilter string, fieldsNeeded
 // createNextActions = false => start of agent to find initial decisions
 // createNextActions = true => mid-agent to decide next steps
 func HandleAiAgentExecutionStart(execution WorkflowExecution, startNode Action, createNextActions bool, callerName string, aiResponseWrapper ...[]byte) (Action, error) {
+
+	if callerName == "" {
+		callerName = "UNKNOWN_CALLER"
+	}
+
+	log.Printf("[INFO][%s] AI Agent: HandleAiAgentExecutionStart invoked by caller: '%s' (createNextActions=%t, node=%s, status=%s)", execution.ExecutionId, callerName, createNextActions, startNode.ID, execution.Status)
+
 	ctx := context.Background()
 	aiStarttime := time.Now().UnixMilli()
 
 	replacedExecution, err := GetWorkflowExecution(ctx, execution.ExecutionId)
 	if err == nil && len(replacedExecution.Results) > 0 && (execution.Status == "EXECUTING" || execution.Status == "WAITING") { 
+		origStatus := execution.Status
+		origCompleted := execution.CompletedAt
+		origResults := execution.Results
 		execution = *replacedExecution
+		if origStatus == "EXECUTING" && (execution.Status == "FINISHED" || execution.Status == "SUCCESS") {
+			log.Printf("[INFO][%s] Preserving EXECUTING status for Agent Continuation over DB %s status", execution.ExecutionId, execution.Status)
+			execution.Status = origStatus
+			execution.CompletedAt = origCompleted
+			execution.Results = origResults
+		}
 	}
 
 	llmResponse := []byte{} 
@@ -9396,7 +9412,7 @@ data_filter:
 
 					// Update the result in cache as actions are self-corrective
 					actionCacheId := fmt.Sprintf("%s_%s_result", execution.ExecutionId, result.Action.ID)
-					err = SetCache(ctx, actionCacheId, []byte(execution.Results[resultIndex].Result), 35)
+					err = SetCache(ctx, actionCacheId, []byte(execution.Results[resultIndex].Result), 600)
 					if err != nil {
 						log.Printf("[ERROR] AI Agent: Failed setting cache for action result %s: %s", actionCacheId, err)
 					}
@@ -9434,7 +9450,6 @@ data_filter:
 		}
 
 		decisionActionRan := false
-		nextActionType := ""
 
 		for decisionIndex, decision := range agentOutput.Decisions {
 			// Random generate an ID that's 10 chars long
@@ -9471,8 +9486,6 @@ data_filter:
 				continue
 			}
 
-			nextActionType = decision.Action
-
 			normalizedAction := strings.ToLower(strings.TrimSpace(decision.Action))
 			normalizedCategory := strings.ToLower(strings.TrimSpace(decision.Category))
 			expected := expectedCategory(normalizedAction)
@@ -9691,32 +9704,23 @@ data_filter:
 
 		// Set the result in cache here as well (just in case)
 		actionCacheId := fmt.Sprintf("%s_%s_result", execution.ExecutionId, resultMapping.Action.ID)
-		err = SetCache(ctx, actionCacheId, []byte(resultMapping.Result), 35)
+		err = SetCache(ctx, actionCacheId, []byte(resultMapping.Result), 600)
 		if err != nil {
 			log.Printf("[ERROR] AI Agent: Failed setting cache for action result %s: %s", actionCacheId, err)
 		}
 
-		// Makes sure ot update the execution itself as well
-		if createNextActions == true {
-			if decisionActionRan {
-			}
-
-			// Initialised from an 'ask' request (question) to user
-			// These aren't properly being updated in the db, so
-			// we need additional logic here to ensure it is being
-			// set/started
-			if nextActionType == "ask" || nextActionType == "question" || nextActionType == "finish" || nextActionType == "answer" {
-				// Ensure we update all of it
-				for resultIndex, result := range execution.Results {
-					if result.Action.ID != startNode.ID {
-						continue
-					}
-
-					execution.Results[resultIndex] = resultMapping
+		// Always update all execution results in DB regardless of action type,
+		// so tools never lose their decisions.
+		if len(execution.Results) > 0 {
+			for resultIndex, result := range execution.Results {
+				if result.Action.ID != startNode.ID {
+					continue
 				}
 
-				SetWorkflowExecution(ctx, execution, true)
+				execution.Results[resultIndex] = resultMapping
 			}
+
+			SetWorkflowExecution(ctx, execution, true)
 		}
 
 		//log.Printf("[INFO] AI_AGENT_FINISH: execution_id=%s status=%s duration=%ds decisions=%d", execution.ExecutionId, agentOutput.Status, time.Now().Unix()-agentOutput.StartedAt, len(agentOutput.Decisions))

cloudSync.go

@@ -2402,11 +2402,11 @@ func RunAgentDecisionAction(execution WorkflowExecution, agentOutput AgentOutput
 
 		// Log individual tool execution result
 		duration := int64(0)
-		if decision.RunDetails.CompletedAt > 0 && decision.RunDetails.StartedAt > 0 {
-			duration = decision.RunDetails.CompletedAt - decision.RunDetails.StartedAt
+		if decision.RunDetails.StartedAt > 0 {
+			duration = (time.Now().UnixMilli() - decision.RunDetails.StartedAt) / 1000
 		}
 
-		log.Printf("[INFO][%s] AI_AGENT_TOOL: org=%s tool=%s action=%s status=%s duration=%ds", execution.ExecutionId, execution.Workflow.OrgId, decision.Tool, decision.Action, decision.RunDetails.Status, duration)
+		log.Printf("[DEBUG][%s] AI_AGENT_TOOL: org=%s tool=%s action=%s status=%s duration=%ds", execution.ExecutionId, execution.Workflow.OrgId, decision.Tool, decision.Action, decision.RunDetails.Status, duration)
 	}
 
 	// when there are late-returning goroutines like more than 5 mins then Fixexecution may have already stamped this decision as FAILURE (5-min timeout) and

db-connector.go

@@ -613,8 +613,21 @@ func SetWorkflowExecution(ctx context.Context, workflowExecution WorkflowExecuti
 		incomingTerminal := workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || workflowExecution.Status == "FAILURE"
 
 		if existingTerminal && !incomingTerminal {
-			log.Printf("[INFO][%s] Existing execution is already %s. Not overriding with incoming %s update.", workflowExecution.ExecutionId, existingExecution.Status, workflowExecution.Status)
-			return nil
+			isAgentContinuation := false
+			if (existingExecution.Status == "FINISHED" || existingExecution.Status == "SUCCESS") && workflowExecution.Status == "EXECUTING" && workflowExecution.CompletedAt == 0 {
+				for _, result := range workflowExecution.Results {
+					if result.Action.AppName == "AI Agent" || result.Action.AppName == "Shuffle Agent" {
+						isAgentContinuation = true
+						break
+					}
+				}
+			}
+
+			if !isAgentContinuation {
+				log.Printf("[INFO][%s] Existing execution is already %s. Not overriding with incoming %s update.", workflowExecution.ExecutionId, existingExecution.Status, workflowExecution.Status)
+				return nil
+			}
+			log.Printf("[INFO][%s] Permitting Agent Continuation: overriding existing %s execution to %s!", workflowExecution.ExecutionId, existingExecution.Status, workflowExecution.Status)
 		}
 
 		if existingTerminal && incomingTerminal && existingExecution.Status != workflowExecution.Status {
@@ -630,7 +643,7 @@ func SetWorkflowExecution(ctx context.Context, workflowExecution WorkflowExecuti
 
 	executionData, err := json.Marshal(workflowExecution)
 	if err == nil {
-		err = SetCache(ctx, cacheKey, executionData, 31)
+		err = SetCache(ctx, cacheKey, executionData, 600)
 		if err != nil {
 			//log.Printf("[WARNING] Failed updating execution cache. Setting DB! %s", err)
 			dbSave = true
@@ -1327,7 +1340,7 @@ func GetWorkflowExecution(ctx context.Context, id string) (*WorkflowExecution, e
 			return workflowExecution, getErr
 		}
 
-		err = SetCache(ctx, id, newexecution, 30)
+		err = SetCache(ctx, id, newexecution, 600)
 		if err != nil {
 			log.Printf("[WARNING] Failed updating execution: %s", err)
 		}
@@ -2337,6 +2350,22 @@ func Fixexecution(ctx context.Context, workflowExecution WorkflowExecution) (Wor
 
 			// Special cleanup for agents
 			if innerresult.Action.AppName == "AI Agent" || innerresult.Action.AppName == "Shuffle Agent" {
+				actionCacheId := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, innerresult.Action.ID)
+				if cachedData, cacheErr := GetCache(ctx, actionCacheId); cacheErr == nil {
+					cachedBytes := []byte(cachedData.([]uint8))
+					var cachedOutput AgentOutput
+					if err := json.Unmarshal(cachedBytes, &cachedOutput); err == nil && len(cachedOutput.Decisions) > 0 {
+						var currentOutput AgentOutput
+						_ = json.Unmarshal([]byte(innerresult.Result), &currentOutput)
+						if len(cachedOutput.Decisions) >= len(currentOutput.Decisions) {
+							if debug {
+								log.Printf("[DEBUG][%s] Fixexecution: upgrading agent result index %d from cache (%d decisions vs %d)", workflowExecution.ExecutionId, resultIndex, len(cachedOutput.Decisions), len(currentOutput.Decisions))
+							}
+							innerresult.Result = string(cachedBytes)
+							workflowExecution.Results[resultIndex].Result = string(cachedBytes)
+						}
+					}
+				}
 
 				// Starting autocorrections
 				mappedOutput := AgentOutput{}
@@ -2401,6 +2430,9 @@ func Fixexecution(ctx context.Context, workflowExecution WorkflowExecution) (Wor
 					// Auto fixing decision data based on cache for better decisionmaking
 					// Map the result into AgentOutput to check decisions
 
+					// Any Unix timestamp under 10 billion is in seconds (valid through year 2286). Milliseconds are > 1 trillion.
+					const maxSecondsTimestamp int64 = 10_000_000_000
+
 					finishedDecisions := []string{}
 					failedFound := false
 					finishDecisionFound := false
@@ -2425,14 +2457,18 @@ func Fixexecution(ctx context.Context, workflowExecution WorkflowExecution) (Wor
 						} else if decision.RunDetails.Status == "RUNNING" && decision.Action != "ask" {
 
 							// Max runtime of a decision at 5 minutes
-							if decision.RunDetails.StartedAt > 0 && time.Now().UnixMilli()-decision.RunDetails.StartedAt > 300000 {
+							startedTs := decision.RunDetails.StartedAt
+							if startedTs > 0 && startedTs < maxSecondsTimestamp {
+								startedTs *= 1000
+							}
+							if startedTs > 0 && time.Now().UnixMilli()-startedTs > 300000 {
 								timeoutFlagKey := fmt.Sprintf("agent-%s-%s-timeout-handled", workflowExecution.ExecutionId, decision.RunDetails.Id)
 								if _, err := GetCache(ctx, timeoutFlagKey); err == nil {
 									// Already handled this timeout in a previous check so just count it as finished.
 									finishedDecisions = append(finishedDecisions, decision.RunDetails.Id)
 									failedFound = true
 								} else {
-									log.Printf("[WARNING] AI_AGENT_DECISION_TIMEOUT: execution_id=%s tool=%s action=%s duration=%ds — marking FAILURE and triggering recovery", workflowExecution.ExecutionId, decision.Tool, decision.Action, (time.Now().UnixMilli()-decision.RunDetails.StartedAt)/1000)
+									log.Printf("[WARNING] AI_AGENT_DECISION_TIMEOUT: execution_id=%s tool=%s action=%s duration=%ds — marking FAILURE and triggering recovery", workflowExecution.ExecutionId, decision.Tool, decision.Action, (time.Now().UnixMilli()-startedTs)/1000)
 									SetCache(ctx, timeoutFlagKey, []byte("1"), 60) // 60 min TTL — long enough to outlive any recovery cycle
 
 									decisionsUpdated = true
@@ -2536,7 +2572,7 @@ func Fixexecution(ctx context.Context, workflowExecution WorkflowExecution) (Wor
 						}
 
 						// Set cache to prevent multiple sends — if cache is down, skip to prevent retry storm
-						if cacheErr := SetCache(ctx, cacheId, []byte("handled"), 1); cacheErr != nil {
+						if cacheErr := SetCache(ctx, cacheId, []byte("handled"), 60); cacheErr != nil {
 							log.Printf("[WARNING][%s] Memcache down — skipping fixexec agent self-request for action %s to prevent retry storm", workflowExecution.ExecutionId, action.ID)
 							continue
 						}
@@ -2555,6 +2591,24 @@ func Fixexecution(ctx context.Context, workflowExecution WorkflowExecution) (Wor
 								go sendAgentActionSelfRequest("SUCCESS", workflowExecution, workflowExecution.Results[resultIndex])
 							}()
 						} else {
+							mostRecentCompletion := int64(0)
+							for _, dec := range mappedOutput.Decisions {
+								ts := dec.RunDetails.CompletedAt
+								if ts > 0 && ts < maxSecondsTimestamp {
+									ts *= 1000
+								}
+								if ts > mostRecentCompletion {
+									mostRecentCompletion = ts
+								}
+							}
+							timeSinceCompletionMs := time.Now().UnixMilli() - mostRecentCompletion
+							if timeSinceCompletionMs < 60000 {
+								if debug {
+									log.Printf("[DEBUG][%s] Skipping fixexecution_timeout_recovery: last decision completed %d ms ago (waiting for LLM response from primary stream handler).", workflowExecution.ExecutionId, timeSinceCompletionMs)
+								}
+								continue
+							}
+
 							log.Printf("[INFO][%s] All decisions finished for agent action %s - but no finish action found, marking as WAITING.", workflowExecution.ExecutionId, action.ID)
 							//log.Printf("[INFO][%s] All decisions finished for agent action %s - but no finish action found. Re-invoking agent to finalize (failedFound: %t).", workflowExecution.ExecutionId, action.ID, failedFound)