From 10f9466545b67930812f340eb6f46ab91b6c5c10 Mon Sep 17 00:00:00 2001 From: alban bertolini Date: Fri, 14 Aug 2026 17:00:16 +0200 Subject: [PATCH 1/3] fix(workflow-executor): report runs that fail to hydrate instead of dropping them /pending-run claims a run before the executor has hydrated it, and only an outcome (or the 60s reaper) clears that claim. A hydration failure that was not a WorkflowExecutorError was only logged customer-side, so the claim was never cleared: reaped after 60s, re-claimed, forever. Two Qonto prod runs cycled ~2 600 times over 3 days this way, blocked on a Camunda connector step (io.camunda:google-sheets:1) the executor cannot map. getAvailableRuns now buckets every hydration failure as malformed, and getAvailableRun wraps non-domain errors in MalformedRunError so the Runner reports on the trigger path too. toMalformedInfo accepts unknown and supplies a generic userMessage for raw errors; its workflowHistory access is now null-safe, since it runs inside the catch block that a non-array history reaches. The remaining unreportable case (toDispatch returning null: no available step to attach an error to) is now logged instead of silently skipped. Behaviour change: on the trigger path a raw hydration error now surfaces as 400 with a user-facing message instead of an opaque 500. It is still logged. fixes PRD-956 Co-Authored-By: Claude Fable 5 --- packages/workflow-executor/CLAUDE.md | 1 + .../adapters/forest-server-workflow-port.ts | 45 +++++++---- .../forest-server-workflow-port.test.ts | 77 ++++++++++++++++++- 3 files changed, 104 insertions(+), 19 deletions(-) diff --git a/packages/workflow-executor/CLAUDE.md b/packages/workflow-executor/CLAUDE.md index 49e30e2fe3..f48b6f1b34 100644 --- a/packages/workflow-executor/CLAUDE.md +++ b/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. diff --git a/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts b/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts index ef7ea8f755..d9e59e2f93 100644 --- a/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts +++ b/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts @@ -43,6 +43,8 @@ const ROUTES = { mcpServerConfigs: '/liana/mcp-server-configs-with-details', }; +const UNHYDRATABLE_RUN_USER_MESSAGE = 'This step could not be loaded and cannot be executed.'; + // Forest sends relatedCollectionName as a `collection.targetKey` reference (e.g. "store.id"); // normalize it to a plain collection name (the related PK comes from the schema's primaryKeyFields). function stripReferenceKey(name: string | undefined): string | undefined { @@ -69,11 +71,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 +112,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 +148,18 @@ 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 { + // Optional chaining is load-bearing: this runs inside a catch block, and a run with a + // non-array workflowHistory is exactly what lands here. Throwing would kill the whole batch. + const pending = run.workflowHistory?.at(-1) ?? null; + const isDomainError = err instanceof WorkflowExecutorError; return { runId: String(run.id), stepId: pending?.stepName ?? null, stepIndex: pending?.stepIndex ?? null, - userMessage: err.userMessage, - technicalMessage: err.message, + userMessage: isDomainError ? err.userMessage : UNHYDRATABLE_RUN_USER_MESSAGE, + technicalMessage: extractErrorMessage(err) ?? 'Unknown error', }; } diff --git a/packages/workflow-executor/test/adapters/forest-server-workflow-port.test.ts b/packages/workflow-executor/test/adapters/forest-server-workflow-port.test.ts index 97fb52c6b3..4d5f2497e0 100644 --- a/packages/workflow-executor/test/adapters/forest-server-workflow-port.test.ts +++ b/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,7 +320,7 @@ 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 @@ -324,13 +331,54 @@ describe('ForestServerWorkflowPort', () => { 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 +441,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: null 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', () => { From b9dba71e3296e5646bf0b8415e8da0f54ad0613d Mon Sep 17 00:00:00 2001 From: alban bertolini Date: Fri, 14 Aug 2026 23:51:46 +0200 Subject: [PATCH 2/3] refactor(workflow-executor): inline the single-use user message and flag No behaviour change. Co-Authored-By: Claude Fable 5 --- .../src/adapters/forest-server-workflow-port.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts b/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts index d9e59e2f93..e994c1c378 100644 --- a/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts +++ b/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts @@ -43,8 +43,6 @@ const ROUTES = { mcpServerConfigs: '/liana/mcp-server-configs-with-details', }; -const UNHYDRATABLE_RUN_USER_MESSAGE = 'This step could not be loaded and cannot be executed.'; - // Forest sends relatedCollectionName as a `collection.targetKey` reference (e.g. "store.id"); // normalize it to a plain collection name (the related PK comes from the schema's primaryKeyFields). function stripReferenceKey(name: string | undefined): string | undefined { @@ -152,13 +150,15 @@ export default class ForestServerWorkflowPort implements WorkflowPort { // Optional chaining is load-bearing: this runs inside a catch block, and a run with a // non-array workflowHistory is exactly what lands here. Throwing would kill the whole batch. const pending = run.workflowHistory?.at(-1) ?? null; - const isDomainError = err instanceof WorkflowExecutorError; return { runId: String(run.id), stepId: pending?.stepName ?? null, stepIndex: pending?.stepIndex ?? null, - userMessage: isDomainError ? err.userMessage : UNHYDRATABLE_RUN_USER_MESSAGE, + userMessage: + err instanceof WorkflowExecutorError + ? err.userMessage + : 'This step could not be loaded and cannot be executed.', technicalMessage: extractErrorMessage(err) ?? 'Unknown error', }; } From 51fd6489a8986d140f92c47bbc8ed79c8109f817 Mon Sep 17 00:00:00 2001 From: alban bertolini Date: Mon, 17 Aug 2026 10:18:27 +0200 Subject: [PATCH 3/3] fix(workflow-executor): guard workflowHistory with Array.isArray, not optional chaining toMalformedInfo runs inside the catch block that a malformed workflowHistory reaches, so a throw there aborts the whole pending-runs batch instead of reporting the run. `?.` only covers null/undefined: it still throws a TypeError on {} and silently indexes a string ("abc".at(-1) === "c"). The two fixtures now use {} instead of null, which fails against the optional chaining form. Co-Authored-By: Claude Fable 5 --- .../src/adapters/forest-server-workflow-port.ts | 6 +++--- .../test/adapters/forest-server-workflow-port.test.ts | 9 +++++---- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts b/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts index e994c1c378..2cd72cb775 100644 --- a/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts +++ b/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts @@ -147,9 +147,9 @@ export default class ForestServerWorkflowPort implements WorkflowPort { } private toMalformedInfo(run: ServerHydratedWorkflowRun, err: unknown): MalformedRunInfo { - // Optional chaining is load-bearing: this runs inside a catch block, and a run with a - // non-array workflowHistory is exactly what lands here. Throwing would kill the whole batch. - const pending = run.workflowHistory?.at(-1) ?? null; + // 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), diff --git a/packages/workflow-executor/test/adapters/forest-server-workflow-port.test.ts b/packages/workflow-executor/test/adapters/forest-server-workflow-port.test.ts index 4d5f2497e0..74bf094652 100644 --- a/packages/workflow-executor/test/adapters/forest-server-workflow-port.test.ts +++ b/packages/workflow-executor/test/adapters/forest-server-workflow-port.test.ts @@ -323,9 +323,10 @@ describe('ForestServerWorkflowPort', () => { 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(); @@ -445,7 +446,7 @@ describe('ForestServerWorkflowPort', () => { 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: null as never }); + mockQuery.mockResolvedValue({ ...makeRun({ id: 112 }), workflowHistory: {} as never }); await expect(portWithLogger.getAvailableRun('112')).rejects.toMatchObject({ name: 'MalformedRunError',