feat(workflow-executor): add UpdateRecordStepExecutor with confirmation flow

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
 ├── 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,8 @@ 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)
 ├── 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

packages/workflow-executor/src/errors.ts

@@ -45,3 +45,9 @@ export class NoResolvedFieldsError extends WorkflowExecutorError {
     super(`None of the requested fields could be resolved: ${fieldNames.join(', ')}`);
   }
 }
+
+export class NoWritableFieldsError extends WorkflowExecutorError {
+  constructor(collectionName: string) {
+    super(`No writable fields on record from collection "${collectionName}"`);
+  }
+}

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

@@ -1,30 +1,66 @@
 import type { ExecutionContext, StepExecutionResult } from '../types/execution';
+import type { CollectionSchema, FieldSchema, RecordRef } from '../types/record';
 import type { StepDefinition } from '../types/step-definition';
-import type { StepExecutionData } from '../types/step-execution-data';
+import type {
+  LoadRelatedRecordStepExecutionData,
+  StepExecutionData,
+} from '../types/step-execution-data';
 import type { StepOutcome } from '../types/step-outcome';
 import type { AIMessage, BaseMessage } from '@langchain/core/messages';
-import type { DynamicStructuredTool } from '@langchain/core/tools';
 
-import { SystemMessage } from '@langchain/core/messages';
+import { HumanMessage, SystemMessage } from '@langchain/core/messages';
+import { DynamicStructuredTool } from '@langchain/core/tools';
+import { z } from 'zod';
 
-import { MalformedToolCallError, MissingToolCallError } from '../errors';
+import {
+  MalformedToolCallError,
+  MissingToolCallError,
+  NoRecordsError,
+  WorkflowExecutorError,
+} from '../errors';
 import { isExecutedStepOnExecutor } from '../types/step-execution-data';
 
 export default abstract class BaseStepExecutor<TStep extends StepDefinition = StepDefinition> {
   protected readonly context: ExecutionContext<TStep>;
 
+  protected readonly schemaCache = new Map<string, CollectionSchema>();
+
   constructor(context: ExecutionContext<TStep>) {
     this.context = context;
   }
 
   abstract execute(): Promise<StepExecutionResult>;
 
+  /** Find a field by displayName first, then fallback to fieldName. */
+  protected findField(schema: CollectionSchema, name: string): FieldSchema | undefined {
+    return (
+      schema.fields.find(f => f.displayName === name) ??
+      schema.fields.find(f => f.fieldName === name)
+    );
+  }
+
+  /** Builds a StepExecutionResult with the given status and optional error. */
+  protected buildOutcomeResult(
+    status: 'success' | 'error' | 'awaiting-input',
+    error?: string,
+  ): StepExecutionResult {
+    return {
+      stepOutcome: {
+        type: 'record-task',
+        stepId: this.context.stepId,
+        stepIndex: this.context.stepIndex,
+        status,
+        ...(error !== undefined && { error }),
+      },
+    };
+  }
+
   /**
    * Returns a SystemMessage array summarizing previously executed steps.
    * Empty array when there is no history. Ready to spread into a messages array.
    */
   protected async buildPreviousStepsMessages(): Promise<SystemMessage[]> {
-    if (!this.context.history.length) return [];
+    if (!this.context.previousSteps.length) return [];
 
     const summary = await this.summarizePreviousSteps();
 
@@ -40,7 +76,7 @@ export default abstract class BaseStepExecutor<TStep extends StepDefinition = St
   private async summarizePreviousSteps(): Promise<string> {
     const allStepExecutions = await this.context.runStore.getStepExecutions(this.context.runId);
 
-    return this.context.history
+    return this.context.previousSteps
       .map(({ stepDefinition, stepOutcome }) => {
         const execution = allStepExecutions.find(e => e.stepIndex === stepOutcome.stepIndex);
 
@@ -61,6 +97,8 @@ export default abstract class BaseStepExecutor<TStep extends StepDefinition = St
     if (isExecutedStepOnExecutor(execution)) {
       if (execution.executionParams !== undefined) {
         lines.push(`  Input: ${JSON.stringify(execution.executionParams)}`);
+      } else if (execution.type === 'update-record' && execution.pendingUpdate) {
+        lines.push(`  Pending: ${JSON.stringify(execution.pendingUpdate)}`);
       }
 
       if (execution.executionResult) {
@@ -107,4 +145,78 @@ export default abstract class BaseStepExecutor<TStep extends StepDefinition = St
 
     throw new MissingToolCallError();
   }
+
+  /** Returns baseRecordRef + any related records loaded by previous steps. */
+  protected async getAvailableRecordRefs(): Promise<RecordRef[]> {
+    const stepExecutions = await this.context.runStore.getStepExecutions();
+    const relatedRecords = stepExecutions
+      .filter((e): e is LoadRelatedRecordStepExecutionData => e.type === 'load-related-record')
+      .map(e => e.record);
+
+    return [this.context.baseRecordRef, ...relatedRecords];
+  }
+
+  /** Selects a record ref via AI when multiple are available, returns directly when only one. */
+  protected async selectRecordRef(
+    records: RecordRef[],
+    prompt: string | undefined,
+  ): Promise<RecordRef> {
+    if (records.length === 0) throw new NoRecordsError();
+    if (records.length === 1) return records[0];
+
+    const identifiers = await Promise.all(records.map(r => this.toRecordIdentifier(r)));
+    const identifierTuple = identifiers as [string, ...string[]];
+
+    const tool = new DynamicStructuredTool({
+      name: 'select-record',
+      description: 'Select the most relevant record for this workflow step.',
+      schema: z.object({
+        recordIdentifier: z.enum(identifierTuple),
+      }),
+      func: undefined,
+    });
+
+    const messages = [
+      ...(await this.buildPreviousStepsMessages()),
+      new SystemMessage(
+        'You are an AI agent selecting the most relevant record for a workflow step.\n' +
+          'Choose the record whose collection best matches the user request.\n' +
+          'Pay attention to the collection name of each record.',
+      ),
+      new HumanMessage(prompt ?? 'Select the most relevant record.'),
+    ];
+
+    const { recordIdentifier } = await this.invokeWithTool<{ recordIdentifier: string }>(
+      messages,
+      tool,
+    );
+
+    const selectedIndex = identifiers.indexOf(recordIdentifier);
+
+    if (selectedIndex === -1) {
+      throw new WorkflowExecutorError(
+        `AI selected record "${recordIdentifier}" which does not match any available record`,
+      );
+    }
+
+    return records[selectedIndex];
+  }
+
+  /** Fetches a collection schema from WorkflowPort, with caching. */
+  protected async getCollectionSchema(collectionName: string): Promise<CollectionSchema> {
+    const cached = this.schemaCache.get(collectionName);
+    if (cached) return cached;
+
+    const schema = await this.context.workflowPort.getCollectionSchema(collectionName);
+    this.schemaCache.set(collectionName, schema);
+
+    return schema;
+  }
+
+  /** Formats a record ref as "Step X - CollectionDisplayName #id". */
+  protected async toRecordIdentifier(record: RecordRef): Promise<string> {
+    const schema = await this.getCollectionSchema(record.collectionName);
+
+    return `Step ${record.stepIndex} - ${schema.collectionDisplayName} #${record.recordId}`;
+  }
 }

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

@@ -5,6 +5,7 @@ import { HumanMessage, SystemMessage } from '@langchain/core/messages';
 import { DynamicStructuredTool } from '@langchain/core/tools';
 import { z } from 'zod';
 
+import { WorkflowExecutorError } from '../errors';
 import BaseStepExecutor from './base-step-executor';
 
 interface GatewayToolArgs {
@@ -66,16 +67,20 @@ export default class ConditionStepExecutor extends BaseStepExecutor<ConditionSte
 
     try {
       args = await this.invokeWithTool<GatewayToolArgs>(messages, tool);
-    } catch (error: unknown) {
-      return {
-        stepOutcome: {
-          type: 'condition',
-          stepId: this.context.stepId,
-          stepIndex: this.context.stepIndex,
-          status: 'error',
-          error: (error as Error).message,
-        },
-      };
+    } catch (error) {
+      if (error instanceof WorkflowExecutorError) {
+        return {
+          stepOutcome: {
+            type: 'condition',
+            stepId: this.context.stepId,
+            stepIndex: this.context.stepIndex,
+            status: 'error',
+            error: error.message,
+          },
+        };
+      }
+
+      throw error;
     }
 
     const { option: selectedOption, reasoning } = args;