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..2cd72cb775 100644 --- a/packages/workflow-executor/src/adapters/forest-server-workflow-port.ts +++ b/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', }; } 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..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 @@ -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', () => {