From 1bb8dbc78794e140bfd202558dabd6ebbda3f45a Mon Sep 17 00:00:00 2001 From: Andrew Macri Date: Fri, 11 Sep 2026 16:37:41 -0400 Subject: [PATCH] ## [AlertZero] Attack Discovery Worker: workflow skeleton and step wiring MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This PR implements [#19276](https://github.com/elastic/security-team/issues/19276) to replace the Attack Discovery worker's `console` stub with two new managed workflows: 1) `Attack Discovery Runner` - Runs Attack Discovery generation, then fans out the generated discoveries to the review workflow 2) `Attack Discovery Review` - Reviews one Attack Discovery: opens an Investigation, runs FP/TP analysis, and applies the resulting verdict Both of the workflows above have stub implementations of their steps. 3) This PR removes the Attack discovery run workflow step's 90s soft deadline. Sync mode now runs to completion instead of returning status: 'pending' at 90s, so a workflow calling this step needs a timeout above the pipeline's 30m budget (`run_example.yaml` updated from 10m to 35m). ### How it works #### 1) Users configure the Attack discovery worker's schedule in AlertZero ![attack_discovery_worker_config ](https://github.com/user-attachments/assets/fc03d155-b610-481b-8b56-5d3465b17f05) _Above: The Attack Discovery worker's schedule config_ #### 2) The per-space `Attack Discovery` workflow is invoked by the schedule It `workflow.execute`s the `Attack Discovery Runner` workflow: ![attack_discovery_worker](https://github.com/user-attachments/assets/b38ad96c-016b-4443-8c14-f87e425ad5ba) _Above: The scheduled `floor_attack_discovery.yaml` workflow executes the `Attack Discovery Runner` workflow_ #### 3) The `Attack Discovery Runner` workflow generates discoveries and batches reviews The `Attack Discovery Runner` workflow generates discoveries via the `security.attack-discovery.run` step. After generating the discoveries, the `Attack Discovery Runner` workflow fans out each generated discovery to a separate invocation of the `Attack Discovery Review` workflow: ![attack_discovery_runner](https://github.com/user-attachments/assets/a2961de9-19d0-476a-b8dc-77ec3ef1551d) _Above: The `attack_discovery_runner.yaml` workflow generates discoveries and invokes the `Attack Discovery Review` workflow for each generation, in batches of parallel runs_ The following example shows the runner generated 8 discoveries, and invoked 8 review workflows: ![run_result](https://github.com/user-attachments/assets/722c11a3-8ae3-4194-9052-e66af39cf25d) _Above: The runner invoked 8 review workflows in paralell_ #### 4) The `Attack Discovery Review` workflow reviews one Attack Discovery: opens an Investigation, runs FP/TP analysis, and applies the resulting verdict ![attack_discovery_review](https://github.com/user-attachments/assets/8fbc507f-74e0-4a4a-bd34-09f32d4ea5d2) _Above: The `attack_discovery_review.yaml` workflow opens an Investigation, runs FP/TP analysis, and applies the resulting verdict_ An example of a single review result is shown below: ![example_review_result](https://github.com/user-attachments/assets/fb775a68-23e9-4443-9bb6-ecd8309ab6d1) _Above: An example review result_ ### Feature flags The following entries in `config/kibana.dev.yml` are required for desk testing: ```yaml xpack.alertzero.enabled: true xpack.agenticInvestigations.enabled: true uiSettings.overrides: securitySolution:enableAttackDiscoveryWorkflows: true workflows:ui:showManagedWorkflows: true ``` ### Out of scope - Alert batching for generation — `#18967` - The real FP/TP analysis workflow — `#19211` - Investigation creation logic — `#19022` - Outcome application — `#19214` _Created with Cursor, Opus 5, Grok 4.6_ --- .../alertzero/attack_discovery_review.yaml | 204 ++++++ .../alertzero/attack_discovery_runner.yaml | 238 +++++++ .../attack_discovery_workflows.test.ts | 635 ++++++++++++++++++ .../alertzero/attack_discovery_workflows.ts | 41 ++ .../alertzero/floor_attack_discovery.ts | 2 +- .../alertzero/floor_attack_discovery.yaml | 41 +- .../managed/definitions/alertzero/index.ts | 20 + .../managed/definitions/discoveries/index.ts | 2 +- .../definitions/discoveries/run_example.yaml | 12 +- .../managed/definitions/index.ts | 7 + .../managed_workflow_definitions.test.ts | 2 +- .../initialize_managed_workflows.test.ts | 7 +- .../initialize_managed_workflows.ts | 9 +- .../security/plugins/discoveries/README.md | 28 +- .../discoveries/common/step_types/run_step.ts | 8 +- .../server/workflows/steps/run_step/README.md | 11 +- .../workflows/steps/run_step/constants.ts | 25 +- .../run_step/get_run_step_definition.test.ts | 71 +- .../steps/run_step/get_run_step_definition.ts | 119 +--- 19 files changed, 1294 insertions(+), 188 deletions(-) create mode 100644 src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_review.yaml create mode 100644 src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_runner.yaml create mode 100644 src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_workflows.test.ts create mode 100644 src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_workflows.ts diff --git a/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_review.yaml b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_review.yaml new file mode 100644 index 0000000000000..c19a7522f0682 --- /dev/null +++ b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_review.yaml @@ -0,0 +1,204 @@ +version: "1" +name: Attack Discovery Review +description: > + Reviews one Attack Discovery: opens an Investigation, runs FP/TP analysis, and + applies the resulting verdict. + + Every action in this workflow is a console stub; the real steps land in the referenced issues. +enabled: true +tags: + - security + - attack-discovery +triggers: + # Manual only: this workflow is always launched by the Attack Discovery worker + # (or by hand for testing). It owns no schedule of its own. + - type: manual + inputs: + properties: + title: + type: string + maxLength: 1024 + description: Title of the Attack Discovery under review. + alert_ids: + type: array + maxItems: 1000 + items: + type: string + maxLength: 512 + description: >- + Document ids of the DETECTION alerts the Attack Discovery correlated. + Not the Attack Discovery alert's `_id`: the run step currently returns + the persist handover, so no persisted document id is available here. + #19022 will consume that id once the run step returns persist-projected + discoveries with document ids in-context. Do not treat the handover's + optional `id` as that value — it is the LLM UUID, not the ES `_id`. + summary_markdown: + type: string + maxLength: 8000 + description: Summary of the Attack Discovery, shown to the analyst and the analysis agent. + autonomy: + type: string + enum: + - manual + - assisted + - supervised + description: >- + Worker autonomy level, which decides how a true-positive or + inconclusive verdict is promoted: `manual` requires an analyst gate, + the others promote without one. Defaults to `manual`. + stub_verdict: + type: string + enum: + - false_positive + - true_positive + - inconclusive + - failed + description: >- + Stands in for the verdict the real FP/TP analysis workflow will return + (#19211). Defaults to + `inconclusive`, which #19214 routes to the same Promote-to-Incident + proposal as true_positive. Overridable by hand so every branch of the + switch below is reachable before the real analysis exists. + parent_run_id: + type: string + maxLength: 128 + description: >- + Workflow execution id of the runner that launched this review, logged by + the investigation step below. It is the only link back to the parent the + engine keeps. Empty on a standalone run. + # Only the attack's identity is required. Everything else defaults from + # `consts` so the worker and this workflow can be upgraded independently + # without a one-sided input-validation failure. + required: + - title + additionalProperties: false +outputs: + - name: title + type: string + - name: verdict + type: string + - name: investigation_opened + type: boolean +consts: + default_verdict: inconclusive + default_autonomy: manual + # Liquid has no array literal, so `| default: []` yields undefined rather than an + # empty array. A const is the only way to spell an empty-array fallback, which is + # what keeps `| size` from erroring when `alert_ids` is omitted. + no_alert_ids: [] +steps: + # STUB — #19022 replaces this + # with `ai.conversation.create` using `template_id: investigation`. Left as a + # console step here on purpose: a real call would open one Investigation per + # attack per Worker run, which is an unbounded side effect for a shape-only + # workflow and the opposite of #19022 (one Investigation per attack, never per + # run). A repeated grouping of the same alerts must not open a second + # Investigation while the first is still open. Dedup is supposed to happen + # upstream at Attack Discovery persistence via the attack hash, so a duplicate + # should not reach this step; confirming that (or adding a Worker-side check) + # is part of #19022. + - name: open_investigation + type: console + with: + message: >- + STUB open_investigation (#19022): would open one Investigation for Attack + Discovery "{{ inputs.title }}" covering + {{ inputs.alert_ids | default: consts.no_alert_ids | size }} detection alert(s). + Parent worker run: {{ inputs.parent_run_id }} + + # STUB — #19211 replaces this + # with the real AlertZero FP/TP analysis workflow, which generates a + # corroborating-evidence report and returns the verdict this workflow switches + # on. Its own built-in retries live in that workflow, not here. + - name: run_fp_tp_analysis + type: console + with: + message: >- + STUB run_fp_tp_analysis (#19211): would run the AlertZero FP/TP analysis + workflow for "{{ inputs.title }}" and return a verdict. + Summary under review: {{ inputs.summary_markdown }} + + # One explicit case per known analysis outcome so an unknown value cannot + # silently take an action — it matches nothing and lands in the default arm. + # Every branch is console-only: applying these outcomes for real is #19214. + # Verdict and autonomy read `inputs` with `| default` rather than a `data.set` + # step: `data.set` outputs are untyped, and the editor warns on every + # `steps.*.output.*` read. + - name: apply_verdict + type: switch + expression: "{{ inputs.stub_verdict | default: consts.default_verdict }}" + cases: + - match: false_positive + steps: + - name: close_as_false_positive + type: console + with: + message: >- + STUB false_positive (#19214): would close the Investigation for + "{{ inputs.title }}" and mark the Attack Discovery as false_positive. + - match: true_positive + steps: + # The workflow's Incident promotion decision. Under `manual` autonomy + # this becomes an analyst gate that can accept (promote) or reject + # (close); under the other levels it promotes without a gate. Both routes + # are one console line here. + - name: promote_per_autonomy + type: console + with: + message: >- + STUB true_positive (#19214): autonomy is + {{ inputs.autonomy | default: consts.default_autonomy }}, so this would + {% assign autonomy = inputs.autonomy | default: consts.default_autonomy %} + {% if autonomy == 'manual' %}present an + analyst gate that promotes the Investigation for "{{ inputs.title }}" + to an Incident on accept and closes it on reject{% else %}promote the + Investigation for "{{ inputs.title }}" to an Incident with no + gate{% endif %}. + - match: inconclusive + steps: + # Same Promote-to-Incident HITL as true_positive + - name: promote_inconclusive_per_autonomy + type: console + with: + message: >- + STUB inconclusive (#19214): would take the same Promote-to-Incident + path as true_positive. Autonomy is + {{ inputs.autonomy | default: consts.default_autonomy }}, so this would + {% assign autonomy = inputs.autonomy | default: consts.default_autonomy %} + {% if autonomy == 'manual' %}present an + analyst gate that promotes the Investigation for "{{ inputs.title }}" + to an Incident on accept and closes it on reject{% else %}promote the + Investigation for "{{ inputs.title }}" to an Incident with no + gate{% endif %}. Not handed to Forensics. + - match: failed + steps: + - name: record_analysis_failure + type: console + with: + message: >- + STUB failed (#19214): would record the failed verdict and its error + explanation on the Investigation for "{{ inputs.title }}" and leave + it open. No additional HITL gate. + default: + # Unreachable through the four cases above; only a verdict outside the enum + # lands here. Reported rather than ignored so a contract drift after #19211 + # is visible instead of silent. + - name: report_unknown_verdict + type: console + with: + message: >- + Attack Discovery review for "{{ inputs.title }}" received an unexpected + verdict "{{ inputs.stub_verdict | default: consts.default_verdict }}" and took no action. + + # The worker reads this payload back per branch, so the contract has to be + # explicit: `workflow.execute` resolves to the child's `context.output`, and a + # child without a `workflow.output` step hands its parent nothing. + - name: emit_result + type: workflow.output + status: completed + with: + title: "{{ inputs.title }}" + verdict: "{{ inputs.stub_verdict | default: consts.default_verdict }}" + # Always true while open_investigation is a console stub. Becomes a real + # assertion about the created Investigation in #19022. + investigation_opened: true diff --git a/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_runner.yaml b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_runner.yaml new file mode 100644 index 0000000000000..9c4d660fac505 --- /dev/null +++ b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_runner.yaml @@ -0,0 +1,238 @@ +version: "1" +name: Attack Discovery Runner +description: > + Runs Attack Discovery generation, then fans out one review workflow per generated + attack so every attack is reviewed concurrently on its own execution. Launched by + the per-space Attack Discovery workflow, which owns the schedule. +enabled: true +tags: + - security + - attack-discovery +settings: + # The space is in the key only so the group reads clearly in execution history: + # concurrency is always checked within a single space. + # `cancel-in-progress` with max 1: the floor Watch Worker uses the same strategy, + # and a new floor dispatch would otherwise be SKIPPED while this slot is still + # held by the runner being cancelled. Latest run wins. The in-flight generation + # is cancelled as cleanly as the engine allows (the run step does not abort the + # LLM pipeline mid-call). Must be an object — unlike a `parallel` step's + # `concurrency`, workflow-level `settings.concurrency` does not accept a bare number. + concurrency: + key: "attack-discovery-worker-{{ workflow.spaceId }}" + strategy: cancel-in-progress + max: 1 +triggers: + # The per-space Attack Discovery workflow owns the schedule. This global workflow + # stays manual-only so there is one scheduler authority and no cross-space task + # collision, and so the whole chain stays exercisable by hand. + - type: manual + inputs: + properties: + connector_id: + type: string + maxLength: 512 + description: >- + LLM connector for generation. Omit to let the run step resolve the + space's default AI connector. + autonomy: + type: string + enum: + - manual + - assisted + - supervised + description: >- + Worker autonomy level, passed through to each review so a true-positive + or inconclusive verdict knows whether it needs an analyst gate. + Defaults to `manual`. + settings_version: + type: integer + minimum: 1 + description: >- + Version of the worker settings that launched this run. Recorded for + traceability only; no step reads it. + schedule_interval: + type: string + maxLength: 32 + description: >- + Schedule interval of the worker that launched this run (for example + `24h`). Recorded for traceability only; no step reads it. Echoed in the + output so a run's cadence is visible in its execution history. + parent_run_id: + type: string + maxLength: 128 + description: >- + Workflow execution id of the per-space Attack Discovery run that launched + this run. No step reads it. It stays readable in this execution's inputs, + which is the only link back to the parent the engine keeps. Empty on a + standalone run. + additionalProperties: false +outputs: + # Declared as string because the editor type-checks the YAML source of + # `workflow.output` `with:` values. A `type: number` field whose value is + # `"${{ ... }}"` is reported as invalid even though Liquid keeps the number + # at runtime. `{{ }}` stringifies so the emitted value matches this schema; + # `${{ }}` would keep the count as a number and fail output validation. + - name: attacks_generated + type: string + - name: reviews_requested + type: string + - name: reviews_failed + type: string + - name: alerts_analyzed + type: string + - name: schedule_interval + type: string + # The Attack Discovery generation id, not this workflow's execution id. It keys + # the AD generations API and lands on every persisted discovery as + # `kibana.alert.rule.execution.uuid`, so a run can be traced to what it produced. + - name: execution_uuid + type: string +consts: + # Liquid has no array literal, so `| default: []` yields undefined rather than an + # empty array. This is an EMPTY-ARRAY FALLBACK ONLY — never synthetic discovery + # data. It covers all three empty shapes the run step can produce: + # `attack_discoveries` omitted, `null` (validation failed), or `[]` (genuine zero). + no_discoveries: [] +steps: + # Reuses the existing generation engine unchanged (#19276 AC2): the run step is + # the composite retrieve -> generate -> validate -> persist pipeline. Alert + # batching (https://github.com/elastic/security-team/issues/18967) will replace + # this call site later, so nothing here is designed around it. + # + # `mode: sync` because the fan-out below needs the discoveries inline. Sync mode + # awaits the pipeline to completion and applies no soft deadline. + # + # No `on-failure`: the default propagation is what #19276 requires — when + # generation fails after its own built-in retries, this run fails and hands off + # nothing. Adding `continue: true` would fan out over a partial result. + - name: run_generation + type: security.attack-discovery.run + # 35m = the 30m pipeline budget + 5m. The pipeline enforces its own 30m budget + # internally (DEFAULT_PIPELINE_TIMEOUT_MS / getRemainingBudgetMs in + # kbn-discoveries run_manual_orchestration: ADR-008's gate 10m + generation 10m + # + validation 5m, shared across phases on one wall clock). This timeout must + # stay ABOVE it so the pipeline's own budget wins the race and fails with an + # attributed PipelineStepError, rather than the engine killing the step with an + # opaque TimeoutError while generation keeps running in the background. + # Must stay a literal: the engine hands `timeout` to the duration parser + # unrendered (elastic/kibana#290258). + timeout: '35m' + with: + connector_id: "{{ inputs.connector_id }}" + mode: sync + + # The fan-out list, resolved once so the batches, the parallel step, and the + # summary agree on it. `| default: consts.no_discoveries` normalizes the three + # empty shapes to an empty array. + - name: resolve_fanout + type: data.set + with: + attacks: "${{ steps.run_generation.output.attack_discoveries | default: consts.no_discoveries }}" + # Computed on its own because comparison operators cannot follow a filter chain. + attack_count: "${{ steps.run_generation.output.attack_discoveries | default: consts.no_discoveries | size }}" + # 100 is DEFAULT_PARALLEL_MAX_FAN_OUT: a single `parallel` foreach throws + # above that. `chunk` drains leftovers in later batches. `concurrency.max` + # (not this) bounds claimed Task Manager workers. + batches: "${{ steps.run_generation.output.attack_discoveries | default: consts.no_discoveries | chunk: 100 }}" + + # A run that generated nothing is a normal outcome, not a failure: there may + # simply be no attack in the alerts. Logged explicitly so an empty execution is + # distinguishable from one that failed before generation. + - name: log_empty_run + type: console + if: "${{ steps.resolve_fanout.output.attack_count == 0 }}" + with: + message: >- + Attack Discovery generation produced no attacks + (alerts analyzed: {{ steps.run_generation.output.alerts_context_count }}, + status: {{ steps.run_generation.output.status }}). Skipping the per-attack + fan-out. + + # Failed-review count starts at 0 so an empty generation and the first batch + # both have a number to add to. Each batch writes the running total back here. + - name: init_review_counts + type: data.set + with: + reviews_failed: 0 + + # One review execution per attack. Each `chunk: 100` batch stays under the + # parallel fan-out ceiling so a 101+ generation cannot throw. Branches still + # run at `concurrency.max` (5). The per-attack body cannot live in a parallel + # branch: that body must be a straight-line sequence, so the verdict `switch` + # sits in a child workflow. `current_batch` copies `foreach.item` so the inner + # parallel's `foreach.item` is the attack, not the batch array. + - name: run_review_batches + type: foreach + foreach: "${{ steps.resolve_fanout.output.batches }}" + steps: + - name: current_batch + type: data.set + with: + attacks: "${{ foreach.item }}" + + - name: run_reviews + type: parallel + foreach: "${{ steps.current_batch.output.attacks }}" + # Settled: one failed review must not abandon the remaining attacks. Failures + # stay visible in the aggregate results and counts. + mode: settled + concurrency: + # Well under DEFAULT_PARALLEL_MAX_CONCURRENCY (20), which the schema enforces + # as a validation error rather than clamping. This is the TM bound; it is not + # a cap on how many attacks are reviewed. + max: 5 + # `count-waiting` is left at its default (true). Every branch here parks as + # soon as it starts its child, because `workflow.execute` is synchronous, so + # this bounds in-flight reviews to `max` and a generation larger than `max` + # starts its reviews in successive waves. Revisit alongside #19214, which + # adds HITL gates and keeps each branch parked far longer. + steps: + - name: run_review + type: workflow.execute + with: + workflow-id: system-security-attack-discovery-review + inputs: + # `${{ }}` preserves the value's type (arrays stay arrays); `{{ }}` + # stringifies. The run step returns the persist handover, so + # `alert_ids` holds DETECTION alert ids. There is no persisted Attack + # Discovery `_id` to pass yet. #19022 / #19214 will use it once the + # run step returns persist-projected discoveries with document ids + # in-context. Do not forward `foreach.item.id`: that optional field + # is the LLM UUID, not the document id. + title: "{{ foreach.item.title }}" + alert_ids: "${{ foreach.item.alert_ids }}" + summary_markdown: "{{ foreach.item.summary_markdown }}" + # `${{ }}` so the editor does not type-check the template source + # against the child's autonomy enum. + autonomy: "${{ inputs.autonomy }}" + parent_run_id: "{{ execution.id }}" + + - name: batch_failed + type: data.set + with: + count: "${{ steps.run_reviews.output.failed | default: 0 }}" + + - name: accumulate_failed + type: data.set + with: + reviews_failed: "${{ variables.reviews_failed | plus: steps.batch_failed.output.count }}" + + # Fan-in: every batch has settled. A branch output is the child's + # `workflow.output` payload (title, verdict, investigation_opened). + - name: summarize_run + type: data.set + with: + attacks_generated: "${{ steps.resolve_fanout.output.attack_count }}" + reviews_requested: "${{ steps.resolve_fanout.output.attack_count }}" + reviews_failed: "${{ variables.reviews_failed | default: 0 }}" + + - name: emit_result + type: workflow.output + status: completed + with: + attacks_generated: "{{ steps.summarize_run.output.attacks_generated }}" + reviews_requested: "{{ steps.summarize_run.output.reviews_requested }}" + reviews_failed: "{{ steps.summarize_run.output.reviews_failed }}" + alerts_analyzed: "{{ steps.run_generation.output.alerts_context_count | default: 0 }}" + schedule_interval: "{{ inputs.schedule_interval }}" + execution_uuid: "{{ steps.run_generation.output.execution_uuid }}" diff --git a/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_workflows.test.ts b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_workflows.test.ts new file mode 100644 index 0000000000000..619511848524c --- /dev/null +++ b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_workflows.test.ts @@ -0,0 +1,635 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the "Elastic License + * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side + * Public License v 1"; you may not use this file except in compliance with, at + * your election, the "Elastic License 2.0", the "GNU Affero General Public + * License v3.0 only", or the "Server Side Public License, v 1". + */ + +import { parse } from 'yaml'; +import { + ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW, + ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW_ID, + ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW, + ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW_ID, + ALERTZERO_WORKER_FLOOR_ATTACK_DISCOVERY_WORKFLOW, +} from '.'; +import { createWorkflowLiquidEngine } from '../../../common/utils'; +import { DEFAULT_PARALLEL_MAX_FAN_OUT, WorkflowSchema } from '../../../spec/schema'; + +/** + * Verdicts the FP/TP analysis workflow may return. Each one must have a dedicated + * switch case in the review workflow, so a verdict never falls through to the + * default arm. + */ +const VERDICTS = ['false_positive', 'true_positive', 'inconclusive', 'failed'] as const; + +interface YamlStep { + name: string; + type: string; + if?: string; + timeout?: string; + mode?: string; + foreach?: string; + expression?: string; + concurrency?: number | { max?: number; 'count-waiting'?: boolean }; + with?: Record; + steps?: YamlStep[]; + else?: YamlStep[]; + cases?: Array<{ match: string; steps: YamlStep[] }>; + default?: YamlStep[]; + 'on-failure'?: { continue?: boolean }; +} + +interface YamlWorkflow { + steps: YamlStep[]; + consts?: Record; + outputs?: Array<{ name: string; type?: string }>; + settings?: { concurrency?: unknown; timeout?: string }; + triggers?: Array<{ + type: string; + inputs?: { + properties?: Record; + required?: string[]; + }; + }>; +} + +interface ConcurrencySettings { + key?: string; + max?: number; + strategy?: string; +} + +const flatten = (steps: YamlStep[]): YamlStep[] => + steps.flatMap((step) => [ + step, + ...flatten(step.steps ?? []), + ...flatten(step.else ?? []), + ...flatten(step.default ?? []), + ...(step.cases ?? []).flatMap((c) => flatten(c.steps)), + ]); + +const worker = parse(ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW.yaml) as YamlWorkflow; +const review = parse(ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW.yaml) as YamlWorkflow; + +// The Watch Floor worker is a template, so it has to be rendered before parsing. +const floor = parse( + ALERTZERO_WORKER_FLOOR_ATTACK_DISCOVERY_WORKFLOW.yamlTemplate({ + autonomyLevel: 'manual', + scheduleInterval: '24h', + settingsVersion: 1, + }) +) as YamlWorkflow; + +const workerSteps = flatten(worker.steps); +const reviewSteps = flatten(review.steps); +const floorSteps = flatten(floor.steps); + +const stepIn = (steps: YamlStep[], name: string) => steps.find((step) => step.name === name); + +const asInputs = (step: YamlStep | undefined): Record => + (step?.with?.inputs ?? {}) as Record; + +const asConcurrency = (value: unknown): ConcurrencySettings => (value ?? {}) as ConcurrencySettings; + +describe('Attack Discovery worker chain', () => { + it.each([ + [ + 'worker', + ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW, + ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW_ID, + ], + [ + 'review', + ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW, + ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW_ID, + ], + ])('registers the %s under its expected id', (_name, workflow, id) => { + expect(workflow.id).toBe(id); + }); + + // Managed install only runs lightweight validation, where `steps` is + // `z.array(z.unknown())`, so an authoring error in these files would otherwise + // surface at execution time rather than in CI. This is the assertion that catches it. + it.each([ + ['worker', ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW.yaml], + ['review', ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW.yaml], + [ + 'floor worker', + ALERTZERO_WORKER_FLOOR_ATTACK_DISCOVERY_WORKFLOW.yamlTemplate({ + autonomyLevel: 'manual', + scheduleInterval: '24h', + settingsVersion: 1, + }), + ], + ])('%s passes strict workflow schema validation', (_name, yaml) => { + const result = WorkflowSchema.safeParse(parse(yaml)); + + expect(result.success ? null : result.error.issues).toBeNull(); + }); + + describe('Watch Floor worker dispatch', () => { + const dispatch = stepIn(floorSteps, 'run_attack_discovery'); + + it('replaces the console stub', () => { + expect(floorSteps.map((step) => step.type)).not.toContain('console'); + }); + + it('dispatches to the Attack Discovery worker', () => { + expect(dispatch?.with?.['workflow-id']).toBe(ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW_ID); + }); + + // Only the sync strategy propagates the child's failure. With executeAsync the + // worker run would report success while generation had failed, which #19276 + // explicitly forbids. + it('dispatches with workflow.execute', () => { + expect(dispatch?.type).toBe('workflow.execute'); + }); + + it('does not dispatch with workflow.executeAsync', () => { + expect(floorSteps.map((step) => step.type)).not.toContain('workflow.executeAsync'); + }); + + it('leaves dispatch failure propagation alone', () => { + expect(dispatch?.['on-failure']).toBeUndefined(); + }); + + // #19276 names all three settings. `scheduleInterval` has no behavioral use in the + // skeleton, so dropping it would be easy and would silently break the contract. + it.each([ + ['autonomy', 'consts.worker_settings.autonomy'], + ['settings_version', 'consts.worker_settings.settingsVersion'], + ['schedule_interval', 'consts.worker_settings.scheduleInterval'], + ])('forwards the %s worker setting', (input, reference) => { + expect(asInputs(dispatch)[input]).toContain(reference); + }); + + it('forwards enum and interval settings with type-preserving templates', () => { + expect(asInputs(dispatch).autonomy).toBe('${{ consts.worker_settings.autonomy }}'); + expect(asInputs(dispatch).schedule_interval).toBe( + '${{ consts.worker_settings.scheduleInterval }}' + ); + }); + + it('keeps the scheduled trigger and adds no second scheduler', () => { + expect((floor.triggers ?? []).map((trigger) => trigger.type)).toEqual(['scheduled']); + }); + + // Workflow-level concurrency must be an object: unlike a `parallel` step's + // `concurrency`, ConcurrencySettingsSchema does not accept a bare number. + it('cancels overlapping scheduled floor runs in progress', () => { + expect(asConcurrency(floor.settings?.concurrency).strategy).toBe('cancel-in-progress'); + }); + + it('limits overlapping scheduled floor runs to one', () => { + expect(asConcurrency(floor.settings?.concurrency).max).toBe(1); + }); + }); + + // The worker is installed globally but executes per space. Concurrency is always + // checked within a single space, so the space in the key is for readability rather + // than isolation. `workflow.spaceId` comes from the execution document and is + // resolvable in a concurrency key, unlike templated `inputs.*`, which are evaluated + // after the key. + it('scopes the global worker concurrency key to the space', () => { + expect(asConcurrency(worker.settings?.concurrency).key).toContain('{{ workflow.spaceId }}'); + }); + + it('cancels overlapping global runner runs in progress', () => { + expect(asConcurrency(worker.settings?.concurrency).strategy).toBe('cancel-in-progress'); + }); + + it('limits overlapping global worker runs to one', () => { + expect(asConcurrency(worker.settings?.concurrency).max).toBe(1); + }); + + describe('generation step', () => { + const run = stepIn(workerSteps, 'run_generation'); + + it('reuses the existing generation engine through the run step', () => { + expect(run?.type).toBe('security.attack-discovery.run'); + }); + + it('runs generation in sync mode', () => { + expect(run?.with?.mode).toBe('sync'); + }); + + // The pipeline enforces its own 30m budget internally. A step timeout at or below + // it would win the race and produce an opaque engine kill that does NOT cancel the + // background pipeline, instead of an attributed PipelineStepError. + it('sets a step timeout above the 30m pipeline budget', () => { + expect(run?.timeout).toBe('35m'); + }); + + it('leaves failure propagation alone so a failed generation fails the run', () => { + expect(run?.['on-failure']).toBeUndefined(); + }); + }); + + describe('fan-out', () => { + const resolve = stepIn(workerSteps, 'resolve_fanout'); + const parallel = stepIn(workerSteps, 'run_reviews'); + const branch = (parallel?.steps ?? [])[0]; + const branchInputs = asInputs(branch); + + it('resolves fan-out from the generated discoveries', () => { + expect(resolve?.with?.attacks).toContain('steps.run_generation.output.attack_discoveries'); + }); + + it('fans out over the resolved attacks', () => { + expect(parallel?.foreach).toContain('steps.current_batch.output.attacks'); + }); + + it('copies each batch so the inner foreach.item is the attack', () => { + expect(stepIn(workerSteps, 'current_batch')?.with?.attacks).toBe('${{ foreach.item }}'); + }); + + it('drains discoveries in batches so parallel never exceeds its fan-out ceiling', () => { + expect(stepIn(workerSteps, 'run_review_batches')?.foreach).toContain( + 'steps.resolve_fanout.output.batches' + ); + }); + + // The ONLY permitted fallback is an empty array. A fallback that substituted + // fabricated attacks would make the artifact synthetic, and would later have #19022 + // opening real Investigations for attacks that never existed. + it('falls back to consts.no_discoveries', () => { + expect(resolve?.with?.attacks).toContain('default: consts.no_discoveries'); + }); + + it('defines no_discoveries as an empty array', () => { + expect(worker.consts?.no_discoveries).toEqual([]); + }); + + it('declares no synthetic discovery data anywhere in either workflow', () => { + const constNames = [...Object.keys(worker.consts ?? {}), ...Object.keys(review.consts ?? {})]; + + expect(constNames.filter((name) => /stub_discover|fake|sample_attack/i.test(name))).toEqual( + [] + ); + }); + + it('does not slice discoveries off the fan-out list', () => { + expect(resolve?.with?.attacks).not.toContain('slice:'); + }); + + it('chunks discoveries to DEFAULT_PARALLEL_MAX_FAN_OUT', () => { + expect(resolve?.with?.batches).toContain(`chunk: ${DEFAULT_PARALLEL_MAX_FAN_OUT}`); + }); + + describe('drain of an oversized generation', () => { + // Evaluates the authored `${{ }}` expressions, not a copy of them, so a YAML + // edit that reintroduces a slice or a tighter chunk is a failed test rather + // than a comment drift. This is the data plane of a 150-discovery run; it + // does not claim Task Manager workers. + const liquid = createWorkflowLiquidEngine(); + const overflowCount = DEFAULT_PARALLEL_MAX_FAN_OUT + 50; + const evaluate = (expression: string, context: Record): unknown => + liquid.evalValueSync(expression.slice(3, -2).trim(), context); + const contextFor = (attackDiscoveries: unknown): Record => ({ + consts: { no_discoveries: worker.consts?.no_discoveries }, + steps: { + run_generation: { + output: + attackDiscoveries === undefined ? {} : { attack_discoveries: attackDiscoveries }, + }, + }, + }); + const discoveries = (count: number): Array<{ title: string }> => + Array.from({ length: count }, (_, index) => ({ title: `Attack ${index}` })); + + it('counts every discovery in an oversized generation', () => { + expect( + evaluate(String(resolve?.with?.attack_count), contextFor(discoveries(overflowCount))) + ).toBe(overflowCount); + }); + + it('splits an oversized generation into a full batch and a remainder', () => { + const batches = evaluate( + String(resolve?.with?.batches), + contextFor(discoveries(overflowCount)) + ) as unknown[][]; + + expect(batches.map((batch) => batch.length)).toEqual([DEFAULT_PARALLEL_MAX_FAN_OUT, 50]); + }); + + it('keeps every discovery across batches', () => { + const batches = evaluate( + String(resolve?.with?.batches), + contextFor(discoveries(overflowCount)) + ) as Array>; + + expect(batches.flat()).toEqual(discoveries(overflowCount)); + }); + + it('fits a generation at the fan-out ceiling into one batch', () => { + const batches = evaluate( + String(resolve?.with?.batches), + contextFor(discoveries(DEFAULT_PARALLEL_MAX_FAN_OUT)) + ) as unknown[][]; + + expect(batches.map((batch) => batch.length)).toEqual([DEFAULT_PARALLEL_MAX_FAN_OUT]); + }); + + it('does not invent a batch when generation returned an empty array', () => { + expect(evaluate(String(resolve?.with?.batches), contextFor([]))).toEqual([]); + }); + + it('does not invent a batch when generation returned null', () => { + expect(evaluate(String(resolve?.with?.batches), contextFor(null))).toEqual([]); + }); + + it('does not invent a batch when generation omitted attack_discoveries', () => { + expect(evaluate(String(resolve?.with?.batches), contextFor(undefined))).toEqual([]); + }); + + it('requests a review for every discovery in an oversized generation', () => { + const attackCount = evaluate( + String(resolve?.with?.attack_count), + contextFor(discoveries(overflowCount)) + ); + + expect( + evaluate(String(stepIn(workerSteps, 'summarize_run')?.with?.reviews_requested), { + steps: { resolve_fanout: { output: { attack_count: attackCount } } }, + }) + ).toBe(overflowCount); + }); + + it('does not skip fan-out for an oversized generation', () => { + expect( + evaluate(String(stepIn(workerSteps, 'log_empty_run')?.if), { + steps: { resolve_fanout: { output: { attack_count: overflowCount } } }, + }) + ).toBe(false); + }); + }); + + // DEFAULT_PARALLEL_MAX_CONCURRENCY is 20, enforced by the schema as an error. + it('sets parallel concurrency above zero', () => { + expect(asConcurrency(parallel?.concurrency).max).toBeGreaterThan(0); + }); + + it('sets parallel concurrency within the schema ceiling', () => { + expect(asConcurrency(parallel?.concurrency).max).toBeLessThanOrEqual(20); + }); + + it('settles every branch so one failed review does not abandon the rest', () => { + expect(parallel?.mode).toBe('settled'); + }); + + it('puts exactly one step in each parallel branch', () => { + expect(parallel?.steps).toHaveLength(1); + }); + + it('launches each review with workflow.execute', () => { + expect(branch?.type).toBe('workflow.execute'); + }); + + it('launches the review workflow once per attack', () => { + expect(branch?.with?.['workflow-id']).toBe(ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW_ID); + }); + + // A parallel branch body must be a straight-line sequence of steps: no nested flow + // control, and no step-level `if`, `timeout`, or `on-failure`. + it.each(['cases', 'foreach', 'if', 'on-failure', 'timeout'] as const)( + 'keeps the branch body free of %s', + (field) => { + expect(flatten(parallel?.steps ?? []).filter((step) => step[field] !== undefined)).toEqual( + [] + ); + } + ); + + it('passes title through to the review', () => { + expect(branchInputs.title).toContain('foreach.item.title'); + }); + + // `${{ }}` preserves the array type; `{{ }}` would stringify it. + it('passes alert_ids through as a typed array', () => { + expect(branchInputs.alert_ids).toBe('${{ foreach.item.alert_ids }}'); + }); + + it('passes autonomy through as a typed enum', () => { + expect(branchInputs.autonomy).toBe('${{ inputs.autonomy }}'); + }); + + // Handover `id` is the LLM UUID, not the persisted ES `_id`. #19022 will + // consume persist ids once the run step returns them in-context. Until then + // this stub must not pass `foreach.item.id`. + it('does not pass the handover id through to the review', () => { + expect(branchInputs.id).toBeUndefined(); + }); + + it('passes parent_run_id through to the review', () => { + expect(branchInputs.parent_run_id).toContain('execution.id'); + }); + + it('logs an empty generation with a console step', () => { + expect(stepIn(workerSteps, 'log_empty_run')?.type).toBe('console'); + }); + + it('skips the fan-out when attack_count is zero', () => { + expect(stepIn(workerSteps, 'log_empty_run')?.if).toContain('attack_count == 0'); + }); + + it('requests a review for every generated attack', () => { + expect(stepIn(workerSteps, 'summarize_run')?.with?.reviews_requested).toContain( + 'steps.resolve_fanout.output.attack_count' + ); + }); + }); + + describe('review workflow', () => { + const verdictSwitch = stepIn(reviewSteps, 'apply_verdict'); + const inconclusiveStep = (verdictSwitch?.cases ?? []).find((c) => c.match === 'inconclusive') + ?.steps[0]; + const actionSteps = reviewSteps.filter( + (step) => !['data.set', 'switch', 'workflow.output'].includes(step.type) + ); + + it('has a switch case for every known verdict', () => { + expect(new Set((verdictSwitch?.cases ?? []).map((c) => c.match))).toEqual(new Set(VERDICTS)); + }); + + it('has no stray switch cases beyond the known verdicts', () => { + expect(verdictSwitch?.cases ?? []).toHaveLength(VERDICTS.length); + }); + + it('still routes an out-of-enum verdict somewhere rather than dropping it', () => { + expect(verdictSwitch?.default ?? []).not.toHaveLength(0); + }); + + it('applies the verdict with a switch', () => { + expect(verdictSwitch?.type).toBe('switch'); + }); + + it('switches on the stub verdict with the default fallback', () => { + expect(verdictSwitch?.expression).toContain('inputs.stub_verdict'); + expect(verdictSwitch?.expression).toContain('default: consts.default_verdict'); + }); + + // #19276 AC1. A default run then exercises the decided Promote-to-Incident + // path (same as true_positive); the branch is still a console stub. + it('defaults the stub verdict from consts.default_verdict', () => { + expect(verdictSwitch?.expression).toContain('default: consts.default_verdict'); + }); + + it('defaults consts.default_verdict to inconclusive', () => { + expect(review.consts?.default_verdict).toBe('inconclusive'); + }); + + it('names the inconclusive branch promote_inconclusive_per_autonomy', () => { + expect(inconclusiveStep?.name).toBe('promote_inconclusive_per_autonomy'); + }); + + it('stubs the inconclusive branch as a console step', () => { + expect(inconclusiveStep?.type).toBe('console'); + }); + + it('routes inconclusive onto the Promote-to-Incident path', () => { + expect(inconclusiveStep?.with?.message).toContain('Promote-to-Incident'); + }); + + it('does not tag inconclusive for a downstream worker', () => { + expect(inconclusiveStep?.with?.message).not.toContain('tagging'); + }); + + it('exposes every verdict as an input so all branches stay reachable', () => { + expect(review.triggers?.[0]?.inputs?.properties?.stub_verdict?.enum).toEqual([...VERDICTS]); + }); + + // The whole point of this PR: nothing here may touch a real Investigation, a real + // proposal, or a real conversation. Those land in #19022 / #19211 / #19214. + it('has at least one action step in the review', () => { + expect(actionSteps).not.toHaveLength(0); + }); + + it('implements every action as a console stub', () => { + expect(actionSteps.map((step) => step.type)).toEqual(actionSteps.map(() => 'console')); + }); + + it.each(['ai.conversation.create', 'kibana.request', 'waitForApproval', 'workflow.execute'])( + 'makes no %s call', + (type) => { + expect(reviewSteps.map((step) => step.type)).not.toContain(type); + } + ); + + it.each(['open_investigation', 'run_fp_tp_analysis'])('stubs %s as a console step', (name) => { + expect(stepIn(reviewSteps, name)?.type).toBe('console'); + }); + }); + + describe('handoff contracts', () => { + // `workflow.execute` resolves to the child's `context.output`, so a child without an + // explicit `workflow.output` step hands its parent nothing. + it.each([ + ['worker', workerSteps], + ['review', reviewSteps], + ])('%s emits an explicit workflow.output', (_name, steps) => { + expect(steps.filter((step) => step.type === 'workflow.output')).toHaveLength(1); + }); + + it.each(['alerts_analyzed', 'attacks_generated', 'reviews_failed', 'reviews_requested'])( + 'reports %s so a caller can tell an empty run from a failed one', + (name) => { + expect((worker.outputs ?? []).map((output) => output.name)).toContain(name); + } + ); + + // The editor type-checks `workflow.output` `with:` source text, so a + // `type: number` field cannot be filled with `"${{ ... }}"`. + it.each(['alerts_analyzed', 'attacks_generated', 'reviews_failed', 'reviews_requested'])( + 'declares %s as a string so the templated emit passes editor type checks', + (name) => { + expect((worker.outputs ?? []).find((output) => output.name === name)?.type).toBe('string'); + } + ); + + // `${{ }}` keeps the Liquid value's type. Runtime output validation then + // rejects a number against `type: string`. `{{ }}` stringifies so the + // emitted value matches the declared schema. + it.each(['alerts_analyzed', 'attacks_generated', 'reviews_failed', 'reviews_requested'])( + 'stringifies %s in emit_result so runtime matches the string output schema', + (name) => { + const value = stepIn(workerSteps, 'emit_result')?.with?.[name]; + + expect(value).toEqual(expect.any(String)); + expect(value).toMatch(/^\{\{/); + expect(value).not.toMatch(/^\$\{\{/); + } + ); + + it('echoes the schedule interval so a run records the cadence that launched it', () => { + expect(stepIn(workerSteps, 'emit_result')?.with?.schedule_interval).toContain( + 'inputs.schedule_interval' + ); + }); + + it('reports the generation execution_uuid so a run can be traced to its discoveries', () => { + expect((worker.outputs ?? []).map((output) => output.name)).toContain('execution_uuid'); + }); + + // The Attack Discovery generation id from the run step, not this workflow's + // execution id: it keys the AD generations API and every persisted discovery. + it('takes execution_uuid from the run step rather than the workflow execution', () => { + expect(stepIn(workerSteps, 'emit_result')?.with?.execution_uuid).toContain( + 'steps.run_generation.output.execution_uuid' + ); + }); + + it.each(['investigation_opened', 'title', 'verdict'])('returns %s from each review', (name) => { + expect((review.outputs ?? []).map((output) => output.name)).toContain(name); + }); + }); + + describe('input bounds', () => { + it.each([ + ['worker', worker], + ['review', review], + ])('%s declares at least one input', (_name, workflow) => { + expect(Object.keys(workflow.triggers?.[0]?.inputs?.properties ?? {}).length).toBeGreaterThan( + 0 + ); + }); + + it.each([ + ['worker', worker], + ['review', review], + ])('%s bounds every free-text input', (_name, workflow) => { + const props = workflow.triggers?.[0]?.inputs?.properties ?? {}; + // A closed enum is inherently bounded; anything else needs an explicit maxLength. + const unbounded = Object.entries(props) + .filter(([, schema]) => schema.type === 'string' && !schema.maxLength && !schema.enum) + .map(([name]) => name); + + expect(unbounded).toEqual([]); + }); + + it.each([ + ['worker', worker], + ['review', review], + ])('%s stays manual-only so the Watch Floor worker owns the schedule', (_name, workflow) => { + expect((workflow.triggers ?? []).map((trigger) => trigger.type)).toEqual(['manual']); + }); + }); + + // Overlapping runner runs are already cancelled by the floor / runner concurrency + // policy. Sequential + // re-generation of the same attack is a #19022 Investigation concern (and AD + // hash-dedup), not a per-attack concurrency key on this review. + it('gives the review workflow no per-attack concurrency policy', () => { + expect(review.settings?.concurrency).toBeUndefined(); + }); + + // The default workflow timeout (6h) must stay above the 35m generation step. + it.each([ + ['floor', floor], + ['worker', worker], + ])('leaves the %s workflow-level timeout at its generous default', (_name, workflow) => { + expect(workflow.settings?.timeout).toBeUndefined(); + }); +}); diff --git a/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_workflows.ts b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_workflows.ts new file mode 100644 index 0000000000000..567ddfaa0a962 --- /dev/null +++ b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/attack_discovery_workflows.ts @@ -0,0 +1,41 @@ +/* + * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one + * or more contributor license agreements. Licensed under the "Elastic License + * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side + * Public License v 1"; you may not use this file except in compliance with, at + * your election, the "Elastic License 2.0", the "GNU Affero General Public + * License v3.0 only", or the "Server Side Public License, v 1". + */ + +import ATTACK_DISCOVERY_REVIEW_YAML from './attack_discovery_review.yaml'; +import ATTACK_DISCOVERY_RUNNER_YAML from './attack_discovery_runner.yaml'; +import { + ALERTZERO_MANAGED_WORKFLOW_PLUGIN_ID, + ALERTZERO_RULE_WORKFLOW_MANAGEMENT, +} from './constants'; +import type { ManagedWorkflowDefinition } from '../../types'; + +// `system-attack-discovery-generation` is already taken by the discoveries plugin, +// so these mirror the `system-security-rule-tuning-worker` / `-review` pair instead. +export const ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW_ID = + 'system-security-attack-discovery-worker'; +export const ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW_ID = + 'system-security-attack-discovery-review'; + +export const ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW = { + billable: false, + id: ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW_ID, + management: ALERTZERO_RULE_WORKFLOW_MANAGEMENT, + pluginId: ALERTZERO_MANAGED_WORKFLOW_PLUGIN_ID, + version: 2, + yaml: ATTACK_DISCOVERY_RUNNER_YAML, +} as const satisfies ManagedWorkflowDefinition; + +export const ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW = { + billable: false, + id: ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW_ID, + management: ALERTZERO_RULE_WORKFLOW_MANAGEMENT, + pluginId: ALERTZERO_MANAGED_WORKFLOW_PLUGIN_ID, + version: 2, + yaml: ATTACK_DISCOVERY_REVIEW_YAML, +} as const satisfies ManagedWorkflowDefinition; diff --git a/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/floor_attack_discovery.ts b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/floor_attack_discovery.ts index bfe4adaa058e1..6772d6a912b0f 100644 --- a/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/floor_attack_discovery.ts +++ b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/floor_attack_discovery.ts @@ -23,7 +23,7 @@ export const ALERTZERO_WORKER_FLOOR_ATTACK_DISCOVERY_WORKFLOW = { id: ALERTZERO_WORKER_FLOOR_ATTACK_DISCOVERY_WORKFLOW_ID, management: ALERTZERO_WORKER_MANAGEMENT, pluginId: ALERTZERO_MANAGED_WORKFLOW_PLUGIN_ID, - version: 2, + version: 5, yamlTemplate: (values: ScheduledWorkerTemplateValues): string => renderScheduledWorkerYaml(FLOOR_ATTACK_DISCOVERY_YAML, values), } as const satisfies ManagedWorkflowDefinition; diff --git a/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/floor_attack_discovery.yaml b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/floor_attack_discovery.yaml index 368ba06a86a66..0e85bdd3fee26 100644 --- a/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/floor_attack_discovery.yaml +++ b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/floor_attack_discovery.yaml @@ -1,10 +1,25 @@ version: "1" name: Attack Discovery -description: Watch Floor Attack Discovery worker stub. +description: > + Watch Floor Attack Discovery worker. Owns the schedule and dispatches to the + global Attack Discovery worker, which runs generation and fans out one review + per generated attack. enabled: false tags: - watch - watch-floor +settings: + # This worker is installed per space, so the key needs no space component. + # `cancel-in-progress` with max 1: a run holds its Task Manager task for as long + # as generation takes (up to the child's 35m step timeout). A newer overlapping + # scheduled run replaces that work instead of skipping itself. The runner uses + # the same strategy so the replacement dispatch is not SKIPPED on the still-held + # child slot. Must be an object — unlike a `parallel` step's `concurrency`, + # workflow-level `settings.concurrency` does not accept a bare number. + concurrency: + key: floor-attack-discovery + strategy: cancel-in-progress + max: 1 triggers: - type: scheduled with: @@ -15,7 +30,25 @@ consts: autonomy: "__WORKER_AUTONOMY_LEVEL__" scheduleInterval: "__WORKER_SCHEDULE_INTERVAL__" steps: - - name: stub - type: console + # Synchronous on purpose: #19276 requires that when generation fails after its + # built-in retries, this worker run remains failed and hands off nothing. + # `workflow.execute` propagates the child's failure, which `workflow.executeAsync` + # would not — the run would report success while generation had failed. The cost + # is that this run's Task Manager task stays claimed for the whole generation. + # + # The child carries no schedule of its own; this worker is the single scheduler + # authority for the space. + - name: run_attack_discovery + type: workflow.execute with: - message: "Attack Discovery worker stub" + workflow-id: system-security-attack-discovery-worker + inputs: + # All three worker settings are forwarded. `autonomy` decides how the child's + # review handles a true-positive verdict; the other two are traceability only + # and are echoed in the child's output. + # `${{ }}` preserves the value's type so the editor does not type-check the + # template source against the child's enum / maxLength constraints. + autonomy: "${{ consts.worker_settings.autonomy }}" + settings_version: "${{ consts.worker_settings.settingsVersion }}" + schedule_interval: "${{ consts.worker_settings.scheduleInterval }}" + parent_run_id: "{{ execution.id }}" diff --git a/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/index.ts b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/index.ts index e90afda7ca98a..749dbf3e07671 100644 --- a/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/index.ts +++ b/src/platform/packages/shared/kbn-workflows/managed/definitions/alertzero/index.ts @@ -8,6 +8,10 @@ */ import { ALERTZERO_ACTION_CREATE_RULE_WORKFLOW_ID } from './actions/action_create_detection_rule'; +import { + ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW_ID, + ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW_ID, +} from './attack_discovery_workflows'; import { ALERTZERO_WORKER_DARK_CONTINUOUS_THREAT_HUNT_WORKFLOW_ID } from './dark_continuous_threat_hunt'; import { ALERTZERO_WORKER_DETECTION_RULE_CREATION_WORKFLOW_ID } from './detection_rule_creation'; import { ALERTZERO_WORKER_DETECTION_RULE_TUNING_WORKFLOW_ID } from './detection_rule_tuning'; @@ -37,6 +41,12 @@ export { ALERTZERO_ACTION_CREATE_RULE_WORKFLOW, ALERTZERO_ACTION_CREATE_RULE_WORKFLOW_ID, } from './actions/action_create_detection_rule'; +export { + ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW, + ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW_ID, + ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW, + ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW_ID, +} from './attack_discovery_workflows'; export { ALERTZERO_WORKER_DARK_CONTINUOUS_THREAT_HUNT_WORKFLOW, ALERTZERO_WORKER_DARK_CONTINUOUS_THREAT_HUNT_WORKFLOW_ID, @@ -74,6 +84,16 @@ export const ALERTZERO_RULE_WORKFLOW_IDS = [ ALERTZERO_RULE_TUNING_WORKER_WORKFLOW_ID, ] as const; +/** + * Attack Discovery worker chain: the global workflow that runs generation and fans + * out per attack, plus the per-attack review workflow it launches. Installed + * globally so the per-space Watch Floor worker can dispatch to them. + */ +export const ALERTZERO_ATTACK_DISCOVERY_WORKFLOW_IDS = [ + ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW_ID, + ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW_ID, +] as const; + /** * Action workflows AlertZero may propose. Discovery is normally by the generic * `action` tag; this list is the install set and the fallback. diff --git a/src/platform/packages/shared/kbn-workflows/managed/definitions/discoveries/index.ts b/src/platform/packages/shared/kbn-workflows/managed/definitions/discoveries/index.ts index 5dc2211f29862..22dacd031ab59 100644 --- a/src/platform/packages/shared/kbn-workflows/managed/definitions/discoveries/index.ts +++ b/src/platform/packages/shared/kbn-workflows/managed/definitions/discoveries/index.ts @@ -83,7 +83,7 @@ export const ATTACK_DISCOVERY_RUN_EXAMPLE_WORKFLOW = { id: ATTACK_DISCOVERY_RUN_EXAMPLE_WORKFLOW_ID, management: MANAGEMENT, pluginId: 'discoveries', - version: 3, + version: 4, yaml: RUN_EXAMPLE_YAML, } as const satisfies ManagedWorkflowDefinition; diff --git a/src/platform/packages/shared/kbn-workflows/managed/definitions/discoveries/run_example.yaml b/src/platform/packages/shared/kbn-workflows/managed/definitions/discoveries/run_example.yaml index ae18009cc2410..e67e23d9f674a 100644 --- a/src/platform/packages/shared/kbn-workflows/managed/definitions/discoveries/run_example.yaml +++ b/src/platform/packages/shared/kbn-workflows/managed/definitions/discoveries/run_example.yaml @@ -177,7 +177,17 @@ triggers: steps: - name: run_attack_discovery type: security.attack-discovery.run - timeout: '10m' + # 35m = the 30m pipeline budget + 5m. In sync mode this step awaits the + # pipeline, which enforces its own 30m budget internally + # (DEFAULT_PIPELINE_TIMEOUT_MS / getRemainingBudgetMs in kbn-discoveries + # run_manual_orchestration: ADR-008's gate 10m + generation 10m + + # validation 5m, shared across phases on one wall clock). This timeout must + # stay ABOVE it so the pipeline's own budget wins the race and fails with an + # attributed PipelineStepError, rather than the engine killing the step with + # an opaque TimeoutError while generation keeps running in the background. + # Must stay a literal: the engine hands `timeout` to the duration parser + # unrendered (elastic/kibana#290258). + timeout: '35m' with: # `alerts` is the primary composability point — when non-empty, # alert_retrieval_mode is automatically set to "provided". diff --git a/src/platform/packages/shared/kbn-workflows/managed/definitions/index.ts b/src/platform/packages/shared/kbn-workflows/managed/definitions/index.ts index ac21f65d8b3d9..416f1362407ba 100644 --- a/src/platform/packages/shared/kbn-workflows/managed/definitions/index.ts +++ b/src/platform/packages/shared/kbn-workflows/managed/definitions/index.ts @@ -11,6 +11,8 @@ import { CREATE_INVESTIGATION_PROPOSAL_WORKFLOW } from './agentic_investigations import { SECURITY_ALERT_ANALYSIS_WORKFLOW } from './alert_analysis'; import { ALERTZERO_ACTION_CREATE_RULE_WORKFLOW, + ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW, + ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW, ALERTZERO_DETECTION_COVERAGE_WORKFLOW, ALERTZERO_RULE_CREATION_WORKFLOW, ALERTZERO_RULE_PREVIEW_WORKFLOW, @@ -103,6 +105,9 @@ export { ALERTZERO_DETECTION_COVERAGE_WORKFLOW_ID, ALERTZERO_ACTION_CREATE_RULE_WORKFLOW_ID, ALERTZERO_ACTION_WORKFLOW_IDS, + ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW_ID, + ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW_ID, + ALERTZERO_ATTACK_DISCOVERY_WORKFLOW_IDS, ALERTZERO_MANAGED_WORKER_WORKFLOW_IDS, ALERTZERO_RULE_CREATION_WORKFLOW_ID, ALERTZERO_RULE_PREVIEW_WORKFLOW_ID, @@ -168,6 +173,8 @@ export const managedWorkflowDefinitions = [ ALERTZERO_RULE_TUNING_REVIEW_WORKFLOW, ALERTZERO_RULE_CREATION_WORKFLOW, ALERTZERO_DETECTION_COVERAGE_WORKFLOW, + ALERTZERO_ATTACK_DISCOVERY_WORKER_WORKFLOW, + ALERTZERO_ATTACK_DISCOVERY_REVIEW_WORKFLOW, // Generic proposal gate, owned by the agenticInvestigations plugin. CREATE_INVESTIGATION_PROPOSAL_WORKFLOW, // AlertZero action catalog. diff --git a/src/platform/packages/shared/kbn-workflows/managed/managed_workflow_definitions.test.ts b/src/platform/packages/shared/kbn-workflows/managed/managed_workflow_definitions.test.ts index ac7caa6b54a7c..16ad083a2c33c 100644 --- a/src/platform/packages/shared/kbn-workflows/managed/managed_workflow_definitions.test.ts +++ b/src/platform/packages/shared/kbn-workflows/managed/managed_workflow_definitions.test.ts @@ -147,7 +147,7 @@ function createContentFingerprint(content: string): string { it.each([ [ALERTZERO_WORKER_FLOOR_ALERT_TRIAGE_WORKFLOW_ID, FLOOR_ALERT_TRIAGE_YAML, '1:d6a82eff'], - [ALERTZERO_WORKER_FLOOR_ATTACK_DISCOVERY_WORKFLOW_ID, FLOOR_ATTACK_DISCOVERY_YAML, '2:d13818a0'], + [ALERTZERO_WORKER_FLOOR_ATTACK_DISCOVERY_WORKFLOW_ID, FLOOR_ATTACK_DISCOVERY_YAML, '5:365e40e0'], [ ALERTZERO_WORKER_DARK_CONTINUOUS_THREAT_HUNT_WORKFLOW_ID, DARK_CONTINUOUS_THREAT_HUNT_YAML, diff --git a/x-pack/solutions/security/plugins/alertzero/server/managed_workflows/initialize_managed_workflows.test.ts b/x-pack/solutions/security/plugins/alertzero/server/managed_workflows/initialize_managed_workflows.test.ts index d972807a170ba..07fcdbba144c8 100644 --- a/x-pack/solutions/security/plugins/alertzero/server/managed_workflows/initialize_managed_workflows.test.ts +++ b/x-pack/solutions/security/plugins/alertzero/server/managed_workflows/initialize_managed_workflows.test.ts @@ -6,7 +6,11 @@ */ import { loggerMock } from '@kbn/logging-mocks'; -import { ALERTZERO_ACTION_WORKFLOW_IDS, ALERTZERO_RULE_WORKFLOW_IDS } from '@kbn/workflows/managed'; +import { + ALERTZERO_ACTION_WORKFLOW_IDS, + ALERTZERO_ATTACK_DISCOVERY_WORKFLOW_IDS, + ALERTZERO_RULE_WORKFLOW_IDS, +} from '@kbn/workflows/managed'; import { GLOBAL_WORKFLOW_SPACE_ID } from '@kbn/workflows/server'; import type { WorkflowsExtensionsServerPluginStart } from '@kbn/workflows-extensions/server'; import { initializeManagedWorkflows } from './initialize_managed_workflows'; @@ -45,6 +49,7 @@ describe('initializeManagedWorkflows', () => { expect(client.install.mock.calls.map(([id]) => id)).toEqual([ ...ALERTZERO_RULE_WORKFLOW_IDS, ...ALERTZERO_ACTION_WORKFLOW_IDS, + ...ALERTZERO_ATTACK_DISCOVERY_WORKFLOW_IDS, ]); expect(client.install).not.toHaveBeenCalledWith( expect.anything(), diff --git a/x-pack/solutions/security/plugins/alertzero/server/managed_workflows/initialize_managed_workflows.ts b/x-pack/solutions/security/plugins/alertzero/server/managed_workflows/initialize_managed_workflows.ts index f28e734e0b4d6..1c8985c6cd827 100644 --- a/x-pack/solutions/security/plugins/alertzero/server/managed_workflows/initialize_managed_workflows.ts +++ b/x-pack/solutions/security/plugins/alertzero/server/managed_workflows/initialize_managed_workflows.ts @@ -6,7 +6,11 @@ */ import type { Logger } from '@kbn/logging'; -import { ALERTZERO_ACTION_WORKFLOW_IDS, ALERTZERO_RULE_WORKFLOW_IDS } from '@kbn/workflows/managed'; +import { + ALERTZERO_ACTION_WORKFLOW_IDS, + ALERTZERO_ATTACK_DISCOVERY_WORKFLOW_IDS, + ALERTZERO_RULE_WORKFLOW_IDS, +} from '@kbn/workflows/managed'; import { GLOBAL_WORKFLOW_SPACE_ID } from '@kbn/workflows/server'; import type { PluginScopedManagedWorkflowsApi } from '@kbn/workflows/server/types'; import type { WorkflowsExtensionsServerPluginStart } from '@kbn/workflows-extensions/server'; @@ -27,10 +31,11 @@ export const initializeManagedWorkflows = async ({ let canReconcile = true; // AlertZero action catalog entries install alongside the rule workflows: - // both are global and static. + // all are global and static. const globalWorkflowIds = [ ...ALERTZERO_RULE_WORKFLOW_IDS, ...ALERTZERO_ACTION_WORKFLOW_IDS, + ...ALERTZERO_ATTACK_DISCOVERY_WORKFLOW_IDS, ] as const; const globalWorkflowInstalls = await Promise.allSettled( diff --git a/x-pack/solutions/security/plugins/discoveries/README.md b/x-pack/solutions/security/plugins/discoveries/README.md index e77a487155a7f..6a499514d1142 100644 --- a/x-pack/solutions/security/plugins/discoveries/README.md +++ b/x-pack/solutions/security/plugins/discoveries/README.md @@ -54,15 +54,16 @@ See the YAML block above. | Surface | Path | |---|---| -| This plugin | [`x-pack/solutions/security/plugins/discoveries/`](.) | +| [`executeGenerationWorkflow()`](../../packages/kbn-discoveries/impl/attack_discovery/generation/execute_generation_workflow.ts) is the single shared entry point for all four ways to run AD (`POST /_generate`, Alerting Framework `workflowExecutor`, workflows `security.attack-discovery.run` step, `Agent Builder` skill ) | [`@kbn/discoveries/impl/attack_discovery/generation/execute_generation_workflow.ts`](../../packages/kbn-discoveries/impl/attack_discovery/generation/execute_generation_workflow.ts) | +| executeGenerationWorkflow() -> [`runManualOrchestration()`](../../packages/kbn-discoveries/impl/attack_discovery/generation/run_manual_orchestration/index.ts) invokes the chain of worklfows with timeout budgets | [`@kbn/discoveries/impl/attack_discovery/generation/run_manual_orchestration/index.ts`](../../packages/kbn-discoveries/impl/attack_discovery/generation/run_manual_orchestration/index.ts) | +| System workflow definitions | [`@kbn/workflows/managed/definitions/discoveries`](../../../../../src/platform/packages/shared/kbn-workflows/managed/definitions/discoveries) | +| UI hook (frontend entry to `_generate`) | [`use_attack_discovery`](../security_solution/public/attack_discovery/pages/use_attack_discovery/) | | Shared server logic (LangGraph, event logging, telemetry definitions) | [`@kbn/discoveries`](../../packages/kbn-discoveries/) | | OpenAPI schemas + generated types | [`@kbn/discoveries-schemas`](../../packages/kbn-discoveries-schemas/) | -| System workflow definitions (the live source-of-truth — seven inline YAML strings) | [`@kbn/workflows/managed/definitions/discoveries/index.ts`](../../../../../src/platform/packages/shared/kbn-workflows/managed/definitions/discoveries/index.ts) | | Plugin-side managed-workflow install + integrity check | [`server/managed_workflows/`](server/managed_workflows/) | | Workflow step common definitions | [`common/step_types/`](common/step_types/) | | Workflow step server handlers | [`server/workflows/steps/`](server/workflows/steps/) | | Step registration | [`server/workflows/register_workflow_steps.ts`](server/workflows/register_workflow_steps.ts) | -| UI hook (frontend entry to `_generate`) | [`use_attack_discovery`](../security_solution/public/attack_discovery/pages/use_attack_discovery/) | ### 3. Run the example workflow @@ -303,7 +304,7 @@ See [Using the `security.attack-discovery.run` Step](#using-the-securityattack-d ## Timeouts -Attack Discovery generation is bounded by **layered** timeouts (see [ADR-008](#adr-008--layered-timeout-architecture-30-min-total-budget)). Timeouts propagate inside-out: a slow LLM call trips the connector timeout, which fails the workflow step, which `runManualOrchestration` catches against the total pipeline budget. The only **hard wall-clock kill** of an in-flight run is the scheduled rule-task timeout; the ad-hoc route, the run step, and the run tool are fire-and-forget / soft-handoff, so the background pipeline keeps running up to the pipeline budget. +Attack Discovery generation is bounded by **layered** timeouts (see [ADR-008](#adr-008--layered-timeout-architecture-30-min-total-budget)). Timeouts propagate inside-out: a slow LLM call trips the connector timeout, which fails the workflow step, which `runManualOrchestration` catches against the total pipeline budget. The only **hard wall-clock kill** of an in-flight run is the scheduled rule-task timeout; the ad-hoc route and the run tool are fire-and-forget / soft-handoff, so the background pipeline keeps running up to the pipeline budget. The run **step** in sync mode instead awaits the pipeline and is bounded by its authored `timeout`. ### Per-method entry timeouts @@ -311,8 +312,9 @@ Attack Discovery generation is bounded by **layered** timeouts (see [ADR-008](#a |---|---|---|---|---| | Scheduled (Alerting Framework) | Rule task timeout | **15m** | `ruleTaskTimeout` — [`register_schedule/definition.ts`](../elastic_assistant/server/lib/attack_discovery/schedules/register_schedule/definition.ts) | **Hard kill**: the Alerting Framework cancels the task; `shouldStopExecution()` flips true and the run is reported failed. | | Ad hoc (`POST /internal/attack_discovery/_generate`) | Route handler `idleSocket` | **10m** | `DEFAULT_ROUTE_HANDLER_TIMEOUT_MS` — [`routes/constants.ts`](server/routes/constants.ts) | Effectively moot: the route is fire-and-forget and returns `execution_uuid` immediately, so this does not bound the generation. | -| Run tool (`security.attack-discovery.run`) & run step, sync mode | Soft deadline | **90s** | `ATTACK_DISCOVERY_RUN_SOFT_DEADLINE_MS` — [`run_step/constants.ts`](server/workflows/steps/run_step/constants.ts) | **Not a kill**: on deadline the call returns `execution_uuid` and the pipeline keeps running in the background for slow-path resume. | -| Agent Builder workflow-tool wrapper | Wait-for-completion ceiling | **120s** | `WAIT_FOR_COMPLETION_TIMEOUT_SEC` — `@kbn/agent-builder-common` | External AB ceiling the 90s soft deadline sits safely under. | +| Run **tool** (Agent Builder builtin), sync mode | Soft deadline | **90s** | `ATTACK_DISCOVERY_RUN_SOFT_DEADLINE_MS` — [`run_step/constants.ts`](server/workflows/steps/run_step/constants.ts) | **Not a kill**: on deadline the call returns `execution_uuid` and the pipeline keeps running in the background for slow-path resume. | +| Run **step** (`security.attack-discovery.run` in workflow YAML), sync mode | Step `timeout` | **authored per call site** | `timeout:` on the step — no soft deadline | Awaits the pipeline to completion. Keep the authored value **above** the 30m pipeline budget so the budget wins the race and fails with an attributed `PipelineStepError`; an engine step timeout is an opaque kill that does **not** cancel the background pipeline. | +| Agent Builder workflow-tool wrapper | Wait-for-completion ceiling | **120s** | `WAIT_FOR_COMPLETION_TIMEOUT_SEC` — `@kbn/agent-builder-common` | Applies only to generic `ToolType.workflow` tools. The AD skill does **not** use one (see ADR-012), so this does not bound the skill's run path. | ### Pipeline (orchestration) layer @@ -764,7 +766,7 @@ The skill registers a single Agent Builder skill. In its conversational role it 1. **Loads the analyst prompt** — same "world-class cyber security analyst" framing used by the LangGraph generate node, plus stricter rules layered on top: a Validation Standard ("when in doubt, discard"), a default-to-split independent-evaluation rule, and Entity Correlation Hygiene guidance that calls out service accounts, shared infrastructure, and same-tactic-different-host coincidence as **not** sufficient correlation evidence. 2. **Tells the agent to corroborate before deciding** — the skill content intentionally does not enumerate which tools to use. It instructs the agent to *enumerate the tools available in this conversation* and call those that gather supporting evidence (threat hunting, threat intelligence, entity context, knowledge base, etc.). The skill exposes a small set of platform tools (`execute_esql`, `generate_esql`, `search`, `get_document_by_id`, `get_index_mapping`, `get_workflow_execution_status`) plus the inline `get_default_esql_query` and `security.attack-discovery.get_status` tools, but other tools active in the session are also fair game. -3. **Mode A — Generate**: once the agent has corroborated, it invokes `security.attack-discovery.run` per [ADR-012](#adr-012--agent-builder-uses-run-in-sync-mode-with-a-soft-deadline). The pipeline handles anonymization, LangGraph generation, hallucination detection, validation, and persistence to the Attack Discovery alerts index. Sync mode races a ~90s soft deadline against the 120s Agent Builder workflow-tool ceiling — fast generations return discoveries inline; slower generations return only an `execution_uuid` and the agent hands off cleanly with an in-progress acknowledgement. +3. **Mode A — Generate**: once the agent has corroborated, it invokes the `security.attack-discovery.run` **tool** per [ADR-012](#adr-012--agent-builder-uses-run-in-sync-mode-with-a-soft-deadline). The pipeline handles anonymization, LangGraph generation, hallucination detection, validation, and persistence to the Attack Discovery alerts index. The tool races a ~90s soft deadline against the 120s Agent Builder ceiling — fast generations return discoveries inline; slower generations return only an `execution_uuid` and the agent hands off cleanly with an in-progress acknowledgement. 4. **Mode B — Status-only**: when the user supplies an `execution_uuid` (or asks about a previously-started generation), the agent calls `security.attack-discovery.get_status` and emits the insights JSON if the run has succeeded, reports progress with the active phase if still running, or reports the failure cleanly. No new generation is started. 5. **Persists discoveries through the shared pipeline** — discoveries are written through the same `defaultValidation` + `persistDiscoveries` chain used by every other execution path (so they appear in the AD UI and via `GET /api/attack_discovery/generations`), regardless of which mode emitted them in the agent reply. @@ -775,7 +777,7 @@ flowchart TB USER["Agent Builder user"] AGENT["Agent + skill content"] CORR["Corroboration phase
(execute_esql, search, threat intel, ...)"] - RUN["security.attack-discovery.run
(sync mode + ~90s soft deadline)"] + RUN["security.attack-discovery.run tool
(sync mode + ~90s soft deadline)"] ORCH["Orchestrator pipeline
(retrieve → generate → validate → persist)"] STATUS["security.attack-discovery.get_status"] AD["Attack Discovery alerts index"] @@ -807,7 +809,7 @@ The skill teaches the agent to pick the `security.attack-discovery.run` mode tha ⛔ The skill explicitly forbids bare connector-ID-only invocations (`{ "connector_id": "..." }`) because they rely on server-side defaults that do not reflect the investigation context. -Sync mode is the default and the only mode the skill actively instructs the agent to use, per [ADR-012](#adr-012--agent-builder-uses-run-in-sync-mode-with-a-soft-deadline). The run step's executor races the pipeline against `ATTACK_DISCOVERY_RUN_SOFT_DEADLINE_MS` (90s) so the wrapping Agent Builder workflow tool — which itself caps at 120s — always gets a clean response well inside its window. When the soft deadline wins, only `execution_uuid` is returned; the pipeline keeps running in the background and the agent resumes via `security.attack-discovery.get_status` when the user asks for status. +Sync mode is the default and the only mode the skill actively instructs the agent to use, per [ADR-012](#adr-012--agent-builder-uses-run-in-sync-mode-with-a-soft-deadline). The **tool's** handler races the pipeline against `ATTACK_DISCOVERY_RUN_SOFT_DEADLINE_MS` (90s) so the agent always gets a clean response well inside the 120s Agent Builder ceiling. When the soft deadline wins, only `execution_uuid` is returned; the pipeline keeps running in the background and the agent resumes via `security.attack-discovery.get_status` when the user asks for status. The workflow **step** of the same name does not apply this soft deadline — see ADR-012. #### Connector resolution @@ -1609,11 +1611,13 @@ gantt ### ADR-012 — Agent Builder uses `run` in sync mode with a soft deadline -**Context.** Agent Builder tools execute as part of a larger agent conversation. The agent needs the result inline to formulate its response. The Agent Builder workflow tool that wraps `security.attack-discovery.run` waits up to `WAIT_FOR_COMPLETION_TIMEOUT_SEC = 120s` for the workflow to complete. Real Attack Discovery generations frequently exceed two minutes, but the run step itself has a 10-minute internal timeout. Without intervention, the wrapping AB tool would hit its own timeout and return only a workflow execution ID — useless for an AD-specific resume path. Async-mode polling is not the current Agent Builder pattern (`platform.core.get_workflow_execution_status` explicitly tells agents not to auto-poll within a turn). +**Context.** Agent Builder tools execute as part of a larger agent conversation. The agent needs the result inline to formulate its response. Real Attack Discovery generations frequently exceed two minutes (ADR-007: routinely 2–5 minutes), so an integration that waits for completion within a turn would stall the conversation. Async-mode polling is not the current Agent Builder pattern (`platform.core.get_workflow_execution_status` explicitly tells agents not to auto-poll within a turn). The generic `ToolType.workflow` wrapper caps its wait at `WAIT_FOR_COMPLETION_TIMEOUT_SEC = 120s`, which sets the ceiling any AB-facing generation path must stay under. + +**Decision.** The Agent Builder AD path uses a **native `ToolType.builtin` tool** — [`run_attack_discovery_tool/index.ts`](server/agent_builder/skills/tools/run_attack_discovery_tool/index.ts), registered as an inline tool of the `attack-discovery-generator` skill — which calls `executeGenerationWorkflow` directly in sync mode and races it against a hard-coded `ATTACK_DISCOVERY_RUN_SOFT_DEADLINE_MS = 90s` soft deadline (≈30s of headroom under the 120s ceiling). If the pipeline finishes first, the tool returns the full sync output (`attack_discoveries`, `execution_uuid`, `alerts_context_count`, `discovery_count`). If the soft deadline wins, it returns `{ execution_uuid }` only and lets the pipeline keep running in the background. The skill exposes a dedicated `security.attack-discovery.get_status` tool so the user can resume by `execution_uuid` on a subsequent prompt. -**Decision.** Agent Builder integrations call `security.attack-discovery.run` in **sync mode**. The run step's executor races the generation pipeline against a hard-coded `ATTACK_DISCOVERY_RUN_SOFT_DEADLINE_MS = 90s` soft deadline (≈30s of headroom under the 120s AB ceiling). If the pipeline finishes first, the step returns the full sync output (`attack_discoveries`, `execution_uuid`, `alerts_context_count`, `discovery_count`). If the soft deadline wins, the step returns `{ execution_uuid }` only and lets the underlying pipeline keep running in the background. The agent skill exposes a dedicated `security.attack-discovery.get_status` tool so the user can resume by `execution_uuid` on a subsequent prompt. +**The soft deadline is the tool's, not the step's.** The Agent Builder tool id and the workflow step type id are the same string, `security.attack-discovery.run`, but they are two separate implementations. Only the **tool** applies the 90s soft deadline. The **step** ([`get_run_step_definition.ts`](server/workflows/steps/run_step/get_run_step_definition.ts)) awaits the pipeline to completion in sync mode, because a workflow caller has no 120s ceiling to respect and needs the discoveries inline to fan out over them. Workflow call sites bound the wait with the step's own `timeout` instead, authored above the 30m pipeline budget (see Timeouts). -**Consequence.** The AB workflow tool always receives a clean response well inside its 120s window — it never times out. Fast generations return inline discoveries (today's behavior). Slow generations return a clean `execution_uuid` handoff; the agent acknowledges the in-progress state, the pipeline persists discoveries automatically when complete, and the user can ask for status to resume. The agent never sees the `replacements` map (excluded by schema). +**Consequence.** The Agent Builder path always receives a clean response well inside its 120s window — it never times out. Fast generations return inline discoveries; slow generations return a clean `execution_uuid` handoff, the pipeline persists discoveries automatically when complete, and the user can ask for status to resume. The agent never sees the `replacements` map (excluded by schema). One residual risk: an operator who hand-configures a generic `ToolType.workflow` tool pointing at a workflow that uses the run **step** gets the step's semantics (no 90s cap), so that tool must respect the 120s ceiling itself. Nothing in the repo registers such a tool. ### ADR-013 — SHA-256 integrity verification of required default workflows *(superseded)* diff --git a/x-pack/solutions/security/plugins/discoveries/common/step_types/run_step.ts b/x-pack/solutions/security/plugins/discoveries/common/step_types/run_step.ts index e4aed3ef288a6..8cb71d98c8c46 100644 --- a/x-pack/solutions/security/plugins/discoveries/common/step_types/run_step.ts +++ b/x-pack/solutions/security/plugins/discoveries/common/step_types/run_step.ts @@ -51,10 +51,10 @@ export const RunStepInputSchema = z.object({ * from both modes for security. * * `status` distinguishes a terminal run (`completed`) from one that is still - * executing in the background (`pending`) — the async branch and the sync - * soft-deadline slow path both return `pending` so consumers poll - * `security.attack-discovery.get_status` instead of reading the absent counts - * as "0 discoveries". + * executing in the background (`pending`). Only async mode returns `pending`, + * so consumers of that mode poll `security.attack-discovery.get_status` instead + * of reading the absent counts as "0 discoveries". Sync mode awaits the pipeline + * to completion and always returns `completed`, bounded by the step's `timeout`. */ export const RunStepOutputSchema = z.object({ alerts_context_count: z.number().int().optional(), diff --git a/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/README.md b/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/README.md index f9773a2473c3b..021b8fb13e2dc 100644 --- a/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/README.md +++ b/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/README.md @@ -50,7 +50,7 @@ The anonymization `replacements` map (which maps anonymized tokens back to real In async mode, pipeline errors are logged but do not propagate to the workflow — the step always returns successfully with the `execution_uuid`. -Sync mode also races the pipeline against a soft deadline (`ATTACK_DISCOVERY_RUN_SOFT_DEADLINE_MS`). If the pipeline does not finish in time, the step returns early with just `execution_uuid` and `status: 'pending'` while the pipeline keeps running in the background (the Agent Builder run tool then resumes via the status tool). +Sync mode awaits the pipeline to completion and always returns `status: 'completed'`. It applies no soft deadline of its own; the only bound is the step's `timeout`, which must stay above the pipeline's own 30m budget (see the example below). `ATTACK_DISCOVERY_RUN_SOFT_DEADLINE_MS` applies to the Agent Builder run tool, not to this step. ### Alert retrieval modes @@ -94,7 +94,10 @@ The `Attack discovery - Run example` workflow (`ATTACK_DISCOVERY_RUN_EXAMPLE_WOR steps: - name: run_attack_discovery type: security.attack-discovery.run - timeout: '10m' + # Must stay above the pipeline's own 30m budget so the pipeline fails with an + # attributed error rather than the engine killing the step. Keep it a literal: + # the engine hands `timeout` to the duration parser unrendered. + timeout: '35m' with: # Primary composability point: when non-empty, retrieval is skipped. alerts: ${{ inputs.alerts }} @@ -126,8 +129,8 @@ The example exposes all inputs as workflow inputs so the same workflow can run s }> | null; discovery_count: number; // 0 if validation failed alerts_context_count: number; // 0 if validation failed - status: 'completed'; // 'pending' on the async / soft-deadline paths + status: 'completed'; // always 'completed' in sync mode } ``` -In async mode (and when the sync soft deadline is exceeded), only `execution_uuid` and `status: 'pending'` are returned. +In async mode, only `execution_uuid` and `status: 'pending'` are returned. diff --git a/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/constants.ts b/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/constants.ts index 8a807f49c97de..799a6eac3f755 100644 --- a/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/constants.ts +++ b/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/constants.ts @@ -6,20 +6,21 @@ */ /** - * Soft deadline for the run step in `sync` mode. + * Soft deadline for the Agent Builder Attack Discovery run **tool** in `sync` mode. * - * Why: Agent Builder workflow tools wait up to `WAIT_FOR_COMPLETION_TIMEOUT_SEC` - * (120s) for a workflow to complete. Real Attack Discovery generations frequently - * exceed that ceiling. If we let the AB wrapper hit its own timeout, the agent - * receives a workflow execution ID without the run step's output, which is not - * useful for the AD-specific resume path. + * Consumed only by `run_attack_discovery_tool`, not by the run step. The step and + * the tool share the `security.attack-discovery.run` name but are separate + * implementations, and the step deliberately applies no soft deadline: a workflow + * caller has no wrapper ceiling to respect and needs the discoveries inline, so it + * bounds the wait with its own step `timeout` instead (see ADR-012 in the plugin + * README). This constant stays here because the tool imports it from this path. * - * Instead, the run step itself watches a soft deadline well below the AB ceiling. - * If the pipeline isn't done by then, the executor returns `{ execution_uuid }` - * (matching async-mode output) and lets the underlying pipeline keep running in - * the background. The AB wrapper sees the workflow as completed and forwards the - * clean result. The agent then handles the slow-path handoff via the dedicated - * AD status tool. + * Why the tool needs one: an Agent Builder turn must not stall. Real Attack + * Discovery generations frequently exceed two minutes, which is over the 120s + * `WAIT_FOR_COMPLETION_TIMEOUT_SEC` ceiling. When the deadline wins, the tool + * returns `{ execution_uuid }` (matching async-mode output) and lets the pipeline + * keep running in the background; the agent handles the slow-path handoff via the + * dedicated AD status tool. * * The 30s of headroom under 120s covers serialization, network, and workflow * engine overhead. diff --git a/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/get_run_step_definition.test.ts b/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/get_run_step_definition.test.ts index 6d8b364742ffb..e2ee564f40ebc 100644 --- a/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/get_run_step_definition.test.ts +++ b/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/get_run_step_definition.test.ts @@ -11,10 +11,6 @@ import { WorkflowExecutionAuthorizationError } from '@kbn/discoveries/impl/attac import { resolveConnectorDetails } from '../../helpers/resolve_connector_details'; import { resolveDefaultConnectorId } from '../../helpers/resolve_default_connector_id'; -jest.mock('./constants', () => ({ - ATTACK_DISCOVERY_RUN_SOFT_DEADLINE_MS: 25, -})); - const mockIsWorkflowsEnabledForSpace = jest.fn(); jest.mock('../../../lib/is_workflows_enabled_for_space', () => ({ @@ -441,39 +437,13 @@ describe('getRunStepDefinition', () => { }); }); - describe('soft deadline (sync mode)', () => { - it('returns execution_uuid only when the pipeline exceeds the soft deadline', async () => { - mockExecuteGenerationWorkflow.mockReturnValue(new Promise(() => {})); - - const stepDefinition = getStepDefinition(); - - const result = await stepDefinition.handler(syncMockContext as never); - - expect(result.output).toEqual({ execution_uuid: 'test-execution-uuid', status: 'pending' }); - }); - - it('returns status pending when the pipeline exceeds the soft deadline', async () => { - mockExecuteGenerationWorkflow.mockReturnValue(new Promise(() => {})); - - const stepDefinition = getStepDefinition(); - - const result = await stepDefinition.handler(syncMockContext as never); - - expect(result.output?.status).toBe('pending'); - }); - - it('omits attack_discoveries when the pipeline exceeds the soft deadline', async () => { - mockExecuteGenerationWorkflow.mockReturnValue(new Promise(() => {})); - - const stepDefinition = getStepDefinition(); - - const result = await stepDefinition.handler(syncMockContext as never); + describe('sync mode has no soft deadline', () => { + /** Resolves after `delayMs`, standing in for a generation slower than the old 90s cap. */ + const slowPipeline = (delayMs: number) => + new Promise((resolve) => setTimeout(() => resolve(mockSuccessOutcome), delayMs)); - expect(result.output).not.toHaveProperty('attack_discoveries'); - }); - - it('returns the full sync output when the pipeline finishes before the soft deadline', async () => { - mockExecuteGenerationWorkflow.mockResolvedValue(mockSuccessOutcome); + it('waits for a slow pipeline instead of returning early', async () => { + mockExecuteGenerationWorkflow.mockReturnValue(slowPipeline(50)); const stepDefinition = getStepDefinition(); @@ -481,36 +451,15 @@ describe('getRunStepDefinition', () => { expect(result.output?.attack_discoveries).toEqual(handoverDiscoveries); }); - }); - - describe('soft deadline timer cleanup (sync mode)', () => { - beforeEach(() => { - jest.useFakeTimers(); - }); - afterEach(() => { - jest.runOnlyPendingTimers(); - jest.useRealTimers(); - }); - - it('clears the soft-deadline timer when the pipeline rejects', async () => { - mockExecuteGenerationWorkflow.mockRejectedValue(new Error('Pipeline failed')); + it('reports completed rather than pending for a slow pipeline', async () => { + mockExecuteGenerationWorkflow.mockReturnValue(slowPipeline(50)); const stepDefinition = getStepDefinition(); - await stepDefinition.handler(syncMockContext as never); - - expect(jest.getTimerCount()).toBe(0); - }); - - it('clears the soft-deadline timer when the pipeline resolves before the deadline', async () => { - mockExecuteGenerationWorkflow.mockResolvedValue(mockSuccessOutcome); - - const stepDefinition = getStepDefinition(); - - await stepDefinition.handler(syncMockContext as never); + const result = await stepDefinition.handler(syncMockContext as never); - expect(jest.getTimerCount()).toBe(0); + expect(result.output?.status).toBe('completed'); }); }); diff --git a/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/get_run_step_definition.ts b/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/get_run_step_definition.ts index baa46854f9ddb..406758d7b1345 100644 --- a/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/get_run_step_definition.ts +++ b/x-pack/solutions/security/plugins/discoveries/server/workflows/steps/run_step/get_run_step_definition.ts @@ -21,10 +21,6 @@ import type { DiscoveriesPluginStartDeps } from '../../../types'; import { resolveConnectorDetails } from '../../helpers/resolve_connector_details'; import { resolveDefaultConnectorId } from '../../helpers/resolve_default_connector_id'; import { checkManagedWorkflowIntegrity } from '../../../managed_workflows/check_managed_workflow_integrity'; -import { ATTACK_DISCOVERY_RUN_SOFT_DEADLINE_MS } from './constants'; - -const SOFT_DEADLINE_SENTINEL = Symbol('attack-discovery-run-soft-deadline'); -type SoftDeadlineSentinel = typeof SOFT_DEADLINE_SENTINEL; /** * Server-side implementation of the Attack Discovery run step. @@ -211,94 +207,49 @@ export const getRunStepDefinition = ({ }; } - // sync mode races the pipeline against a soft deadline (see constants.ts). - // If the pipeline doesn't finish in time, return execution_uuid only and - // let the pipeline keep running in the background — the AB workflow tool - // wrapper then receives a clean response well inside its own 120s ceiling, - // and the agent resumes via the dedicated AD status tool. - const pipelinePromise = executeGenerationWorkflow(executeParams); - - let softDeadlineTimer: NodeJS.Timeout | undefined; - const softDeadlinePromise = new Promise((resolve) => { - softDeadlineTimer = setTimeout( - () => resolve(SOFT_DEADLINE_SENTINEL), - ATTACK_DISCOVERY_RUN_SOFT_DEADLINE_MS - ); - }); - - try { - const raced = await Promise.race([pipelinePromise, softDeadlinePromise]); - - if (raced === SOFT_DEADLINE_SENTINEL) { - context.logger.info( - `Attack Discovery sync pipeline exceeded soft deadline of ${ATTACK_DISCOVERY_RUN_SOFT_DEADLINE_MS}ms; returning execution_uuid for slow-path resume (execution=${executionUuid})` - ); - - // The pipeline keeps running in the background; surface any later - // rejection so it doesn't surface as an unhandled promise rejection. - pipelinePromise.catch((err) => { - logger.error( - `Attack Discovery sync pipeline rejected after returning early (execution=${executionUuid}): ${ - err instanceof Error ? err.message : String(err) - }` - ); - }); - - return { - output: { - execution_uuid: executionUuid, - status: 'pending' as const, - }, - }; - } - - const outcome = raced; - - if (outcome.outcome === 'validation_succeeded') { - const { alertRetrievalResult, generationResult, validationResult } = outcome; - - return { - output: { - alerts_context_count: alertRetrievalResult.alertsContextCount, - // R3: the run step persists via the persist step and returns exactly the - // discoveries it was handed (`[]` when the persist step did not run). - attack_discoveries: (validationResult.discoveriesToPersist ?? []) as Array<{ - alert_ids: string[]; - details_markdown: string; - entity_summary_markdown?: string; - id?: string; - mitre_attack_tactics?: string[]; - summary_markdown: string; - timestamp?: string; - title: string; - }>, - discovery_count: validationResult.generatedCount, - execution_uuid: generationResult.executionUuid, - status: 'completed' as const, - }, - }; - } + // sync mode (the default): await the pipeline to completion and return + // the discoveries inline. The step's own `timeout` bounds the wait for a + // slow generation. The Agent Builder run tool needs to stay under its + // 120s wrapper ceiling and applies its own soft deadline separately — it + // does not use this step, so no soft deadline is applied here. + const outcome = await executeGenerationWorkflow(executeParams); - context.logger.warn(`Attack Discovery validation failed (execution=${executionUuid})`); + if (outcome.outcome === 'validation_succeeded') { + const { alertRetrievalResult, generationResult, validationResult } = outcome; return { output: { - alerts_context_count: 0, - attack_discoveries: null, - discovery_count: 0, - execution_uuid: executionUuid, + alerts_context_count: alertRetrievalResult.alertsContextCount, + // R3: the run step persists via the persist step and returns exactly the + // discoveries it was handed (`[]` when the persist step did not run). + attack_discoveries: (validationResult.discoveriesToPersist ?? []) as Array<{ + alert_ids: string[]; + details_markdown: string; + entity_summary_markdown?: string; + id?: string; + mitre_attack_tactics?: string[]; + summary_markdown: string; + timestamp?: string; + title: string; + }>, + discovery_count: validationResult.generatedCount, + execution_uuid: generationResult.executionUuid, status: 'completed' as const, }, }; - } finally { - // Always clear the soft-deadline timer so it does not leak into the - // event loop when the pipeline promise rejects (the timer would - // otherwise stay pending until it fired). This also covers the - // success and soft-deadline-exceeded paths. - if (softDeadlineTimer != null) { - clearTimeout(softDeadlineTimer); - } } + + context.logger.warn(`Attack Discovery validation failed (execution=${executionUuid})`); + + return { + output: { + alerts_context_count: 0, + attack_discoveries: null, + discovery_count: 0, + execution_uuid: executionUuid, + status: 'completed' as const, + }, + }; } catch (error) { context.logger.error( `Attack Discovery run step failed: ${