feat(workflow-executor): add buildInMemoryExecutor and buildDatabaseExecutor factories (#1510)

Author: Scra3

WORKFLOW-EXECUTOR-CONTRACT.md

@@ -1,7 +1,7 @@
 # Workflow Executor — Contract Types
 
 > Types exchanged between the **orchestrator (server)**, the **executor (agent-nodejs)**, and the **frontend**.
-> Last updated: 2026-03-25
+> Last updated: 2026-03-26
 
 ---
 
@@ -12,18 +12,59 @@
 The executor polls for the current pending step of a run. The server must return **one object** (not an array), or `null` if the run is not found.
 
 ```typescript
+interface StepUser {
+  id:              number;
+  email:           string;
+  firstName:       string;
+  lastName:        string;
+  team:            string;
+  renderingId:     number;
+  role:            string;
+  permissionLevel: string;
+  tags:            Record<string, string>;
+}
+
 interface PendingStepExecution {
   runId:          string;
   stepId:         string;
   stepIndex:      number;
   baseRecordRef:  RecordRef;
   stepDefinition: StepDefinition;
   previousSteps:  Step[];
+  user:           StepUser;  // identity of the user who initiated the step
 }
 ```
 
 > **`null` response** → executor throws `RunNotFoundError` → HTTP 404 returned to caller.
 
+### CollectionSchema
+
+Schema of a collection, returned by the orchestrator via `GET /liana/v1/collections/:collectionName`. Used by the executor to resolve primary keys and action endpoints.
+
+```typescript
+interface CollectionSchema {
+  collectionName:        string;
+  collectionDisplayName: string;
+  primaryKeyFields:      string[];
+  fields:                FieldSchema[];
+  actions:               ActionSchema[];
+}
+
+interface FieldSchema {
+  fieldName:              string;
+  displayName:            string;
+  isRelationship:         boolean;
+  relationType?:          "BelongsTo" | "HasMany" | "HasOne";
+  relatedCollectionName?: string;
+}
+
+interface ActionSchema {
+  name:        string;
+  displayName: string;
+  endpoint:    string;  // route path used by the agent to execute the action
+}
+```
+
 ### RecordRef
 
 Lightweight pointer to a specific record.
@@ -97,7 +138,7 @@ interface ConditionStepOutcome {
   type:            "condition";
   stepId:          string;
   stepIndex:       number;
-  status:          "success" | "error" | "manual-decision";
+  status:          "success" | "error";
   selectedOption?: string; // present when status = "success"
   error?:          string; // present when status = "error"
 }

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

@@ -5,8 +5,13 @@ import type {
   GetRelatedDataQuery,
   UpdateRecordQuery,
 } from '../ports/agent-port';
-import type { CollectionSchema } from '../types/record';
-import type { RemoteAgentClient, SelectOptions } from '@forestadmin/agent-client';
+import type SchemaCache from '../schema-cache';
+import type { StepUser } from '../types/execution';
+import type { CollectionSchema, RecordData } from '../types/record';
+import type { SelectOptions } from '@forestadmin/agent-client';
+
+import { createRemoteAgentClient } from '@forestadmin/agent-client';
+import jsonwebtoken from 'jsonwebtoken';
 
 import { RecordNotFoundError } from '../errors';
 
@@ -41,20 +46,20 @@ function extractRecordId(
 }
 
 export default class AgentClientAgentPort implements AgentPort {
-  private readonly client: RemoteAgentClient;
-  private readonly collectionSchemas: Record<string, CollectionSchema>;
-
-  constructor(params: {
-    client: RemoteAgentClient;
-    collectionSchemas: Record<string, CollectionSchema>;
-  }) {
-    this.client = params.client;
-    this.collectionSchemas = params.collectionSchemas;
+  private readonly agentUrl: string;
+  private readonly authSecret: string;
+  private readonly schemaCache: SchemaCache;
+
+  constructor(params: { agentUrl: string; authSecret: string; schemaCache: SchemaCache }) {
+    this.agentUrl = params.agentUrl;
+    this.authSecret = params.authSecret;
+    this.schemaCache = params.schemaCache;
   }
 
-  async getRecord({ collection, id, fields }: GetRecordQuery) {
+  async getRecord({ collection, id, fields }: GetRecordQuery, user: StepUser): Promise<RecordData> {
+    const client = this.createClient(user);
     const schema = this.resolveSchema(collection);
-    const records = await this.client.collection(collection).list<Record<string, unknown>>({
+    const records = await client.collection(collection).list<Record<string, unknown>>({
       filters: buildPkFilter(schema.primaryKeyFields, id),
       pagination: { size: 1, number: 1 },
       ...(fields?.length && { fields }),
@@ -67,18 +72,26 @@ export default class AgentClientAgentPort implements AgentPort {
     return { collectionName: collection, recordId: id, values: records[0] };
   }
 
-  async updateRecord({ collection, id, values }: UpdateRecordQuery) {
-    const updatedRecord = await this.client
+  async updateRecord(
+    { collection, id, values }: UpdateRecordQuery,
+    user: StepUser,
+  ): Promise<RecordData> {
+    const client = this.createClient(user);
+    const updatedRecord = await client
       .collection(collection)
       .update<Record<string, unknown>>(encodePk(id), values);
 
     return { collectionName: collection, recordId: id, values: updatedRecord };
   }
 
-  async getRelatedData({ collection, id, relation, limit, fields }: GetRelatedDataQuery) {
+  async getRelatedData(
+    { collection, id, relation, limit, fields }: GetRelatedDataQuery,
+    user: StepUser,
+  ): Promise<RecordData[]> {
+    const client = this.createClient(user);
     const relatedSchema = this.resolveSchema(relation);
 
-    const records = await this.client
+    const records = await client
       .collection(collection)
       .relation(relation, encodePk(id))
       .list<Record<string, unknown>>({
@@ -93,26 +106,55 @@ export default class AgentClientAgentPort implements AgentPort {
     }));
   }
 
-  async executeAction({ collection, action, id }: ExecuteActionQuery): Promise<unknown> {
+  async executeAction(
+    { collection, action, id }: ExecuteActionQuery,
+    user: StepUser,
+  ): Promise<unknown> {
+    const client = this.createClient(user);
     const encodedId = id?.length ? [encodePk(id)] : [];
-    const act = await this.client.collection(collection).action(action, { recordIds: encodedId });
+    const act = await client.collection(collection).action(action, { recordIds: encodedId });
 
     return act.execute();
   }
 
-  private resolveSchema(collectionName: string): CollectionSchema {
-    const schema = this.collectionSchemas[collectionName];
+  private createClient(user: StepUser) {
+    const token = jsonwebtoken.sign({ ...user, scope: 'step-execution' }, this.authSecret, {
+      expiresIn: '5m',
+    });
+
+    return createRemoteAgentClient({
+      url: this.agentUrl,
+      token,
+      actionEndpoints: this.buildActionEndpoints(),
+    });
+  }
+
+  private buildActionEndpoints() {
+    const endpoints: Record<string, Record<string, { name: string; endpoint: string }>> = {};
+
+    for (const [collectionName, schema] of this.schemaCache) {
+      endpoints[collectionName] = {};
 
-    if (!schema) {
-      return {
+      for (const action of schema.actions) {
+        endpoints[collectionName][action.name] = {
+          name: action.name,
+          endpoint: action.endpoint,
+        };
+      }
+    }
+
+    return endpoints;
+  }
+
+  private resolveSchema(collectionName: string): CollectionSchema {
+    return (
+      this.schemaCache.get(collectionName) ?? {
         collectionName,
         collectionDisplayName: collectionName,
         primaryKeyFields: ['id'],
         fields: [],
         actions: [],
-      };
-    }
-
-    return schema;
+      }
+    );
   }
 }

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

@@ -1,5 +1,5 @@
 import type { McpConfiguration, WorkflowPort } from '../ports/workflow-port';
-import type { PendingStepExecution } from '../types/execution';
+import type { PendingStepExecution, StepUser } from '../types/execution';
 import type { CollectionSchema } from '../types/record';
 import type { StepOutcome } from '../types/step-outcome';
 import type { HttpOptions } from '@forestadmin/forestadmin-client';
@@ -62,10 +62,8 @@ export default class ForestServerWorkflowPort implements WorkflowPort {
   }
 
   // eslint-disable-next-line @typescript-eslint/no-unused-vars
-  async hasRunAccess(_runId: string, _userToken: string): Promise<boolean> {
+  async hasRunAccess(_runId: string, _user: StepUser): Promise<boolean> {
     // TODO: implement once GET /liana/v1/workflow-runs/:runId/access is available.
-    // When live: call ServerUtils.query with extra header 'forest-user-token': userToken
-    // to let the orchestrator verify ownership.
     return true;
   }
 }

packages/workflow-executor/src/build-workflow-executor.ts

@@ -0,0 +1,85 @@
+import type { Logger } from './ports/logger-port';
+import type { AiConfiguration } from '@forestadmin/ai-proxy';
+import type { Options as SequelizeOptions } from 'sequelize';
+
+import { AiClient } from '@forestadmin/ai-proxy';
+import { Sequelize } from 'sequelize';
+
+import AgentClientAgentPort from './adapters/agent-client-agent-port';
+import ForestServerWorkflowPort from './adapters/forest-server-workflow-port';
+import Runner from './runner';
+import SchemaCache from './schema-cache';
+import DatabaseStore from './stores/database-store';
+import InMemoryStore from './stores/in-memory-store';
+
+const DEFAULT_FOREST_SERVER_URL = 'https://api.forestadmin.com';
+const DEFAULT_POLLING_INTERVAL_MS = 5000;
+
+export interface WorkflowExecutor {
+  start(): Promise<void>;
+  stop(): Promise<void>;
+}
+
+export interface ExecutorOptions {
+  envSecret: string;
+  authSecret: string;
+  agentUrl: string;
+  forestServerUrl?: string;
+  aiConfigurations: AiConfiguration[];
+  pollingIntervalMs?: number;
+  httpPort?: number;
+  logger?: Logger;
+}
+
+export type DatabaseExecutorOptions = ExecutorOptions &
+  ({ database: SequelizeOptions & { uri: string } } | { database: SequelizeOptions });
+
+function buildCommonDependencies(options: ExecutorOptions) {
+  const forestServerUrl = options.forestServerUrl ?? DEFAULT_FOREST_SERVER_URL;
+
+  const workflowPort = new ForestServerWorkflowPort({
+    envSecret: options.envSecret,
+    forestServerUrl,
+  });
+
+  const aiClient = new AiClient({
+    aiConfigurations: options.aiConfigurations,
+  });
+
+  const schemaCache = new SchemaCache();
+
+  const agentPort = new AgentClientAgentPort({
+    agentUrl: options.agentUrl,
+    authSecret: options.authSecret,
+    schemaCache,
+  });
+
+  return {
+    agentPort,
+    schemaCache,
+    workflowPort,
+    aiClient,
+    pollingIntervalMs: options.pollingIntervalMs ?? DEFAULT_POLLING_INTERVAL_MS,
+    envSecret: options.envSecret,
+    authSecret: options.authSecret,
+    httpPort: options.httpPort,
+    logger: options.logger,
+  };
+}
+
+export function buildInMemoryExecutor(options: ExecutorOptions): WorkflowExecutor {
+  return new Runner({
+    ...buildCommonDependencies(options),
+    runStore: new InMemoryStore(),
+  });
+}
+
+export function buildDatabaseExecutor(options: DatabaseExecutorOptions): WorkflowExecutor {
+  const { uri, ...sequelizeOptions } = options.database as SequelizeOptions & { uri?: string };
+  const sequelize = uri ? new Sequelize(uri, sequelizeOptions) : new Sequelize(sequelizeOptions);
+
+  return new Runner({
+    ...buildCommonDependencies(options),
+    runStore: new DatabaseStore({ sequelize }),
+  });
+}

packages/workflow-executor/src/errors.ts

@@ -216,6 +216,13 @@ export class RunNotFoundError extends Error {
   }
 }
 
+export class UserMismatchError extends Error {
+  constructor(runId: string) {
+    super(`User not authorized for run "${runId}"`);
+    this.name = 'UserMismatchError';
+  }
+}
+
 export class PendingDataNotFoundError extends Error {
   constructor(runId: string, stepIndex: number) {
     super(`Step ${stepIndex} in run "${runId}" not found or has no pending data`);

packages/workflow-executor/src/executors/base-step-executor.ts

@@ -29,8 +29,6 @@ export default abstract class BaseStepExecutor<TStep extends StepDefinition = St
 
   protected readonly agentPort: AgentPort;
 
-  protected readonly schemaCache = new Map<string, CollectionSchema>();
-
   constructor(context: ExecutionContext<TStep>) {
     this.context = context;
     this.agentPort = new SafeAgentPort(context.agentPort);
@@ -260,13 +258,13 @@ export default abstract class BaseStepExecutor<TStep extends StepDefinition = St
     return records[selectedIndex];
   }
 
-  /** Fetches a collection schema from WorkflowPort, with caching. */
+  /** Fetches a collection schema from WorkflowPort, with TTL-based caching. */
   protected async getCollectionSchema(collectionName: string): Promise<CollectionSchema> {
-    const cached = this.schemaCache.get(collectionName);
+    const cached = this.context.schemaCache.get(collectionName);
     if (cached) return cached;
 
     const schema = await this.context.workflowPort.getCollectionSchema(collectionName);
-    this.schemaCache.set(collectionName, schema);
+    this.context.schemaCache.set(collectionName, schema);
 
     return schema;
   }

packages/workflow-executor/src/executors/condition-step-executor.ts

@@ -1,6 +1,6 @@
 import type { StepExecutionResult } from '../types/execution';
 import type { ConditionStepDefinition } from '../types/step-definition';
-import type { ConditionStepStatus } from '../types/step-outcome';
+import type { BaseStepStatus } from '../types/step-outcome';
 
 import { DynamicStructuredTool, HumanMessage, SystemMessage } from '@forestadmin/ai-proxy';
 import { z } from 'zod';
@@ -38,7 +38,7 @@ const GATEWAY_SYSTEM_PROMPT = `You are an AI agent selecting the correct option
 
 export default class ConditionStepExecutor extends BaseStepExecutor<ConditionStepDefinition> {
   protected buildOutcomeResult(outcome: {
-    status: ConditionStepStatus;
+    status: BaseStepStatus;
     error?: string;
     selectedOption?: string;
   }): StepExecutionResult {
@@ -97,7 +97,10 @@ export default class ConditionStepExecutor extends BaseStepExecutor<ConditionSte
     }
 
     if (!selectedOption) {
-      return this.buildOutcomeResult({ status: 'manual-decision' });
+      return this.buildOutcomeResult({
+        status: 'error',
+        error: "The AI couldn't decide. Try rephrasing the step's prompt.",
+      });
     }
 
     return this.buildOutcomeResult({ status: 'success', selectedOption });