Kibana Anomaly Detection

{
  "name": "kibana-anomaly-detection-skills",
  "version": "0.1.0",
  "lockfileVersion": 3,
  "requires": true,
  "packages": {
    "": {
      "name": "kibana-anomaly-detection-skills",
      "version": "0.1.0"
    }
  }
}

{
  "name": "kibana-anomaly-detection-skills",
  "version": "0.1.0",
  "private": true,
  "type": "module",
  "scripts": {
    "tools:register": "node scripts/kibana-agent-builder.mjs tools register",
    "workflows:register": "node scripts/kibana-agent-builder.mjs workflows register",
    "skills:register": "node scripts/kibana-agent-builder.mjs skills register",
    "all:register": "node scripts/kibana-agent-builder.mjs all register"
  },
  "dependencies": {}
}

{
  "tools": [
    "ad_get_available_metadata",
    "ad_get_jobs",
    "ad_discover_related_jobs",
    "ad_rca_cross_job_entity_match",
    "ad_query_anomaly_records",
    "ad_query_anomaly_timeline",
    "ad_query_influencers",
    "ad_rca_entity_profile",
    "ad_rca_detector_fingerprint",
    "ad_rca_correlation",
    "ad_rca_blast_radius",
    "ad_rca_multi_job_entities",
    "ad_rca_source_evidence",
    "ad_discover_jobs_by_datafeed_index",
    "ad_get_job_datafeed_config",
    "ad_get_categories",
    "ad_search_log_category_examples"
  ]
}

{
  "tools": [
    "ad_get_available_metadata",
    "ad_get_jobs",
    "ad_query_anomaly_records",
    "ad_query_influencers",
    "ad_rca_score_reassessment",
    "ad_get_model_plot",
    "ad_get_categories",
    "ad_get_forecast_results",
    "ad_wf_troubleshoot_anomaly_score",
    "ad_get_job_datafeed_config"
  ]
}

{
  "tools": [
    "ad_get_available_metadata",
    "ad_get_jobs",
    "ad_get_job_messages",
    "ad_get_model_snapshots",
    "ad_ts_bucket_event_gaps",
    "ad_ts_delayed_data_annotations",
    "ad_ts_ingest_latency_estimate",
    "ad_ts_model_memory_health",
    "ad_wf_ts_field_cardinality",
    "ad_get_job_datafeed_config",
    "ad_manage_datafeed",
    "ad_preview_datafeed_with_latency",
    "ad_create_job",
    "ad_revert_model_snapshot",
    "ad_create_calendar_event",
    "ad_get_calendar_events",
    "ad_update_datafeed_query_delay",
    "ad_update_delayed_data_check_config",
    "ad_update_model_memory_limit",
    "ad_estimate_memory_requirement",
    "ad_validate_ml_tool_permissions",
    "ad_ts_ccs_diagnostics",
    "ad_wf_troubleshoot_query_delay",
    "ad_wf_troubleshoot_memory_limit"
  ]
}

{
  "name": "ad_discover_related_jobs",
  "description": "Given a job of interest, find all other jobs that share the same entity field name (partition, by, or over field) by reading job configs from .ml-config. Returns the shared field name, the full list of sibling jobs in that group, their configured influencer fields, and the total sibling count. A job appears in its primary entity field group (partition > by > over). Call this early in investigation to find sibling jobs even when job names and prefixes differ completely. Complement with ad_discover_jobs_by_datafeed_index for source index overlap.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-config | WHERE job_type == \"anomaly_detector\" | EVAL entity_field = COALESCE(`analysis_config.detectors.partition_field_name`, `analysis_config.detectors.by_field_name`, `analysis_config.detectors.over_field_name`) | MV_EXPAND entity_field | WHERE entity_field IS NOT NULL | STATS job_count = COUNT_DISTINCT(job_id), jobs = VALUES(job_id), influencers = VALUES(`analysis_config.influencers`) BY entity_field | WHERE MV_CONTAINS(jobs, ?job_id) | SORT job_count DESC | LIMIT 50"
  },
  "parameters": {
    "job_id": {
      "type": "string",
      "description": "The job ID of interest. Returns all other jobs that share the same entity split field (partition, by, or over field) as this job."
    }
  }
}

{
  "name": "ad_get_available_metadata",
  "description": "Discover all anomaly detection jobs and their configured metadata: job IDs, detector functions, monitored field names, entity split fields (by/over/partition), influencer field names, and bucket spans. Reads from .ml-config so all jobs are visible even if they have never produced an anomaly. Returns a single summary row. Call this FIRST before other tools to learn valid parameter values for job_id, function, field_name, and entity field filters.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-config | WHERE job_type == \"anomaly_detector\" | STATS job_count = COUNT(*), job_ids = VALUES(job_id), functions = VALUES(`analysis_config.detectors.function`), fields = VALUES(`analysis_config.detectors.field_name`), by_fields = VALUES(`analysis_config.detectors.by_field_name`), over_fields = VALUES(`analysis_config.detectors.over_field_name`), partition_fields = VALUES(`analysis_config.detectors.partition_field_name`), influencers = VALUES(`analysis_config.influencers`), bucket_spans = VALUES(`analysis_config.bucket_span`)"
  },
  "parameters": {}
}

{
  "name": "ad_get_categories",
  "description": "Get log categories for categorization jobs: category definitions, regex patterns, and examples. Useful for understanding what types of log messages the model has learned and which categories are generating anomalies.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"category_definition\" AND job_id == ?job_id | SORT category_id ASC | KEEP job_id, category_id, terms, regex, max_matching_length, examples | LIMIT 100"
  },
  "parameters": {
    "job_id": {
      "type": "string",
      "description": "The anomaly detection job ID"
    }
  }
}

{
  "name": "ad_get_forecast_results",
  "description": "Retrieve forecast predictions with upper/lower bounds for capacity planning. Queries model_forecast results from .ml-anomalies-*.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"model_forecast\" AND job_id == ?job_id | SORT timestamp ASC | KEEP job_id, timestamp, forecast_prediction, forecast_upper, forecast_lower, bucket_span | LIMIT 200"
  },
  "parameters": {
    "job_id": {
      "type": "string",
      "description": "The anomaly detection job ID"
    }
  }
}

{
  "name": "ad_get_job_messages",
  "description": "Retrieve all notifications for a job from the ML notifications index, including all severity levels (info, warning, error). Covers datafeed warnings, delayed data alerts, missing document warnings, query errors, timeout messages, CCS connectivity issues, memory limit warnings, and job lifecycle events. Returns all levels so the agent can filter by level in its reasoning. Use this for both general notification browsing and targeted datafeed warning investigation.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-notifications-* | WHERE job_id == ?job_id | SORT timestamp DESC | KEEP timestamp, level, message, node_name, job_id | LIMIT 50"
  },
  "parameters": {
    "job_id": {
      "type": "string",
      "description": "The anomaly detection job ID"
    }
  }
}

{
  "name": "ad_get_jobs",
  "description": "List all anomaly detection jobs with their full configuration: bucket_span, detector functions, monitored field names, entity split fields (partition/by/over), influencers, memory limit, groups, and description. Reads from .ml-config so ALL jobs appear regardless of whether they have run or produced any anomaly results. Use this early in any investigation to understand what jobs exist and how they are configured.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-config | WHERE job_type == \"anomaly_detector\" | KEEP job_id, `analysis_config.bucket_span`, `analysis_config.detectors.function`, `analysis_config.detectors.field_name`, `analysis_config.detectors.partition_field_name`, `analysis_config.detectors.by_field_name`, `analysis_config.detectors.over_field_name`, `analysis_config.influencers`, `analysis_limits.model_memory_limit`, groups, description | SORT job_id ASC | LIMIT 100"
  },
  "parameters": {}
}

{
  "name": "ad_get_model_plot",
  "description": "Get model bounds (upper/lower/median) to explain why something was or was not flagged as anomalous. Shows the model's confidence interval at each time point. If actual value is within bounds, no anomaly; if outside, anomaly score depends on distance from bounds.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"model_plot\" AND job_id == ?job_id AND timestamp >= ?start_time AND timestamp <= ?end_time | SORT timestamp ASC | KEEP job_id, timestamp, model_lower, model_upper, model_median, actual, partition_field_value, by_field_value | LIMIT 500"
  },
  "parameters": {
    "job_id": {
      "type": "string",
      "description": "The anomaly detection job ID"
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_get_model_snapshots",
  "description": "List available model snapshots for a job, including timestamp, description, and size. Used for model revert operations when data quality issues have corrupted the model.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"model_snapshot\" AND job_id == ?job_id | SORT timestamp DESC | KEEP job_id, timestamp, description, snapshot_doc_count | LIMIT 20"
  },
  "parameters": {
    "job_id": {
      "type": "string",
      "description": "The anomaly detection job ID"
    }
  }
}

{
  "name": "ad_query_anomaly_records",
  "description": "Search anomaly records by score threshold and time range. Use job_id_pattern='*' for cross-job search (default) or an exact job ID for single-job drill-down. Supports global cross-job search, job-scoped deep dive, absence detection (actual << typical), and value trend analysis. This is the primary tool for questions like 'find anomalies related to entity X' as well as per-job investigation.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"record\" AND job_id LIKE ?job_id_pattern AND record_score >= ?min_score AND timestamp >= ?start_time AND timestamp <= ?end_time | SORT record_score DESC | LIMIT 50 | KEEP job_id, timestamp, record_score, function, field_name, actual, typical, by_field_name, by_field_value, over_field_name, over_field_value, partition_field_name, partition_field_value, multi_bucket_impact, initial_record_score, detector_index, probability"
  },
  "parameters": {
    "job_id_pattern": {
      "type": "string",
      "description": "Job ID pattern using LIKE wildcards (* = multi-char). Use '*' for all jobs (cross-job search), or an exact job ID to scope to a single job (e.g. 'rcaeval-ob-cpu'). Use 'rcaeval-*' to scope to a prefix group."
    },
    "min_score": {
      "type": "string",
      "description": "Minimum record_score threshold (0-100). Use 50 for significant anomalies, 25 for broader search, 75 for critical only."
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format (e.g. 2024-01-01T00:00:00Z)"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_query_anomaly_timeline",
  "description": "Get a time-series summary of anomaly severity across all or selected jobs, bucketed by time. Use for building cross-job swimlanes, detecting anomaly storms (multiple jobs firing together), and computing composite signal scores. When multiple jobs have anomaly_score > threshold in the same bucket, it indicates a common external trigger. Scope to a job group with job_id_pattern (e.g. 'rcaeval-*') or use '*' for all jobs.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"bucket\" AND job_id LIKE ?job_id_pattern AND anomaly_score >= ?min_score AND timestamp >= ?start_time AND timestamp <= ?end_time | STATS max_score = MAX(anomaly_score), job_count = COUNT_DISTINCT(job_id), jobs = VALUES(job_id), composite_score = SUM(anomaly_score), avg_score = AVG(anomaly_score) BY timestamp | SORT timestamp"
  },
  "parameters": {
    "job_id_pattern": {
      "type": "string",
      "description": "Job ID pattern using LIKE wildcards. Use '*' for all jobs, or a prefix pattern to scope to a related group (e.g. 'rcaeval-*')."
    },
    "min_score": {
      "type": "string",
      "description": "Minimum bucket anomaly_score. Use 25 for signal boosting (catch coordinated low-severity events), 50 for standard, 75 for critical."
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_query_influencers",
  "description": "Find the most unusual entities across all or selected jobs for a time range. Answers 'What entities are most anomalous right now?' and 'Which entities appear as influencers in MULTIPLE jobs simultaneously?' When require_multi_job semantics are needed, filter results where job_count > 1 to find shared influencers for cross-job RCA. Scope to a job group with job_id_pattern (e.g. 'rcaeval-*') or use '*' for all jobs.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"influencer\" AND job_id LIKE ?job_id_pattern AND influencer_score >= ?min_score AND timestamp >= ?start_time AND timestamp <= ?end_time | STATS total_score = SUM(influencer_score), job_count = COUNT_DISTINCT(job_id), jobs = VALUES(job_id), max_score = MAX(influencer_score) BY influencer_field_name, influencer_field_value | SORT total_score DESC | LIMIT 30"
  },
  "parameters": {
    "job_id_pattern": {
      "type": "string",
      "description": "Job ID pattern using LIKE wildcards. Use '*' for all jobs, or a prefix pattern to scope to a related group (e.g. 'rcaeval-*')."
    },
    "min_score": {
      "type": "string",
      "description": "Minimum influencer_score threshold. Use 25 for broad search (shared influencer discovery), 50 for significant entities."
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_rca_blast_radius",
  "description": "Measure how widespread a threat or issue is by counting how many hosts, partitions, or entities are affected by a specific anomalous value. Given a value (e.g. process.name: node-backdoor), counts distinct affected partitions across all jobs. Also aggregates weak signals: entities with multiple low-score anomalies across different jobs that individually are below alerting thresholds but collectively indicate a real problem.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"record\" AND record_score >= ?min_score AND timestamp >= ?start_time AND timestamp <= ?end_time AND (by_field_value == ?anomalous_value OR over_field_value == ?anomalous_value OR partition_field_value == ?anomalous_value) | STATS affected_partitions = COUNT_DISTINCT(partition_field_value), affected_jobs = COUNT_DISTINCT(job_id), total_anomalies = COUNT(*), max_score = MAX(record_score), time_span_hours = DATE_DIFF(\"hour\", MIN(timestamp), MAX(timestamp)) BY by_field_value | SORT affected_partitions DESC"
  },
  "parameters": {
    "anomalous_value": {
      "type": "string",
      "description": "The specific anomalous value to measure blast radius for (e.g. 'node-backdoor', 'suspicious-script.ps1')"
    },
    "min_score": {
      "type": "string",
      "description": "Minimum record_score. Use low values (10-25) for weak signal aggregation."
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_rca_correlation",
  "description": "Find temporally correlated anomalies across different jobs. Supports two modes: (1) co_occurrence — anomalies from different jobs in overlapping time windows regardless of shared influencers, essential for cross-domain correlation (K8s + APM + logs); (2) ordered_sequence — anomalies sorted by time to detect cascading failures (network -> app -> DB propagation). The agent examines temporal ordering and job types to infer causality. Use job_id_pattern to scope to a subset of jobs (e.g. 'rcaeval-ob-*') when many jobs exist.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"record\" AND job_id LIKE ?job_id_pattern AND record_score >= ?min_score AND timestamp >= ?start_time AND timestamp <= ?end_time | SORT timestamp ASC | KEEP job_id, timestamp, record_score, function, field_name, by_field_name, by_field_value, partition_field_name, partition_field_value, over_field_name, over_field_value, multi_bucket_impact | LIMIT 100"
  },
  "parameters": {
    "job_id_pattern": {
      "type": "string",
      "description": "Job ID pattern using LIKE wildcards (* = multi-char). Use '*' for all jobs, 'rcaeval-*' for all RCAEval jobs, 'rcaeval-ob-*' for Online Boutique only, 'nab-*' for NAB jobs, etc."
    },
    "min_score": {
      "type": "string",
      "description": "Minimum record_score threshold"
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_rca_cross_job_entity_match",
  "description": "Starting from a specific entity value (e.g. a service name extracted from an alert), find ALL other jobs where that entity appears as anomalous — across partition, by, and over fields. Returns per-job summary with max score, anomaly count, detector functions, field names, and the first/last anomaly timestamps. Use first_anomaly to reconstruct chronology: the job that detected the entity earliest is closest to the root cause. Unlike ad_rca_multi_job_entities (which groups by partition_field_value only), this matches the entity value across ALL split field types.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"record\" AND record_score >= ?min_score AND timestamp >= ?start_time AND timestamp <= ?end_time AND (partition_field_value == ?entity_value OR by_field_value == ?entity_value OR over_field_value == ?entity_value) | STATS max_score = MAX(record_score), anomaly_count = COUNT(*), functions = VALUES(function), fields = VALUES(field_name), first_anomaly = MIN(timestamp), last_anomaly = MAX(timestamp) BY job_id | SORT max_score DESC"
  },
  "parameters": {
    "entity_value": {
      "type": "string",
      "description": "The entity value to search for across all jobs (e.g. 'frontend', 'server-01', 'payment-service'). Extract this from the alerting anomaly's partition_field_value, by_field_value, or over_field_value."
    },
    "min_score": {
      "type": "string",
      "description": "Minimum record_score threshold. Use 10-25 for comprehensive search (catches weak cascade signals), 50 for significant anomalies only."
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format. Use a wider window than the alert (e.g. alert_time minus 6 hours) to catch the full cascade."
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format. Use alert_time plus 1-2 hours to catch delayed effects."
    }
  }
}

{
  "name": "ad_rca_detector_fingerprint",
  "description": "For a specific incident time window, produce a fingerprint showing exactly which detectors fired across all or selected jobs and what they monitor. Shows which aspects of the system are anomalous (CPU? Network? Error rate? Latency?). Group by job_id + function + field_name to understand the incident signature. Scope to a job group with job_id_pattern (e.g. 'rcaeval-*') or use '*' for all jobs.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"record\" AND job_id LIKE ?job_id_pattern AND record_score >= ?min_score AND timestamp >= ?start_time AND timestamp <= ?end_time | STATS count = COUNT(*), max_score = MAX(record_score), avg_score = AVG(record_score) BY job_id, function, field_name, detector_index | SORT max_score DESC"
  },
  "parameters": {
    "job_id_pattern": {
      "type": "string",
      "description": "Job ID pattern using LIKE wildcards. Use '*' for all jobs, or a prefix pattern to scope to a related group (e.g. 'rcaeval-*')."
    },
    "min_score": {
      "type": "string",
      "description": "Minimum record_score threshold"
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_rca_entity_profile",
  "description": "Build a complete anomaly dossier for a suspect entity across ALL jobs. Given an entity value (e.g. host.name: server-01), shows every anomaly where it appeared as an influencer, by_field, partition_field, or over_field value. Use after identifying a suspect via ad_query_influencers to build full context.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"record\" AND timestamp >= ?start_time AND timestamp <= ?end_time AND (by_field_value == ?entity_value OR over_field_value == ?entity_value OR partition_field_value == ?entity_value) | SORT timestamp ASC | LIMIT 100 | KEEP job_id, timestamp, record_score, function, field_name, by_field_name, by_field_value, over_field_name, over_field_value, partition_field_name, partition_field_value, actual, typical, multi_bucket_impact"
  },
  "parameters": {
    "entity_value": {
      "type": "string",
      "description": "The entity value to profile (e.g. 'server-01', 'payment-service', 'user@example.com')"
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_rca_multi_job_entities",
  "description": "Find entities that are anomalous in MULTIPLE jobs simultaneously — the strongest root cause signal. Returns entities ranked by the number of distinct jobs they appear in, with per-job max scores and detector functions. Entities in 2+ jobs are prime root cause candidates (e.g., a service with CPU AND latency anomalies); entities in only 1 job are likely secondary effects or victims. This is the key discriminator for RCA: resource faults produce multi-job anomalies, network faults produce single-job anomalies.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"record\" AND record_score >= ?min_score AND timestamp >= ?start_time AND timestamp <= ?end_time | STATS job_count = COUNT_DISTINCT(job_id), jobs = VALUES(job_id), max_score = MAX(record_score), total_records = COUNT(*), functions = VALUES(function), fields = VALUES(field_name) BY partition_field_value | WHERE job_count >= ?min_job_count | SORT job_count DESC, max_score DESC | LIMIT 20"
  },
  "parameters": {
    "min_score": {
      "type": "string",
      "description": "Minimum record_score threshold. Use 25 for broader search (catches weak cascade signals), 50 for significant anomalies."
    },
    "min_job_count": {
      "type": "string",
      "description": "Minimum number of distinct jobs the entity must appear in. Use 2 to find cross-job root cause candidates, 1 to include all entities."
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_rca_score_reassessment",
  "description": "Find anomalies where the model has significantly changed its assessment over time due to renormalization. This tool defines score_drift = initial_record_score - record_score. When renormalization lowers the current score, initial_record_score stays higher → large positive drift (initial >> current). Large negative drift means the current score rose versus the initial snapshot (upward reconsideration). Records where scores stayed high indicate persistent anomalies the model never explained away. Scope to a specific job with job_id_pattern or use '*' for all jobs.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"record\" AND job_id LIKE ?job_id_pattern AND timestamp >= ?start_time AND timestamp <= ?end_time | EVAL score_drift = initial_record_score - record_score | WHERE ABS(score_drift) >= ?min_drift | SORT ABS(score_drift) DESC | LIMIT 30 | KEEP job_id, timestamp, initial_record_score, record_score, score_drift, function, field_name, by_field_value, partition_field_value"
  },
  "parameters": {
    "job_id_pattern": {
      "type": "string",
      "description": "Job ID pattern using LIKE wildcards. Use '*' for all jobs, or an exact job ID to scope to a single job (e.g. 'rcaeval-ob-cpu')."
    },
    "min_drift": {
      "type": "string",
      "description": "Minimum score point difference between initial and current score (default 20)"
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_rca_source_evidence",
  "description": "After identifying an anomaly, retrieve raw source documents from the ORIGINAL data index for the anomaly's time window. This is the evidence drilldown — see the actual log lines, traces, or metrics that caused the statistical deviation. Without this, the agent can say 'something unusual happened' but cannot say 'here is what actually happened'.\n\nUsage: First find the source index from the job's datafeed configuration (via ad_get_job_datafeed_config). Pass the index name as source_index using LIKE wildcards (* for multi-char, ? for single-char). The tool returns raw documents with all original fields from that index.",
  "type": "esql",
  "configuration": {
    "query": "FROM * METADATA _index | WHERE _index LIKE ?source_index AND @timestamp >= ?start_time AND @timestamp <= ?end_time | SORT @timestamp DESC | LIMIT 50"
  },
  "parameters": {
    "source_index": {
      "type": "string",
      "description": "Source data index name or LIKE pattern (* = multi-char wildcard). Get this from the job's datafeed config. Examples: 'rcaeval-re1-ob', 'nab', 'otel-flat-*', 'smd'"
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_search_log_category_examples",
  "description": "Search for log message examples in the source data for a specific time window. Used for two-window comparison in log categorization RCA: run once for a baseline window (e.g., 24h before anomaly) and once for the anomaly window. Compare the samples to identify what changed in the variable parts (IPs, hostnames, error codes, paths) that caused the category count anomaly. Returns raw log documents so you can inspect the categorization_field_name field specified in the job config.",
  "type": "esql",
  "configuration": {
    "query": "FROM * METADATA _index | WHERE _index LIKE ?source_index AND @timestamp >= ?start_time AND @timestamp <= ?end_time | SORT @timestamp DESC | LIMIT 50"
  },
  "parameters": {
    "source_index": {
      "type": "string",
      "description": "Source data index name or LIKE pattern from the job's datafeed config. Get this from ad_get_job_datafeed_config. Examples: 'logs-*', 'filebeat-*', 'otel-logs-*'"
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format. For baseline window, use a period before the anomaly (e.g., 24h before)."
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_ts_bucket_event_gaps",
  "description": "Find buckets with zero or suspiciously low event counts for a specific job. These are the buckets actually affected by missing data. Compare event_count across buckets to identify time ranges where data was lost. Correlate with delayed data annotations and anomaly scores to confirm false positives from missing data.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"bucket\" AND job_id == ?job_id AND timestamp >= ?start_time AND timestamp <= ?end_time | SORT timestamp ASC | KEEP timestamp, event_count, anomaly_score, bucket_span, is_interim | LIMIT 500"
  },
  "parameters": {
    "job_id": {
      "type": "string",
      "description": "The anomaly detection job ID"
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_ts_delayed_data_annotations",
  "description": "Retrieve all delayed data annotations for a job, showing exactly when and how many documents were missed. The annotation field contains text like 'Datafeed has missed 30 documents due to ingest latency...' \u2014 frequent annotations indicate chronic ingest latency. This is the starting point for any 'missing documents' investigation.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-annotations-* | WHERE job_id == ?job_id AND event == \"delayed_data\" | SORT timestamp DESC | KEEP job_id, timestamp, end_timestamp, annotation | LIMIT 100"
  },
  "parameters": {
    "job_id": {
      "type": "string",
      "description": "The anomaly detection job ID"
    }
  }
}

{
  "name": "ad_ts_ingest_latency_estimate",
  "description": "Measure actual ingest latency by comparing event timestamps with ingestion timestamps in the SOURCE data. Determines whether the current query_delay is sufficient. If P95(event.ingested - @timestamp) > query_delay, data will be lost. Requires the source index to have an event.ingested or _ingest.timestamp field.\n\nUsage: Get the source index from the job's datafeed config (via ad_get_job_datafeed_config). Pass it as source_index using LIKE wildcards if needed.",
  "type": "esql",
  "configuration": {
    "query": "FROM * METADATA _index | WHERE _index LIKE ?source_index AND @timestamp >= ?start_time AND @timestamp <= ?end_time AND event.ingested IS NOT NULL | EVAL latency_seconds = DATE_DIFF(\"second\", @timestamp, event.ingested) | STATS p50_latency = PERCENTILE(latency_seconds, 50), p95_latency = PERCENTILE(latency_seconds, 95), p99_latency = PERCENTILE(latency_seconds, 99), max_latency = MAX(latency_seconds), doc_count = COUNT(*) | LIMIT 1"
  },
  "parameters": {
    "source_index": {
      "type": "string",
      "description": "Source data index name or LIKE pattern (* = multi-char wildcard). Get this from the job's datafeed config. Examples: 'rcaeval-re1-ob', 'nab', 'otel-flat-*', 'smd'"
    },
    "start_time": {
      "type": "string",
      "description": "Start of time range in ISO 8601 format"
    },
    "end_time": {
      "type": "string",
      "description": "End of time range in ISO 8601 format"
    }
  }
}

{
  "name": "ad_ts_model_memory_health",
  "description": "Get memory health and growth trend for a job. Returns model_size_stats records in reverse-chronological order. Use limit=1 for a fast current-state snapshot (hard_limit/soft_limit check); use limit=500 for full memory growth trend analysis (stable plateau / linear / exponential). Interpretation: hard_limit = CRITICAL (job blind to new entities), soft_limit = WARNING (aggressive pruning), model_bytes/model_bytes_memory_limit > 0.8 = APPROACHING LIMIT.",
  "type": "esql",
  "configuration": {
    "query": "FROM .ml-anomalies-* | WHERE result_type == \"model_size_stats\" AND job_id == ?job_id | SORT timestamp DESC | KEEP job_id, timestamp, model_bytes, peak_model_bytes, model_bytes_memory_limit, model_bytes_exceeded, memory_status, total_by_field_count, total_over_field_count, total_partition_field_count, bucket_allocation_failures_count | LIMIT ?limit"
  },
  "parameters": {
    "job_id": {
      "type": "string",
      "description": "The anomaly detection job ID"
    },
    "limit": {
      "type": "string",
      "description": "Number of historical records to return. Use 1 for current memory status (fast snapshot), 500 for full memory growth trend analysis."
    }
  }
}

name: ad_create_calendar_event
description: >
  Add a scheduled event to a calendar to suppress false positives during known downtime, maintenance windows, or
  holidays.
enabled: true
tags: ["anomaly-detection", "remediation"]

inputs:
  - name: calendar_id
    type: string
    description: The calendar ID
  - name: event_description
    type: string
    description: "Event description (e.g., 'Planned maintenance window')"
  - name: start_time
    type: string
    description: Event start time in ISO 8601 format
  - name: end_time
    type: string
    description: Event end time in ISO 8601 format

triggers:
  - type: manual

steps:
  - name: create_event
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/calendars/{{ inputs.calendar_id }}/events
      body:
        events:
          - description: "{{ inputs.event_description }}"
            start_time: "{{ inputs.start_time }}"
            end_time: "{{ inputs.end_time }}"

  - name: result
    type: console
    with:
      message: |
        Created event '{{ inputs.event_description }}' on calendar {{ inputs.calendar_id }}
        from {{ inputs.start_time }} to {{ inputs.end_time }}.

name: ad_create_datafeed
description: >
  Create or replace a datafeed for an anomaly detection job via PUT _ml/datafeeds/{datafeed_id}. Use after job creation
  and before opening the job / starting the datafeed.
enabled: true
tags: ["anomaly-detection", "management"]

inputs:
  - name: datafeed_id
    type: string
    description: Datafeed ID (typically datafeed-{job_id})
  - name: datafeed_body
    type: string
    description: >
      Full datafeed configuration as JSON text. Parsed with json_parse so the PUT body is a structured object for the ML
      API, not a JSON-encoded string.

triggers:
  - type: manual

steps:
  - name: put_datafeed
    type: elasticsearch.request
    with:
      method: PUT
      path: /_ml/datafeeds/{{ inputs.datafeed_id }}
      body: "${{ inputs.datafeed_body | json_parse }}"

  - name: result
    type: console
    with:
      message: |
        Datafeed {{ inputs.datafeed_id }}:
        {{ steps.put_datafeed.output | json:2 }}

name: ad_create_job
description: >
  Create a new anomaly detection job from a configuration. Stretch goal — for advanced agent use cases where the agent
  helps design and create jobs based on data exploration. job_body is JSON text; the step uses Liquid json_parse and typed
  interpolation (${{ }}) so the PUT body is a structured JSON object for the ML API, not a JSON-encoded string.
enabled: true
tags: ["anomaly-detection", "management"]

inputs:
  - name: job_id
    type: string
    description: The new job ID to create
  - name: job_body
    type: string
    description: >
      Full job configuration as JSON text (same shape as PUT /_ml/anomaly_detectors/{job_id}). Parsed with json_parse
      before the request so the HTTP body is an object. Runners that support a native object input may still pass JSON
      text here.

triggers:
  - type: manual

steps:
  - name: create_job
    type: elasticsearch.request
    with:
      method: PUT
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}
      body: "${{ inputs.job_body | json_parse }}"

  - name: result
    type: console
    with:
      message: |
        Created job {{ inputs.job_id }}:
        {{ steps.create_job.output | json:2 }}

name: ad_discover_jobs_by_datafeed_index
description: >
  Given a job of interest, find all other jobs whose datafeed reads from overlapping source indices. Step 1 retrieves
  the target job config (including datafeed_config.indices). Step 2 logs the target job summary. Step 3 iterates over
  each index in that list and queries .ml-config for other datafeed documents containing the same index, logging matches
  per index pattern. Jobs reading from the same indices monitor the same system and are strong candidates for cross-job
  correlation.
enabled: true
tags: ["anomaly-detection", "rca", "discovery"]

inputs:
  - name: job_id
    type: string
    description: >
      The job ID of interest whose source indices you want to match against all other jobs.

triggers:
  - type: manual

steps:
  - name: get_target_job
    type: elasticsearch.request
    with:
      method: GET
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}

  - name: log_target
    type: console
    with:
      message: |
        TARGET JOB: {{ inputs.job_id }}
        Source indices: {{ steps.get_target_job.output.jobs[0].datafeed_config.indices }}

        Searching for jobs with overlapping source indices...

  - name: find_related_jobs
    type: foreach
    foreach: "${{ steps.get_target_job.output.jobs[0].datafeed_config.indices }}"
    steps:
      - name: search_matching_datafeeds
        type: elasticsearch.search
        with:
          index: .ml-config
          size: 200
          _source: ["job_id", "datafeed_id", "indices"]
          query:
            bool:
              filter:
                - term:
                    config_type: datafeed
                - term:
                    indices: "{{ foreach.item }}"
              must_not:
                - term:
                    job_id: "{{ inputs.job_id }}"

      - name: log_matches
        type: console
        with:
          message: |
            [{{ foreach.index | plus: 1 }}/{{ foreach.total }}] Index: {{ foreach.item }}
            Matching jobs ({{ steps.search_matching_datafeeds.output.hits.total.value }}):
            {{ steps.search_matching_datafeeds.output.hits.hits | json:2 }}

name: ad_estimate_memory_requirement
description: >
  Compute a principled model_memory_limit estimate by automatically sampling cardinality from source data and calling
  the Estimate Model Memory API. Dramatically better than guessing or peak_model_bytes * 1.3 because it uses the exact
  same estimation algorithm Elasticsearch uses internally. Steps: (1) retrieve job+datafeed config, (2) identify fields
  requiring cardinality estimates, (3) compute overall_cardinality via cardinality aggregations, (4) compute
  max_bucket_cardinality via date_histogram + cardinality + max_bucket pipeline, (5) call Estimate Model Memory API, (6)
  compare with current state, (7) produce recommendation.
enabled: true
tags: ["anomaly-detection", "troubleshooting", "memory"]

inputs:
  - name: job_id
    type: string
    description: The anomaly detection job ID

triggers:
  - type: manual

steps:
  # Step 1: Retrieve job configuration
  - name: get_job_config
    type: elasticsearch.request
    with:
      method: GET
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}

  - name: get_job_stats
    type: elasticsearch.request
    with:
      method: GET
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}/_stats

  # Step 2: Log what we found
  # The agent (or manual reviewer) inspects the config to identify:
  #   - overall_fields: all partition/by/over field names from detectors
  #   - pure_influencers: influencer fields NOT used in any detector split
  #   - datafeed source indices and query
  - name: log_config
    type: console
    with:
      message: |
        Job config retrieved for {{ inputs.job_id }}.
        Analysis config: {{ steps.get_job_config.output | json:2 }}
        Current stats: {{ steps.get_job_stats.output | json:2 }}

        MANUAL STEP REQUIRED:
        1. From analysis_config.detectors, extract all unique by_field_name,
           over_field_name, partition_field_name values → these are "overall_fields"
        2. From analysis_config.influencers, find fields NOT in any detector →
           these are "pure_influencers"
        3. From datafeed_config, get indices[] and query{}
        4. Run cardinality aggregations on those indices for each field
        5. Run date_histogram(bucket_span) + cardinality sub-agg + max_bucket
           pipeline for each pure influencer
        6. Call POST _ml/anomaly_detectors/_estimate_model_memory with the
           analysis_config and computed cardinalities
        7. Compare the estimate with current model_memory_limit and model_bytes

        See tools/workflow/ad_estimate_memory_requirement.json for the full
        step-by-step algorithm.

name: ad_get_calendar_events
description: >
  Get scheduled events from calendars (maintenance windows, holidays). These suppress anomaly detection during known
  downtime periods.
enabled: true
tags: ["anomaly-detection", "config"]

inputs:
  - name: calendar_id
    type: string
    description: The calendar ID

triggers:
  - type: manual

steps:
  - name: get_events
    type: elasticsearch.request
    with:
      method: GET
      path: /_ml/calendars/{{ inputs.calendar_id }}/events

  - name: result
    type: console
    with:
      message: |
        Calendar events for {{ inputs.calendar_id }}:
        {{ steps.get_events.output | json:2 }}

name: ad_get_job_datafeed_config
description: >
  Fetch complete job and datafeed configuration in one call: detectors, by/over/partition fields, bucket_span,
  frequency, query_delay, delayed_data_check_config, source indices, and datafeed query. Essential for troubleshooting
  and for ad_estimate_memory_requirement.
enabled: true
tags: ["anomaly-detection", "config"]

inputs:
  - name: job_id
    type: string
    description: The anomaly detection job ID

triggers:
  - type: manual

steps:
  - name: get_job
    type: elasticsearch.request
    with:
      method: GET
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}

  - name: get_job_stats
    type: elasticsearch.request
    with:
      method: GET
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}/_stats

  - name: summary
    type: console
    with:
      message: |
        Job: {{ inputs.job_id }}
        Config: {{ steps.get_job.output | json:2 }}
        Stats: {{ steps.get_job_stats.output | json:2 }}

name: ad_get_log_categories
description: >
  Retrieve ML log category details — terms, regex pattern, and example messages — for a specific category from a log
  categorization job. Use when investigating anomalies where by_field_name == "mlcategory" to understand what type of
  log message the category represents before comparing samples across time windows.
enabled: true
tags: ["anomaly-detection", "log-categorization"]

inputs:
  - name: job_id
    type: string
    description: The categorization job ID
  - name: category_id
    type: string
    description: "The category ID from the anomaly record's by_field_value"

triggers:
  - type: manual

steps:
  - name: get_categories
    type: elasticsearch.request
    with:
      method: GET
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}/results/categories/{{ inputs.category_id }}

  - name: result
    type: console
    with:
      message: |
        Log category {{ inputs.category_id }} for job {{ inputs.job_id }}:
        {{ steps.get_categories.output | json:2 }}

name: ad_manage_datafeed
description: >
  Start or stop a datafeed via POST _ml/datafeeds/{id}/{_start|_stop}. Used in remediation sequences (for example stop
  before updating query_delay, then restart). For payload preview use ad_preview_datafeed_with_latency (GET
  _ml/datafeeds/{id}/_preview); preview is not supported here because it requires GET, not POST.
enabled: true
tags: ["anomaly-detection", "remediation"]

inputs:
  - name: datafeed_id
    type: string
    description: "The datafeed ID (typically 'datafeed-{job_id}')"
  - name: action
    type: string
    description: "Action to perform: _start or _stop only"

triggers:
  - type: manual

steps:
  - name: manage_datafeed
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/datafeeds/{{ inputs.datafeed_id }}/{{ inputs.action }}

  - name: result
    type: console
    with:
      message: |
        Datafeed {{ inputs.datafeed_id }} action {{ inputs.action }}: {{ steps.manage_datafeed.output | json:2 }}

name: ad_open_job
description: >
  Open an anomaly detection job so it can receive data and run analysis (POST _ml/anomaly_detectors/{job_id}/_open).
enabled: true
tags: ["anomaly-detection", "management"]

inputs:
  - name: job_id
    type: string
    description: The anomaly detection job ID

triggers:
  - type: manual

steps:
  - name: open_job
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}/_open

  - name: result
    type: console
    with:
      message: |
        Opened job {{ inputs.job_id }}:
        {{ steps.open_job.output | json:2 }}

name: ad_preview_datafeed_with_latency
description: >
  Preview a datafeed's source payload and measure effective latency before tuning query_delay. Shows what data the
  datafeed would see at query time, helping identify fields available for latency measurement.
enabled: true
tags: ["anomaly-detection", "troubleshooting"]

inputs:
  - name: datafeed_id
    type: string
    description: The datafeed ID to preview

triggers:
  - type: manual

steps:
  - name: preview_datafeed
    type: elasticsearch.request
    with:
      method: GET
      path: /_ml/datafeeds/{{ inputs.datafeed_id }}/_preview

  - name: result
    type: console
    with:
      message: |
        Datafeed preview for {{ inputs.datafeed_id }}:
        {{ steps.preview_datafeed.output | json:2 }}

name: ad_revert_model_snapshot
description: >
  Revert a job's model to a previous snapshot to 'unlearn' bad data. Stops the datafeed, closes the job, reverts to the
  snapshot, reopens, and restarts the datafeed from the snapshot timestamp.
enabled: true
tags: ["anomaly-detection", "remediation"]

inputs:
  - name: job_id
    type: string
    description: The anomaly detection job ID
  - name: snapshot_id
    type: string
    description: The snapshot ID to revert to

triggers:
  - type: manual

steps:
  - name: stop_datafeed
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/datafeeds/datafeed-{{ inputs.job_id }}/_stop

  - name: close_job
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}/_close

  - name: revert_snapshot
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}/model_snapshots/{{ inputs.snapshot_id }}/_revert

  - name: open_job
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}/_open

  - name: start_datafeed
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/datafeeds/datafeed-{{ inputs.job_id }}/_start

  - name: result
    type: console
    with:
      message: |
        Reverted {{ inputs.job_id }} to snapshot {{ inputs.snapshot_id }}.
        Job reopened and datafeed restarted from snapshot timestamp.

name: ad_search_log_category_examples
description: >
  Search a source log index for messages matching specific ML category terms within a time window. Returns concrete log
  examples belonging to a category during a specific period. Call twice — once for the anomaly window, once for a
  baseline period (e.g. 24h prior) — to compare log content and identify what changed in the variable parts (IPs,
  hostnames, error codes) that may reveal the root cause.
enabled: true
tags: ["anomaly-detection", "log-categorization", "evidence"]

inputs:
  - name: source_index
    type: string
    description: "The source log index (from ad_get_job_datafeed_config)"
    default: "it_ops_logs"
  - name: search_terms
    type: string
    description: "Category terms copied from ad_get_log_categories output"
  - name: start_time
    type: string
    description: "Start of the time window (ISO 8601)"
  - name: end_time
    type: string
    description: "End of the time window (ISO 8601)"

triggers:
  - type: manual

steps:
  - name: search_logs
    type: elasticsearch.search
    with:
      index: "{{ inputs.source_index }}"
      size: 20
      sort: "@timestamp:desc"
      query:
        bool:
          must:
            - match:
                message:
                  query: "{{ inputs.search_terms }}"
                  minimum_should_match: "70%"
          filter:
            - range:
                "@timestamp":
                  gte: "{{ inputs.start_time }}"
                  lte: "{{ inputs.end_time }}"

  - name: result
    type: console
    with:
      message: |
        Log examples matching category terms in {{ inputs.source_index }}
        Time window: {{ inputs.start_time }} to {{ inputs.end_time }}
        Results: {{ steps.search_logs.output.hits.hits | json:2 }}

name: ad_ts_ccs_diagnostics
description: >
  Diagnose cross-cluster search (CCS) issues for datafeeds that query remote clusters. Checks remote cluster
  connectivity, latency, and error rates. Helps identify per-cluster skew contributing to missing data.
enabled: true
tags: ["anomaly-detection", "troubleshooting", "ccs"]

triggers:
  - type: manual

steps:
  - name: check_remote_clusters
    type: elasticsearch.request
    with:
      method: GET
      path: /_remote/info

  - name: check_cluster_health
    type: elasticsearch.request
    with:
      method: GET
      path: /_cluster/health

  - name: result
    type: console
    with:
      message: |
        Remote clusters: {{ steps.check_remote_clusters.output | json:2 }}
        Cluster health: {{ steps.check_cluster_health.output | json:2 }}

name: ad_update_datafeed_query_delay
description: >
  Update the query_delay setting on a datafeed. The datafeed must be stopped first. Larger query_delay captures more
  late-arriving data but delays anomaly alerts. Recommended: set to P95 ingest latency + buffer.
enabled: true
tags: ["anomaly-detection", "remediation"]

inputs:
  - name: datafeed_id
    type: string
    description: The datafeed ID
  - name: new_query_delay
    type: string
    description: "New query_delay value (e.g., '3m', '120s', '5m')"

triggers:
  - type: manual

steps:
  - name: stop_datafeed
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/datafeeds/{{ inputs.datafeed_id }}/_stop

  - name: update_query_delay
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/datafeeds/{{ inputs.datafeed_id }}/_update
      body:
        query_delay: "{{ inputs.new_query_delay }}"

  - name: start_datafeed
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/datafeeds/{{ inputs.datafeed_id }}/_start

  - name: result
    type: console
    with:
      message: |
        Updated query_delay on {{ inputs.datafeed_id }} to {{ inputs.new_query_delay }}.
        Datafeed restarted.

name: ad_update_delayed_data_check_config
description: >
  Update the delayed_data_check_config on a job to control how aggressively delayed data is detected. POST nests under
  analysis_config.delayed_data_check_config. A data.parseJson step builds JSON so enabled is a boolean (not a quoted
  string) and check_window is omitted when the input is empty or whitespace-only after trim.
enabled: true
tags: ["anomaly-detection", "remediation"]

inputs:
  - name: job_id
    type: string
    description: The anomaly detection job ID
  - name: enabled
    type: boolean
    description: Whether to enable delayed data checks
  - name: check_window
    type: string
    description: "Time window to check for delayed data (e.g., '2h'). Leave empty, omit, or whitespace-only to exclude check_window from the update body."

triggers:
  - type: manual

steps:
  - name: compose_payload
    type: data.parseJson
    source: |
      {%- assign cw = inputs.check_window | default: "" | strip -%}
      {%- if cw != "" -%}
      {"analysis_config":{"delayed_data_check_config":{"enabled":{{ inputs.enabled }},"check_window":{{ cw | json }}}}
      {%- else -%}
      {"analysis_config":{"delayed_data_check_config":{"enabled":{{ inputs.enabled }}}}
      {%- endif -%}
    with: {}

  - name: update_delayed_data
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}/_update
      body: "${{ steps.compose_payload.output }}"

  - name: result
    type: console
    with:
      message: |
        Updated delayed_data_check_config on {{ inputs.job_id }}:
        {{ steps.update_delayed_data.output | json:2 }}

name: ad_update_model_memory_limit
description: >
  Remediation workflow to change analysis_limits.model_memory_limit on an anomaly detector (for example after
  ad_estimate_memory_requirement). Stops the datafeed, closes the job, POSTs /_ml/anomaly_detectors/{job_id}/_update with
  the new limit, opens the job, and starts the datafeed again. Expects the default datafeed id datafeed-{job_id}. You
  cannot decrease model_memory_limit below current model_bytes — clone the job to shrink.
enabled: true
tags: ["anomaly-detection", "remediation"]

inputs:
  - name: job_id
    type: string
    description: The anomaly detection job ID
  - name: model_memory_limit
    type: string
    description: "New model_memory_limit value (e.g., '512mb', '1gb')"

triggers:
  - type: manual

steps:
  - name: stop_datafeed
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/datafeeds/datafeed-{{ inputs.job_id }}/_stop

  - name: close_job
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}/_close

  - name: update_model_memory_limit
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}/_update
      body:
        analysis_limits:
          model_memory_limit: "{{ inputs.model_memory_limit }}"

  - name: open_job
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}/_open

  - name: start_datafeed
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/datafeeds/datafeed-{{ inputs.job_id }}/_start

  - name: result
    type: console
    with:
      message: |
        Updated model_memory_limit on {{ inputs.job_id }} to {{ inputs.model_memory_limit }}.
        Job reopened and datafeed restarted.

name: ad_validate_job_spec
description: >
  Validate an anomaly detection job configuration before creation. POSTs to the ML validate endpoint with the same JSON
  document shape as PUT job creation. Requires manage_ml (cluster).
enabled: true
tags: ["anomaly-detection", "management"]

inputs:
  - name: job_body
    type: string
    description: >
      Full job configuration as JSON text (same shape as job creation). Parsed with json_parse so the POST body is a
      structured object for _validate, not a JSON string value.

triggers:
  - type: manual

steps:
  - name: validate_job
    type: elasticsearch.request
    with:
      method: POST
      path: /_ml/anomaly_detectors/_validate
      body: "${{ inputs.job_body | json_parse }}"

  - name: result
    type: console
    with:
      message: |
        Validation result:
        {{ steps.validate_job.output | json:2 }}

name: ad_validate_ml_tool_permissions
description: >
  Preflight check for core ML result and config indices via _has_privileges. Verifies read + view_index_metadata on
  .ml-anomalies-*, .ml-config, .ml-annotations-*, and .ml-notifications-*. Does not check privileges on job-specific
  source data indices — validate those separately before ad_rca_source_evidence, datafeed preview, or ad_wf_ts_field_cardinality.
enabled: true
tags: ["anomaly-detection", "diagnostics"]

triggers:
  - type: manual

steps:
  - name: check_security
    type: elasticsearch.request
    with:
      method: GET
      path: /_security/_authenticate

  - name: check_ml_privileges
    type: elasticsearch.request
    with:
      method: POST
      path: /_security/user/_has_privileges
      body:
        index:
          - names: [".ml-anomalies-*"]
            privileges: ["read", "view_index_metadata"]
          - names: [".ml-config"]
            privileges: ["read", "view_index_metadata"]
          - names: [".ml-annotations-*"]
            privileges: ["read", "view_index_metadata"]
          - names: [".ml-notifications-*"]
            privileges: ["read", "view_index_metadata"]

  - name: result
    type: console
    with:
      message: |
        User: {{ steps.check_security.output.username }}
        Roles: {{ steps.check_security.output.roles | json }}
        ML index privileges: {{ steps.check_ml_privileges.output | json:2 }}

name: ad_wf_troubleshoot_anomaly_score
description: >
  Stored workflow for troubleshooting unexpectedly high or low anomaly scores. Implements a branching decision tree: (0)
  gate checks (sufficient data, memory status, delayed data, UI aggregation), (1) UI display vs real score
  (renormalization), (2) job configuration analysis (bucket_span, detector function, partition/influencer
      fields, custom rules),
  (3) model learning and data characteristics (insufficient history, high variance
      penalty, model adaptation),
  (4) score factor education (anomaly_score_explanation breakdown). Trigger: 'Why is my score low?', 'Expected anomaly
  not detected', 'Score too high/low'.
enabled: true
tags: ["anomaly-detection", "troubleshooting", "scores"]

inputs:
  - name: job_id
    type: string
    description: The anomaly detection job ID
  - name: record_timestamp
    type: string
    description: "Optional: ISO 8601 timestamp of the specific anomaly record to investigate"

triggers:
  - type: manual

steps:
  # Gate check 0a: Has the job processed enough data?
  - name: get_job_stats
    type: elasticsearch.request
    with:
      method: GET
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}/_stats

  - name: gate_check
    type: console
    with:
      message: |
        GATE CHECKS for {{ inputs.job_id }}:
        Job stats: {{ steps.get_job_stats.output | json:2 }}

        CHECK 1 - Sufficient data:
          Model needs >= 3 weeks for weekly seasonality, >= 2 full cycles.
          Check data_counts.processed_record_count and earliest_record_timestamp.

        CHECK 2 - Memory status:
          If memory_status is soft_limit or hard_limit, the model is degraded.
          Fix memory first before investigating scores.

        CHECK 3 - Delayed data:
          If the job has delayed data warnings, bucket scores may be based on
          incomplete data. Check .ml-annotations-* for delayed data annotations.

  # Step 1: Compare record_score vs initial_record_score
  - name: get_anomaly_records
    type: elasticsearch.search
    with:
      index: .ml-anomalies-*
      size: 10
      sort: "record_score:desc"
      query:
        bool:
          filter:
            - term:
                job_id: "{{ inputs.job_id }}"
            - term:
                result_type: record

  - name: score_comparison
    type: console
    with:
      message: |
        SCORE ANALYSIS for {{ inputs.job_id }}:
        Top records: {{ steps.get_anomaly_records.output.hits.hits | json:2 }}

        Compare initial_record_score vs record_score:
        - If initial >> current: renormalization lowered the score after more
          extreme anomalies appeared later. This is EXPECTED behavior.
        - initial_record_score is the score at detection time.
        - record_score is the current (renormalized) score.

  # Step 2: Get job configuration for analysis
  - name: get_job_config
    type: elasticsearch.request
    with:
      method: GET
      path: /_ml/anomaly_detectors/{{ inputs.job_id }}

  - name: config_analysis
    type: console
    with:
      message: |
        JOB CONFIGURATION ANALYSIS:
        Config: {{ steps.get_job_config.output | json:2 }}

        Check these factors:
        1. bucket_span: Too large → dilutes anomalies. Too small → noisy.
        2. Detector function: mean vs high_mean vs low_mean affects directionality.
        3. Partition fields: High cardinality partitions split the model thin.
        4. custom_rules: May be suppressing valid anomalies.
        5. use_null: If false (default), missing entities produce no anomalies.

  # Step 3-4: Score factor explanation
  - name: score_education
    type: console
    with:
      message: |
        ANOMALY SCORE FACTORS (from anomaly_score_explanation):

        1. anomaly_length: How many consecutive buckets are anomalous.
           Longer sequences → higher scores.

        2. single_bucket_impact: How extreme this single bucket is.
           Driven by probability (lower p → higher impact).

        3. multi_bucket_impact: Positive (0-5) when anomaly spans multiple
           buckets. Values >= 3 suggest genuine behavioral shift.

        4. anomaly_characteristics_impact: Nature of the anomaly (mean shift
           vs variance change).

        5. high_variance_penalty: REDUCES score when the model's confidence
           bounds are wide. Common early in model training or with noisy data.
           Wide bounds → model is uncertain → anomalies appear less surprising.

        6. incomplete_bucket_penalty: REDUCES score when the bucket doesn't
           have the expected amount of data (e.g., due to delayed data).

        To see these factors, query the specific record from .ml-anomalies-*
        and inspect the anomaly_score_explanation field.

name: ad_wf_ts_field_cardinality
description: >
  Estimate cardinality of a split field (by_field, over_field, partition_field) in SOURCE data via ES|QL
  POST /_query. The column for COUNT_DISTINCT is interpolated into the query text (not a ? parameter - those bind only
  literals). Pass split_field_esql as a valid ES|QL column reference (for example service.keyword or `host.name.keyword`)
  from ad_get_job_datafeed_config. Compare distinct_count with total_*_count from ad_ts_model_memory_health; if source
  cardinality is much larger, entities may be dropped. For CCS, run per cluster and sum. Prefer ad_estimate_memory_requirement
  for full sizing; this workflow answers how many distinct values the field has in the window.
enabled: true
tags: ["anomaly-detection", "diagnostics"]

inputs:
  - name: source_index
    type: string
    description: >
      Source index name or LIKE pattern (* wildcard). From the job datafeed config (same as ES|QL FROM * METADATA _index filter).
  - name: split_field_esql
    type: string
    description: >
      Exact ES|QL column expression for COUNT_DISTINCT (not quoted as a string). Examples: service.keyword, `host.name.keyword`.
      Must match a field on matched documents; use only values taken from job analysis config to avoid query injection.
  - name: start_time
    type: string
    description: Start of time range in ISO 8601 format
  - name: end_time
    type: string
    description: End of time range in ISO 8601 format

triggers:
  - type: manual

steps:
  - name: cardinality_esql
    type: elasticsearch.request
    with:
      method: POST
      path: /_query
      body:
        query: "FROM * METADATA _index | WHERE _index LIKE ?source_index AND @timestamp >= ?start_time AND @timestamp <= ?end_time | STATS distinct_count = COUNT_DISTINCT({{ inputs.split_field_esql }}) | LIMIT 1"
        params:
          source_index: "{{ inputs.source_index }}"
          start_time: "{{ inputs.start_time }}"
          end_time: "{{ inputs.end_time }}"

  - name: result
    type: console
    with:
      message: |
        Split-field cardinality ({{ inputs.split_field_esql }}) on indices matching {{ inputs.source_index }} ({{ inputs.start_time }}-{{ inputs.end_time }}):
        {{ steps.cardinality_esql.output | json:2 }}

{
  "default_agent_id": "elastic-ai-agent",
  "default_agent_enable_elastic_capabilities": true,
  "workflow_tool_exclusions": [
    "ad_get_job_datafeed_config",
    "ad_ts_ccs_diagnostics",
    "ad_get_calendar_events",
    "ad_create_calendar_event",
    "ad_discover_jobs_by_datafeed_index"
  ],
  "fallback_tools": {
    "ad_get_index_mappings": "platform.core.get_index_mapping"
  },
  "workflow_prefixes": [
    "ad_wf_",
    "ad_create_",
    "ad_manage_",
    "ad_open_",
    "ad_update_",
    "ad_revert_",
    "ad_preview_",
    "ad_validate_",
    "ad_estimate_"
  ],
  "builtin_tools": [
    "platform.core.search",
    "platform.core.list_indices",
    "platform.core.get_index_mapping",
    "platform.core.execute_esql",
    "platform.core.generate_esql",
    "platform.core.product_documentation"
  ]
}