fix(workflow-executor): report runs that fail to hydrate instead of dropping them

packages/workflow-executor/CLAUDE.md

@@ -48,6 +48,7 @@ Front  ◀──▶  Orchestrator  ◀──pull/push──▶  Executor (this p
 - **`displayName` vs technical name** — AI tools/prompts use `displayName` (the admin-configured label end users write against), never `fieldName`. Map AI-returned display names back to technical names before any datasource op.
 - **Idempotency (mutating steps: update-record, trigger-action, mcp)** — write-ahead log in the RunStore: save `idempotencyPhase: 'executing'` before the side effect, `'done'` + `executionResult` after. On re-dispatch `(runId, stepIndex)`: `done` → rebuild success outcome without re-running or re-logging; `executing` → throw `StepStateError`. `checkIdempotency()` runs before `doExecute()`; the `executing` marker is set in the `beforeCall` thunk passed to `AgentWithLog` (after `createPending`) so a log-creation failure leaves no orphan marker. Non-mutating steps don't override it (replay is safe).
 - **Fetched steps must execute** — any step from `getAvailableRuns()` must run; silently dropping one breaks the orchestrator contract. The only allowed pre-filter is `inFlightRuns` dedup (keyed by `runId`, not step — a chain advances `stepId`).
+- **A run we can't execute must be reported, not dropped** — `/pending-run` claims a run *before* the executor has hydrated it, and only an outcome (or the 60s reaper) clears that claim. So every hydration failure in the port buckets as `malformed` — **including non-`WorkflowExecutorError` ones** (raw ZodError/TypeError get a generic `userMessage`) — and `getAvailableRun` wraps them in `MalformedRunError` so the Runner reports on the trigger path too. Dropping instead means: reaped after 60s, re-claimed, forever (PRD-956: two prod runs cycled ~2 600 times over 3 days). The one unreportable case is `toDispatch` returning `null` (no available step to attach an error to) — log it loudly.
 - **Auto-chain** — `WorkflowPort.updateStepExecution` returns the next dispatch (or `null`); the Runner runs it inline instead of waiting for the next poll. Exits on `null` / non-progressing `stepIndex` / `maxChainDepth` (default 50) / `stop()`. Each step uses its own dispatch's `forestServerToken`. `/update-step` is retried on transient failures → the orchestrator **must** dedupe identical `(runId, stepIndex)` outcomes (server-side idempotency) to avoid double side-effects.
 - **Revise-safety** — on revision the orchestrator marks the pivot `revised`, later entries `cancelled`, then appends clones (`originalStepIndex` → source) + a fresh re-exec of the revised step. Consumers of `workflowHistory` must keep only the live path (`!revised && !cancelled`). To find a step's RunStore record: own `stepIndex` first, then fall back to `originalStepIndex`. Never key on `stepName` (LinkTo loops repeat names).
 - **Boundary validation** — wire/mapper types live in `types/validated/` as zod. Strictness by origin: executor-produced + frontend bodies use `.strict()`; the orchestrator collection schema **strips** unknowns and asserts step-specific props at use-time (resilient to orchestrator drift). Parse failure → `DomainValidationError`/`InvalidStepDefinitionError`. `StepOutcome` is validated only when it arrives via `previousSteps`; executor outputs are trusted by construction.

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

@@ -69,11 +69,22 @@ export default class ForestServerWorkflowPort implements WorkflowPort {
     for (const run of runs) {
       try {
         const dispatch = this.toDispatch(run);
-        if (dispatch) pending.push(dispatch);
-      } catch (error) {
-        if (error instanceof WorkflowExecutorError) {
-          malformed.push(this.toMalformedInfo(run, error));
+
+        if (dispatch) {
+          pending.push(dispatch);
         } else {
+          // Reporting is impossible here: there is no available step to attach an error to.
+          this.logger('Error', 'Pending run served with no executable step — dropped', {
+            runId: run.id,
+            lastStepIndex: run.workflowHistory?.at(-1)?.stepIndex,
+          });
+        }
+      } catch (error) {
+        // Every hydration failure must be reported, including non-domain ones: an unreported run
+        // keeps its claim, gets reaped after 60s and re-claimed forever (PRD-956).
+        malformed.push(this.toMalformedInfo(run, error));
+
+        if (!(error instanceof WorkflowExecutorError)) {
           this.logger('Error', 'Failed to hydrate pending run — unexpected error', {
             runId: run.id,
             error: extractErrorMessage(error),
@@ -99,12 +110,14 @@ export default class ForestServerWorkflowPort implements WorkflowPort {
     try {
       return this.toDispatch(run);
     } catch (error) {
-      if (error instanceof WorkflowExecutorError) {
-        throw new MalformedRunError(this.toMalformedInfo(run, error));
+      if (!(error instanceof WorkflowExecutorError)) {
+        this.logger('Error', 'Failed to hydrate run — unexpected error', {
+          runId: run.id,
+          error: extractErrorMessage(error),
+        });
       }
 
-      /* istanbul ignore next — defensive fallback for unexpected non-domain errors */
-      throw error;
+      throw new MalformedRunError(this.toMalformedInfo(run, error));
     }
   }
 
@@ -133,18 +146,20 @@ export default class ForestServerWorkflowPort implements WorkflowPort {
     return { step, auth: { forestServerToken: token } };
   }
 
-  private toMalformedInfo(
-    run: ServerHydratedWorkflowRun,
-    err: WorkflowExecutorError,
-  ): MalformedRunInfo {
-    const pending = run.workflowHistory.at(-1) ?? null;
+  private toMalformedInfo(run: ServerHydratedWorkflowRun, err: unknown): MalformedRunInfo {
+    // Array.isArray, not `?.`: this runs inside a catch block and a malformed workflowHistory is
+    // exactly what lands here. `?.` would still throw on {} and silently index a string.
+    const pending = Array.isArray(run.workflowHistory) ? run.workflowHistory.at(-1) ?? null : null;
 
     return {
       runId: String(run.id),
       stepId: pending?.stepName ?? null,
       stepIndex: pending?.stepIndex ?? null,
-      userMessage: err.userMessage,
-      technicalMessage: err.message,
+      userMessage:
+        err instanceof WorkflowExecutorError
+          ? err.userMessage
+          : 'This step could not be loaded and cannot be executed.',
+      technicalMessage: extractErrorMessage(err) ?? 'Unknown error',
     };
   }
 

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

@@ -116,7 +116,9 @@ describe('ForestServerWorkflowPort', () => {
       expect(result.malformed).toEqual([]);
     });
 
-    it('filters out runs with no available step', async () => {
+    it('drops runs with no available step and logs it (no step to report an error against)', async () => {
+      const logger = jest.fn();
+      const portWithLogger = new ForestServerWorkflowPort({ ...options, logger });
       const terminalRun = makeRun({
         workflowHistory: [
           {
@@ -133,10 +135,15 @@ describe('ForestServerWorkflowPort', () => {
       });
       mockQuery.mockResolvedValue([terminalRun]);
 
-      const result = await port.getAvailableRuns();
+      const result = await portWithLogger.getAvailableRuns();
 
       expect(result.pending).toEqual([]);
       expect(result.malformed).toEqual([]);
+      expect(logger).toHaveBeenCalledWith(
+        'Error',
+        'Pending run served with no executable step — dropped',
+        { runId: 42, lastStepIndex: 0 },
+      );
     });
 
     it('bucketizes malformed runs and keeps valid ones in the pending bucket', async () => {
@@ -313,24 +320,66 @@ describe('ForestServerWorkflowPort', () => {
       );
     });
 
-    it('logs and skips when the mapping throws a non-WorkflowExecutorError', async () => {
+    it('reports and logs when the mapping throws a non-WorkflowExecutorError', async () => {
       const logger = jest.fn();
       const portWithLogger = new ForestServerWorkflowPort({ ...options, logger });
-      // Simulate a non-domain error by passing a run whose workflowHistory will
-      // blow up a pure JS operation inside the mapper (missing `find` on non-array).
-      const brokenRun = { ...makeRun({ id: 111 }), workflowHistory: null as never };
+      // A non-array object, not null: it blows up the mapper the same way, but also makes
+      // toMalformedInfo throw a second time if that guards with `?.` instead of Array.isArray —
+      // which would abort the whole batch rather than reporting this run.
+      const brokenRun = { ...makeRun({ id: 111 }), workflowHistory: {} as never };
       mockQuery.mockResolvedValue([brokenRun]);
 
       const result = await portWithLogger.getAvailableRuns();
 
       expect(result.pending).toEqual([]);
-      expect(result.malformed).toEqual([]);
+      expect(result.malformed).toEqual([
+        {
+          runId: '111',
+          stepId: null,
+          stepIndex: null,
+          userMessage: 'This step could not be loaded and cannot be executed.',
+          technicalMessage: expect.any(String),
+        },
+      ]);
       expect(logger).toHaveBeenCalledWith(
         'Error',
         'Failed to hydrate pending run — unexpected error',
         expect.objectContaining({ runId: 111 }),
       );
     });
+
+    it('reports a non-domain failure against the pending step so the claim can be cleared', async () => {
+      const brokenRun = makeRun({
+        id: 956,
+        workflowHistory: [
+          {
+            stepName: 'done-step',
+            stepIndex: 0,
+            done: true,
+            stepDefinition: makeConditionStepDef(),
+          },
+          {
+            stepName: 'Task_UpdateGoogleSheet',
+            stepIndex: 1,
+            done: false,
+            stepDefinition: undefined as never,
+          },
+        ],
+      });
+      mockQuery.mockResolvedValue([brokenRun]);
+
+      const result = await port.getAvailableRuns();
+
+      expect(result.pending).toEqual([]);
+      expect(result.malformed[0]).toEqual(
+        expect.objectContaining({
+          runId: '956',
+          stepId: 'Task_UpdateGoogleSheet',
+          stepIndex: 1,
+          userMessage: 'This step could not be loaded and cannot be executed.',
+        }),
+      );
+    });
   });
 
   describe('getAvailableRun', () => {
@@ -393,6 +442,27 @@ describe('ForestServerWorkflowPort', () => {
 
       await expect(port.getAvailableRun('66')).rejects.toBeInstanceOf(MalformedRunError);
     });
+
+    it('wraps a non-WorkflowExecutorError in MalformedRunError so the Runner still reports', async () => {
+      const logger = jest.fn();
+      const portWithLogger = new ForestServerWorkflowPort({ ...options, logger });
+      mockQuery.mockResolvedValue({ ...makeRun({ id: 112 }), workflowHistory: {} as never });
+
+      await expect(portWithLogger.getAvailableRun('112')).rejects.toMatchObject({
+        name: 'MalformedRunError',
+        info: {
+          runId: '112',
+          stepId: null,
+          stepIndex: null,
+          userMessage: 'This step could not be loaded and cannot be executed.',
+        },
+      });
+      expect(logger).toHaveBeenCalledWith(
+        'Error',
+        'Failed to hydrate run — unexpected error',
+        expect.objectContaining({ runId: 112 }),
+      );
+    });
   });
 
   describe('updateStepExecution', () => {