feat(agent): embed workflow executor in-process via addWorkflowExecutor()

[PMerlet]

Why

The agent and the workflow executor ship as two separate packages so the executor stays compatible with any agent version. For Node agents that want a turnkey setup, this adds an opt-in way to run the executor in the same process — no separate deployment, no second service to operate.

What

createAgent(options)
  .addDataSource(...)
  .addWorkflowExecutor({ database: { uri: process.env.DATABASE_URL } })
  .mountOnStandaloneServer(3351)
  .start();

addWorkflowExecutor() boots the executor on start() and drains it on stop(). It reuses the existing /_internal/executor/* proxy by wiring workflowExecutorUrl to a loopback port, so the agent's auth layer and the raw-bytes passthrough apply unchanged.

Design decisions

• Port known up-front (option → HTTP_PORT env → 3400): the proxy route is built before the listener exists, so a random :0 port wasn't viable.
• agentUrl derived from the standalone server port + prefix (the executor reaches the agent over HTTP via agent-client). An explicit agentUrl is required when the agent is mounted on Express/Fastify/NestJS, since the agent can't know the host app's address.
• Database-backed run store: database option → DATABASE_URL env → clear startup error. No silent in-memory fallback (it would lose runs on restart).
• Mutually exclusive with the workflowExecutorUrl option (remote executor) — throws if both are set.
• @forestadmin/workflow-executor is an optionalDependency, loaded via a guarded dynamic import(). Agents that don't embed it pull none of its (heavy) deps; a missing package yields an actionable error.

OpenTelemetry

No incompatibility. The executor's OTel setup is an entrypoint/Docker concern (--require tracing.js), fully decoupled from the library consumed here. In embedded mode, instrumentation is governed by the agent process — if the customer instruments it, the embedded executor is auto-traced in the same process for free.

Tests

• 13 unit tests (packages/agent/test/agent-workflow-executor.test.ts): port wiring, HTTP_PORT fallback, chaining, double-call guard, mutual exclusion, agentUrl derivation, DATABASE_URL fallback, missing-database error, lifecycle (start/stop).
• Full agent suite green (951/951), build + lint clean.
• Manual end-to-end against a local Postgres (real agent + real buildDatabaseExecutor, not mocked): executor boots, migrations create the forest schema + workflow_step_executions table, /health returns 200, agent probe passes over loopback, clean shutdown. (Not committed as a CI test — jest's resolver can't load the executor's @langchain/anthropic dep tree; Node resolves it fine, so the unit tests mock the package.)

🤖 Generated with Claude Code

Note

Add addWorkflowExecutor() to run a workflow executor in-process within the agent

• Adds addWorkflowExecutor(options?) to the Agent class in agent.ts, which configures and starts an embedded @forestadmin/workflow-executor tied to the agent lifecycle (started after router mount, stopped before shutdown).
• Introduces embedded-workflow-executor.ts to build and run a database-backed executor on a loopback port, forwarding logs through the agent logger and deriving agentUrl from the standalone server's host/port when not explicitly provided.
• Adds a manageProcessSignals flag to buildDatabaseExecutor and buildInMemoryExecutor in the workflow-executor package; when false, the executor skips SIGTERM/SIGINT handler registration so the host process manages signals.
• addWorkflowExecutor() and workflowExecutorUrl in agent options are mutually exclusive; calling both throws an error.
• Risk: @forestadmin/workflow-executor is an optional peer dependency; if missing, agent.start() throws with a user-facing error.

^{Macroscope summarized 0f23209.}

[qltysh]

Coverage Impact

⬆️ Merging this pull request will increase total coverage on main by 0.16%.

Modified Files with Diff Coverage (4)

Rating	File	% Diff	Uncovered Line #s
	packages/workflow-executor/src/build-workflow-executor.ts	100.0%
	packages/agent/src/agent.ts	100.0%
	packages/agent/src/framework-mounter.ts	100.0%
	packages/agent/src/embedded-workflow-executor.ts	97.4%	20
	Total	98.2%

🤖 Increase coverage with AI coding...

In the `feat/agent-embed-workflow-executor` branch, add test coverage for this new code:

- `packages/agent/src/embedded-workflow-executor.ts` -- Line 20

🚦 See full report on Qlty Cloud »

🛟 Help

• Diff Coverage: Coverage for added or modified lines of code (excludes deleted files). Learn more.

• Total Coverage: Coverage for the whole repository, calculated as the sum of all File Coverage. Learn more.

• File Coverage: Covered Lines divided by Covered Lines plus Missed Lines. (Excludes non-executable lines including blank lines and comments.)

  • Indirect Changes: Changes to File Coverage for files that were not modified in this PR. Learn more.

[qltysh]

1 new issue

Tool	Category	Rule	Count
qlty	Structure	Function with many parameters (count = 4): createWorkflowExecutor	1

[macroscopeapp]

🟡 Medium

agent-nodejs/packages/agent/src/agent.ts

Line 96 in 8b98527

try {

If this.embeddedExecutor.start() throws after this.mount(router) succeeds, the catch block only rethrows without unmounting or closing the server. The agent is left mounted and serving routes with no running executor, and a subsequent start() retry fails with EADDRINUSE because the listener was never cleaned up. Consider stopping/unmounting the agent in the catch block before rethrowing so startup failures don't leave a half-started server bound to the port.

Also found in 1 other location(s)

packages/agent/src/embedded-workflow-executor.ts:99

start() does not clean up a partially started executor when this.executor.start() rejects. In @forestadmin/workflow-executor, WorkflowExecutor.start() starts the runner before binding the HTTP port, so a port conflict or listen error leaves the polling timer and database connections running. Here that rejection is just propagated from line 99, which means agent.start() fails but the embedded executor keeps background resources alive and can hang the process until it is killed.

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @packages/agent/src/agent.ts around line 96:

If `this.embeddedExecutor.start()` throws after `this.mount(router)` succeeds, the `catch` block only rethrows without unmounting or closing the server. The agent is left mounted and serving routes with no running executor, and a subsequent `start()` retry fails with `EADDRINUSE` because the listener was never cleaned up. Consider stopping/unmounting the agent in the `catch` block before rethrowing so startup failures don't leave a half-started server bound to the port.

Also found in 1 other location(s):
- packages/agent/src/embedded-workflow-executor.ts:99 -- `start()` does not clean up a partially started executor when `this.executor.start()` rejects. In `@forestadmin/workflow-executor`, `WorkflowExecutor.start()` starts the runner before binding the HTTP port, so a port conflict or listen error leaves the polling timer and database connections running. Here that rejection is just propagated from line 99, which means `agent.start()` fails but the embedded executor keeps background resources alive and can hang the process until it is killed.

[macroscopeapp]

🟡 Medium src/embedded-workflow-executor.ts:146

importPackage() catches every rejection from import('@forestadmin/workflow-executor') and replaces it with a generic "package not installed" message. When the package is installed but throws during module evaluation (e.g. a broken transitive dependency or a runtime error), agent.start() surfaces the false missing-dependency error, hiding the real cause. Consider detecting MODULE_NOT_FOUND specifically and rethrowing all other errors unchanged.

  private async importPackage() {
    try {
      return await import('@forestadmin/workflow-executor');
-    } catch (error) {
-      throw new Error(
-        'The embedded workflow executor requires the `@forestadmin/workflow-executor` package. ' +
-          'Install it with `npm install @forestadmin/workflow-executor`.',
-      );
+    } catch (error: any) {
+      if (error?.code === 'MODULE_NOT_FOUND') {
+        throw new Error(
+          'The embedded workflow executor requires the `@forestadmin/workflow-executor` package. ' +
+            'Install it with `npm install @forestadmin/workflow-executor`.',
+        );
+      }
+
+      throw error;
    }
  }

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @packages/agent/src/embedded-workflow-executor.ts around lines 146-155:

`importPackage()` catches every rejection from `import('@forestadmin/workflow-executor')` and replaces it with a generic "package not installed" message. When the package is installed but throws during module evaluation (e.g. a broken transitive dependency or a runtime error), `agent.start()` surfaces the false missing-dependency error, hiding the real cause. Consider detecting `MODULE_NOT_FOUND` specifically and rethrowing all other errors unchanged.

[macroscopeapp]

🟡 Medium src/agent.ts:118

stop() awaits this.embeddedExecutor?.stop() without a finally, so if that promise rejects, forestAdminClient.close() and super.stop() are skipped. The agent's server socket stays open and the Forest Admin client remains subscribed after a failed executor shutdown. Wrap the executor drain in try/finally (or .catch()) so the remaining cleanup always runs.

  override async stop(): Promise<void> {
-    // Drain the embedded executor first, while the agent it depends on is still serving.
-    await this.embeddedExecutor?.stop();
-    // Close anything related to ForestAdmin client
-    this.options.forestAdminClient.close();
-    // Stop at framework level
-    await super.stop();
+    try {
+      // Drain the embedded executor first, while the agent it depends on is still serving.
+      await this.embeddedExecutor?.stop();
+    } finally {
+      // Close anything related to ForestAdmin client
+      this.options.forestAdminClient.close();
+      // Stop at framework level
+      await super.stop();
+    }
  }

🚀 Reply "fix it for me" or copy this AI Prompt for your agent:

In file @packages/agent/src/agent.ts around lines 118-125:

`stop()` awaits `this.embeddedExecutor?.stop()` without a `finally`, so if that promise rejects, `forestAdminClient.close()` and `super.stop()` are skipped. The agent's server socket stays open and the Forest Admin client remains subscribed after a failed executor shutdown. Wrap the executor drain in `try/finally` (or `.catch()`) so the remaining cleanup always runs.