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: ${