diff --git a/ai.go b/ai.go index 80346cf0..e39bca21 100644 --- a/ai.go +++ b/ai.go @@ -7644,7 +7644,259 @@ 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 getTemplateContext(ctx context.Context, template string, execution WorkflowExecution) (string, string, error) { + switch template { + case "workflow-edit": + return buildWorkflowEditContext(ctx, execution) + default: + return "", "", nil + } +} + +func buildWorkflowEditContext(ctx context.Context, execution WorkflowExecution) (string, string, error) { + targetWorkflowId := execution.ExecutionArgument + + user := User{ + ActiveOrg: OrgMini{Id: execution.ExecutionOrg}, + } + appSummaries, err := getOrgAppSummaries(ctx, user) + if err != nil { + log.Printf("[WARNING] buildWorkflowEditContext: failed getting app summaries for org %s: %s", execution.ExecutionOrg, err) + } + appsJson, _ := json.Marshal(appSummaries) + + // What we tell the agent about its workflow_id + workflowIdLine := "CRITICAL: You do NOT have an existing workflow yet. Your FIRST action MUST be to call create_workflow to create one, then ALWAYS use the returned workflow_id for ALL subsequent operations." + if len(targetWorkflowId) > 0 { + workflowIdLine = fmt.Sprintf(`CRITICAL: Working Workflow ID: %s +You MUST use this exact workflow_id in every single agent_update operation. DO NOT try to guess or create a new one.`, targetWorkflowId) + } + + systemRule := `[SECONDARY]: You are an autonomous workflow-building agent. You build or edit workflows that will eventually automate the task. Only use the special actions built for you — agent_update is the primary one you will use. If the user requests an integration that is not in your current context, you can still fetch its actions by passing the requested app name to your get_workflow_app_actions tool. The system will automatically search for the app and return its actions. + +Only ask the user for information when it is genuinely unavailable through any allowed action. Use get_minimal_action to fetch the latest workflow state whenever you need to inspect or verify what is currently in the workflow — prefer this over relying on stale context. + +Payload Structure + +Send your incremental steps in the operations array. You can send one operation at a time, or multiple if you are building a connected sequence. +Key is "body" and Value is { +"workflow_id": "", +"operations": [ ...array of step-by-step operations... ] +} + +CRITICAL: The "operations" field MUST be a JSON array of objects. NEVER write a natural language plan, summary, or string inside the "operations" field. This overrides the generic "value": "literal_value" example shown elsewhere in your instructions — for agent_update, "value" is ALWAYS an object, and "operations" inside it is ALWAYS an array, never a string. + +Concrete example of a correct full decision using agent_update: +{ + "action": "agent_update", + "category": "singul", + "tool": "agent_update", + "fields": [ + { + "key": "body", + "value": { + "workflow_id": "", + "operations": [ + { "op": "edit_node", "id": "", "data": { "parameters": [ { "name": "method", "value": "GET" }, { "name": "url", "value": "https://example.com" } ] } } + ] + } + } + ] +} + +INCORRECT (do NOT do this): { "key": "body", "value": { "operations": "Update the workflow actions as follows: ...", "workflow_id": "" } } + +Supported Step-by-Step Operations + +A workflow has two types of nodes: Actions and Triggers. If you need a trigger, use either Webhook or Schedule. There is no rigid rule that triggers must always exist — only use a trigger if the workflow solution genuinely needs it. + +Format: Webhook Trigger +{ +"op": "add_node", +"node_type": "trigger", +"temp_id": "", +"data": { + "app_name": "Webhook", + "label": "", + "x": 100, + "y": 100 +} +} + +Format: Schedule Trigger +{ +"op": "add_node", +"node_type": "trigger", +"temp_id": "", +"data": { + "app_name": "Schedule", + "label": "", + "parameters": [ + { "name": "cron", "value": "*/15 * * * *" } + ], + "x": 100, + "y": 100 +} +} + +1. ADDING A NODE (Action) +{ +"op": "add_node", +"node_type": "action", +"temp_id": "a_temp_id_string", +"data": { + "app_id": "", + "action_name": "", + "label": "Authenticate User", + "parameters": [ + { "name": "param_1", "value": "some_value" } + ], + "x": 100, + "y": 100 +} +} + +Every action’s response is stored under its label. You can reference it using: +$label_name this itself gives you the parsed JSON output of the action +* $jira_action_1.id +* $python_2.message.email +* For triggers use "$exec" for example $exec.field + +Note: You can vertically position the node between existing ones by adding "insert_before": "" or "insert_after": "" alongside the data field. + +2. CONNECTING NODES (Adding a Branch) +Connect your nodes using their real IDs (if they already exist) or the temp_id you assigned when creating them in the same payload. You can include an array of conditions if you are adding multiple to the same branch. +{ +"op": "add_branch", +"temp_id": "branch_1", +"data": { + "source_id": "", + "destination_id": "", + "conditions": [ + { "source": "$some_var", "condition": "equals", "destination": "true" } + ] +} +} + +3. EDITING A NODE +Only provide the fields you actually want to change. +{ +"op": "edit_node", +"id": "", +"data": { + "label": "New Name", + "parameters": [ + { "name": "param_1", "value": "new_value" } + ] +} +} + +4. DELETING OR MOVING +{ "op": "delete_node", "node_type": "action", "id": "" } +{ "op": "delete_branch", "id": "" } +{ "op": "move_node", "id": "", "data": { "x": 250, "y": 300 } } + +5. MANAGING CONDITIONS +You can add, edit, or delete conditions on an EXISTING branch. +NOTE: To edit or delete, you must use a Get/Read action first to find the exact condition ID. + +Available operators: equals, larger than, less than, does not equal, startswith, endswith, is empty, contains, contains_any_of. + +Add conditions to an existing branch: +{ +"op": "add_condition", +"branch_id": "", +"data": { + "conditions": [ + { "source": "$var", "condition": "equals", "destination": "val" } + ] +} +} + +Edit existing conditions (only include fields you want to change): +{ +"op": "edit_condition", +"branch_id": "", +"data": { + "conditions": [ + { "id": "", "source": "$new_var", "condition": "contains", "destination": "new_val" } + ] +} +} + +Delete conditions: +{ +"op": "delete_condition", +"branch_id": "", +"data": { + "condition_ids": [""] +} +} + +6. SETTING THE START NODE +Defines the entry point of the workflow. You can use a real ID or a temp_id from the same payload. +{ +"op": "set_start_node", +"id": "" +} + +7. SAVING THE WORKFLOW +Your edits are stored as a real-time draft. When you are completely finished building or modifying the workflow, you MUST append this operation to your final payload to permanently save the workflow to the database. Call this only after all changes are complete — do not forget it. +{ +"op": "save_workflow" +} + +To use data from another node, reference its label. + +CRITICAL RULES FOR THE AGENT + +1. Handling Errors: If you make a mistake, the API will reject your request and tell you EXACTLY which operation failed and why (e.g., "Operation 1 failed: app_id is required"). Read the error carefully, fix the specific missing or incorrect field, and try again. +2. Special actions are built exclusively for you. agent_update is the core action that does most of the heavy lifting. +3. Use get_minimal_action to fetch the current workflow state only whenever you need it. +4. Try not to use parallel decision calling; always do sequential tool calls.` + + templateContext := fmt.Sprintf(`%s + + +%s +`, workflowIdLine, string(appsJson)) + + return systemRule, templateContext, nil +} + + +func getWorkflowEditPromptRemovals() []string { + return []string{ + ` - **Destructive Guard:** + - If action is DESTRUCTIVE (stop/delete/remove) -> Set "approval_required": true on the action/tool.`, + `// true IF the action seems risky or destructive and requires user approval. Otherwise false`, + `### DATA REDUCTION: +data_filter: +- "full": The default value of the data_filter is full. Use for all non-data-returning calls or when you need the entire response. +- "list": Use for ALL data calls. Request ONLY essential fields. If the schema is completely unknown, fallback to "full"`, + } +} + +func filterSystemPromptByTemplate(template string, systemMessage string) string { + var removals []string + + switch template { + case "workflow-edit": + removals = getWorkflowEditPromptRemovals() + } + + for _, removal := range removals { + if removal != "" { + systemMessage = strings.ReplaceAll(systemMessage, removal, "") + } + } + + return systemMessage +} + func HandleAiAgentExecutionStart(execution WorkflowExecution, startNode Action, createNextActions bool, callerName string, aiResponseWrapper ...[]byte) (Action, error) { + template := "" ctx := context.Background() aiStarttime := time.Now().UnixMilli() @@ -7837,6 +8089,10 @@ func HandleAiAgentExecutionStart(execution WorkflowExecution, startNode Action, imagesIncluded := []string{} imageDetail := openai.ImageURLDetailAuto // low, high, original, auto (let the model decide) for _, param := range startNode.Parameters { + if param.Name == "template" && len(template) == 0 { + template = param.Value + } + if param.Name == "input" { userMessage = param.Value @@ -8576,6 +8832,15 @@ data_filter: } ]`, enableQuestionsString) + systemMessage = filterSystemPromptByTemplate(template, systemMessage) + + // If a template is set, get secondary system rules + extra context. + templateSystemRule := "" + templateContext := "" + if len(template) > 0 { + templateSystemRule, templateContext, _ = getTemplateContext(ctx, template, execution) + } + agentReasoningEffort := "low" newReasoningEffort := os.Getenv("AI_AGENT_REASONING_EFFORT") if len(newReasoningEffort) > 0 { @@ -8614,7 +8879,7 @@ data_filter: } // Set model based on environment - aiModel := "gpt-5-mini" + aiModel := "gpt-5.4-2026-03-05" newAiModel := os.Getenv("AI_MODEL") if newAiModel == "" { newAiModel = os.Getenv("OPENAI_MODEL") @@ -8624,26 +8889,38 @@ data_filter: aiModel = newAiModel } - completionRequest := openai.ChatCompletionRequest{ - Model: aiModel, - Messages: []openai.ChatCompletionMessage{ - { - Role: openai.ChatMessageRoleSystem, - Content: systemMessage, - }, + primaryMessages := []openai.ChatCompletionMessage{ + { + Role: openai.ChatMessageRoleSystem, + Content: systemMessage, }, + } - // Move towards determinism - Temperature: 0, + // Inject template-specific rules as a secondary system message + if len(templateSystemRule) > 0 { + primaryMessages = append(primaryMessages, openai.ChatCompletionMessage{ + Role: openai.ChatMessageRoleSystem, + Content: templateSystemRule, + }) + } + + completionRequest := openai.ChatCompletionRequest{ + Model: aiModel, + Messages: primaryMessages, + Temperature: 0, ReasoningEffort: agentReasoningEffort, + } - // Reasoning control - // MaxCompletionTokens: 5000, - // ReasoningEffort: agentReasoningEffort, - // Store: true, + // Inject template-specific context as a completely separate user message + if len(templateContext) > 0 { + completionRequest.Messages = append(completionRequest.Messages, openai.ChatCompletionMessage{ + Role: openai.ChatMessageRoleUser, + Content: "TEMPLATE CONTEXT:\n" + templateContext, + }) } preparedContent := fmt.Sprintf("USER CONTEXT:\n%s\n", metadata) + if len(imagesIncluded) == 0 { completionRequest.Messages = append(completionRequest.Messages, openai.ChatCompletionMessage{ Role: openai.ChatMessageRoleUser, diff --git a/blobs.go b/blobs.go index 95242824..f481da96 100644 --- a/blobs.go +++ b/blobs.go @@ -22,8 +22,8 @@ import ( func IsShuffleApp(app WorkflowApp) bool { parsedAppname := strings.ReplaceAll(strings.ToLower(app.Name), " ", "_") - skipAuthAppnames := []string{"openai", "shuffle_datastore", "shuffle_workflows", "shuffle_detection", "shuffle_sensors", "shuffle_monitors", "shuffle_host_monitors", "shuffle_apps"} - skipAuthAppIds := []string{"5d19dd82517870c68d40cacad9b5ca91", "b82668d868f6dc7ac1dc14caa92c674b", "b598b078fd5c531699fca803c172ce72", "afda48b8d1f7dc7ac3caae87b2c072e9", "7f12d725c356677d28db042170444448", "48a954b9440b3913b8a2620e57b94a75", "7db43ccd25261967b095cfbd467a75cc"} + skipAuthAppnames := []string{"openai", "shuffle_datastore", "shuffle_workflows", "shuffle_detection", "shuffle_sensors", "shuffle_monitors", "shuffle_host_monitors", "shuffle_apps", "shuffle_workflows_builder"} + skipAuthAppIds := []string{"5d19dd82517870c68d40cacad9b5ca91", "b82668d868f6dc7ac1dc14caa92c674b", "b598b078fd5c531699fca803c172ce72", "afda48b8d1f7dc7ac3caae87b2c072e9", "7f12d725c356677d28db042170444448", "48a954b9440b3913b8a2620e57b94a75", "7db43ccd25261967b095cfbd467a75cc", "de4ef2287bd41b9d5563e39989643ee6"} isShuffleApp := false if project.Environment == "cloud" && len(app.ID) > 0 { diff --git a/cloudSync.go b/cloudSync.go index 9578099b..d4ddc8a9 100755 --- a/cloudSync.go +++ b/cloudSync.go @@ -2231,7 +2231,25 @@ func RunAgentDecisionSingulActionHandler(execution WorkflowExecution, decision A } if outputMapped.Success == false { - return originalBody, debugUrl, appname, []string{}, "", errors.New("Failed running agent decision (3). Success false for Singul action") + // The outer Singul wrapper said success=false, but the actual underlying + // HTTP call may have still worked (this happens with new/custom apps that + // Singul doesn't fully recognise at the schema level). + // By this point `body` already holds the extracted inner raw_response content + // (decoded above). Check if that inner result was actually successful before + // giving up — we only fail hard if the inner call also failed. + var innerResult struct { + Status int `json:"status"` + Success bool `json:"success"` + } + innerUnmarshalErr := json.Unmarshal(body, &innerResult) + + innerCallSucceeded := innerUnmarshalErr == nil && (innerResult.Success || innerResult.Status == 200) + if !innerCallSucceeded { + log.Printf("[ERROR][%s] AI Agent: Singul action failed (success=false) and inner raw_response also shows failure. Returning error.", execution.ExecutionId) + return originalBody, debugUrl, appname, []string{}, "", errors.New("Failed running agent decision (3). Success false for Singul action") + } + + log.Printf("[INFO][%s] AI Agent: Singul outer wrapper returned success=false but inner raw_response shows success (status=%d). Treating as success for custom/new app.", execution.ExecutionId, innerResult.Status) } /* diff --git a/shared.go b/shared.go index 4df41eb2..3b38052b 100644 --- a/shared.go +++ b/shared.go @@ -6842,12 +6842,36 @@ func SetNewWorkflow(resp http.ResponseWriter, request *http.Request) { } } - workflowjson, err := json.Marshal(workflow) - if err != nil { - log.Printf("Failed workflow json setting marshalling: %s", err) - resp.WriteHeader(http.StatusInternalServerError) - resp.Write([]byte(`{"success": false}`)) - return + type minimalWorkflowWithId struct { + *MinimalWorkflow + WorkflowId string `json:"workflow_id"` + } + + var workflowjson []byte + if request.URL.Query().Get("minimal") == "true" { + minimalWorkflow := buildMinimalWorkflow(&workflow) + if minimalWorkflow == nil { + log.Printf("[ERROR] Failed building minimal workflow %s", workflow.ID) + resp.WriteHeader(http.StatusInternalServerError) + resp.Write([]byte(`{"success": false}`)) + return + } + + workflowjson, err = json.Marshal(minimalWorkflowWithId{minimalWorkflow, workflow.ID}) + if err != nil { + log.Printf("Failed minimal workflow json setting marshalling: %s", err) + resp.WriteHeader(http.StatusInternalServerError) + resp.Write([]byte(`{"success": false}`)) + return + } + } else { + workflowjson, err = json.Marshal(workflow) + if err != nil { + log.Printf("Failed workflow json setting marshalling: %s", err) + resp.WriteHeader(http.StatusInternalServerError) + resp.Write([]byte(`{"success": false}`)) + return + } } err = SetWorkflow(ctx, workflow, workflow.ID) @@ -37788,6 +37812,79 @@ func getOrgAppSummaries(ctx context.Context, user User) ([]AppSummary, error) { return appSummaries, nil } +func buildAppActionResponses(matchedApps []WorkflowApp) []AppActionResponse { + responses := []AppActionResponse{} + + for _, app := range matchedApps { + if len(app.Actions) == 0 { + continue + } + + appDesc := app.Description + if len(appDesc) > 150 { + appDesc = appDesc[:150] + "..." + } + + appResp := AppActionResponse{ + AppName: app.Name, + AppDescription: appDesc, + AppID: app.ID, + Actions: []ActionSummary{}, + } + + for _, action := range app.Actions { + if len(action.Name) == 0 { + continue + } + + params := []ActionParameter{} + for _, param := range action.Parameters { + if len(param.Name) == 0 { + continue + } + + params = append(params, ActionParameter{ + Name: param.Name, + Required: param.Required, + Description: param.Description, + }) + } + + desc := action.Description + if len(desc) > 100 { + desc = desc[:100] + "..." + } + + appResp.Actions = append(appResp.Actions, ActionSummary{ + Name: action.Name, + Description: desc, + Parameters: params, + }) + } + + if len(appResp.Actions) > 0 { + responses = append(responses, appResp) + } + } + + return responses +} + +func getOrgAppActionSummaries(ctx context.Context, user User) ([]AppActionResponse, error) { + apps, err := GetPrioritizedApps(ctx, user) + if err != nil { + log.Printf("[WARNING] Failed getting apps for agent: %s", err) + return nil, err + } + + maxApps := 150 + if len(apps) > maxApps { + apps = apps[:maxApps] + } + + return buildAppActionResponses(apps), nil +} + func GetOrgAppsSummary(resp http.ResponseWriter, request *http.Request) { cors := HandleCors(resp, request) if cors { @@ -37891,12 +37988,35 @@ func GetWorkflowAppActions(resp http.ResponseWriter, request *http.Request) { // Search by IDs first (more direct) if len(actionReq.AppIDs) > 0 { for _, appID := range actionReq.AppIDs { + foundInOrg := false for _, app := range allApps { if app.ID == appID && len(app.Name) > 0 { + if debug { + log.Printf("[DEBUG] Found app by ID %s in org apps list", appID) + } matchedApps = append(matchedApps, app) + foundInOrg = true break } } + + // If not found in org's active apps, fetch it globally by ID! + if !foundInOrg { + if debug { + log.Printf("[DEBUG] App ID %s not in org, trying global GetApp", appID) + } + globalApp, err := GetApp(ctx, appID, user, false) + if err == nil && globalApp != nil { + if debug { + log.Printf("[DEBUG] Successfully found app %s globally", appID) + } + matchedApps = append(matchedApps, *globalApp) + } else { + if debug { + log.Printf("[DEBUG] Failed to fetch app globally by ID %s: %v", appID, err) + } + } + } } } @@ -37904,8 +38024,12 @@ func GetWorkflowAppActions(resp http.ResponseWriter, request *http.Request) { if len(actionReq.AppNames) > 0 { for _, appName := range actionReq.AppNames { lowerName := strings.ToLower(strings.TrimSpace(appName)) + foundInOrg := false for _, app := range allApps { if strings.ToLower(app.Name) == lowerName && len(app.ID) > 0 { + if debug { + log.Printf("[DEBUG] Found app by name '%s' (ID %s) in org apps list", appName, app.ID) + } // Check if already added via ID search alreadyAdded := false for _, matched := range matchedApps { @@ -37917,9 +38041,39 @@ func GetWorkflowAppActions(resp http.ResponseWriter, request *http.Request) { if !alreadyAdded { matchedApps = append(matchedApps, app) } + foundInOrg = true break } } + + // If not found in org, search Algolia + if !foundInOrg { + if debug { + log.Printf("[DEBUG] App Name '%s' not in org, searching Algolia", appName) + } + algoliaApp, err := HandleAlgoliaAppSearch(ctx, appName) + if err == nil && algoliaApp.ObjectID != "" { + if debug { + log.Printf("[DEBUG] Found Algolia match for '%s': ObjectID %s", appName, algoliaApp.ObjectID) + } + discoveredApp, err := GetApp(ctx, algoliaApp.ObjectID, user, false) + if err == nil && discoveredApp != nil { + alreadyAdded := false + for _, matched := range matchedApps { + if matched.ID == discoveredApp.ID { + alreadyAdded = true + break + } + } + if !alreadyAdded { + if debug { + log.Printf("[DEBUG] Appended Algolia app %s to matched apps", discoveredApp.ID) + } + matchedApps = append(matchedApps, *discoveredApp) + } + } + } + } } } @@ -37934,57 +38088,7 @@ func GetWorkflowAppActions(resp http.ResponseWriter, request *http.Request) { } // Build response with actions for each matched app - responses := []AppActionResponse{} - - for _, app := range matchedApps { - if len(app.Actions) == 0 { - if debug { - log.Printf("[DEBUG] Skipping app %s because len(Actions) is 0", app.Name) - } - continue - } - - appResp := AppActionResponse{ - AppName: app.Name, - AppID: app.ID, - Actions: []ActionSummary{}, - } - - // Extract minimal action info - for _, action := range app.Actions { - if len(action.Name) == 0 { - continue - } - - // Build parameter list - params := []ActionParameter{} - for _, param := range action.Parameters { - if len(param.Name) == 0 { - continue - } - - params = append(params, ActionParameter{ - Name: param.Name, - Required: param.Required, - }) - } - - desc := action.Description - if len(desc) > 100 { - desc = desc[:100] + "..." - } - - appResp.Actions = append(appResp.Actions, ActionSummary{ - Name: action.Name, - Description: desc, - Parameters: params, - }) - } - - if len(appResp.Actions) > 0 { - responses = append(responses, appResp) - } - } + responses := buildAppActionResponses(matchedApps) if len(responses) == 0 { if debug { @@ -38129,216 +38233,125 @@ func AgentWorkflowEditor(resp http.ResponseWriter, request *http.Request) { return } - //type agentContextRequest struct { - // Input string `json:"input"` - // WorkflowId string `json:"workflow_id"` - //} - var req MCPRequest err = json.Unmarshal(body, &req) + if err != nil || len(strings.TrimSpace(req.Params.Input.Text)) == 0 { - log.Printf("[WARNING] Bad body in AgentWorkflowEditor: %s", err) + log.Printf("[WARNING] Bad body in AgentWorkflowEditor. Error: %v, Body: %s", err, string(body)) resp.WriteHeader(400) resp.Write([]byte(`{"success": false, "reason": "input field is required"}`)) return } - // Get org apps summary - appSummaries, err := getOrgAppSummaries(ctx, user) + // Auth check: verify user has access to the target workflow before kicking off the agent + workflow, err := GetWorkflow(ctx, req.Params.Input.WorkflowId) if err != nil { - log.Printf("[WARNING] Failed getting apps in AgentWorkflowEditor for user %s: %s", user.Username, err) - resp.WriteHeader(500) - resp.Write([]byte(`{"success": false}`)) + log.Printf("[WARNING] Failed getting workflow %s in AgentWorkflowEditor: %s", req.Params.Input.WorkflowId, err) + resp.WriteHeader(404) + resp.Write([]byte(`{"success": false, "reason": "Workflow not found"}`)) return } - // Build system prompt with apps context - appsJson, _ := json.Marshal(appSummaries) - - systemPrompt := fmt.Sprintf(`You are an autonomous workflow-building agent. You build or edit workflows that will eventually automates the task. Only use the special actions build for you, like get_minimal_workflow, agent_update are the ones you have to use. - -The workflow you are currently working on has the ID: %s — always use this as the "workflow_id" in every payload you send. - -Only ask the user for information when it is genuinely unavailable through any allowed action. - -Payload Structure - -Send your incremental steps in the operations array. You can send one operation at a time, or multiple if you are building a connected sequence. -{ -"workflow_id": "%s", -"operations": [ ...array of step-by-step operations... ] // api just supports bulk operations but its optional, just mention 1 op at a time in the array -} - -Supported Step-by-Step Operations + if user.Id != workflow.Owner && workflow.OrgId != user.ActiveOrg.Id { + log.Printf("[WARNING] User %s (%s) unauthorized to edit workflow %s (owner: %s, org: %s)", user.Username, user.Id, req.Params.Input.WorkflowId, workflow.Owner, workflow.OrgId) + resp.WriteHeader(403) + resp.Write([]byte(`{"success": false, "reason": "Unauthorized"}`)) + return + } -A workflow has two types of nodes: Actions and Triggers. If you need a trigger, you use either Webhook or Schedule. Keep in mind that there is no rigid rule that triggers must exists everytime. Only use trigger if the workflow solution you are building really needs it. + // All context building (workflow state, app actions, rules) is handled inside + // buildWorkflowEditContext which is called by getTemplateContext inside HandleAiAgentExecutionStart + toolApps := "app:7db43ccd25261967b095cfbd467a75cc:shuffle_apps,app:de4ef2287bd41b9d5563e39989643ee6:shuffle_workflows_builder" -Format: Webhook Trigger + action := Action{ + ID: uuid.NewV4().String(), + Name: "agent", + AppName: "AI Agent", + AppID: "shuffle_agent", + AppVersion: "1.0.0", + Environment: "cloud", + Parameters: []WorkflowAppActionParameter{ + { + Name: "input", + Value: req.Params.Input.Text, + }, + { + Name: "action", + Value: toolApps, + }, + { + Name: "template", + Value: "workflow-edit", + }, + }, + } -{ -"op": "add_node", -"node_type": "trigger", -"temp_id": "", -"data": { -"app_name": "Webhook", -"label": "", -"x": 100, -"y": 100 -} -} + workflowId := uuid.NewV4().String() + action.SourceWorkflow = workflowId -Format: Schedule Trigger + exec := WorkflowExecution{ + Workflow: Workflow{ + ID: workflowId, + Actions: []Action{ + action, + }, + OrgId: user.ActiveOrg.Id, + Owner: user.Username, + UpdatedBy: user.Username, + Start: action.ID, + }, + Type: "AGENT", + Start: action.ID, + Status: "EXECUTING", + WorkflowId: workflowId, + ExecutionId: workflowId, + ExecutionOrg: user.ActiveOrg.Id, + StartedAt: int64(time.Now().Unix()), + Authorization: uuid.NewV4().String(), + // Store the target workflow_id here so buildWorkflowEditContext can fetch it + ExecutionArgument: req.Params.Input.WorkflowId, + } -{ -"op": "add_node", -"node_type": "trigger", -"temp_id": "", -"data": { -"app_name": "Schedule", -"label": "", -"parameters": [ -{ "name": "cron", "value": "*/15 * * * *" } -], -"x": 100, -"y": 100 -} -} + SetWorkflowExecution(ctx, exec, true) -1. ADDING A NODE (Action) -{ -"op": "add_node", -"node_type": "action", -"temp_id": "a_temp_id_string", -"data": { -"app_id": "", -"action_name": "", -"label": "Authenticate User", -"parameters": [ -{ "name": "param_1", "value": "some_value" } -], -"x": 100, -"y": 100 -} -} -Note: You can vertically position the node between existing ones by adding "insert_before": "" or "insert_after": "" alongside the data field. -2. CONNECTING NODES (Adding a Branch) -Connect your nodes using their real IDs (if they already exist) or the temp_id you assigned when creating them in the same payload. -{ -"op": "add_branch", -"data": { -"source_id": "", -"destination_id": "" -} -} -3. EDITING A NODE -Only provide the fields you actually want to change. -{ -"op": "edit_node", -"id": "", -"data": { -"label": "New Name", -"parameters": [ -{ "name": "param_1", "value": "new_value" } -] -} -} -4. DELETING OR MOVING -{ "op": "delete_node", "node_type": "action", "id": "" } -{ "op": "delete_branch", "id": "" } -{ "op": "move_node", "id": "", "data": { "x": 250, "y": 300 } } -5. SETTING THE START NODE -Defines the entry point of the workflow. You can use a real ID or a temp_id from the same payload. -{ -"op": "set_start_node", -"id": "" -} -6. SAVING THE WORKFLOW -Your edits are stored as a real-time draft. When you are completely finished building or modifying the workflow, you MUST append this operation to your final payload to permanently save the workflow to the database. Make sure you do this after fully finishing all the changes you want to do to the workflow, but don't forget to call it. -{ -"op": "save_workflow" -} + log.Printf("[INFO] AgentWorkflowEditor: calling HandleAiAgentExecutionStart for user %s (%s), target_workflow_id=%s, execution_id=%s", user.Username, user.Id, req.Params.Input.WorkflowId, exec.ExecutionId) -To use data from another node, reference its label. - -CRITICAL RULES FOR THE AGENT - -1. Handling Errors: If you make a mistake, the API will reject your request and tell you EXACTLY which operation failed and why (e.g., Operation 1 failed: app_id is required). Read this error carefully, fix the specific missing or incorrect parameter, and try again. -2. Special actions are built exculsively for you. get_minimal_workflow is enough to the info about how the workflow is structured and generally there is no need of other meta data info of the worklfow as often times its not that useful. agent_update is another beautiful action that do a lot of heavy lifting for you. -3. Try not to use parallel decision calling as much as possible; always do sequential tool calls. - - -%s - - - -%s -`, req.Params.Input.WorkflowId, req.Params.Input.WorkflowId, string(appsJson), req.Params.Input.Text) - - // Build MCPRequest with the full system prompt as input - mcpReq := MCPRequest{} - mcpReq.Jsonrpc = "2.0" - mcpReq.Method = "tools/call" - mcpReq.Params.ToolName = "app:9f05339c05f9aaca4eeb35e6f13e41e6:shuffles_app_management,app:b598b078fd5c531699fca803c172ce72:shuffle_workflows" - mcpReq.Params.Input.Text = systemPrompt - - mcpBody, err := json.Marshal(mcpReq) - if err != nil { - log.Printf("[ERROR] Failed marshalling MCPRequest in AgentWorkflowEditor: %s", err) + returnAction, err := HandleAiAgentExecutionStart(exec, action, false, "AgentWorkflowEditor") + if err != nil { + log.Printf("[ERROR] HandleAiAgentExecutionStart failed in AgentWorkflowEditor: %s", err) resp.WriteHeader(500) - resp.Write([]byte(`{"success": false}`)) + resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "%s"}`, err.Error()))) return } - // Call /api/v1/agent directly - backendUrl := os.Getenv("BASE_URL") - if len(backendUrl) == 0 { - if len(os.Getenv("SHUFFLE_CLOUDRUN_URL")) > 0 { - backendUrl = os.Getenv("SHUFFLE_CLOUDRUN_URL") - } else { - port := os.Getenv("PORT") - if len(port) == 0 { - port = "5001" - } - backendUrl = fmt.Sprintf("http://localhost:%s", port) - } - } - - // FIXME: Change this to a function. This is stupid // Fred - agentReq, err := http.NewRequest("POST", fmt.Sprintf("%s/api/v1/agent", backendUrl), strings.NewReader(string(mcpBody))) + // Fetch the updated execution to return + newExec, err := GetWorkflowExecution(ctx, exec.ExecutionId) if err != nil { - log.Printf("[ERROR] Failed creating agent request in AgentWorkflowEditor: %s", err) + log.Printf("[ERROR] Failed to get workflow execution after agent start in AgentWorkflowEditor: %s", err) resp.WriteHeader(500) resp.Write([]byte(`{"success": false}`)) return } - agentReq.Header = request.Header.Clone() - agentReq.Header.Set("Content-Type", "application/json") - agentReq.Header.Set("X-Internal-Caller", "AgentWorkflowEditor") + _ = returnAction - log.Printf("[INFO] AgentWorkflowEditor: calling /api/v1/agent for user %s (%s), workflow_id=%s, apps=%d", user.Username, user.Id, req.Params.Input.WorkflowId, len(appSummaries)) - - client := &http.Client{} - agentResp, err := client.Do(agentReq) - if err != nil { - log.Printf("[ERROR] Failed calling /api/v1/agent in AgentWorkflowEditor: %s", err) - resp.WriteHeader(500) - resp.Write([]byte(`{"success": false}`)) - return + respObj := agentResponse{ + Success: true, + ExecutionId: newExec.ExecutionId, + Authorization: newExec.Authorization, } - defer agentResp.Body.Close() - agentRespBody, err := ioutil.ReadAll(agentResp.Body) + responseData, err := json.Marshal(respObj) if err != nil { - log.Printf("[ERROR] Failed reading agent response in AgentWorkflowEditor: %s", err) + log.Printf("[ERROR] Failed marshalling execution response in AgentWorkflowEditor: %s", err) resp.WriteHeader(500) resp.Write([]byte(`{"success": false}`)) return } resp.Header().Set("Content-Type", "application/json") - resp.WriteHeader(agentResp.StatusCode) - resp.Write(agentRespBody) + resp.WriteHeader(200) + resp.Write(responseData) } func generateNodeID() string { @@ -38346,20 +38359,21 @@ func generateNodeID() string { } func createCondition(sourceVal, conditionVal, destVal string) Condition { + sharedID := generateNodeID() return Condition{ Source: WorkflowAppActionParameter{ - ID: generateNodeID(), + ID: sharedID, Name: "source", Variant: "STATIC_VALUE", Value: sourceVal, }, Condition: WorkflowAppActionParameter{ - ID: generateNodeID(), + ID: sharedID, Name: "condition", Value: conditionVal, }, Destination: WorkflowAppActionParameter{ - ID: generateNodeID(), + ID: sharedID, Name: "destination", Variant: "STATIC_VALUE", Value: destVal, @@ -38378,8 +38392,55 @@ func findAppByID(ctx context.Context, appID string, user User) (*WorkflowApp, er return app, err } + // 1. Search Org's Prioritized Apps first + prioritizedApps, err := GetPrioritizedApps(ctx, user) + if err == nil { + for _, pApp := range prioritizedApps { + if pApp.ID == appID { + return &pApp, nil + } + } + } + + // 2. If not found in org apps, try getting it globally app, err := GetApp(ctx, appID, user, false) - return app, err + if err != nil { + return nil, err + } + + // 3. If we got it globally but it wasn't in prioritized apps, auto-activate it in parallel + go func(activateAppID string, apiKey string) { + baseURL := os.Getenv("BASE_URL") + if len(baseURL) == 0 { + if len(os.Getenv("SHUFFLE_CLOUDRUN_URL")) > 0 { + baseURL = os.Getenv("SHUFFLE_CLOUDRUN_URL") + } else { + port := os.Getenv("PORT") + if len(port) == 0 { + port = "5001" + } + baseURL = fmt.Sprintf("http://localhost:%s", port) + } + } + + activateURL := fmt.Sprintf("%s/api/v1/apps/%s/activate", baseURL, activateAppID) + req, err := http.NewRequest("GET", activateURL, nil) + if err != nil { + log.Printf("[WARNING] Failed building auto-activate request for app %s: %s", activateAppID, err) + return + } + + req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", apiKey)) + httpResp, err := http.DefaultClient.Do(req) + if err != nil { + log.Printf("[WARNING] Failed auto-activating app %s: %s", activateAppID, err) + return + } + httpResp.Body.Close() + //log.Printf("[INFO] Auto-activated app %s in the background", activateAppID) + }(appID, user.ApiKey) + + return app, nil } func enrichActionFromApp(ctx context.Context, minAct *MinimalAction, realApp *WorkflowApp, environment string) (Action, error) { @@ -38531,7 +38592,7 @@ func enrichTriggerFromApp(minTrig *MinimalTrigger, environment string) (Trigger, } } -func HandleAgentWorkflowSave(resp http.ResponseWriter, request *http.Request) { +func HandleAgentWorkflowOperations(resp http.ResponseWriter, request *http.Request) { cors := HandleCors(resp, request) if cors { return @@ -38540,7 +38601,7 @@ func HandleAgentWorkflowSave(resp http.ResponseWriter, request *http.Request) { ctx := GetContext(request) user, userErr := HandleApiAuthentication(resp, request) if userErr != nil { - log.Printf("[WARNING] Api authentication failed in HandleAgentWorkflowSave: %s", userErr) + log.Printf("[WARNING] Api authentication failed in HandleAgentWorkflowOperations: %s", userErr) resp.WriteHeader(401) resp.Write([]byte(`{"success": false, "reason": "Authentication failed"}`)) return @@ -38553,17 +38614,6 @@ func HandleAgentWorkflowSave(resp http.ResponseWriter, request *http.Request) { return } - // Extract workflow ID from URL - location := strings.Split(request.URL.String(), "/") - var urlWorkflowID string - if len(location) > 4 && location[1] == "api" { - urlWorkflowID = location[4] - if strings.Contains(urlWorkflowID, "?") { - urlWorkflowID = strings.Split(urlWorkflowID, "?")[0] - } - } - - // Parse request first so we can fallback to body's WorkflowID body, err := ioutil.ReadAll(request.Body) if err != nil { log.Printf("[WARNING] Failed reading request body: %s", err) @@ -38582,17 +38632,12 @@ func HandleAgentWorkflowSave(resp http.ResponseWriter, request *http.Request) { return } - workflowID := urlWorkflowID + workflowID := setOpsReq.WorkflowID if len(workflowID) != 36 { - // Fallback to body's WorkflowID if URL ID is invalid (e.g., %7Bkey%7D from MCP) - if len(setOpsReq.WorkflowID) == 36 { - workflowID = setOpsReq.WorkflowID - } else { - log.Printf("[WARNING] Invalid workflow ID: %s", urlWorkflowID) - resp.WriteHeader(400) - resp.Write([]byte(`{"success": false, "reason": "Invalid workflow ID"}`)) - return - } + log.Printf("[WARNING] Invalid or missing workflow ID in body: %s", workflowID) + resp.WriteHeader(400) + resp.Write([]byte(`{"success": false, "reason": "Invalid workflow ID"}`)) + return } // Validate request @@ -38648,6 +38693,10 @@ func HandleAgentWorkflowSave(resp http.ResponseWriter, request *http.Request) { tempIDMap := make(map[string]string) // Maps temp_id → real_id shouldSaveDB := false + // Track triggers to start/stop AFTER all operations are applied. + var addedTriggers []Trigger + var deletedTriggers []Trigger + // Collect stream ops as we apply each operation. var streamOps []StreamWorkflowOperation @@ -38662,10 +38711,23 @@ func HandleAgentWorkflowSave(resp http.ResponseWriter, request *http.Request) { // For delete_node: capture which branch IDs are connected to this node BEFORE apply, because opDeleteNode silently prunes them from wf.Branches. We'll stream an edge:remove for each one auto-cleaned. var prunedBranchIDs []string - if operation.Op == "delete_node" { - for _, br := range workflow.Branches { - if br.SourceID == operation.ID || br.DestinationID == operation.ID { - prunedBranchIDs = append(prunedBranchIDs, br.ID) + if operation.Op == "delete_node" || operation.Op == "remove_node" { + for _, branch := range workflow.Branches { + if branch.SourceID == operation.ID || branch.DestinationID == operation.ID { + prunedBranchIDs = append(prunedBranchIDs, branch.ID) + } + } + } + + // Track triggers being added/deleted so we can start/stop them AFTER + // the full workflow is constructed (start node + branches all resolved). + var triggerToStop *Trigger + if (operation.Op == "delete_node" || operation.Op == "remove_node") && operation.NodeType == "trigger" { + for _, existingTrigger := range workflow.Triggers { + if existingTrigger.ID == operation.ID { + copy := existingTrigger + triggerToStop = © + break } } } @@ -38679,6 +38741,17 @@ func HandleAgentWorkflowSave(resp http.ResponseWriter, request *http.Request) { return } streamOps = append(streamOps, collectStreamOps(workflow, &operation, tempIDMap, branchCountBefore, prunedBranchIDs)...) + + // After apply: track triggers that were just added + if operation.Op == "add_node" && operation.NodeType == "trigger" { + // The newly added trigger is always the last in the Triggers slice + if len(workflow.Triggers) > 0 { + addedTriggers = append(addedTriggers, workflow.Triggers[len(workflow.Triggers)-1]) + } + } + if triggerToStop != nil { + deletedTriggers = append(deletedTriggers, *triggerToStop) + } } // Save to cache (volatile, 30 min TTL) @@ -38696,6 +38769,83 @@ func HandleAgentWorkflowSave(resp http.ResponseWriter, request *http.Request) { // Don't fail the request, cache is best-effort } + // Fire trigger start/stop goroutines NOW — workflow is fully constructed + // with all nodes, branches and start node resolved. + if len(addedTriggers) > 0 || len(deletedTriggers) > 0 { + go func(triggersToStart []Trigger, triggersToStop []Trigger, finalWorkflow Workflow, currentUser User) { + for _, trigger := range triggersToStart { + // Resolve start node: prefer a branch from this trigger, else use workflow start. + startNode := finalWorkflow.Start + for _, branch := range finalWorkflow.Branches { + if branch.SourceID == trigger.ID { + startNode = branch.DestinationID + break + } + } + if len(startNode) == 0 { + log.Printf("[INFO] Auto-start skipped for trigger %s: no start node in fully-constructed workflow", trigger.ID) + continue + } + + switch strings.ToUpper(trigger.TriggerType) { + case "WEBHOOK": + auth := "" + customResponse := "" + version := "v1" + versionTimeout := 15 + for _, param := range trigger.Parameters { + switch param.Name { + case "auth_headers": + auth = param.Value + case "custom_response_body": + customResponse = param.Value + case "await_response": + version = param.Value + case "version_timeout": + if parsedTimeout, convErr := strconv.Atoi(param.Value); convErr == nil { + versionTimeout = parsedTimeout + } + } + } + if startErr := startWebhookTrigger(context.Background(), finalWorkflow.ID, trigger.ID, trigger.Label, startNode, trigger.Environment, auth, customResponse, version, versionTimeout, currentUser, currentUser.ActiveOrg.Id); startErr != nil { + log.Printf("[WARNING] Auto-start webhook trigger %s failed: %s", trigger.ID, startErr) + } else { + log.Printf("[INFO] Auto-started webhook trigger %s for workflow %s", trigger.ID, finalWorkflow.ID) + } + case "SCHEDULE": + if startErr := startSchedule(trigger, currentUser.ApiKey, finalWorkflow); startErr != nil { + log.Printf("[WARNING] Auto-start schedule trigger %s failed: %s", trigger.ID, startErr) + } else { + log.Printf("[INFO] Auto-started schedule trigger %s for workflow %s", trigger.ID, finalWorkflow.ID) + } + } + } + + for _, trigger := range triggersToStop { + switch strings.ToUpper(trigger.TriggerType) { + case "WEBHOOK": + hook, err := GetHook(context.Background(), trigger.ID) + if err != nil { + log.Printf("[WARNING] Auto-stop webhook: could not find hook %s: %s", trigger.ID, err) + continue + } + hook.Status = "stopped" + hook.Running = false + if setErr := SetHook(context.Background(), *hook); setErr != nil { + log.Printf("[WARNING] Auto-stop webhook: failed updating hook %s status: %s", trigger.ID, setErr) + } + log.Printf("[INFO] Auto-stopped webhook trigger %s for workflow %s", trigger.ID, finalWorkflow.ID) + case "SCHEDULE": + if stopErr := deleteScheduleGeneral(context.Background(), trigger.ID); stopErr != nil { + log.Printf("[WARNING] Auto-stop schedule trigger %s failed: %s", trigger.ID, stopErr) + } else { + log.Printf("[INFO] Auto-stopped schedule trigger %s for workflow %s", trigger.ID, finalWorkflow.ID) + } + } + } + }(addedTriggers, deletedTriggers, *workflow, user) + } + if shouldSaveDB { env := workflow.ExecutionEnvironment if len(env) == 0 { @@ -38742,6 +38892,20 @@ func HandleAgentWorkflowSave(resp http.ResponseWriter, request *http.Request) { } } + // After hydration, update any stream operations with the final state of the actions, + // so the frontend receives the auto-assigned AuthenticationId in real-time. + for s := range streamOps { + if streamOps[s].Item == "node" && (streamOps[s].Type == "add" || streamOps[s].Type == "configure") { + for _, action := range workflow.Actions { + if action.ID == streamOps[s].ID { + dataBytes, _ := json.Marshal(action) + streamOps[s].Data = dataBytes + break + } + } + } + } + err = SetWorkflow(ctx, *workflow, workflow.ID) if err != nil { log.Printf("[ERROR] Failed saving workflow to DB: %s", err) @@ -38953,6 +39117,23 @@ func collectStreamOps(wf *Workflow, op *WorkflowOperation, tempIDMap map[string] ID: realID, }} + case "add_condition", "edit_condition", "delete_condition": + // The branch already has its updated conditions applied. + // Stream edge:configure with the full branch so the frontend re-renders it. + branchID := op.BranchID + for _, branch := range wf.Branches { + if branch.ID == branchID { + dataBytes, _ := json.Marshal(branch) + return []StreamWorkflowOperation{{ + Item: "edge", + Type: "configure", + ID: branch.ID, + Data: dataBytes, + }} + } + } + return nil + default: return nil } @@ -39005,7 +39186,7 @@ func applyWorkflowOperationWithMapping(ctx context.Context, user User, wf *Workf return opEditNode(wf, op) case "move_node": return opMoveNode(wf, op) - case "delete_node": + case "delete_node", "remove_node": return opDeleteNode(wf, op) // ====== BRANCH OPERATIONS ====== @@ -39013,16 +39194,24 @@ func applyWorkflowOperationWithMapping(ctx context.Context, user User, wf *Workf return opAddBranchWithMapping(wf, op, tempIDMap) case "edit_branch": return opEditBranch(wf, op) - case "delete_branch": + case "delete_branch", "remove_branch": return opDeleteBranch(wf, op) // ====== CONDITION OPERATIONS ====== - case "add_condition": - return opAddCondition(wf, op) - case "edit_condition": - return opEditCondition(wf, op) - case "delete_condition": - return opDeleteCondition(wf, op) + case "add_condition", "edit_condition", "delete_condition": + if realBranchID, exists := tempIDMap[op.BranchID]; exists { + op.BranchID = realBranchID + } + + switch op.Op { + case "add_condition": + return opAddCondition(wf, op) + case "edit_condition": + return opEditCondition(wf, op) + case "delete_condition": + return opDeleteCondition(wf, op) + } + return nil // ====== WORKFLOW OPERATIONS ====== case "set_start_node": @@ -39056,6 +39245,50 @@ func findNodePosition(wf *Workflow, nodeID string) (string, int, error) { func opAddNodeWithMapping(ctx context.Context, user User, wf *Workflow, op *WorkflowOperation, tempIDMap map[string]string) error { + // IDEMPOTENCY: if this temp_id was already resolved in a prior agent loop, + // the node already exists in the workflow (loaded from DB). Skip the add and + // re-populate the tempIDMap entry so downstream add_branch ops still work. + if len(op.TempID) > 0 { + if _, alreadyMapped := tempIDMap[op.TempID]; alreadyMapped { + return nil + } + } + + // Always check by label for idempotency, because labels must be unique in a workflow. + // This prevents duplicates if the workflow was loaded from DB after a prior save, + // regardless of whether a temp_id was provided or not. + var minLabel string + if op.NodeType == "action" { + var minAct MinimalAction + if jsonErr := json.Unmarshal(op.Data, &minAct); jsonErr == nil { + minLabel = minAct.Label + } + } else if op.NodeType == "trigger" { + var minTrig MinimalTrigger + if jsonErr := json.Unmarshal(op.Data, &minTrig); jsonErr == nil { + minLabel = minTrig.Label + } + } + + if len(minLabel) > 0 { + for _, existing := range wf.Actions { + if strings.EqualFold(existing.Label, minLabel) { + if len(op.TempID) > 0 { + tempIDMap[op.TempID] = existing.ID + } + return nil + } + } + for _, existing := range wf.Triggers { + if strings.EqualFold(existing.Label, minLabel) { + if len(op.TempID) > 0 { + tempIDMap[op.TempID] = existing.ID + } + return nil + } + } + } + err := opAddNode(ctx, user, wf, op) if err != nil { return err @@ -39227,6 +39460,7 @@ func opAddNode(ctx context.Context, user User, wf *Workflow, op *WorkflowOperati } else { wf.Triggers = append(wf.Triggers, newTrigger) } + return nil default: @@ -39340,7 +39574,6 @@ func opDeleteNode(wf *Workflow, op *WorkflowOperation) error { if debug { log.Printf("[DEBUG] delete_node(action): action %s not found, already removed - skipping", op.ID) } - return nil } @@ -39349,9 +39582,9 @@ func opDeleteNode(wf *Workflow, op *WorkflowOperation) error { // Remove branches connected to this node var newBranches []Branch - for _, br := range wf.Branches { - if br.SourceID != op.ID && br.DestinationID != op.ID { - newBranches = append(newBranches, br) + for _, branch := range wf.Branches { + if branch.SourceID != op.ID && branch.DestinationID != op.ID { + newBranches = append(newBranches, branch) } } wf.Branches = newBranches @@ -39359,7 +39592,7 @@ func opDeleteNode(wf *Workflow, op *WorkflowOperation) error { case "trigger": idx := findTriggerIndexByID(wf, op.ID) if idx == -1 { - // Already gone idempotent no-op + // Already gone, idempotent no-op if debug { log.Printf("[DEBUG] delete_node(trigger): trigger %s not found, already removed - skipping", op.ID) } @@ -39370,9 +39603,9 @@ func opDeleteNode(wf *Workflow, op *WorkflowOperation) error { // Remove branches connected to this trigger (both source and destination) var newBranches []Branch - for _, br := range wf.Branches { - if br.SourceID != op.ID && br.DestinationID != op.ID { - newBranches = append(newBranches, br) + for _, branch := range wf.Branches { + if branch.SourceID != op.ID && branch.DestinationID != op.ID { + newBranches = append(newBranches, branch) } } wf.Branches = newBranches @@ -39408,6 +39641,21 @@ func opAddBranchWithMapping(wf *Workflow, op *WorkflowOperation, tempIDMap map[s resolvedData, _ := json.Marshal(branchData) op.Data = resolvedData + // IDEMPOTENCY: if a branch between this source to destination already exists + // (e.g. a second agent loop re-issued the same add_branch), skip the add + // and re-map the temp_id so downstream ops still resolve correctly. + for _, existing := range wf.Branches { + if existing.SourceID == branchData.SourceID && existing.DestinationID == branchData.DestinationID { + if len(op.TempID) > 0 { + tempIDMap[op.TempID] = existing.ID + } + if len(op.ID) > 0 { + tempIDMap[op.ID] = existing.ID + } + return nil + } + } + err := opAddBranch(wf, op) if err != nil { return err @@ -39432,6 +39680,11 @@ func opAddBranch(wf *Workflow, op *WorkflowOperation) error { SourceID string `json:"source_id"` DestinationID string `json:"destination_id"` Label string `json:"label"` + Conditions []struct { + Source string `json:"source"` + Condition string `json:"condition"` + Destination string `json:"destination"` + } `json:"conditions"` } if err := json.Unmarshal(op.Data, &branchData); err != nil { @@ -39457,6 +39710,10 @@ func opAddBranch(wf *Workflow, op *WorkflowOperation) error { Conditions: []Condition{}, } + for _, c := range branchData.Conditions { + newBranch.Conditions = append(newBranch.Conditions, createCondition(c.Source, c.Condition, c.Destination)) + } + // Detect circular references before adding branch if hasCircularBranch(wf, newBranch) { return fmt.Errorf("circular branch detected: would create loop from %s → %s", branchData.SourceID, branchData.DestinationID) @@ -39506,9 +39763,11 @@ func opDeleteBranch(wf *Workflow, op *WorkflowOperation) error { func opAddCondition(wf *Workflow, op *WorkflowOperation) error { var condData struct { - Source string `json:"source"` - Condition string `json:"condition"` - Destination string `json:"destination"` + Conditions []struct { + Source string `json:"source"` + Condition string `json:"condition"` + Destination string `json:"destination"` + } `json:"conditions"` } if err := json.Unmarshal(op.Data, &condData); err != nil { @@ -39520,8 +39779,10 @@ func opAddCondition(wf *Workflow, op *WorkflowOperation) error { return fmt.Errorf("branch %s not found", op.BranchID) } - newCond := createCondition(condData.Source, condData.Condition, condData.Destination) - wf.Branches[branchIdx].Conditions = append(wf.Branches[branchIdx].Conditions, newCond) + for _, c := range condData.Conditions { + newCond := createCondition(c.Source, c.Condition, c.Destination) + wf.Branches[branchIdx].Conditions = append(wf.Branches[branchIdx].Conditions, newCond) + } return nil } @@ -39532,25 +39793,35 @@ func opEditCondition(wf *Workflow, op *WorkflowOperation) error { return fmt.Errorf("branch %s not found", op.BranchID) } - if op.ConditionIndex < 0 || op.ConditionIndex >= len(wf.Branches[branchIdx].Conditions) { - return fmt.Errorf("condition index %d out of range", op.ConditionIndex) - } - var condData struct { - Source string `json:"source"` - Condition string `json:"condition"` - Destination string `json:"destination"` + Conditions []struct { + ID string `json:"id"` + Source *string `json:"source"` + Condition *string `json:"condition"` + Destination *string `json:"destination"` + } `json:"conditions"` } if err := json.Unmarshal(op.Data, &condData); err != nil { return fmt.Errorf("invalid condition update data: %w", err) } - wf.Branches[branchIdx].Conditions[op.ConditionIndex] = createCondition( - condData.Source, - condData.Condition, - condData.Destination, - ) + for _, update := range condData.Conditions { + for cIdx := range wf.Branches[branchIdx].Conditions { + if wf.Branches[branchIdx].Conditions[cIdx].Condition.ID == update.ID { + if update.Source != nil { + wf.Branches[branchIdx].Conditions[cIdx].Source.Value = *update.Source + } + if update.Condition != nil { + wf.Branches[branchIdx].Conditions[cIdx].Condition.Value = *update.Condition + } + if update.Destination != nil { + wf.Branches[branchIdx].Conditions[cIdx].Destination.Value = *update.Destination + } + break + } + } + } return nil } @@ -39561,14 +39832,28 @@ func opDeleteCondition(wf *Workflow, op *WorkflowOperation) error { return fmt.Errorf("branch %s not found", op.BranchID) } - if op.ConditionIndex < 0 || op.ConditionIndex >= len(wf.Branches[branchIdx].Conditions) { - return fmt.Errorf("condition index %d out of range", op.ConditionIndex) + var deleteData struct { + ConditionIDs []string `json:"condition_ids"` } - wf.Branches[branchIdx].Conditions = append( - wf.Branches[branchIdx].Conditions[:op.ConditionIndex], - wf.Branches[branchIdx].Conditions[op.ConditionIndex+1:]..., - ) + if err := json.Unmarshal(op.Data, &deleteData); err != nil { + return fmt.Errorf("invalid delete condition data: %w", err) + } + + var newConds []Condition + for _, c := range wf.Branches[branchIdx].Conditions { + deleted := false + for _, delID := range deleteData.ConditionIDs { + if c.Condition.ID == delID { + deleted = true + break + } + } + if !deleted { + newConds = append(newConds, c) + } + } + wf.Branches[branchIdx].Conditions = newConds return nil } diff --git a/structs.go b/structs.go index c9a8fc1f..53f2ae0a 100755 --- a/structs.go +++ b/structs.go @@ -4902,6 +4902,31 @@ type MinimalParameter struct { Value string `json:"value"` } +func (m *MinimalParameter) UnmarshalJSON(data []byte) error { + type Alias MinimalParameter + var aux struct { + Value json.RawMessage `json:"value"` + *Alias + } + aux.Alias = (*Alias)(m) + if err := json.Unmarshal(data, &aux); err != nil { + return err + } + + if len(aux.Value) > 0 { + if aux.Value[0] == '"' { + var str string + if err := json.Unmarshal(aux.Value, &str); err != nil { + return err + } + m.Value = str + } else { + m.Value = string(aux.Value) + } + } + return nil +} + // MinimalAction - action with position and basic info type MinimalAction struct { AppName string `json:"app_name"` @@ -5743,9 +5768,10 @@ type ActionSummary struct { // AppActionResponse - actions grouped by app type AppActionResponse struct { - AppName string `json:"app_name"` - AppID string `json:"app_id"` - Actions []ActionSummary `json:"actions"` + AppName string `json:"app_name"` + AppDescription string `json:"app_description"` + AppID string `json:"app_id"` + Actions []ActionSummary `json:"actions"` } // WorkflowOperation represents a single modification operation @@ -5781,6 +5807,13 @@ type WorkflowSetOpsResponse struct { } type rawField struct { - Name string `json:"name"` + Name string `json:"name,omitempty"` + Key string `json:"key,omitempty"` Value interface{} `json:"value"` } + +type agentResponse struct { + Success bool `json:"success"` + ExecutionId string `json:"execution_id"` + Authorization string `json:"authorization"` +} \ No newline at end of file