[One Workflow] refactor: Split WorkflowExecutionRuntimeManager into workflow and step concerns (elastic#237904)

## Summary

Closes: elastic/security-team#14137

## 🔄 **CHANGELOG**

### ✨ **New Features**

#### **Step Execution Runtime Extraction**
- **Created `StepExecutionRuntime` class** - Extracted step-specific
execution logic from `WorkflowExecutionRuntimeManager`
- **Added `StepExecutionRuntimeFactory`** - Factory pattern for creating
isolated step execution contexts
- **Enhanced scope management** - Added guard conditions for proper
scope entry/exit validation

#### **Improved Workflow Architecture**
- **Separated concerns** - Clear separation between workflow-level and
step-level execution management
- **Better error handling** - Enhanced error propagation with proper
step failure tracking

### 🛠 **Breaking Changes**

#### **Interface Updates**
- **`NodeImplementation` interface** - Now accepts
`StepExecutionRuntime` instead of `WorkflowContextManager`
- **Node factory pattern** - Updated to work with new step execution
runtime pattern
- **Workflow execution loop** - Refactored to use factory pattern for
step runtime creation

### 🔧 **Refactoring**

#### **WorkflowExecutionRuntimeManager Cleanup**
- **Removed step-specific methods**:
  - `getCurrentStepExecutionId()`
  - `getCurrentStepResult()` / `setCurrentStepResult()`
  - `getCurrentStepState()` / `setCurrentStepState()`
  - `startStep()` / `finishStep()` / `failStep()` / `setWaitStep()`
- **Simplified navigation** - Navigation methods now use internal
`nextNodeId` tracking
- **Enhanced scope management** - Added comprehensive documentation and
validation guards

#### **Step Implementation Updates**
- **All step implementations** updated to use `StepExecutionRuntime`
instead of `WorkflowContextManager`
- **Improved abort handling** - Better support for step cancellation via
`AbortController`
- **Centralized step logging** - Step-specific logging moved to
`StepExecutionRuntime`

### 🧰 **Utilities**

#### **New Helper Functions**
- **`stringifyStackFrames()`** - Utility for debugging workflow
execution paths

### 📋 **Technical Improvements**

- **Reduced coupling** - Clear separation between workflow and step
execution concerns
- **Enhanced maintainability** - Factory pattern improves code
organization and testability
- **Better error isolation** - Step failures are now properly contained
within step execution context
- **Improved debugging** - Enhanced logging and tracing capabilities for
workflow execution analysis

---

**Migration Impact**: This is an internal refactoring that improves code
architecture without changing public APIs. No external changes required
for workflow definitions or execution.

### Checklist

Check the PR satisfies following conditions. 

Reviewers should verify this PR satisfies this list as well.

- [ ] Any text added follows [EUI's writing
guidelines](https://elastic.github.io/eui/#/guidelines/writing), uses
sentence case text and includes [i18n
support](https://github.com/elastic/kibana/blob/main/src/platform/packages/shared/kbn-i18n/README.md)
- [ ]
[Documentation](https://www.elastic.co/guide/en/kibana/master/development-documentation.html)
was added for features that require explanation or tutorials
- [ ] [Unit or functional
tests](https://www.elastic.co/guide/en/kibana/master/development-tests.html)
were updated or added to match the most common scenarios
- [ ] If a plugin configuration key changed, check if it needs to be
allowlisted in the cloud and added to the [docker
list](https://github.com/elastic/kibana/blob/main/src/dev/build/tasks/os_packages/docker_generator/resources/base/bin/kibana-docker)
- [ ] This was checked for breaking HTTP API changes, and any breaking
changes have been approved by the breaking-change committee. The
`release_note:breaking` label should be applied in these situations.
- [ ] [Flaky Test
Runner](https://ci-stats.kibana.dev/trigger_flaky_test_runner/1) was
used on any tests changed
- [ ] The PR description includes the appropriate Release Notes section,
and the correct `release_note:*` label is applied per the
[guidelines](https://www.elastic.co/guide/en/kibana/master/contributing.html#kibana-release-notes-process)
- [ ] Review the [backport
guidelines](https://docs.google.com/document/d/1VyN5k91e5OVumlc0Gb9RPa3h1ewuPE705nRtioPiTvY/edit?usp=sharing)
and apply applicable `backport:*` labels.

### Identify risks

Does this PR introduce any risks? For example, consider risks like hard
to test bugs, performance regression, potential of data loss.

Describe the risk, its severity, and mitigation for each identified
risk. Invite stakeholders and evaluate how to proceed before merging.

- [ ] [See some risk
examples](https://github.com/elastic/kibana/blob/main/RISK_MATRIX.mdx)
- [ ] ...

---------

Co-authored-by: kibanamachine <[email protected]>
Co-authored-by: Elastic Machine <[email protected]>

Author: 3 people

src/platform/packages/shared/kbn-workflows/graph/build_execution_graph/build_execution_graph.ts

@@ -595,6 +595,7 @@ function createFallback(
   const enterTryBlockNodeId = `enterTryBlock_${stepId}`;
   const exitTryBlockNodeId = `exitTryBlock_${stepId}`;
   const enterNormalPathNodeId = `enterNormalPath_${stepId}`;
+  const enterFallbackPathNodeId = `enterFallbackPath_${stepId}`;
 
   const enterTryBlockNode: EnterTryBlockNode = {
     id: enterTryBlockNodeId,
@@ -603,6 +604,7 @@ function createFallback(
     stepType: 'fallback',
     type: 'enter-try-block',
     enterNormalPathNodeId,
+    enterFallbackPathNodeId,
   };
   graph.setNode(enterTryBlockNodeId, enterTryBlockNode);
   const exitTryBlockNode: ExitTryBlockNode = {

src/platform/packages/shared/kbn-workflows/graph/build_execution_graph/tests/on_failure_graph.test.ts

@@ -255,6 +255,7 @@ describe('on_failure graph', () => {
           stepType: 'fallback',
           type: 'enter-try-block',
           enterNormalPathNodeId: 'enterNormalPath_testRetryConnectorStep',
+          enterFallbackPathNodeId: 'enterFallbackPath_testRetryConnectorStep',
         });
       });
 

src/platform/packages/shared/kbn-workflows/graph/types/nodes/on_failure_nodes.ts

@@ -44,6 +44,7 @@ export const EnterTryBlockNodeSchema = GraphNodeSchema.extend({
   id: z.string(),
   type: z.literal('enter-try-block'),
   enterNormalPathNodeId: z.string(),
+  enterFallbackPathNodeId: z.string(),
   exitNodeId: z.string(),
 });
 export type EnterTryBlockNode = z.infer<typeof EnterTryBlockNodeSchema>;

src/platform/plugins/shared/workflows_execution_engine/server/plugin.ts

@@ -48,6 +48,7 @@ import type {
   StartWorkflowExecutionParams,
 } from './workflow_task_manager/types';
 import { WorkflowTaskManager } from './workflow_task_manager/workflow_task_manager';
+import { StepExecutionRuntimeFactory } from './workflow_context_manager/step_execution_runtime_factory';
 
 export class WorkflowsExecutionEnginePlugin
   implements Plugin<WorkflowsExecutionEnginePluginSetup, WorkflowsExecutionEnginePluginStart>
@@ -88,6 +89,7 @@ export class WorkflowsExecutionEnginePlugin
 
               const {
                 workflowRuntime,
+                stepExecutionRuntimeFactory,
                 workflowExecutionState,
                 workflowLogger,
                 nodesFactory,
@@ -111,6 +113,7 @@ export class WorkflowsExecutionEnginePlugin
 
               await workflowExecutionLoop({
                 workflowRuntime,
+                stepExecutionRuntimeFactory,
                 workflowExecutionState,
                 workflowExecutionRepository,
                 workflowLogger,
@@ -152,6 +155,7 @@ export class WorkflowsExecutionEnginePlugin
 
               const {
                 workflowRuntime,
+                stepExecutionRuntimeFactory,
                 workflowExecutionState,
                 workflowLogger,
                 nodesFactory,
@@ -175,6 +179,7 @@ export class WorkflowsExecutionEnginePlugin
 
               await workflowExecutionLoop({
                 workflowRuntime,
+                stepExecutionRuntimeFactory,
                 workflowExecutionState,
                 workflowExecutionRepository,
                 workflowLogger,
@@ -241,6 +246,7 @@ export class WorkflowsExecutionEnginePlugin
 
         const {
           workflowRuntime,
+          stepExecutionRuntimeFactory,
           workflowExecutionState,
           workflowLogger,
           nodesFactory,
@@ -264,6 +270,7 @@ export class WorkflowsExecutionEnginePlugin
         await workflowRuntime.start();
         await workflowExecutionLoop({
           workflowRuntime,
+          stepExecutionRuntimeFactory,
           workflowExecutionState,
           workflowExecutionRepository,
           workflowLogger,
@@ -487,19 +494,29 @@ async function createContainer(
     allowedHosts: config.http.allowedHosts,
   });
 
+  const stepExecutionRuntimeFactory = new StepExecutionRuntimeFactory({
+    workflowExecutionGraph,
+    workflowExecutionState,
+    workflowLogger,
+    esClient,
+    fakeRequest,
+    coreStart,
+  });
+
   const nodesFactory = new NodesFactory(
     connectorExecutor,
     workflowRuntime,
-    workflowExecutionState,
     workflowLogger,
     workflowTaskManager,
     urlValidator,
-    workflowExecutionGraph
+    workflowExecutionGraph,
+    stepExecutionRuntimeFactory
   );
 
   return {
     workflowExecutionGraph,
     workflowRuntime,
+    stepExecutionRuntimeFactory,
     workflowExecutionState,
     connectorExecutor,
     workflowLogger,

src/platform/plugins/shared/workflows_execution_engine/server/step/atomic_step/atomic_step_impl.ts

@@ -9,7 +9,7 @@
 import type { AtomicGraphNode } from '@kbn/workflows/graph';
 import type { NodeImplementation } from '../node_implementation';
 import { ConnectorStepImpl } from '../connector_step';
-import type { WorkflowContextManager } from '../../workflow_context_manager/workflow_context_manager';
+import type { StepExecutionRuntime } from '../../workflow_context_manager/step_execution_runtime';
 import type { ConnectorExecutor } from '../../connector_executor';
 import type { WorkflowExecutionRuntimeManager } from '../../workflow_context_manager/workflow_execution_runtime_manager';
 import type { IWorkflowEventLogger } from '../../workflow_event_logger/workflow_event_logger';
@@ -33,7 +33,7 @@ import type { IWorkflowEventLogger } from '../../workflow_event_logger/workflow_
 export class AtomicStepImpl implements NodeImplementation {
   constructor(
     private node: AtomicGraphNode,
-    private contextManager: WorkflowContextManager,
+    private stepExecutionRuntime: StepExecutionRuntime,
     private connectorExecutor: ConnectorExecutor,
     private workflowState: WorkflowExecutionRuntimeManager,
     private workflowLogger: IWorkflowEventLogger
@@ -45,7 +45,7 @@ export class AtomicStepImpl implements NodeImplementation {
     // for now it only calls ConnectorStepImpl
     await new ConnectorStepImpl(
       this.node.configuration,
-      this.contextManager,
+      this.stepExecutionRuntime,
       this.connectorExecutor,
       this.workflowState,
       this.workflowLogger

src/platform/plugins/shared/workflows_execution_engine/server/step/connector_step.ts

@@ -8,11 +8,11 @@
  */
 
 import type { ConnectorExecutor } from '../connector_executor';
-import type { WorkflowContextManager } from '../workflow_context_manager/workflow_context_manager';
 import type { WorkflowExecutionRuntimeManager } from '../workflow_context_manager/workflow_execution_runtime_manager';
 import type { IWorkflowEventLogger } from '../workflow_event_logger/workflow_event_logger';
 import type { RunStepResult, BaseStep } from './node_implementation';
 import { BaseAtomicNodeImplementation } from './node_implementation';
+import type { StepExecutionRuntime } from '../workflow_context_manager/step_execution_runtime';
 
 // Extend BaseStep for connector-specific properties
 export interface ConnectorStep extends BaseStep {
@@ -23,17 +23,17 @@ export interface ConnectorStep extends BaseStep {
 export class ConnectorStepImpl extends BaseAtomicNodeImplementation<ConnectorStep> {
   constructor(
     step: ConnectorStep,
-    contextManager: WorkflowContextManager,
+    stepExecutionRuntime: StepExecutionRuntime,
     connectorExecutor: ConnectorExecutor,
     workflowState: WorkflowExecutionRuntimeManager,
     private workflowLogger: IWorkflowEventLogger
   ) {
-    super(step, contextManager, connectorExecutor, workflowState);
+    super(step, stepExecutionRuntime, connectorExecutor, workflowState);
   }
 
   public getInput() {
     // Get current context for templating
-    const context = this.contextManager.getContext();
+    const context = this.stepExecutionRuntime.contextManager.getContext();
     // Render inputs from 'with'
     return Object.entries(this.step.with ?? {}).reduce((acc: Record<string, any>, [key, value]) => {
       if (typeof value === 'string') {

src/platform/plugins/shared/workflows_execution_engine/server/step/elasticsearch_action_step.ts

@@ -8,7 +8,7 @@
  */
 
 import { buildRequestFromConnector } from '@kbn/workflows';
-import type { WorkflowContextManager } from '../workflow_context_manager/workflow_context_manager';
+import type { StepExecutionRuntime } from '../workflow_context_manager/step_execution_runtime';
 import type { WorkflowExecutionRuntimeManager } from '../workflow_context_manager/workflow_execution_runtime_manager';
 import type { IWorkflowEventLogger } from '../workflow_event_logger/workflow_event_logger';
 import type { RunStepResult, BaseStep } from './node_implementation';
@@ -23,7 +23,7 @@ export interface ElasticsearchActionStep extends BaseStep {
 export class ElasticsearchActionStepImpl extends BaseAtomicNodeImplementation<ElasticsearchActionStep> {
   constructor(
     step: ElasticsearchActionStep,
-    contextManager: WorkflowContextManager,
+    contextManager: StepExecutionRuntime,
     workflowRuntime: WorkflowExecutionRuntimeManager,
     private workflowLogger: IWorkflowEventLogger
   ) {
@@ -32,7 +32,7 @@ export class ElasticsearchActionStepImpl extends BaseAtomicNodeImplementation<El
 
   public getInput() {
     // Get current context for templating
-    const context = this.contextManager.getContext();
+    const context = this.stepExecutionRuntime.contextManager.getContext();
     // Render inputs from 'with' - support both direct step.with and step.configuration.with
     const stepWith = this.step.with || (this.step as any).configuration?.with || {};
     return this.renderObjectTemplate(stepWith, context);
@@ -78,7 +78,7 @@ export class ElasticsearchActionStepImpl extends BaseAtomicNodeImplementation<El
       });
 
       // Get ES client (user-scoped if available, fallback otherwise)
-      const esClient = this.contextManager.getEsClientAsUser();
+      const esClient = this.stepExecutionRuntime.contextManager.getEsClientAsUser();
 
       // Generic approach like Dev Console - just forward the request to ES
       const result = await this.executeElasticsearchRequest(esClient, stepType, stepWith);

src/platform/plugins/shared/workflows_execution_engine/server/step/foreach_step/enter_foreach_node_impl.ts

@@ -11,27 +11,27 @@ import type { EnterForeachNode } from '@kbn/workflows/graph';
 import type { NodeImplementation } from '../node_implementation';
 import type { WorkflowExecutionRuntimeManager } from '../../workflow_context_manager/workflow_execution_runtime_manager';
 import type { IWorkflowEventLogger } from '../../workflow_event_logger/workflow_event_logger';
-import type { WorkflowContextManager } from '../../workflow_context_manager/workflow_context_manager';
+import type { StepExecutionRuntime } from '../../workflow_context_manager/step_execution_runtime';
 
 export class EnterForeachNodeImpl implements NodeImplementation {
   constructor(
     private node: EnterForeachNode,
     private wfExecutionRuntimeManager: WorkflowExecutionRuntimeManager,
-    private contextManager: WorkflowContextManager,
+    private stepExecutionRuntime: StepExecutionRuntime,
     private workflowLogger: IWorkflowEventLogger
   ) {}
 
   public async run(): Promise<void> {
-    if (!this.wfExecutionRuntimeManager.getCurrentStepState()) {
+    if (!this.stepExecutionRuntime.getCurrentStepState()) {
       await this.enterForeach();
     } else {
       await this.advanceIteration();
     }
   }
 
   private async enterForeach(): Promise<void> {
-    let foreachState = this.wfExecutionRuntimeManager.getCurrentStepState();
-    await this.wfExecutionRuntimeManager.startStep();
+    let foreachState = this.stepExecutionRuntime.getCurrentStepState();
+    await this.stepExecutionRuntime.startStep();
     const evaluatedItems = this.getItems();
 
     if (evaluatedItems.length === 0) {
@@ -41,11 +41,11 @@ export class EnterForeachNodeImpl implements NodeImplementation {
           workflow: { step_id: this.node.stepId },
         }
       );
-      await this.wfExecutionRuntimeManager.setCurrentStepState({
+      await this.stepExecutionRuntime.setCurrentStepState({
         items: [],
         total: 0,
       });
-      await this.wfExecutionRuntimeManager.finishStep();
+      await this.stepExecutionRuntime.finishStep();
       this.wfExecutionRuntimeManager.navigateToNode(this.node.exitNodeId);
       return;
     }
@@ -65,14 +65,14 @@ export class EnterForeachNodeImpl implements NodeImplementation {
       total: evaluatedItems.length,
     };
 
-    await this.wfExecutionRuntimeManager.setCurrentStepState(foreachState);
+    await this.stepExecutionRuntime.setCurrentStepState(foreachState);
     // Enter a new scope for the first iteration
     this.wfExecutionRuntimeManager.enterScope(foreachState.index!.toString());
     this.wfExecutionRuntimeManager.navigateToNextNode();
   }
 
   private async advanceIteration(): Promise<void> {
-    let foreachState = this.wfExecutionRuntimeManager.getCurrentStepState()!;
+    let foreachState = this.stepExecutionRuntime.getCurrentStepState()!;
     // Update items and index if they have changed
     const items = foreachState.items;
     const index = foreachState.index + 1;
@@ -85,7 +85,7 @@ export class EnterForeachNodeImpl implements NodeImplementation {
       total,
     };
     // Enter a new scope for the new iteration
-    await this.wfExecutionRuntimeManager.setCurrentStepState(foreachState);
+    await this.stepExecutionRuntime.setCurrentStepState(foreachState);
     this.wfExecutionRuntimeManager.enterScope(foreachState.index!.toString());
     this.wfExecutionRuntimeManager.navigateToNextNode();
   }
@@ -100,7 +100,7 @@ export class EnterForeachNodeImpl implements NodeImplementation {
     try {
       items = JSON.parse(this.node.configuration.foreach);
     } catch (error) {
-      const { value, pathExists } = this.contextManager.readContextPath(
+      const { value, pathExists } = this.stepExecutionRuntime.contextManager.readContextPath(
         this.node.configuration.foreach
       );
 

src/platform/plugins/shared/workflows_execution_engine/server/step/foreach_step/exit_foreach_node_impl.ts

@@ -11,18 +11,18 @@ import type { ExitForeachNode } from '@kbn/workflows/graph';
 import type { NodeImplementation } from '../node_implementation';
 import type { WorkflowExecutionRuntimeManager } from '../../workflow_context_manager/workflow_execution_runtime_manager';
 import type { IWorkflowEventLogger } from '../../workflow_event_logger/workflow_event_logger';
+import type { StepExecutionRuntime } from '../../workflow_context_manager/step_execution_runtime';
 
 export class ExitForeachNodeImpl implements NodeImplementation {
   constructor(
     private node: ExitForeachNode,
+    private stepExecutionRuntime: StepExecutionRuntime,
     private wfExecutionRuntimeManager: WorkflowExecutionRuntimeManager,
     private workflowLogger: IWorkflowEventLogger
   ) {}
 
   public async run(): Promise<void> {
-    // Exit the scope of the current iteration
-    this.wfExecutionRuntimeManager.exitScope();
-    const foreachState = this.wfExecutionRuntimeManager.getCurrentStepState();
+    const foreachState = this.stepExecutionRuntime.getCurrentStepState();
 
     if (!foreachState) {
       throw new Error(`Foreach state for step ${this.node.stepId} not found`);
@@ -33,7 +33,7 @@ export class ExitForeachNodeImpl implements NodeImplementation {
       return;
     }
 
-    await this.wfExecutionRuntimeManager.finishStep();
+    await this.stepExecutionRuntime.finishStep();
     this.workflowLogger.logDebug(
       `Exiting foreach step ${this.node.stepId} after processing all items.`,
       {