diff --git a/ai.go b/ai.go index cccbff85..195927a5 100644 --- a/ai.go +++ b/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)) diff --git a/cloudSync.go b/cloudSync.go index 1bee07af..27f7f2c2 100755 --- a/cloudSync.go +++ b/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 diff --git a/db-connector.go b/db-connector.go index 214a064c..8c82a2fb 100755 --- a/db-connector.go +++ b/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), ¤tOutput) + 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) diff --git a/shared.go b/shared.go index 37ba98fc..ae4c64ef 100644 --- a/shared.go +++ b/shared.go @@ -17919,6 +17919,24 @@ func handleAgentDecisionStreamResult(workflowExecution WorkflowExecution, action ctx := context.Background() + // re-fetch the freshest execution state from DB/cache before doing ANY decision counting. + if freshExec, fetchErr := GetWorkflowExecution(ctx, workflowExecution.ExecutionId); fetchErr == nil && freshExec != nil { + if len(freshExec.Results) >= len(workflowExecution.Results) { + origStatus := workflowExecution.Status + origCompleted := workflowExecution.CompletedAt + origResults := workflowExecution.Results + workflowExecution = *freshExec + if origStatus == "EXECUTING" && (workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS") { + log.Printf("[INFO][%s] Preserving EXECUTING status and continuation results over stale DB FINISHED status in handleAgentDecisionStreamResult", workflowExecution.ExecutionId) + workflowExecution.Status = origStatus + workflowExecution.CompletedAt = origCompleted + workflowExecution.Results = origResults + } else { + log.Printf("[DEBUG][%s] handleAgentDecisionStreamResult: refreshed execution from DB (results: %d)", workflowExecution.ExecutionId, len(workflowExecution.Results)) + } + } + } + foundActionResultIndex := -1 for actionIndex, result := range workflowExecution.Results { if result.Action.ID == actionResult.Action.ID { @@ -17927,6 +17945,7 @@ func handleAgentDecisionStreamResult(workflowExecution WorkflowExecution, action } } + if foundActionResultIndex < 0 { // In test mode, Singul doesn't create sub-executions, so we need to handle this gracefully if os.Getenv("AGENT_TEST_MODE") == "true" { @@ -18014,6 +18033,26 @@ func handleAgentDecisionStreamResult(workflowExecution WorkflowExecution, action } } + if decisionIdResultIndex < 0 { + actionCacheId := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.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 { + for resultDecisionIndex, resultDecision := range cachedOutput.Decisions { + if resultDecision.RunDetails.Id == decisionId { + log.Printf("[INFO][%s] Found decision ID '%s' in action result cache during fallback!", workflowExecution.ExecutionId, decisionId) + mappedResult = cachedOutput + workflowExecution.Results[foundActionResultIndex].Result = string(cachedBytes) + decisionIdResultIndex = resultDecisionIndex + decisionIndex = resultDecision.I + break + } + } + } + } + } + if decisionIdResultIndex < 0 { log.Printf("[ERROR][%s] Decision ID %s was not found. Skipping.", workflowExecution.ExecutionId, decisionId) return &workflowExecution, false, errors.New(fmt.Sprintf("decisionIdResultIndex: Agent node ID for decision ID %s not found", decisionId)) @@ -18178,9 +18217,9 @@ func handleAgentDecisionStreamResult(workflowExecution WorkflowExecution, action if marshalledResult, marshalErr := json.Marshal(mappedResult); marshalErr == nil { workflowExecution.Results[foundActionResultIndex].Result = string(marshalledResult) - // push to the action result cache so GetWorkflowExecution inside HandleAiAgentExecutionStart picks up the fresh copy. + // push to the action result cache so GetWorkflowExecution inside HandleAiAgentExecutionStart picks up the fresh copy. actionCacheId := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) - go SetCache(ctx, actionCacheId, marshalledResult, 35) + go SetCache(ctx, actionCacheId, marshalledResult, 600) } else { log.Printf("[WARNING][%s] Failed to marshal updated mappedResult before HandleAiAgentExecutionStart: %s", workflowExecution.ExecutionId, marshalErr) } @@ -18190,8 +18229,14 @@ func handleAgentDecisionStreamResult(workflowExecution WorkflowExecution, action executionCacheKey := fmt.Sprintf("workflowexecution_%s", workflowExecution.ExecutionId) DeleteCache(ctx, executionCacheKey) if marshalledExec, marshalErr := json.Marshal(workflowExecution); marshalErr == nil { - SetCache(ctx, executionCacheKey, marshalledExec, 30) + SetCache(ctx, executionCacheKey, marshalledExec, 600) } + SetWorkflowExecution(ctx, workflowExecution, true) + + // Set the fixexec cache lock so fixExecution doesn't race and start a duplicate LLM call + + cacheId := fmt.Sprintf("agent-%s-%s-fixexec-finished-check", workflowExecution.ExecutionId, originalAction.ID) + _ = SetCache(ctx, cacheId, []byte("handled"), 60) callerName := "handleAgentDecisionStreamResult" returnAction, err := HandleAiAgentExecutionStart(workflowExecution, originalAction, true, callerName) @@ -18302,7 +18347,7 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut } if !skipAgentContinue { - callerName := "LLMResponse" + callerName := "ParsedExecutionResult" marshalledResult, err := json.Marshal(actionResult) if err != nil { log.Printf("[ERROR] AI Agent (10): Failed marshalling actionResult: %s", err) @@ -18395,7 +18440,7 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut actionResultBody, err := json.Marshal(actionResult) if err == nil { cacheId := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) - err = SetCache(ctx, cacheId, actionResultBody, 35) + err = SetCache(ctx, cacheId, actionResultBody, 600) if err != nil { log.Printf("[WARNING] Couldn't find in fix exec %s (2): %s", cacheId, err) continue @@ -18412,7 +18457,7 @@ func ParsedExecutionResult(ctx context.Context, workflowExecution WorkflowExecut actionResultBody, err := json.Marshal(actionResult) if err == nil { cacheId := fmt.Sprintf("%s_%s_result", workflowExecution.ExecutionId, actionResult.Action.ID) - err = SetCache(ctx, cacheId, actionResultBody, 35) + err = SetCache(ctx, cacheId, actionResultBody, 600) if err != nil { log.Printf("[ERROR][%s] Failed to update cache for %s", workflowExecution.ExecutionId, cacheId) } @@ -26687,8 +26732,9 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h // The only key we care about in this case if key == "continue" { // Overwrite everything - if workflowExecution.Status == "FINISHED" { + if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "SUCCESS" { workflowExecution.Status = "EXECUTING" + workflowExecution.CompletedAt = 0 } unmarshalledDecision.Status = "RUNNING" @@ -26734,6 +26780,7 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h // Both start counting from 0 if len(decision.Fields) <= fieldNumber { + log.Printf("[ERROR][%s] Decision '%s' field number '%d' exceeds maximum length %d", workflowExecution.ExecutionId, decisionId, fieldNumber, len(decision.Fields)) continue } @@ -26747,7 +26794,7 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h if fieldsChanged { decision.RunDetails.Status = "FINISHED" - decision.RunDetails.CompletedAt = time.Now().Unix() + decision.RunDetails.CompletedAt = time.Now().UnixMilli() unmarshalledDecision.Decisions[decisionIndex] = decision // Updates cache live @@ -26756,7 +26803,7 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h if err != nil { log.Printf("[ERROR][%s] Failed marshalling decision during agentic decision handling: %s", oldExecution.ExecutionId, err) } else { - SetCache(ctx, decisionId, marshalledDecision, 60) + SetCache(ctx, decisionId, marshalledDecision, 600) } } @@ -26782,7 +26829,7 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h if err != nil { log.Printf("[ERROR][%s] Failed marshalling decision during agentic decision handling: %s", oldExecution.ExecutionId, err) } else { - SetCache(ctx, decisionId, marshalledDecision, 60) + SetCache(ctx, decisionId, marshalledDecision, 600) } fieldsChanged = true @@ -26807,7 +26854,7 @@ func PrepareWorkflowExecution(ctx context.Context, workflow Workflow, request *h } actionCacheId := fmt.Sprintf("%s_%s_result", oldExecution.ExecutionId, result.Action.ID) - err = SetCache(ctx, actionCacheId, []byte(result.Result), 35) + err = SetCache(ctx, actionCacheId, []byte(result.Result), 600) if err != nil { log.Printf("[ERROR] Failed setting cache for action result %s: %s", actionCacheId, err) } diff --git a/structs.go b/structs.go index c9a8fc1f..d403e9a9 100755 --- a/structs.go +++ b/structs.go @@ -5782,5 +5782,6 @@ type WorkflowSetOpsResponse struct { type rawField struct { Name string `json:"name"` + Key string `json:"key,omitempty"` Value interface{} `json:"value"` }