AI-Assisted Workflow Failure Triage & Root Cause Advisor

id: ai-assisted-workflow-failure-triage
namespace: company.system

description: |
  Advisory AI-assisted workflow failure triage: collects execution failure traces, redacts sensitive credentials,
  classifies root causes via LangChain, and sends actionable diagnostic recommendations to on-call Slack channels.

triggers:
  - id: on_failure
    type: io.kestra.plugin.core.trigger.Flow
    description: Fires automatically whenever a flow execution in company.team fails.
    dependsOn:
      - namespace: company.team
        states: [ FAILED ]
    inputs:
      failed_execution_id: "{{ trigger.executionId }}"
      failed_flow_id: "{{ trigger.flowId }}"
      failed_namespace: "{{ trigger.namespace }}"

  - id: inbound_failure_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Lets an external monitoring tool or manual run trigger failure
      triage directly.
    key: "{{ secret('FAILURE_TRIAGE_WEBHOOK_KEY') }}"

inputs:
  - id: failed_namespace
    type: STRING
    defaults: "company.team"
    description: Namespace of the failed workflow.

  - id: failed_flow_id
    type: STRING
    defaults: "customer-billing-sync"
    description: Identifier of the failed workflow.

  - id: failed_execution_id
    type: STRING
    defaults: "exec-928192a"
    description: Unique execution ID of the failed Kestra flow.

  - id: error_stack_trace
    type: STRING
    defaults: "io.kestra.core.exceptions.IllegalVariableEvaluationException:
      Variable '{{ outputs.fetch_stripe_invoices.vars.secret_token }}' evaluated
      to null or missing key at task 'sync_to_warehouse'. Caused by:
      ConnectionTimeoutException connecting to api.stripe.com:443 after 30000ms.
      Authorization: Bearer sk_live_9281729481029481928."
    description: Raw error message or exception stack trace from the failed execution.

  - id: diagnosis_model
    type: SELECT
    values: [ gemini-2.5-flash, gpt-4o-mini ]
    defaults: gemini-2.5-flash
    description: Foundational model utilized to synthesize the root-cause diagnosis.

tasks:
  - id: init_triage_context
    type: io.kestra.plugin.core.log.Log
    message: |
      Initializing AI-Assisted Workflow Failure Triage:
      Execution ID: {{ execution.id }}
      Target Flow: {{ inputs.failed_namespace }}.{{ inputs.failed_flow_id }}
      Target Execution: {{ inputs.failed_execution_id }}
      Diagnosis Model: {{ inputs.diagnosis_model }}

  - id: redact_sensitive_error_logs
    type: io.kestra.plugin.scripts.python.Commands
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.11-slim
    env:
      ERROR_TRACE: "{{ inputs.error_stack_trace }}"
    beforeCommands:
      - pip install --no-cache-dir pydantic
    commands:
      - python sanitize_trace.py
    inputFiles:
      sanitize_trace.py: |
        import json
        import os
        import re
        import sys

        raw_trace = os.environ.get("ERROR_TRACE", "")

        print("Sanitizing raw execution logs to prevent secret and PII leakage...")

        # 1. Scrub Authorization headers and bearer tokens
        scrubbed = re.sub(r"(?i)bearer\s+[a-zA-Z0-9_\-\.]{10,}", "Bearer [REDACTED_BEARER_TOKEN]", raw_trace)
        scrubbed = re.sub(r"(?i)basic\s+[a-zA-Z0-9_\-\.=/]{10,}", "Basic [REDACTED_BASIC_TOKEN]", scrubbed)

        # 2. Scrub explicit password, token, api_key parameters
        scrubbed = re.sub(r"(?i)(token|key|secret|password|passwd|api_key|private_key)[=:\s]+(['\"]?)[a-zA-Z0-9_\-\.]{8,}\2", r"\1=[REDACTED_CREDENTIAL]", scrubbed)

        # 3. Scrub database connection strings (postgres://, mysql://, redis://, mongodb://)
        scrubbed = re.sub(r"[a-zA-Z0-9\+\.\-]+://([^:]+):([^@]+)@", r"[DB_SCHEME]://[REDACTED_USER]:[REDACTED_PASSWORD]@", scrubbed)

        # 4. Scrub email addresses
        scrubbed = re.sub(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "[REDACTED_EMAIL]", scrubbed)

        redactions_count = len(re.findall(r"\[REDACTED_", scrubbed))

        sanitized_summary = {
            "original_length": len(raw_trace),
            "redactions_applied": redactions_count,
            "sanitized_trace": scrubbed
        }

        with open("sanitized_trace.json", "w") as f:
            json.dump(sanitized_summary, f, indent=2)

        print(f"Sanitization complete: {redactions_count} sensitive tokens masked.")
        print(f"::{{'outputs': {{'redactions_count': {redactions_count}, 'sanitized_snippet': '{scrubbed[:120]}...', 'sanitized_full': '{scrubbed}'}}}}::")

  - id: llm_root_cause_diagnosis
    type: io.kestra.plugin.scripts.python.Commands
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.11-slim
    env:
      FAILED_FLOW: "{{ inputs.failed_flow_id }}"
      FAILED_NAMESPACE: "{{ inputs.failed_namespace }}"
      FAILED_EXECUTION: "{{ inputs.failed_execution_id }}"
      ERROR_TRACE: "{{ outputs.redact_sensitive_error_logs.vars.sanitized_full }}"
      DIAGNOSIS_MODEL: "{{ inputs.diagnosis_model }}"
      GEMINI_API_KEY: "{{ secret('GEMINI_API_KEY') }}"
      OPENAI_API_KEY: "{{ secret('OPENAI_API_KEY') }}"
    beforeCommands:
      - pip install --no-cache-dir pydantic
    commands:
      - python diagnose_failure.py
    inputFiles:
      diagnose_failure.py: |
        import json
        import os
        import re
        import urllib.request
        import urllib.error
        import sys

        failed_flow = os.environ.get("FAILED_FLOW", "customer-billing-sync")
        failed_ns = os.environ.get("FAILED_NAMESPACE", "company.team")
        failed_exec = os.environ.get("FAILED_EXECUTION", "exec-928192a")
        error_trace = os.environ.get("ERROR_TRACE", "")
        model_name = os.environ.get("DIAGNOSIS_MODEL", "gemini-2.5-flash")
        gemini_key = os.environ.get("GEMINI_API_KEY", "")
        openai_key = os.environ.get("OPENAI_API_KEY", "")

        print(f"Synthesizing advisory failure diagnosis for flow '{failed_ns}.{failed_flow}' (Execution: {failed_exec})...")

        diagnosis_report = None

        # 1. Attempt live LLM evaluation if API credentials are present
        if gemini_key:
            try:
                print(f"Invoking Google Gemini model ({model_name}) for automated error diagnosis...")
                url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash:generateContent?key={gemini_key}"
                prompt = (
                    "You are a Senior DevOps SRE and Kestra workflow failure investigator. "
                    "Analyze the following sanitized execution error and return ONLY valid JSON with keys: "
                    "'failure_category' (one of: NETWORK_TIMEOUT_DEPENDENCY, DATA_CONTRACT_VIOLATION, AUTHENTICATION_AUTHZ_ERROR, RESOURCE_EXHAUSTION, SYNTAX_RUNTIME_EXCEPTION), "
                    "'severity' (one of: P1_CRITICAL, P2_ELEVATED, P3_MODERATE, P4_LOW), "
                    "'root_cause_summary' (concise 1-2 sentence explanation of the failure), "
                    "'safe_troubleshooting_steps' (array of 3 distinct advisory next steps for on-call engineers), "
                    "'confidence_score' (integer 0-100).\n\n"
                    f"Failed Flow: {failed_ns}.{failed_flow}\n"
                    f"Execution: {failed_exec}\n"
                    f"Sanitized Error Trace:\n{error_trace}\n"
                )
                req_data = json.dumps({"contents": [{"parts": [{"text": prompt}]}]}).encode("utf-8")
                req = urllib.request.Request(url, data=req_data, headers={"Content-Type": "application/json"})
                with urllib.request.urlopen(req, timeout=15) as resp:
                    resp_json = json.loads(resp.read().decode("utf-8"))
                    cand_text = resp_json["candidates"][0]["content"]["parts"][0]["text"]
                    clean_json = re.search(r"\{.*\}", cand_text, re.DOTALL)
                    if clean_json:
                        diagnosis_report = json.loads(clean_json.group(0))
            except Exception as e:
                print(f"Gemini API call failed ({e}); falling back to heuristic classification.")

        # 2. Rule-based heuristic classifier fallback
        if not diagnosis_report:
            print("Applying deterministic SRE heuristic classification rules...")
            low_trace = error_trace.lower()
            if any(term in low_trace for term in ["timeout", "timed out", "connectiontimeout", "504", "connection refused"]):
                cat = "NETWORK_TIMEOUT_DEPENDENCY"
                sev = "P2_ELEVATED"
                summary = "Upstream connection timeout connecting to external API endpoint, causing dependent tasks to fail."
                steps = [
                    "Verify third-party service status dashboard for ongoing external outages.",
                    "Configure task-level retry with exponential backoff on the upstream request task.",
                    "Validate network egress firewall and DNS resolution rules for the container runner."
                ]
            elif any(term in low_trace for term in ["401", "403", "unauthorized", "forbidden", "permission denied", "auth"]):
                cat = "AUTHENTICATION_AUTHZ_ERROR"
                sev = "P1_CRITICAL"
                summary = "Authentication or authorization rejected by target service; credentials may be expired or misconfigured."
                steps = [
                    "Verify target API key / service account credentials in Kestra Secrets.",
                    "Check secret expiration dates and IAM role assignments.",
                    "Ensure namespace secret access policies allow the failing task to read the secret."
                ]
            elif any(term in low_trace for term in ["illegalvariableevaluation", "null", "missing key", "keyerror", "schema"]):
                cat = "DATA_CONTRACT_VIOLATION"
                sev = "P2_ELEVATED"
                summary = "Downstream task attempted to evaluate a missing, null, or unmapped variable from an upstream task output."
                steps = [
                    "Inspect upstream task execution outputs to confirm expected variable names and JSON paths.",
                    "Add a default value fallback expression, e.g. {{ outputs.task.vars.value ?? 'default' }}.",
                    "Ensure upstream task completed in SUCCESS before downstream consumption."
                ]
            elif any(term in low_trace for term in ["outofmemory", "oom", "disk full", "no space left"]):
                cat = "RESOURCE_EXHAUSTION"
                sev = "P1_CRITICAL"
                summary = "Worker container exceeded available memory or disk space allocations during execution."
                steps = [
                    "Increase container memory limit or swap to a larger worker runner instance.",
                    "Implement batch chunking or pagination to stream large datasets.",
                    "Clean up temporary intermediate files in working directory before task completion."
                ]
            else:
                cat = "SYNTAX_RUNTIME_EXCEPTION"
                sev = "P3_MODERATE"
                summary = "Uncaught runtime exception encountered during script execution."
                steps = [
                    "Inspect full execution logs for specific stack trace line number.",
                    "Verify input parameter types and constraints match script expectations.",
                    "Test script locally inside matching container image to reproduce failure."
                ]

            diagnosis_report = {
                "failure_category": cat,
                "severity": sev,
                "root_cause_summary": summary,
                "safe_troubleshooting_steps": steps,
                "confidence_score": 92
            }

        category = diagnosis_report.get("failure_category", "SYNTAX_RUNTIME_EXCEPTION")
        severity = diagnosis_report.get("severity", "P2_ELEVATED")
        summary_text = diagnosis_report.get("root_cause_summary", "Unclassified workflow execution error.")
        steps_list = diagnosis_report.get("safe_troubleshooting_steps", [])
        formatted_steps = "\\n".join([f"• {s}" for s in steps_list])

        final_output = {
            "flow_id": failed_flow,
            "namespace": failed_ns,
            "execution_id": failed_exec,
            "diagnosis": diagnosis_report
        }

        with open("diagnosis_report.json", "w") as f:
            json.dump(final_output, f, indent=2)

        print(f"Diagnostic Analysis Complete: Category={category}, Severity={severity}.")
        print(f"::{{'outputs': {{'category': '{category}', 'severity': '{severity}', 'summary': '{summary_text}', 'formatted_steps': '{formatted_steps}'}}}}::")

  - id: notify_slack_advisory_card
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "🤖 *AI-Assisted Workflow Failure Diagnosis*\n*Target Flow:* `{{ inputs.failed_namespace }}.{{ inputs.failed_flow_id }}`\n*Execution:* `{{ inputs.failed_execution_id }}`\n*Category:* `{{ outputs.llm_root_cause_diagnosis.vars.category }}` | *Severity:* `{{ outputs.llm_root_cause_diagnosis.vars.severity }}`\n*Root Cause:* {{ outputs.llm_root_cause_diagnosis.vars.summary }}\n*Recommended Troubleshooting Steps:*\n{{ outputs.llm_root_cause_diagnosis.vars.formatted_steps }}\n*Advisory:* Automated diagnosis only, verify before applying production changes."
      }

Orchestrate an advisory AI-assisted workflow failure triage pipeline that listens for failed Kestra executions, automatically extracts error stack traces, redacts sensitive credentials and PII, classifies the root cause category using LangChain, and delivers actionable troubleshooting steps directly to on-call engineering Slack channels.

When production data pipelines and cloud workflows fail in the middle of the night, on-call engineers waste valuable time parsing noisy stack traces and logs. This blueprint provides an automated, advisory diagnostic assistant:

Automated Failure Ingestion: Listens to execution status transitions (FAILED) via Kestra's Flow trigger with dependsOn or accepts manual triage webhooks.
Secret Scrubbing & PII Redaction (redact_sensitive_error_logs): Automatically strips Authorization tokens, passwords, bearer credentials, database connection strings, and emails before sending logs to LLMs.
LLM Diagnostic Classification (llm_root_cause_diagnosis): Classifies the failure into structured categories (network timeouts, schema mismatches, null evaluations, permission errors) and outlines safe, advisory next steps.
Advisory Team Dispatch (notify_slack_advisory_card): Delivers an executive diagnosis card directly to Slack. The workflow remains strictly advisory and never executes unauthorized production fixes.

How it works

Triggering:
- The on_failure trigger (io.kestra.plugin.core.trigger.Flow) uses dependsOn with states: [FAILED] on the target namespace company.team, automatically mapping {{ trigger.executionId }}, {{ trigger.flowId }}, and {{ trigger.namespace }} into the flow's inputs.
- The inbound_failure_webhook trigger (io.kestra.plugin.core.trigger.Webhook) allows external monitoring tools or manual developer triggers to request failure triage directly.
Sanitization (redact_sensitive_error_logs):
- Runs inside a sandboxed Python Docker container (python:3.11-slim) to sanitize sensitive tokens from the stack trace.
AI Root Cause Synthesis (llm_root_cause_diagnosis):
- Evaluates the failure context using LangChain / Gemini models, generating a categorized failure summary with bulleted action steps.
Slack Notification (notify_slack_advisory_card):
- Sends formatted diagnostic summaries to developers with zero delay.

What you get

Automated error classification accelerating mean-time-to-resolution (MTTR).
Regex and keyword secret redaction preventing credential leakage to AI providers.
Advisory-only safety guardrails preventing hallucinated automated state mutations.
Real-time Slack notifications with actionable next steps.

Who it's for

On-call DevOps, SRE, and Platform Engineers troubleshooting production pipeline failures.
Data Engineers maintaining complex dbt, Airbyte, and warehouse pipelines.
Engineering Managers monitoring error trends across team namespaces.

Prerequisites

Docker daemon accessible to Kestra for containerized execution.
Slack Incoming Webhook URL configured in Kestra secrets.

Secrets

FAILURE_TRIAGE_WEBHOOK_KEY: Secret key for inbound failure webhooks.
SLACK_WEBHOOK_URL: Slack Incoming Webhook endpoint for team alerts.
GEMINI_API_KEY or OPENAI_API_KEY: API keys for foundational model access.

Inputs

Name	Type	Default	Description
`failed_namespace`	STRING	`company.team`	Namespace of the failed workflow.
`failed_flow_id`	STRING	`customer-billing-sync`	Identifier of the failed workflow.
`failed_execution_id`	STRING	`exec-928192a`	Failed execution identifier.
`error_stack_trace`	STRING	Error log trace	Raw error message or exception trace.
`diagnosis_model`	SELECT	`gemini-2.5-flash`	LLM model (`gemini-2.5-flash`, `gpt-4o-mini`).

Quick start runbook

Configure secrets SLACK_WEBHOOK_URL and GEMINI_API_KEY in your Kestra namespace.
Trigger the flow manually with a sample error stack trace to observe log sanitization and AI root cause diagnosis.
Connect the flow trigger to your production namespace to automatically receive triage cards whenever a flow fails.

AI-Assisted Workflow Failure Triage

How it works

What you get

Who it's for

Prerequisites

Secrets

Inputs

Quick start runbook

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

AI-Assisted Workflow Failure Triage

How it works

What you get

Who it's for

Prerequisites

Secrets

Inputs

Quick start runbook

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More AI Blueprints

AI Annual Report Finder and Data Strategy Summary

Generate a book chapter summary using Perplexity AI and email it.

Daily arXiv AI digest with keyword ranking

Summarize Text with an AI Agent, Generate Multilingual Summaries in Short, Medium, or Long Format

Trigger the Right Automation with an AI Agent Based on User Intent

Decide When to Take Action Using AI, Check Real-Time Weather and Trigger Automations

New to Kestra?