feat(workflow-executor): workflow executor DOING#1493
Conversation
…premature deps, add smoke test - Rewrite CLAUDE.md with project overview and architecture principles, remove changelog - Remove unused dependencies (ai-proxy, sequelize, zod) per YAGNI - Add smoke test so CI passes Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Scra3
force-pushed
the
feat/prd-214-setup-workflow-executor-package
branch
from
083894b to
4510b7b
Compare
… document system architecture
- Lint now covers src and test directories
- Replace require() with import, use stronger assertion (toHaveLength)
- Add System Architecture section describing Front/Orchestrator/Executor/Agent
- Mark Architecture Principles as planned (not yet implemented)
- Remove redundant test/.gitkeep
- Make index.ts a valid module with export {}
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
| export type McpConfiguration = unknown; | ||
|
|
||
| export interface WorkflowPort { | ||
| getPendingStepExecutions(): Promise<PendingStepExecution[]>; |
There was a problem hiding this comment.
this method will retrieve the workflowRun and workflowSteps of one pending run. It will take an optional runId, and return an object with complete workflowRun, workflowSteps, workflowRecords
| interface BaseStepHistory { | ||
| stepId: string; | ||
| stepIndex: number; | ||
| status: StepStatus; | ||
| /** Present when status is 'error'. */ | ||
| error?: string; | ||
| } | ||
|
|
||
| export interface ConditionStepHistory extends BaseStepHistory { | ||
| type: 'condition'; | ||
| /** Present when status is 'success'. */ | ||
| selectedOption?: string; | ||
| } | ||
|
|
||
| export interface AiTaskStepHistory extends BaseStepHistory { | ||
| type: 'ai-task'; | ||
| } |
There was a problem hiding this comment.
what are those types ? this does not correspond to anything
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
16 new issues
|
| function extractRecordId( | ||
| primaryKeyFields: string[], | ||
| record: Record<string, unknown>, | ||
| ): Array<string | number> { | ||
| return primaryKeyFields.map(field => record[field] as string | number); | ||
| } |
There was a problem hiding this comment.
🟢 Low adapters/agent-client-agent-port.ts:30
extractRecordId returns undefined for missing primary key fields, but the string | number type assertion hides this. When passed to encodePk, undefined becomes the literal string "undefined", causing lookups to silently fail with wrong record IDs. Consider adding a runtime check for missing fields and throwing an error, or return the actual values and let encodePk validate them.
function extractRecordId(
primaryKeyFields: string[],
record: Record<string, unknown>,
): Array<string | number> {
- return primaryKeyFields.map(field => record[field] as string | number);
+ return primaryKeyFields.map(field => {
+ const value = record[field];
+ if (value === undefined || value === null) {
+ throw new Error(`Missing primary key field: ${field}`);
+ }
+ return value as string | number;
+ });
}🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/src/adapters/agent-client-agent-port.ts around lines 30-35:
`extractRecordId` returns `undefined` for missing primary key fields, but the `string | number` type assertion hides this. When passed to `encodePk`, `undefined` becomes the literal string `"undefined"`, causing lookups to silently fail with wrong record IDs. Consider adding a runtime check for missing fields and throwing an error, or return the actual values and let `encodePk` validate them.
Evidence trail:
packages/workflow-executor/src/adapters/agent-client-agent-port.ts lines 25-33 (REVIEWED_COMMIT): `encodePk` uses `String(v)` which converts undefined to "undefined"; `extractRecordId` uses type assertion `as string | number` on `record[field]` which can be undefined at runtime. Line 83 shows `extractRecordId` is called in `getRelatedData` to create record IDs that are returned and could be used in subsequent operations.
| // TODO: finalize route paths with the team — these are placeholders | ||
| const ROUTES = { | ||
| pendingStepExecutions: '/liana/v1/workflow-step-executions/pending', | ||
| updateStepExecution: (runId: string) => `/liana/v1/workflow-step-executions/${runId}/complete`, | ||
| collectionRef: (collectionName: string) => `/liana/v1/collections/${collectionName}`, | ||
| mcpServerConfigs: '/liana/mcp-server-configs-with-details', | ||
| }; |
There was a problem hiding this comment.
🟢 Low adapters/forest-server-workflow-port.ts:9
ROUTES.updateStepExecution(runId) and ROUTES.collectionRef(collectionName) interpolate raw values into URL paths without encoding, so special characters like /, ?, or % in runId or collectionName produce malformed URLs (e.g., collectionName="a/b" becomes /liana/v1/collections/a/b with three path segments). Consider wrapping the parameters with encodeURIComponent() to ensure they are safely encoded.
-// TODO: finalize route paths with the team — these are placeholders
const ROUTES = {
pendingStepExecutions: '/liana/v1/workflow-step-executions/pending',
- updateStepExecution: (runId: string) => `/liana/v1/workflow-step-executions/${runId}/complete`,
- collectionRef: (collectionName: string) => `/liana/v1/collections/${collectionName}`,
+ updateStepExecution: (runId: string) => `/liana/v1/workflow-step-executions/${encodeURIComponent(runId)}/complete`,
+ collectionRef: (collectionName: string) => `/liana/v1/collections/${encodeURIComponent(collectionName)}`,
mcpServerConfigs: '/liana/mcp-server-configs-with-details',
};🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/src/adapters/forest-server-workflow-port.ts around lines 9-15:
`ROUTES.updateStepExecution(runId)` and `ROUTES.collectionRef(collectionName)` interpolate raw values into URL paths without encoding, so special characters like `/`, `?`, or `%` in `runId` or `collectionName` produce malformed URLs (e.g., `collectionName="a/b"` becomes `/liana/v1/collections/a/b` with three path segments). Consider wrapping the parameters with `encodeURIComponent()` to ensure they are safely encoded.
Evidence trail:
packages/workflow-executor/src/adapters/forest-server-workflow-port.ts lines 11-13 (ROUTES definitions with template literals), packages/forestadmin-client/src/utils/server.ts lines 63-70 (queryWithHeaders passes path directly to new URL() without encoding)
| expect(selectRecordTool.schema.shape.recordIdentifier.options).not.toContain( | ||
| expect.stringContaining('stepIndex: 3'), | ||
| ); |
There was a problem hiding this comment.
🟡 Medium executors/load-related-record-step-executor.test.ts:1492
The assertion at lines 1492-1494 checks that recordIdentifier.options does not contain 'stepIndex: 3', but the actual format used in select-record tool identifiers is 'Step N - ...' (as shown at line 1447). Since 'stepIndex: 3' never appears in the options, this assertion always passes regardless of whether the pending execution is correctly excluded from the pool.
- expect(selectRecordTool.schema.shape.recordIdentifier.options).not.toContain(
- expect.stringContaining('stepIndex: 3'),
- );🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/test/executors/load-related-record-step-executor.test.ts around lines 1492-1494:
The assertion at lines 1492-1494 checks that `recordIdentifier.options` does not contain `'stepIndex: 3'`, but the actual format used in `select-record` tool identifiers is `'Step N - ...'` (as shown at line 1447). Since `'stepIndex: 3'` never appears in the options, this assertion always passes regardless of whether the pending execution is correctly excluded from the pool.
Evidence trail:
packages/workflow-executor/test/executors/load-related-record-step-executor.test.ts lines 1492-1494 (assertion checking 'stepIndex: 3'), line 1446 (mock args using format 'Step 2 - Orders #99'), packages/workflow-executor/src/executors/base-step-executor.ts lines 263-267 (toRecordIdentifier method defining format as `Step ${record.stepIndex} - ${schema.collectionDisplayName} #${record.recordId}`), packages/workflow-executor/test/executors/read-record-step-executor.test.ts lines 565-567 (expected options format showing 'Step 3 - Customers #42')
| return selectedDisplayNames.map( | ||
| dn => nonRelationFields.find(f => f.displayName === dn)?.fieldName ?? dn, | ||
| ); |
There was a problem hiding this comment.
🟢 Low executors/load-related-record-step-executor.ts:345
In selectRelevantFields, when the AI returns a displayName that doesn't exist in nonRelationFields, the find returns undefined and the fallback ?? dn preserves the raw display name. This invalid field name then fails to match any key in c.values during selectBestRecordIndex, causing the filtered candidates to have empty values objects and making the AI's record selection arbitrary. Consider throwing InvalidAIResponseError when find returns undefined instead of falling back to the raw display name.
+ return selectedDisplayNames.map(dn => {
+ const field = nonRelationFields.find(f => f.displayName === dn);
+ if (!field) {
+ throw new InvalidAIResponseError(
+ `AI returned unknown field name "${dn}" for collection "${schema.collectionName}"`,
+ );
+ }
+ return field.fieldName;
+ });🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/src/executors/load-related-record-step-executor.ts around lines 345-347:
In `selectRelevantFields`, when the AI returns a `displayName` that doesn't exist in `nonRelationFields`, the `find` returns `undefined` and the fallback `?? dn` preserves the raw display name. This invalid field name then fails to match any key in `c.values` during `selectBestRecordIndex`, causing the filtered candidates to have empty `values` objects and making the AI's record selection arbitrary. Consider throwing `InvalidAIResponseError` when `find` returns `undefined` instead of falling back to the raw display name.
Evidence trail:
1. packages/workflow-executor/src/executors/load-related-record-step-executor.ts lines 337-347: Comment states 'Zod's .min(1) shapes the prompt but is NOT validated against the AI response' and the mapping logic `nonRelationFields.find(f => f.displayName === dn)?.fieldName ?? dn`.
2. packages/workflow-executor/src/executors/base-step-executor.ts lines 149-163: `invokeWithTools` returns `toolCall.args as T` without runtime Zod validation.
3. packages/workflow-executor/src/executors/load-related-record-step-executor.ts lines 357-365: `filteredCandidates` filters `c.values` entries by checking `fieldNames.includes(k)`.
4. packages/workflow-executor/src/types/record.ts line 37: `RecordData` definition shows `values: Record<string, unknown>` keyed by technical field names.
|
|
||
| export default class ConsoleLogger implements Logger { | ||
| error(message: string, context: Record<string, unknown>): void { | ||
| console.error(JSON.stringify({ message, timestamp: new Date().toISOString(), ...context })); |
There was a problem hiding this comment.
🟡 Medium adapters/console-logger.ts:5
The spread ...context comes after message and timestamp, so any message or timestamp keys in context overwrite the explicit parameters. Callers logging { message: "other" } will see their context value instead of the actual error message. Consider moving the spread first so explicit parameters take precedence.
| console.error(JSON.stringify({ message, timestamp: new Date().toISOString(), ...context })); | |
| console.error(JSON.stringify({ ...context, message, timestamp: new Date().toISOString() })); |
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/src/adapters/console-logger.ts around line 5:
The spread `...context` comes after `message` and `timestamp`, so any `message` or `timestamp` keys in `context` overwrite the explicit parameters. Callers logging `{ message: "other" }` will see their context value instead of the actual error message. Consider moving the spread first so explicit parameters take precedence.
Evidence trail:
packages/workflow-executor/src/adapters/console-logger.ts:5 - The code `{ message, timestamp: new Date().toISOString(), ...context }` shows explicit properties defined first and spread operator last. In JavaScript, object spread puts later properties after earlier ones, so `context.message` would overwrite the `message` parameter if present. This is standard JavaScript object spread behavior documented at MDN and in the ECMAScript specification.
…erver (#1504) Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com> Co-authored-by: alban bertolini <albanb@forestadmin.com>
packages/workflow-executor/src/executors/mcp-task-step-executor.ts
Outdated
Show resolved
Hide resolved
…+ DatabaseStore) (#1506)
packages/ai-proxy/src/ai-client.ts
Outdated
| async loadRemoteTools(mcpConfig: McpConfiguration): Promise<McpClient['tools']> { | ||
| await this.closeMcpClient('Error closing previous MCP connection'); | ||
|
|
||
| const newClient = new McpClient(mcpConfig, this.logger); | ||
| const tools = await newClient.loadTools(); | ||
| this.mcpClient = newClient; | ||
|
|
||
| return tools; | ||
| } |
There was a problem hiding this comment.
🟡 Medium src/ai-client.ts:39
In loadRemoteTools, if newClient.loadTools() throws, the McpClient instance is orphaned and never closed. Since this.mcpClient is only assigned after success, closeConnections() cannot reach the failed client, leaving resources leaked. Consider wrapping loadTools() in a try/finally to ensure cleanup on failure.
async loadRemoteTools(mcpConfig: McpConfiguration): Promise<McpClient['tools']> {
await this.closeMcpClient('Error closing previous MCP connection');
const newClient = new McpClient(mcpConfig, this.logger);
- const tools = await newClient.loadTools();
- this.mcpClient = newClient;
+ try {
+ const tools = await newClient.loadTools();
+ this.mcpClient = newClient;
- return tools;
+ return tools;
+ } catch (error) {
+ await newClient.closeConnections();
+ throw error;
+ }
}🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/ai-proxy/src/ai-client.ts around lines 39-47:
In `loadRemoteTools`, if `newClient.loadTools()` throws, the `McpClient` instance is orphaned and never closed. Since `this.mcpClient` is only assigned after success, `closeConnections()` cannot reach the failed client, leaving resources leaked. Consider wrapping `loadTools()` in a try/finally to ensure cleanup on failure.
Evidence trail:
packages/ai-proxy/src/ai-client.ts lines 39-45 (loadRemoteTools function) and lines 52-60 (closeMcpClient function) at REVIEWED_COMMIT. The code shows newClient is instantiated at line 42, loadTools() is called at line 43, and this.mcpClient is assigned at line 44. If line 43 throws, line 44 never executes, leaving newClient orphaned with no cleanup path.
|
|
||
| import { RecordNotFoundError } from '../errors'; | ||
|
|
||
| function buildPkFilter( |
There was a problem hiding this comment.
🟢 Low adapters/agent-client-agent-port.ts:13
In buildPkFilter, when id.length < primaryKeyFields.length, id[i] evaluates to undefined for missing indices, producing filter conditions with value: undefined. This creates an And aggregator with Equal conditions comparing against undefined, which likely matches zero records or causes unexpected query behavior instead of the intended record lookup.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/src/adapters/agent-client-agent-port.ts around line 13:
In `buildPkFilter`, when `id.length < primaryKeyFields.length`, `id[i]` evaluates to `undefined` for missing indices, producing filter conditions with `value: undefined`. This creates an `And` aggregator with `Equal` conditions comparing against `undefined`, which likely matches zero records or causes unexpected query behavior instead of the intended record lookup.
Evidence trail:
packages/workflow-executor/src/adapters/agent-client-agent-port.ts lines 13-27 at REVIEWED_COMMIT - the `buildPkFilter` function uses `primaryKeyFields.map((field, i) => ({ field, operator: 'Equal', value: id[i] }))` which iterates over `primaryKeyFields` length and accesses `id[i]` without checking if `id` has enough elements. JavaScript array access beyond bounds returns `undefined`.
|
|
||
| throw new MalformedToolCallError(toolCall.name ?? 'unknown', 'args field is missing or null'); | ||
| } | ||
|
|
||
| const invalidCall = response.invalid_tool_calls?.[0]; | ||
|
|
||
| if (invalidCall) { |
There was a problem hiding this comment.
🟢 Low executors/base-step-executor.ts:172
invokeWithTools returns toolCall.name directly at line 174 without validating it exists. When the AI returns a tool call with defined args but undefined name, the function returns { toolName: undefined, args }, violating the declared return type { toolName: string; args: T }. This can cause downstream failures when code expects a valid string tool name. Consider checking toolCall.name before returning and throwing MalformedToolCallError if it's missing, consistent with how args is validated.
if (toolCall !== undefined) {
- if (toolCall.args !== undefined && toolCall.args !== null) {
- return { toolName: toolCall.name, args: toolCall.args as T };
+ if (toolCall.name !== undefined && toolCall.name !== null && toolCall.args !== undefined && toolCall.args !== null) {
+ return { toolName: toolCall.name, args: toolCall.args as T };
}
- throw new MalformedToolCallError(toolCall.name ?? 'unknown', 'args field is missing or null');
+ throw new MalformedToolCallError(toolCall.name ?? 'unknown', 'name or args field is missing or null');
}🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/src/executors/base-step-executor.ts around lines 172-178:
`invokeWithTools` returns `toolCall.name` directly at line 174 without validating it exists. When the AI returns a tool call with defined `args` but undefined `name`, the function returns `{ toolName: undefined, args }`, violating the declared return type `{ toolName: string; args: T }`. This can cause downstream failures when code expects a valid string tool name. Consider checking `toolCall.name` before returning and throwing `MalformedToolCallError` if it's missing, consistent with how `args` is validated.
Evidence trail:
packages/workflow-executor/src/executors/base-step-executor.ts lines 160-175 (REVIEWED_COMMIT) - shows the `invokeWithTools` function with return type `Promise<{ toolName: string; args: T }>` returning `toolCall.name` without validation. packages/ai-proxy/package.json shows `@langchain/core: 1.1.15`. @langchain/core ToolCall type definition (https://github.com/langchain-ai/langchainjs/blob/main/libs/langchain-core/src/messages/tool.ts) shows `name?: string` is optional.
…xecutor factories (#1510)
| describe('no records available', () => { | ||
| it('returns error when no records are available', () => { | ||
| const error = new NoRecordsError(); | ||
|
|
||
| expect(error).toBeInstanceOf(NoRecordsError); | ||
| expect(error.message).toBe('No records available'); | ||
| }); | ||
| }); |
There was a problem hiding this comment.
🟢 Low executors/read-record-step-executor.test.ts:317
The test 'returns error when no records are available' only instantiates NoRecordsError and checks its properties — it never creates a ReadRecordStepExecutor or calls execute(). This gives false confidence that the executor handles the "no records" scenario when it actually tests nothing about executor behavior. Consider removing this test or implementing it to verify the executor throws when no records are available.
- describe('no records available', () => {
- it('returns error when no records are available', () => {
- const error = new NoRecordsError();
-
- expect(error).toBeInstanceOf(NoRecordsError);
- expect(error.message).toBe('No records available');
- });
- });🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/test/executors/read-record-step-executor.test.ts around lines 317-324:
The test `'returns error when no records are available'` only instantiates `NoRecordsError` and checks its properties — it never creates a `ReadRecordStepExecutor` or calls `execute()`. This gives false confidence that the executor handles the "no records" scenario when it actually tests nothing about executor behavior. Consider removing this test or implementing it to verify the executor throws when no records are available.
Evidence trail:
packages/workflow-executor/test/executors/read-record-step-executor.test.ts lines 317-323 (the test in question - only creates NoRecordsError and checks properties)
packages/workflow-executor/test/executors/read-record-step-executor.test.ts lines 326-340 (adjacent test showing proper executor test pattern with ReadRecordStepExecutor instantiation and execute() call)
packages/workflow-executor/test/executors/read-record-step-executor.test.ts lines 1-12 (imports showing both NoRecordsError and ReadRecordStepExecutor are available)
…ain (#1512) Co-authored-by: alban bertolini <albanb@forestadmin.com>
| return promise; | ||
| } | ||
|
|
||
| private async doExecuteStep(step: PendingStepExecution, key: string): Promise<void> { |
There was a problem hiding this comment.
🟡 Medium src/runner.ts:236
In doExecuteStep, the step is deleted from inFlightSteps in the finally block (line 253) before updateStepExecution completes (lines 256-267). If runPollCycle executes between the delete and updateStepExecution completing, the same step is refetched from the server, passes the !this.inFlightSteps.has(...) filter on line 204, and gets executed a second time concurrently. Move the delete to after updateStepExecution succeeds or fails, or wrap both in a single try/finally.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/src/runner.ts around line 236:
In `doExecuteStep`, the step is deleted from `inFlightSteps` in the `finally` block (line 253) before `updateStepExecution` completes (lines 256-267). If `runPollCycle` executes between the delete and `updateStepExecution` completing, the same step is refetched from the server, passes the `!this.inFlightSteps.has(...)` filter on line 204, and gets executed a second time concurrently. Move the `delete` to after `updateStepExecution` succeeds or fails, or wrap both in a single try/finally.
Evidence trail:
packages/workflow-executor/src/runner.ts lines 235-268 (REVIEWED_COMMIT): The `doExecuteStep` method shows the `finally` block with `this.inFlightSteps.delete(key)` at line 253, followed by the `updateStepExecution` call in a separate try block at lines 256-267. The `runPollCycle` method at lines 200-212 shows the filter `!this.inFlightSteps.has(Runner.stepKey(s))` at line 203.
- Remove McpClient.tools property, loadTools() returns local array - Rename closeConnections() → dispose() - Rename testConnections() → checkConnection() - Add McpServers type export - Rename mcpServerConfigs → toolConfigs in create-ai-provider - Update all tests accordingly Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…eability Add sourceId to McpToolRef so that persisted execution data (pendingData, executionParams) tracks which MCP server provided the tool. This fixes tool lookup on re-entry (confirmation flow) when multiple servers expose tools with the same name. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
… collection schema - Replace non-null assertion with explicit McpToolNotFoundError when AI selects a tool name that doesn't match any available tool - Resolve related collection name from parent schema before looking up the related schema in cache, fixing cases where relation name differs from target collection name Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
matthv
force-pushed
the
feat/prd-214-setup-workflow-executor-package
branch
from
d43f168 to
6744696
Compare
|
|
||
| return injectOauthTokens({ | ||
| mcpConfigs: args.mcpServerConfigs as McpConfiguration | undefined, | ||
| mcpConfigs: args.toolConfigs as McpConfiguration | undefined, |
There was a problem hiding this comment.
🟢 Low src/create-ai-provider.ts:15
resolveMcpConfigs accesses args.toolConfigs, but the AiRouter.route interface defines the property as mcpServerConfigs. Since toolConfigs is not defined on the args type, it will always be undefined, causing MCP configurations to never be passed through to injectOauthTokens.
- mcpConfigs: args.toolConfigs as McpConfiguration | undefined,
+ mcpConfigs: args.mcpServerConfigs as McpConfiguration | undefined,🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/ai-proxy/src/create-ai-provider.ts around line 15:
`resolveMcpConfigs` accesses `args.toolConfigs`, but the `AiRouter.route` interface defines the property as `mcpServerConfigs`. Since `toolConfigs` is not defined on the args type, it will always be `undefined`, causing MCP configurations to never be passed through to `injectOauthTokens`.
Evidence trail:
1. packages/ai-proxy/src/create-ai-provider.ts line 15: accesses `args.toolConfigs`
2. packages/ai-proxy/src/create-ai-provider.ts line 9: `args` is typed as `Parameters<AiRouter['route']>[0]`
3. packages/agent-toolkit/src/interfaces/ai.ts lines 21-26: `AiRouter.route` interface defines the property as `mcpServerConfigs?: unknown;`, not `toolConfigs`
| async start() { | ||
| await runner.start(); | ||
| await server.start(); | ||
|
|
||
| process.on('SIGTERM', onSignal); | ||
| process.on('SIGINT', onSignal); | ||
| }, |
There was a problem hiding this comment.
🟢 Low src/build-workflow-executor.ts:123
Calling start() multiple times registers duplicate SIGTERM and SIGINT handlers, but stop() only removes one handler per signal. The remaining stale handlers continue to trigger shutdown behavior after stop() completes, causing unexpected process exits. Consider tracking whether listeners are already registered and skipping registration on subsequent start() calls.
async start() {
+ if (process.listenerCount('SIGTERM') > 0 || process.listenerCount('SIGINT') > 0) {
+ throw new Error('WorkflowExecutor already started');
+ }
await runner.start();
await server.start();
process.on('SIGTERM', onSignal);
process.on('SIGINT', onSignal);
},🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/src/build-workflow-executor.ts around lines 123-129:
Calling `start()` multiple times registers duplicate `SIGTERM` and `SIGINT` handlers, but `stop()` only removes one handler per signal. The remaining stale handlers continue to trigger shutdown behavior after `stop()` completes, causing unexpected process exits. Consider tracking whether listeners are already registered and skipping registration on subsequent `start()` calls.
Evidence trail:
packages/workflow-executor/src/build-workflow-executor.ts lines 88-131 at REVIEWED_COMMIT - `start()` method at lines 118-123 registers handlers via `process.on()` without checking if already registered; `stop()` at lines 125-131 uses `process.removeListener()` which removes only one listener; `onSignal()` at lines 88-109 schedules force exit timer that can cause unexpected process exits.
packages/workflow-executor/src/executors/load-related-record-step-executor.ts
Show resolved
Hide resolved
Merge main into feature branch, resolve conflicts by taking main's versions of router.ts, create-ai-provider.ts and their tests. Remove deleted mcp-config-checker.ts. Bump workflow-executor internal deps to match main. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Macroscope review — triage (post-merge with main)Fixed:
Not applicable:
|
| stepId: this.context.stepId, | ||
| stepIndex: this.context.stepIndex, | ||
| error: error instanceof Error ? error.message : String(error), | ||
| cause: errorCause instanceof Error ? errorCause.message : undefined, |
There was a problem hiding this comment.
🟢 Low executors/base-step-executor.ts:80
On line 80, when errorCause is not an Error, it is logged as undefined instead of preserving the actual value. This causes loss of debugging information when the unexpected error has a non-Error cause (e.g., a string or object). Consider using String(errorCause) to match the pattern used on line 66.
| cause: errorCause instanceof Error ? errorCause.message : undefined, | |
| cause: errorCause instanceof Error ? errorCause.message : String(errorCause), |
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/src/executors/base-step-executor.ts around line 80:
On line 80, when `errorCause` is not an `Error`, it is logged as `undefined` instead of preserving the actual value. This causes loss of debugging information when the unexpected error has a non-Error cause (e.g., a string or object). Consider using `String(errorCause)` to match the pattern used on line 66.
Evidence trail:
packages/workflow-executor/src/executors/base-step-executor.ts lines 60-90 at REVIEWED_COMMIT. Line 66 shows `cause: error.cause instanceof Error ? error.cause.message : String(error.cause)` while line 80 shows `cause: errorCause instanceof Error ? errorCause.message : undefined`. The inconsistency is clear - one preserves non-Error values via String(), the other discards them as undefined.
| const configs = await this.config.workflowPort.getMcpServerConfigs(); | ||
| if (configs.length === 0) return []; | ||
|
|
||
| const mergedConfig: McpConfiguration = { |
There was a problem hiding this comment.
🟢 Low src/runner.ts:221
In fetchRemoteTools(), the merge logic spreads only configs[0] as the base and then overwrites the configs property with merged values from all configs. If McpConfiguration has other top-level properties (server URLs, authentication tokens, etc.) in configs[1], configs[2], etc., they are silently discarded. The merged configuration may therefore use incorrect connection settings from configs[0] when multiple MCP servers are configured.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/src/runner.ts around line 221:
In `fetchRemoteTools()`, the merge logic spreads only `configs[0]` as the base and then overwrites the `configs` property with merged values from all configs. If `McpConfiguration` has other top-level properties (server URLs, authentication tokens, etc.) in `configs[1]`, `configs[2]`, etc., they are silently discarded. The merged configuration may therefore use incorrect connection settings from `configs[0]` when multiple MCP servers are configured.
Evidence trail:
- packages/workflow-executor/src/runner.ts:217-226 (REVIEWED_COMMIT) - merge logic that spreads only configs[0]
- packages/ai-proxy/src/mcp-client.ts:13-15 (REVIEWED_COMMIT) - McpConfiguration type definition showing it includes properties from MultiServerMCPClient['config']
- https://github.com/langchain-ai/langchainjs/tree/main/libs/langchain-mcp-adapters/ - @langchain/mcp-adapters documentation showing MultiServerMCPClient config has many top-level options beyond mcpServers
| } catch (error) { | ||
| this.isRunning = false; | ||
| this._state = 'idle'; | ||
| throw error; | ||
| } |
There was a problem hiding this comment.
🟡 Medium src/runner.ts:69
If stop() is called while start() is awaiting runStore.init(), and init() then fails, the catch block at lines 70-71 unconditionally sets _state = 'idle'. This overwrites the 'stopped' state that stop() had already set, defeating the protection at lines 56-58. A subsequent start() call would pass the stopped-runner check and attempt to re-initialize already-closed resources.
Consider checking the current state before overwriting it: only set 'idle' if the state hasn't already moved to 'draining' or 'stopped'.
} catch (error) {
- this.isRunning = false;
- this._state = 'idle';
+ this.isRunning = false;
+ if (this._state !== 'draining' && this._state !== 'stopped') {
+ this._state = 'idle';
+ }
throw error;
}🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/src/runner.ts around lines 69-73:
If `stop()` is called while `start()` is awaiting `runStore.init()`, and `init()` then fails, the catch block at lines 70-71 unconditionally sets `_state = 'idle'`. This overwrites the `'stopped'` state that `stop()` had already set, defeating the protection at lines 56-58. A subsequent `start()` call would pass the stopped-runner check and attempt to re-initialize already-closed resources.
Consider checking the current state before overwriting it: only set `'idle'` if the state hasn't already moved to `'draining'` or `'stopped'`.
Evidence trail:
packages/workflow-executor/src/runner.ts at REVIEWED_COMMIT:
- Lines 55-57: stopped/draining check in start()
- Line 65: `_state = 'running'`
- Lines 67-72: try/catch with unconditional `_state = 'idle'` in catch block
- Line 78: stop() early return check
- Line 80: `_state = 'draining'`
- Line 113: `runStore.close()` called
- Line 119: `_state = 'stopped'` in finally block
| const rawId = (ctx.state.user as { id?: unknown })?.id; | ||
| const bearerUserId = typeof rawId === 'number' ? rawId : Number(rawId); | ||
|
|
||
| if (!Number.isFinite(bearerUserId)) { |
There was a problem hiding this comment.
🟢 Low http/executor-http-server.ts:163
Number.isFinite(bearerUserId) on line 166 returns true for 0, so malformed JWTs with id: null, id: "", or id: 0 all validate as user ID 0. If 0 is not a valid user ID in your system, these tokens bypass validation and reach triggerPoll with a forged identity. Consider rejecting bearerUserId <= 0 or adding an explicit null/undefined check before the numeric conversion.
- const rawId = (ctx.state.user as { id?: unknown })?.id;
- const bearerUserId = typeof rawId === 'number' ? rawId : Number(rawId);
-
- if (!Number.isFinite(bearerUserId)) {
+ const rawId = (ctx.state.user as { id?: unknown })?.id;
+ const bearerUserId = typeof rawId === 'number' ? rawId : Number(rawId);
+
+ if (!Number.isFinite(bearerUserId) || bearerUserId <= 0) {🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file packages/workflow-executor/src/http/executor-http-server.ts around lines 163-166:
`Number.isFinite(bearerUserId)` on line 166 returns `true` for `0`, so malformed JWTs with `id: null`, `id: ""`, or `id: 0` all validate as user ID `0`. If `0` is not a valid user ID in your system, these tokens bypass validation and reach `triggerPoll` with a forged identity. Consider rejecting `bearerUserId <= 0` or adding an explicit `null`/`undefined` check before the numeric conversion.
Evidence trail:
packages/workflow-executor/src/http/executor-http-server.ts lines 163-176 (REVIEWED_COMMIT): Line 163 extracts `rawId` from JWT, line 164 converts to number using `typeof rawId === 'number' ? rawId : Number(rawId)`, line 166 validates with `Number.isFinite(bearerUserId)`. JavaScript behavior: `Number(null) = 0`, `Number("") = 0`, `Number.isFinite(0) = true`. All three malformed inputs pass validation and reach `triggerPoll` at line 175 with `bearerUserId: 0`.
3 participants
Summary
@forestadmin/workflow-executorpackage (tsconfig, jest, eslint, CI matrix)fixes PRD-214
Test plan
yarn workspace @forestadmin/workflow-executor buildpassesyarn workspace @forestadmin/workflow-executor lintpassesyarn workspace @forestadmin/workflow-executor testpasses🤖 Generated with Claude Code
Note
Add
workflow-executorpackage with step executors, HTTP server, run stores, and builder functions@forestadmin/workflow-executorpackage that polls for pending workflow steps, executes them via AI-assisted step executors, and persists results to an in-memory or database-backed run store.ConditionStepExecutor,ReadRecordStepExecutor,UpdateRecordStepExecutor,TriggerRecordActionStepExecutor,LoadRelatedRecordStepExecutor,McpStepExecutor) plus aStepExecutorFactorythat selects the correct executor at runtime.buildInMemoryExecutorandbuildDatabaseExecutorfactory functions that wire all dependencies (JWT auth, AI client, workflow port, agent port, schema cache) and register SIGTERM/SIGINT handlers for graceful shutdown.ExecutorHttpServerwith JWT auth,/health,GET /runs/:runId, andPOST /runs/:runId/triggerendpoints.AiClient,createBaseChatModel, andgetAiConfigurationto@forestadmin/ai-proxy, and re-exportsServerUtilsfrom@forestadmin/forestadmin-client.hasRunAccessinForestServerWorkflowPortalways returnstrue(placeholder implementation).Changes since #1493 opened
LoadRelatedRecordStepExecutorinternal recordIndex selection method [3c51658]AiModelPortinterface and replaced allAiClientusage throughout theworkflow-executorpackage with the new port abstraction [f815eca]ServerAiAdapterthat routes LLM invocations through Forest Admin server using JWT authentication [f815eca]AiClientAdapterwrapping@forestadmin/ai-proxyAiClientto conform toAiModelPortinterface [f815eca]buildCommonDependenciesto conditionally instantiateAiClientAdapterwhenaiConfigurationsare provided, otherwise instantiateServerAiAdapter[f815eca]envIdfield toPendingStepExecutioninterface [f815eca]@langchain/openaiversion1.2.5as runtime dependency [f815eca]AiModelPortabstraction and added test coverage for new adapters [f815eca]BaseStepExecutor.buildContextMessagemethod that constructs aSystemMessagecontaining executing user context and UTC timestamp, then prepended this context message to all LLM invocations acrossConditionStepExecutor.doExecute,LoadRelatedRecordStepExecutorrelation selection flow,LoadRelatedRecordStepExecutorfield identification flow,LoadRelatedRecordStepExecutorcandidate record selection flow,McpStepExecutor.selectTool,McpStepExecutorsummary generation flow,ReadRecordStepExecutor.selectFields,TriggerRecordActionStepExecutor.selectAction,UpdateRecordStepExecutor.selectFieldAndValue, and the record selection tool invocation inBaseStepExecutorwithin theworkflow-executorpackage [22450d1]condition-step-executor.test.ts,load-related-record-step-executor.test.ts,mcp-step-executor.test.ts,read-record-step-executor.test.ts,trigger-record-action-step-executor.test.ts, andupdate-record-step-executor.test.tsto account for the additional leadingSystemMessagecontaining user and datetime context [22450d1]Runner.patchPendingDatamethod to individual executors via newBaseStepExecutor.patchAndReloadPendingDataprotected method [1798709]guidancestep type with executor, type definitions, validation schema, and formatter [1798709]Macroscope summarized dd78bf5.