feat(workflow-executor): run steps once per record of a Multiple related-records step

CLAUDE.md

@@ -145,6 +145,7 @@ Decisions live in `docs/adr/`. Run `/adr` to record one.
 - [Bedrock is executor-only, and its models pass an allowlist](docs/adr/2026-09-15-bedrock-is-executor-only-and-its-models-pass-an-allowlist.md) — 2026-09-15-bedrock-is-executor-only-and-its-models-pass-an-allowlist [accepted]
 - [One Gateway serves the MCP and the API behind a shared OAuth route](docs/adr/2026-10-12-one-gateway-serves-the-mcp-and-the-api-behind-a-shared-oauth-route.md) — 2026-10-12-one-gateway-serves-the-mcp-and-the-api-behind-a-shared-oauth-route [accepted]
 - [The automation sweep sends record ids to the orchestrator, and decides nothing itself](docs/adr/2026-10-15-the-automation-sweep-sends-record-ids-to-the-orchestrator-and-decides.md) — 2026-10-15-the-automation-sweep-sends-record-ids-to-the-orchestrator-and-decides [accepted]
+- [A step running once per record renews its run lock with the orchestrator before each record](docs/adr/2026-10-21-a-step-running-once-per-record-renews-its-run-lock-with-the-orchestrat.md) — 2026-10-21-a-step-running-once-per-record-renews-its-run-lock-with-the-orchestrat [accepted]
 - [The MCP server's single `mcp:read` scope means "may use the MCP", not read-only](docs/adr/2026-10-26-the-mcp-server-s-single-mcp-read-scope-means-may-use-the-mcp-not-read.md) — 2026-10-26-the-mcp-server-s-single-mcp-read-scope-means-may-use-the-mcp-not-read [accepted]
 - [BFF OAuth sessions travel sealed in the client's tokens, and the Forest server owns refresh safety](docs/adr/2026-11-03-bff-oauth-sessions-travel-sealed-in-the-client-s-tokens-and-the-forest.md) — 2026-11-03-bff-oauth-sessions-travel-sealed-in-the-client-s-tokens-and-the-forest [accepted]
 <!-- ADR:INDEX:END -->

docs/adr/2026-10-21-a-step-running-once-per-record-renews-its-run-lock-with-the-orchestrat.md

@@ -0,0 +1,10 @@
+---
+status: accepted
+date: 2026-10-21
+tags: [workflow-executor, run-lock, multiple-related-records]
+affected_components: [workflow-executor]
+---
+
+# A step running once per record renews its run lock with the orchestrator before each record
+
+A step running once per record can outlast the orchestrator's 7-minute run lock, and an operator's abort must stop it before its next record. So before each record the runtime calls `POST /api/workflow-orchestrator/renew-run-lock` on forestadmin-server with the lock it holds: a 409 (the run was aborted or its lock lost) stops the step without reporting, and an orchestrator that cannot answer leaves the step running under the lock it already holds. Accepted cost: released runtimes call this route, so the orchestrator must keep serving it, and a run whose renewals go unanswered for 7 minutes can still be claimed by another runtime. Rejected: handing the run back and claiming it again about once a minute, which needs no route but lets an abort wait up to a minute while more records run, and one record per dispatch, the simplest loop but up to 30 seconds between records, 25 minutes for 50.

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

@@ -4,6 +4,7 @@ import type {
   AvailableRunDispatch,
   AvailableRunsBatch,
   MalformedRunInfo,
+  RenewRunLockResult,
   WorkflowPort,
 } from '../ports/workflow-port';
 import type { CollectionSchema } from '../types/validated/collection';
@@ -36,6 +37,7 @@ const ROUTES = {
   updateStep: '/api/workflow-orchestrator/update-step',
   executorMetadata: '/api/workflow-orchestrator/executor-metadata',
   releaseRun: '/api/workflow-orchestrator/release-run',
+  renewRunLock: '/api/workflow-orchestrator/renew-run-lock',
   collectionSchema: (collectionName: string, runId: string) =>
     `/api/workflow-orchestrator/collection-schema/${encodeURIComponent(
       collectionName,
@@ -51,6 +53,8 @@ function stripReferenceKey(name: string | undefined): string | undefined {
   return name?.includes('.') ? name.slice(0, name.lastIndexOf('.')) : name;
 }
 
+const LOCK_REFUSED_STATUS = 409;
+
 const RELEASE_ROUTE_MISSING =
   'The orchestrator cannot take back the runs this executor lets go; they resume once their ' +
   'lock expires. Expected while the executor runs ahead of the server.';
@@ -169,6 +173,7 @@ export default class ForestServerWorkflowPort implements WorkflowPort {
   async updateStepExecution(
     runId: string,
     stepOutcome: StepOutcome,
+    options?: { clearsError?: boolean },
   ): Promise<AvailableRunDispatch | null> {
     const run = await this.callPort(
       'updateStepExecution',
@@ -178,7 +183,7 @@ export default class ForestServerWorkflowPort implements WorkflowPort {
           'post',
           ROUTES.updateStep,
           {},
-          toUpdateStepRequest(runId, stepOutcome),
+          toUpdateStepRequest(runId, stepOutcome, options),
         ),
       { retry: true },
     );
@@ -227,6 +232,31 @@ export default class ForestServerWorkflowPort implements WorkflowPort {
     }
   }
 
+  async renewRunLock(runId: string, lockedAt: string): Promise<RenewRunLockResult> {
+    try {
+      const renewed = await ServerUtils.query<{ lockedAt: string }>(
+        this.options,
+        'post',
+        ROUTES.renewRunLock,
+        {},
+        { runId: Number(runId), lockedAt },
+      );
+
+      return { status: 'renewed', lockedAt: renewed.lockedAt };
+    } catch (error) {
+      if ((error as { status?: number })?.status === LOCK_REFUSED_STATUS) {
+        return { status: 'refused' };
+      }
+
+      this.logger('Warn', 'Failed to renew the lock of a run', {
+        runId,
+        error: extractErrorMessage(error),
+      });
+
+      return { status: 'unavailable' };
+    }
+  }
+
   async getCollectionSchema(collectionName: string, runId: string): Promise<CollectionSchema> {
     return this.callPort(
       'getCollectionSchema',

packages/workflow-executor/src/adapters/run-to-available-step-mapper.ts

@@ -221,6 +221,12 @@ function toStepUser(runId: number, profile: ServerUserProfile): StepUser {
   return toSharedStepUser(profile);
 }
 
+function carriesError(pending: ServerStepHistory): boolean {
+  const { error } = (pending.context ?? {}) as { error?: unknown };
+
+  return typeof error === 'string' && error.length > 0;
+}
+
 // Returns null when the run has no available step (terminal state or all done/cancelled).
 // Throws InvalidStepDefinitionError on missing required fields (collectionId, collectionName,
 // userProfile) or an unmappable step definition.
@@ -274,6 +280,7 @@ export default function toAvailableStepExecution(
     // user, since that one follows the browser.
     timezone: toProjectTimezone(run.timezone),
     ...(callScope && { callScope }),
+    ...(carriesError(pending) && { carriesError: true as const }),
   };
 
   // Defense against mapper bugs: zod asserts the shape we produce is what the domain expects,

packages/workflow-executor/src/adapters/step-outcome-to-update-step-mapper.ts

@@ -24,8 +24,18 @@ function toExecutionStatus(outcome: StepOutcome): ServerExecutionStatus {
 export default function toUpdateStepRequest(
   runId: string,
   outcome: StepOutcome,
+  options?: { clearsError?: boolean },
 ): ServerUpdateStepRequest {
   const context: Record<string, unknown> = { status: outcome.status };
+
+  // The orchestrator merges each report into the context the error left, and keeps a null where
+  // it drops an undefined key, so only a null clears it.
+  if (options?.clearsError && outcome.status !== 'error') {
+    context.error = null;
+    context.errorKind = null;
+    context.errorSourceStepIndex = null;
+  }
+
   if (outcome.error !== undefined) context.error = outcome.error;
   if (outcome.errorKind !== undefined) context.errorKind = outcome.errorKind;
 

packages/workflow-executor/src/errors.ts

@@ -687,6 +687,59 @@ export class SourceRecordMissingError extends WorkflowExecutorError {
   }
 }
 
+// One level of fan-out only: the editor and the orchestrator refuse it too, so only a hand-edited
+// or stale workflow gets here.
+export class MultipleOnPerRecordSourceError extends WorkflowConfigurationError {
+  constructor(stepTitle?: string) {
+    const step = stepTitle ? `"${stepTitle}"` : 'This step';
+    super(
+      `Load Related Record ${step} is set to Multiple on a source that already runs once per record`,
+      `${step} loads multiple records from a source that already runs once per record. Set it to "Single record".`,
+    );
+  }
+}
+
+export class MissingConditionQuantifierError extends WorkflowConfigurationError {
+  constructor(option: string, fieldName: string) {
+    super(
+      `Condition on "${fieldName}" for option "${option}" reads per-record data without a quantifier`,
+      `The option "${option}" reads "${fieldName}" from a step that runs once per record without saying whether any, every or no record must match. Reconfigure this decision step.`,
+    );
+  }
+}
+
+export class RecordNotOfferedError extends WorkflowExecutorError {
+  constructor(recordId: string) {
+    super(
+      `Record "${recordId}" was not among the candidates offered`,
+      'A record picked is not among those this step offered. Pick the records again.',
+    );
+  }
+}
+
+// The write-ahead marker was still set: the change was cut off after it started, by a restart or
+// the step timeout, so it may have landed and running it again could apply it twice.
+export class RecordChangeInterruptedError extends WorkflowExecutorError {
+  constructor() {
+    super(
+      'Record change was cut off after it started',
+      'The change on this record was cut off after it started and may already be applied. Check the record before you retry.',
+    );
+  }
+}
+
+// Not SourceRecordMissingError: the source loaded several records, which a Sub-workflow cannot take
+// until it runs once per record, so the remedy is the calling Sub-workflow step.
+export class SourceLoadsSeveralRecordsError extends WorkflowCallSiteError {
+  constructor(stepTitle?: string) {
+    const step = stepTitle ? `"${stepTitle}"` : 'a step';
+    super(
+      `Sub-workflow call pins ${step}, which loads several records`,
+      `The Sub-workflow step that called this workflow takes its record from ${step}, which loads several records. Set its record to a step that loads one.`,
+    );
+  }
+}
+
 // Not SourceRecordMissingError: a record was found and it is the wrong one, so the remedy is the
 // calling Sub-workflow step rather than anything in this workflow.
 export class SourceRecordCollectionMismatchError extends WorkflowCallSiteError {

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

@@ -31,6 +31,24 @@ import {
 import patchBodySchemas from '../http/pending-data-validators';
 import StepSummaryBuilder from './summary/step-summary-builder';
 
+// A step the executor ran has its execution at its own stepIndex. A revision clone never ran, so
+// it inherits the record of the step it copied (originalStepIndex).
+export function findStepExecution(
+  step: Step,
+  executions: StepExecutionData[],
+): StepExecutionData | undefined {
+  const own = executions.find(e => e.stepIndex === step.stepOutcome.stepIndex);
+  if (own) return own;
+
+  // An errored clone persists nothing, so inheriting the copied record would resurface the
+  // superseded original's — only fall back when the clone didn't run-and-fail.
+  if (step.originalStepIndex !== undefined && step.stepOutcome.status !== 'error') {
+    return executions.find(e => e.stepIndex === step.originalStepIndex);
+  }
+
+  return undefined;
+}
+
 export default abstract class BaseStepExecutor<TStep extends StepDefinition = StepDefinition>
   implements IStepExecutor
 {
@@ -257,6 +275,10 @@ export default abstract class BaseStepExecutor<TStep extends StepDefinition = St
       lines.push(`Step title: "${stepDefinition.title}"`);
     }
 
+    if (this.context.perRecordLabel) {
+      lines.push(`This step runs once per record. This run is for ${this.context.perRecordLabel}.`);
+    }
+
     return new SystemMessage(lines.join('\n'));
   }
 
@@ -283,16 +305,7 @@ export default abstract class BaseStepExecutor<TStep extends StepDefinition = St
     step: Step,
     executions: StepExecutionData[],
   ): StepExecutionData | undefined {
-    const own = executions.find(e => e.stepIndex === step.stepOutcome.stepIndex);
-    if (own) return own;
-
-    // An errored clone persists nothing, so inheriting the copied record would resurface the
-    // superseded original's — only fall back when the clone didn't run-and-fail.
-    if (step.originalStepIndex !== undefined && step.stepOutcome.status !== 'error') {
-      return executions.find(e => e.stepIndex === step.originalStepIndex);
-    }
-
-    return undefined;
+    return findStepExecution(step, executions);
   }
 
   private static mergeLeadingSystemMessages(messages: BaseMessage[]): BaseMessage[] {