refactor(workflow-executor): workflow steps

packages/ai-proxy/src/index.ts

@@ -8,6 +8,7 @@ export { default as ProviderDispatcher } from './provider-dispatcher';
 export * from './provider-dispatcher';
 export * from './ai-client';
 export * from './remote-tools';
+export { default as RemoteTool } from './remote-tool';
 export * from './router';
 export * from './mcp-client';
 export * from './oauth-token-injector';
@@ -16,3 +17,9 @@ export * from './errors';
 export function validMcpConfigurationOrThrow(mcpConfig: McpConfiguration) {
   return McpConfigChecker.check(mcpConfig);
 }
+
+export type { BaseChatModel } from '@langchain/core/language_models/chat_models';
+export type { BaseMessage } from '@langchain/core/messages';
+export { HumanMessage, SystemMessage } from '@langchain/core/messages';
+export type { StructuredToolInterface } from '@langchain/core/tools';
+export { DynamicStructuredTool } from '@langchain/core/tools';

packages/ai-proxy/test/ai-client.test.ts

@@ -23,9 +23,7 @@ describe('Model validation', () => {
     expect(
       () =>
         new AiClient({
-          aiConfigurations: [
-            { name: 'test', provider: 'openai', apiKey: 'dev', model: 'gpt-4' },
-          ],
+          aiConfigurations: [{ name: 'test', provider: 'openai', apiKey: 'dev', model: 'gpt-4' }],
         }),
     ).toThrow(AIModelNotSupportedError);
   });
@@ -34,9 +32,7 @@ describe('Model validation', () => {
     expect(
       () =>
         new AiClient({
-          aiConfigurations: [
-            { name: 'test', provider: 'openai', apiKey: 'dev', model: 'gpt-4o' },
-          ],
+          aiConfigurations: [{ name: 'test', provider: 'openai', apiKey: 'dev', model: 'gpt-4o' }],
         }),
     ).not.toThrow();
   });
@@ -143,9 +139,7 @@ describe('getModel', () => {
       'Warn',
       expect.stringContaining("AI configuration 'non-existent' not found"),
     );
-    expect(createBaseChatModelMock).toHaveBeenCalledWith(
-      expect.objectContaining({ name: 'gpt4' }),
-    );
+    expect(createBaseChatModelMock).toHaveBeenCalledWith(expect.objectContaining({ name: 'gpt4' }));
     expect(result).toBe(fakeModel);
   });
 });

packages/ai-proxy/test/errors.test.ts

@@ -17,9 +17,9 @@ import {
   AIProviderError,
   AIProviderUnavailableError,
   AITooManyRequestsError,
-  AIUnauthorizedError,
   AIToolNotFoundError,
   AIToolUnprocessableError,
+  AIUnauthorizedError,
   McpConfigError,
   McpConflictError,
   McpConnectionError,

packages/ai-proxy/test/langchain-adapter.test.ts

@@ -90,15 +90,11 @@ describe('LangChainAdapter', () => {
           {
             role: 'assistant',
             content: '',
-            tool_calls: [
-              { id: 'call_1', function: { name: 'my_tool', arguments: 'not-json' } },
-            ],
+            tool_calls: [{ id: 'call_1', function: { name: 'my_tool', arguments: 'not-json' } }],
           },
         ]),
       ).toThrow(
-        new AIBadRequestError(
-          "Invalid JSON in tool_calls arguments for tool 'my_tool': not-json",
-        ),
+        new AIBadRequestError("Invalid JSON in tool_calls arguments for tool 'my_tool': not-json"),
       );
     });
   });
@@ -256,5 +252,4 @@ describe('LangChainAdapter', () => {
       );
     });
   });
-
 });

packages/ai-proxy/test/provider-dispatcher.test.ts

@@ -181,9 +181,7 @@ describe('ProviderDispatcher', () => {
         const thrown = await dispatcher.dispatch(buildBody()).catch(e => e);
 
         expect(thrown).toBeInstanceOf(AIProviderUnavailableError);
-        expect(thrown.message).toBe(
-          'OpenAI server error (HTTP 500): Internal Server Error',
-        );
+        expect(thrown.message).toBe('OpenAI server error (HTTP 500): Internal Server Error');
         expect(thrown.provider).toBe('OpenAI');
         expect(thrown.providerStatusCode).toBe(500);
         expect(thrown.baseBusinessErrorName).toBe('InternalServerError');
@@ -468,9 +466,7 @@ describe('ProviderDispatcher', () => {
           .catch(e => e);
 
         expect(thrown).toBeInstanceOf(AIProviderUnavailableError);
-        expect(thrown.message).toBe(
-          'Anthropic server error (HTTP 503): Service Unavailable',
-        );
+        expect(thrown.message).toBe('Anthropic server error (HTTP 503): Service Unavailable');
         expect(thrown.provider).toBe('Anthropic');
         expect(thrown.providerStatusCode).toBe(503);
         expect(thrown.baseBusinessErrorName).toBe('InternalServerError');

packages/workflow-executor/CLAUDE.md

@@ -42,7 +42,7 @@ Front  ◀──▶  Orchestrator  ◀──pull/push──▶  Executor  ──
 
 ```
 src/
-├── errors.ts               # WorkflowExecutorError, MissingToolCallError, MalformedToolCallError, NoRecordsError, NoReadableFieldsError
+├── errors.ts               # WorkflowExecutorError, MissingToolCallError, MalformedToolCallError, NoRecordsError, NoReadableFieldsError, NoWritableFieldsError, NoActionsError, StepPersistenceError, NoRelationshipFieldsError, RelatedRecordNotFoundError
 ├── runner.ts               # Runner class — main entry point (start/stop/triggerPoll, HTTP server wiring)
 ├── types/                  # Core type definitions (@draft)
 │   ├── step-definition.ts  # StepType enum + step definition interfaces
@@ -60,7 +60,10 @@ src/
 ├── executors/              # Step executor implementations
 │   ├── base-step-executor.ts       # Abstract base class (context injection + shared helpers)
 │   ├── condition-step-executor.ts  # AI-powered condition step (chooses among options)
-│   └── read-record-step-executor.ts # AI-powered record field reading step
+│   ├── read-record-step-executor.ts  # AI-powered record field reading step
+│   ├── update-record-step-executor.ts # AI-powered record field update step (with confirmation flow)
+│   ├── trigger-record-action-step-executor.ts  # AI-powered action trigger step (with confirmation flow)
+│   └── load-related-record-step-executor.ts  # AI-powered relation loading step (with confirmation flow)
 ├── http/                   # HTTP server (optional, for frontend data access)
 │   └── executor-http-server.ts  # Koa server: GET /runs/:runId, POST /runs/:runId/trigger
 └── index.ts                # Barrel exports
@@ -73,7 +76,11 @@ 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.
+- **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.
+- **Fetched steps must be executed** — Any step retrieved from the orchestrator via `getPendingStepExecutions()` must be executed. Silently discarding a fetched step (e.g. filtering it out by `runId` after fetching) violates the executor contract: the orchestrator assumes execution is guaranteed once the step is dispatched. The only valid filter before executing is deduplication via `inFlightSteps` (to avoid running the same step twice concurrently).
 
 ## Commands
 

packages/workflow-executor/jest.config.ts

@@ -1,8 +1,17 @@
 /* eslint-disable import/no-relative-packages */
+import path from 'path';
+
 import jestConfig from '../../jest.config';
 
+// Jest < 30 doesn't resolve wildcard exports in package.json.
+// @anthropic-ai/sdk uses "./lib/*" exports that need this workaround.
+const anthropicSdkDir = path.dirname(require.resolve('@anthropic-ai/sdk'));
+
 export default {
   ...jestConfig,
   collectCoverageFrom: ['<rootDir>/src/**/*.ts'],
   testMatch: ['<rootDir>/test/**/*.test.ts'],
+  moduleNameMapper: {
+    '^@anthropic-ai/sdk/(.*)$': `${anthropicSdkDir}/$1`,
+  },
 };

packages/workflow-executor/package.json

@@ -23,10 +23,10 @@
     "test": "jest"
   },
   "dependencies": {
+    "@forestadmin/ai-proxy": "1.6.1",
     "@forestadmin/agent-client": "1.4.13",
     "@forestadmin/forestadmin-client": "1.37.17",
     "@koa/router": "^13.1.0",
-    "@langchain/core": "1.1.33",
     "koa": "^3.0.1",
     "zod": "4.3.6"
   },

packages/workflow-executor/src/adapters/agent-client-agent-port.ts

@@ -1,30 +1,36 @@
-import type { AgentPort } from '../ports/agent-port';
+import type {
+  AgentPort,
+  ExecuteActionQuery,
+  GetRecordQuery,
+  GetRelatedDataQuery,
+  UpdateRecordQuery,
+} from '../ports/agent-port';
 import type { CollectionSchema } from '../types/record';
 import type { RemoteAgentClient, SelectOptions } from '@forestadmin/agent-client';
 
 import { RecordNotFoundError } from '../errors';
 
 function buildPkFilter(
   primaryKeyFields: string[],
-  recordId: Array<string | number>,
+  id: Array<string | number>,
 ): SelectOptions['filters'] {
   if (primaryKeyFields.length === 1) {
-    return { field: primaryKeyFields[0], operator: 'Equal', value: recordId[0] };
+    return { field: primaryKeyFields[0], operator: 'Equal', value: id[0] };
   }
 
   return {
     aggregator: 'And',
     conditions: primaryKeyFields.map((field, i) => ({
       field,
       operator: 'Equal',
-      value: recordId[i],
+      value: id[i],
     })),
   };
 }
 
 // agent-client methods (update, relation, action) still expect the pipe-encoded string format
-function encodePk(recordId: Array<string | number>): string {
-  return recordId.map(v => String(v)).join('|');
+function encodePk(id: Array<string | number>): string {
+  return id.map(v => String(v)).join('|');
 }
 
 function extractRecordId(
@@ -46,44 +52,39 @@ export default class AgentClientAgentPort implements AgentPort {
     this.collectionSchemas = params.collectionSchemas;
   }
 
-  async getRecord(collectionName: string, recordId: Array<string | number>, fieldNames?: string[]) {
-    const schema = this.resolveSchema(collectionName);
-    const records = await this.client.collection(collectionName).list<Record<string, unknown>>({
-      filters: buildPkFilter(schema.primaryKeyFields, recordId),
+  async getRecord({ collection, id, fields }: GetRecordQuery) {
+    const schema = this.resolveSchema(collection);
+    const records = await this.client.collection(collection).list<Record<string, unknown>>({
+      filters: buildPkFilter(schema.primaryKeyFields, id),
       pagination: { size: 1, number: 1 },
-      ...(fieldNames?.length && { fields: fieldNames }),
+      ...(fields?.length && { fields }),
     });
 
     if (records.length === 0) {
-      throw new RecordNotFoundError(collectionName, encodePk(recordId));
+      throw new RecordNotFoundError(collection, encodePk(id));
     }
 
-    return { collectionName, recordId, values: records[0] };
+    return { collectionName: collection, recordId: id, values: records[0] };
   }
 
-  async updateRecord(
-    collectionName: string,
-    recordId: Array<string | number>,
-    values: Record<string, unknown>,
-  ) {
+  async updateRecord({ collection, id, values }: UpdateRecordQuery) {
     const updatedRecord = await this.client
-      .collection(collectionName)
-      .update<Record<string, unknown>>(encodePk(recordId), values);
+      .collection(collection)
+      .update<Record<string, unknown>>(encodePk(id), values);
 
-    return { collectionName, recordId, values: updatedRecord };
+    return { collectionName: collection, recordId: id, values: updatedRecord };
   }
 
-  async getRelatedData(
-    collectionName: string,
-    recordId: Array<string | number>,
-    relationName: string,
-  ) {
-    const relatedSchema = this.resolveSchema(relationName);
+  async getRelatedData({ collection, id, relation, limit, fields }: GetRelatedDataQuery) {
+    const relatedSchema = this.resolveSchema(relation);
 
     const records = await this.client
-      .collection(collectionName)
-      .relation(relationName, encodePk(recordId))
-      .list<Record<string, unknown>>();
+      .collection(collection)
+      .relation(relation, encodePk(id))
+      .list<Record<string, unknown>>({
+        ...(limit !== null && { pagination: { size: limit, number: 1 } }),
+        ...(fields?.length && { fields }),
+      });
 
     return records.map(record => ({
       collectionName: relatedSchema.collectionName,
@@ -92,17 +93,11 @@ export default class AgentClientAgentPort implements AgentPort {
     }));
   }
 
-  async executeAction(
-    collectionName: string,
-    actionName: string,
-    recordIds: Array<string | number>[],
-  ): Promise<unknown> {
-    const encodedIds = recordIds.map(id => encodePk(id));
-    const action = await this.client
-      .collection(collectionName)
-      .action(actionName, { recordIds: encodedIds });
-
-    return action.execute();
+  async executeAction({ collection, action, id }: ExecuteActionQuery): Promise<unknown> {
+    const encodedId = id?.length ? [encodePk(id)] : [];
+    const act = await this.client.collection(collection).action(action, { recordIds: encodedId });
+
+    return act.execute();
   }
 
   private resolveSchema(collectionName: string): CollectionSchema {

packages/workflow-executor/src/adapters/console-logger.ts

@@ -0,0 +1,7 @@
+import type { Logger } from '../ports/logger-port';
+
+export default class ConsoleLogger implements Logger {
+  error(message: string, context: Record<string, unknown>): void {
+    console.error(JSON.stringify({ message, timestamp: new Date().toISOString(), ...context }));
+  }
+}

packages/workflow-executor/src/adapters/forest-server-workflow-port.ts

@@ -9,6 +9,8 @@ import { ServerUtils } from '@forestadmin/forestadmin-client';
 // TODO: finalize route paths with the team — these are placeholders
 const ROUTES = {
   pendingStepExecutions: '/liana/v1/workflow-step-executions/pending',
+  pendingStepExecutionForRun: (runId: string) =>
+    `/liana/v1/workflow-step-executions/pending?runId=${encodeURIComponent(runId)}`,
   updateStepExecution: (runId: string) => `/liana/v1/workflow-step-executions/${runId}/complete`,
   collectionSchema: (collectionName: string) => `/liana/v1/collections/${collectionName}`,
   mcpServerConfigs: '/liana/mcp-server-configs-with-details',
@@ -29,6 +31,14 @@ export default class ForestServerWorkflowPort implements WorkflowPort {
     );
   }
 
+  async getPendingStepExecutionsForRun(runId: string): Promise<PendingStepExecution | null> {
+    return ServerUtils.query<PendingStepExecution | null>(
+      this.options,
+      'get',
+      ROUTES.pendingStepExecutionForRun(runId),
+    );
+  }
+
   async updateStepExecution(runId: string, stepOutcome: StepOutcome): Promise<void> {
     await ServerUtils.query(
       this.options,