refactor(workflow-executor): encapsulate pending-data business logic in Runner (#1503)

Author: Scra3

packages/workflow-executor/CLAUDE.md

@@ -76,7 +76,9 @@ src/
 - **Privacy** — Zero client data leaves the client's infrastructure. `StepOutcome` is sent to the orchestrator and must NEVER contain client data. Privacy-sensitive information (e.g. AI reasoning) must stay in `StepExecutionData` (persisted in the RunStore, client-side only).
 - **Ports (IO injection)** — All external IO goes through injected port interfaces, keeping the core pure and testable.
 - **AI integration** — Uses `@langchain/core` (`BaseChatModel`, `DynamicStructuredTool`) for AI-powered steps. `ExecutionContext.model` is a `BaseChatModel`.
-- **Error hierarchy** — All domain errors must extend `WorkflowExecutorError` (defined in `src/errors.ts`). This ensures executors can distinguish domain errors (caught → error outcome) from infrastructure errors (uncaught → propagate to caller). Never throw a plain `Error` for a domain error case.
+- **Error hierarchy** — Two families of errors coexist in `src/errors.ts`:
+  - **Domain errors** (`extends WorkflowExecutorError`) — Thrown during step execution (e.g. `RecordNotFoundError`, `MissingToolCallError`). Caught by `base-step-executor.ts` and converted into `stepOutcome.error` sent to the orchestrator. All domain errors must extend `WorkflowExecutorError`.
+  - **Boundary errors** (`extends Error`) — Thrown outside step execution, at the HTTP or Runner layer (e.g. `RunNotFoundError`, `PendingDataNotFoundError`, `ConfigurationError`). Caught by the HTTP server and translated into HTTP status codes (404, 400, etc.). These intentionally do NOT extend `WorkflowExecutorError` to prevent `base-step-executor` from catching them as step failures.
 - **Dual error messages** — `WorkflowExecutorError` carries two messages: `message` (technical, for dev logs) and `userMessage` (human-readable, surfaced to the Forest Admin UI via `stepOutcome.error`). The mapping happens in a single place: `base-step-executor.ts` uses `error.userMessage` when building the error outcome. When adding a new error subclass, always provide a distinct `userMessage` oriented toward end-users (no collection names, field names, or AI internals). If `userMessage` is omitted in the constructor call, it falls back to `message`.
 - **displayName in AI tools** — All `DynamicStructuredTool` schemas and system message prompts must use `displayName`, never `fieldName`. `displayName` is a Forest Admin frontend feature that replaces the technical field/relation/action name with a product-oriented label configured by the Forest Admin admin. End users write their workflow prompts using these display names, not the underlying technical names. After an AI tool call returns display names, map them back to `fieldName`/`name` before using them in datasource operations (e.g. filtering record values, calling `getRecord`).
 - **No recovery/retry** — Once the executor returns a step result to the orchestrator, the step is considered executed. There is no mechanism to re-dispatch a step, so executors must NOT include recovery checks (e.g. checking the RunStore for cached results before executing). Each step executes exactly once.

packages/workflow-executor/src/errors.ts

@@ -215,3 +215,26 @@ export class RunNotFoundError extends Error {
     if (cause !== undefined) this.cause = cause;
   }
 }
+
+export class PendingDataNotFoundError extends Error {
+  constructor(runId: string, stepIndex: number) {
+    super(`Step ${stepIndex} in run "${runId}" not found or has no pending data`);
+    this.name = 'PendingDataNotFoundError';
+  }
+}
+
+/** Minimal mirror of ZodIssue — avoids importing Zod types into errors.ts. */
+export interface ValidationIssue {
+  path: (string | number)[];
+  message: string;
+  code: string;
+}
+
+export class InvalidPendingDataError extends WorkflowExecutorError {
+  readonly issues: ValidationIssue[];
+
+  constructor(issues: ValidationIssue[]) {
+    super('Invalid pending data', 'The request body is invalid.');
+    this.issues = issues;
+  }
+}

packages/workflow-executor/src/http/executor-http-server.ts

@@ -1,8 +1,6 @@
 import type { Logger } from '../ports/logger-port';
-import type { RunStore } from '../ports/run-store';
 import type { WorkflowPort } from '../ports/workflow-port';
 import type Runner from '../runner';
-import type { StepExecutionData } from '../types/step-execution-data';
 import type { Server } from 'http';
 
 import bodyParser from '@koa/bodyparser';
@@ -11,12 +9,10 @@ import http from 'http';
 import Koa from 'koa';
 import koaJwt from 'koa-jwt';
 
-import { RunNotFoundError } from '../errors';
-import patchBodySchemas from './pending-data-validators';
+import { InvalidPendingDataError, PendingDataNotFoundError, RunNotFoundError } from '../errors';
 
 export interface ExecutorHttpServerOptions {
   port: number;
-  runStore: RunStore;
   runner: Runner;
   authSecret: string;
   workflowPort: WorkflowPort;
@@ -142,9 +138,7 @@ export default class ExecutorHttpServer {
   }
 
   private async handleGetRun(ctx: Koa.Context): Promise<void> {
-    const { runId } = ctx.params;
-    const steps = await this.options.runStore.getStepExecutions(runId);
-
+    const steps = await this.options.runner.getRunStepExecutions(ctx.params.runId);
     ctx.body = { steps };
   }
 
@@ -179,36 +173,26 @@ export default class ExecutorHttpServer {
       return;
     }
 
-    const stepExecutions = await this.options.runStore.getStepExecutions(runId);
-    const execution = stepExecutions.find(e => e.stepIndex === stepIndex);
-    const schema = execution ? patchBodySchemas[execution.type] : undefined;
-
-    if (
-      !execution ||
-      !schema ||
-      !('pendingData' in execution) ||
-      execution.pendingData === undefined
-    ) {
-      ctx.status = 404;
-      ctx.body = { error: 'Step execution not found or has no pending data' };
+    try {
+      await this.options.runner.patchPendingData(runId, stepIndex, ctx.request.body);
+    } catch (err) {
+      if (err instanceof PendingDataNotFoundError) {
+        ctx.status = 404;
+        ctx.body = { error: 'Step execution not found or has no pending data' };
 
-      return;
-    }
+        return;
+      }
 
-    const parsed = schema.safeParse(ctx.request.body);
+      if (err instanceof InvalidPendingDataError) {
+        ctx.status = 400;
+        ctx.body = { error: 'Invalid request body', details: err.issues };
 
-    if (!parsed.success) {
-      ctx.status = 400;
-      ctx.body = { error: 'Invalid request body', details: parsed.error.issues };
+        return;
+      }
 
-      return;
+      throw err;
     }
 
-    await this.options.runStore.saveStepExecution(runId, {
-      ...execution,
-      pendingData: { ...(execution.pendingData as object), ...(parsed.data as object) },
-    } as StepExecutionData);
-
     ctx.status = 204;
   }
 }

…utor/src/http/pending-data-validators.ts …-executor/src/pending-data-validators.tspackages/workflow-executor/src/http/pending-data-validators.ts renamed to packages/workflow-executor/src/pending-data-validators.ts packages/workflow-executor/src/http/pending-data-validators.ts renamed to packages/workflow-executor/src/pending-data-validators.ts

@@ -1,4 +1,4 @@
-import type { StepExecutionData } from '../types/step-execution-data';
+import type { StepExecutionData } from './types/step-execution-data';
 
 import { z } from 'zod';
 

packages/workflow-executor/src/runner.ts

@@ -4,12 +4,19 @@ import type { Logger } from './ports/logger-port';
 import type { RunStore } from './ports/run-store';
 import type { McpConfiguration, WorkflowPort } from './ports/workflow-port';
 import type { PendingStepExecution, StepExecutionResult } from './types/execution';
+import type { StepExecutionData } from './types/step-execution-data';
 import type { AiClient, RemoteTool } from '@forestadmin/ai-proxy';
 
 import ConsoleLogger from './adapters/console-logger';
-import { RunNotFoundError, causeMessage } from './errors';
+import {
+  InvalidPendingDataError,
+  PendingDataNotFoundError,
+  RunNotFoundError,
+  causeMessage,
+} from './errors';
 import StepExecutorFactory from './executors/step-executor-factory';
 import ExecutorHttpServer from './http/executor-http-server';
+import patchBodySchemas from './pending-data-validators';
 import validateSecrets from './validate-secrets';
 
 export interface RunnerConfig {
@@ -64,7 +71,6 @@ export default class Runner {
       if (this.config.httpPort !== undefined && !this.httpServer) {
         const server = new ExecutorHttpServer({
           port: this.config.httpPort,
-          runStore: this.config.runStore,
           runner: this,
           authSecret: this.config.authSecret,
           workflowPort: this.config.workflowPort,
@@ -102,6 +108,41 @@ export default class Runner {
     // TODO: graceful drain of in-flight steps (out of scope PRD-223)
   }
 
+  async getRunStepExecutions(runId: string): Promise<StepExecutionData[]> {
+    return this.config.runStore.getStepExecutions(runId);
+  }
+
+  async patchPendingData(runId: string, stepIndex: number, body: unknown): Promise<void> {
+    const stepExecutions = await this.config.runStore.getStepExecutions(runId);
+    const execution = stepExecutions.find(e => e.stepIndex === stepIndex);
+    const schema = execution ? patchBodySchemas[execution.type] : undefined;
+
+    // pendingData is typed as T | undefined; null is not expected (RunStore never persists null)
+    // but `== null` guards against both for safety.
+    if (!execution || !schema || !('pendingData' in execution) || execution.pendingData == null) {
+      throw new PendingDataNotFoundError(runId, stepIndex);
+    }
+
+    const parsed = schema.safeParse(body);
+
+    if (!parsed.success) {
+      throw new InvalidPendingDataError(
+        parsed.error.issues.map(({ path, message, code }) => ({
+          path: path as (string | number)[],
+          message,
+          code,
+        })),
+      );
+    }
+
+    // Cast is safe: the type guard above ensures `execution` is the correct union branch,
+    // and patchBodySchemas[execution.type] only accepts keys valid for that branch.
+    await this.config.runStore.saveStepExecution(runId, {
+      ...execution,
+      pendingData: { ...(execution.pendingData as object), ...(parsed.data as object) },
+    } as StepExecutionData);
+  }
+
   async triggerPoll(runId: string): Promise<void> {
     const step = await this.config.workflowPort.getPendingStepExecutionsForRun(runId);