diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 7f3538f6..88709e7d 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -135,6 +135,20 @@ Basic code style guidelines are generally enforced by ESLint, but there are addi - Typings exposed to consumers should always attempt to maintain consistency. - Typings for tests are less of a focus than functionality checks. +#### Public API and deprecations + +Public API changes use this lifecycle: + +1. **Introduce** the replacement in a minor release (or a patch when appropriate). +2. **Deprecate** the old export with `@deprecated` JSDoc, example updates, and add a row to the deprecations table in [development.md](./docs/development.md#deprecations). +3. **Remove** deprecated exports only in a **major** release. + +Deprecated APIs remain functional through minor releases. Do not remove them in minors unless the export was never public or keeping it poses a security or correctness risk. + +Before each major release, audit `src/*.ts` and update deprecations that were announced in the prior cycle. + +See [development.md](./docs/development.md#public-api-and-imports) for current deprecations, import paths and authoring guidance. + ### Testing Current testing is based on Jest. diff --git a/docs/development.md b/docs/development.md index 83de9920..23a17354 100644 --- a/docs/development.md +++ b/docs/development.md @@ -177,15 +177,32 @@ const server: PfMcpInstance = await start({ ### Public API and imports -To ensure stability and a predictable developer experience, this package currently enforces a strict public API. All supported programmatic functions and types are exported directly from the root entry point: +This package exposes a curated public API through `package.json` exports: + +| Entry | Import | Purpose | +|---|---|---| +| Root | `@patternfly/patternfly-mcp` | `start()`, server instance types, programmatic options | +| Tools | `@patternfly/patternfly-mcp/tools` | `createMcpTool` and tool-authoring types for plugins and inline tools | ```typescript import { start, type PfMcpInstance } from '@patternfly/patternfly-mcp'; +import { createMcpTool, type ToolModule } from '@patternfly/patternfly-mcp/tools'; ``` -**Deep imports are not supported.** Accessing internal modules (e.g., `@patternfly/patternfly-mcp/dist/server`) is restricted by our package configuration. This "flattened" export strategy allows us to refactor internal code and move logic between files without impacting your programmatic integrations, as long as the root exports remain stable. +**Deep imports are not supported.** Accessing internal modules (e.g., `@patternfly/patternfly-mcp/dist/server`) is restricted by our package configuration. This export strategy allows us to refactor internal code without impacting supported integrations, as long as documented entry points remain stable. + +If you require access to a type or utility that is not currently exported from a supported entry, please open an issue to discuss your use case. + +#### Deprecations -If you require access to a type or utility that is not currently exported from the root, please open an issue to discuss your use case for extending the public API. +Deprecated APIs remain available through minor releases. **Removal is targeted for major releases** (semver). See [CONTRIBUTING.md](../CONTRIBUTING.md#public-api-and-deprecations). + +Current consumer deprecations: + +| Deprecated | Replacement (Use instead) | Removal | +|-------------------------------------|------------------------------------|-----------------| +| `createMcpTool` from the root entry | `@patternfly/patternfly-mcp/tools` | Planned **3.0** | +| `CliOptions` type alias | `PfMcpCliOptions` | Planned **3.0** | ### Server instance @@ -269,7 +286,8 @@ Reference typings are exported from the package. The full listing can be found i You can embed the MCP server inside your application using the `start()` function and provide **Tool Modules** directly. ```ts -import { start, createMcpTool, type PfMcpInstance, type ToolModule } from '@patternfly/patternfly-mcp'; +import { start, type PfMcpInstance } from '@patternfly/patternfly-mcp'; +import { createMcpTool, type ToolModule } from '@patternfly/patternfly-mcp/tools'; const echoTool: ToolModule = createMcpTool({ name: 'echoAMessage', @@ -321,7 +339,7 @@ You can extend the server's capabilities by loading **Tool Plugins** at startup. - **Node.js >= 22**: Loading external tool plugins (`--tool`) requires Node.js version 22 or higher due to the use of advanced process isolation and ESM module loading features. - **ESM**: Plugins MUST be authored as ECMAScript Modules. -- **Dependency Resolution**: Plugins importing from `@patternfly/patternfly-mcp` require the package to be resolvable in the execution environment. This may require a local `npm install` in the plugin's directory or project root if the package is not available globally. +- **Dependency Resolution**: Plugins importing from `@patternfly/patternfly-mcp/tools` require the package to be resolvable in the execution environment. This may require a local `npm install` in the plugin's directory or project root if the package is not available globally. ### Security & isolation @@ -339,7 +357,7 @@ We recommend using the `createMcpTool` helper to define tools. It ensures your t #### Authoring a single tool module ```ts -import { createMcpTool } from '@patternfly/patternfly-mcp'; +import { createMcpTool } from '@patternfly/patternfly-mcp/tools'; export default createMcpTool({ name: 'hello', @@ -360,7 +378,7 @@ export default createMcpTool({ #### Authoring multiple tools in one module ```ts -import { createMcpTool } from '@patternfly/patternfly-mcp'; +import { createMcpTool } from '@patternfly/patternfly-mcp/tools'; export default createMcpTool([ { name: 'hi', description: 'Greets', inputSchema: {}, handler: () => ({ content: [{ type: 'text', text: 'hi' }] }) }, diff --git a/docs/examples/README.md b/docs/examples/README.md index 69a98548..1acd97d4 100644 --- a/docs/examples/README.md +++ b/docs/examples/README.md @@ -16,9 +16,10 @@ Examples should follow the basic guidelines: 2. Filenames are lowerCamelCased 3. Keep examples short; this is an introduction to the project 4. Examples are either JS or TS with ESM import/exports -5. Comments/annotations are used to explain key concepts -6. Examples are linted from the project's linting configs with +5. Tool plugins import `createMcpTool` from `@patternfly/patternfly-mcp/tools`; use the root entry for `start()` and server types +6. Comments/annotations are used to explain key concepts +7. Examples are linted from the project's linting configs with - `npm run test:lint` - `npm run test:types` - `npm run test:spell-docs` -7. Examples are tested and can be run without errors +8. Examples are tested and can be run without errors diff --git a/docs/examples/embeddedInlineTool.ts b/docs/examples/embeddedInlineTool.ts index 4517e67c..09442bf1 100644 --- a/docs/examples/embeddedInlineTool.ts +++ b/docs/examples/embeddedInlineTool.ts @@ -5,7 +5,9 @@ * inside your application with custom tools. */ // @ts-expect-error: Cannot find module '@patternfly/patternfly-mcp' - Remove this line if you're copying this example -import { start, createMcpTool, type PfMcpInstance, type PfMcpLogEvent, type PfMcpStats, type ToolModule } from '@patternfly/patternfly-mcp'; +import { start, type PfMcpInstance, type PfMcpLogEvent, type PfMcpStats } from '@patternfly/patternfly-mcp'; +// @ts-expect-error: Cannot find module '@patternfly/patternfly-mcp/tools' - Remove this line if you're copying this example +import { createMcpTool, type ToolModule } from '@patternfly/patternfly-mcp/tools'; /** * Echo tool - A custom tool that echoes back the provided user message. diff --git a/docs/examples/toolPluginGitStatus.js b/docs/examples/toolPluginGitStatus.js index 879326c3..b5f2cc1c 100644 --- a/docs/examples/toolPluginGitStatus.js +++ b/docs/examples/toolPluginGitStatus.js @@ -11,7 +11,7 @@ * - Requires ESM default export. */ import { spawn } from 'node:child_process'; -import { createMcpTool } from '@patternfly/patternfly-mcp'; +import { createMcpTool } from '@patternfly/patternfly-mcp/tools'; /** * Helper, execute a command using spawn with argument handling. diff --git a/docs/examples/toolPluginHelloWorld.js b/docs/examples/toolPluginHelloWorld.js index 9738a454..0e141341 100644 --- a/docs/examples/toolPluginHelloWorld.js +++ b/docs/examples/toolPluginHelloWorld.js @@ -10,7 +10,7 @@ * - JS support only. TypeScript is only supported for embedding the server. * - Requires ESM default export. */ -import { createMcpTool } from '@patternfly/patternfly-mcp'; +import { createMcpTool } from '@patternfly/patternfly-mcp/tools'; export default createMcpTool({ name: 'helloWorld', diff --git a/guidelines/agent_coding.md b/guidelines/agent_coding.md index 074442b2..db989c49 100644 --- a/guidelines/agent_coding.md +++ b/guidelines/agent_coding.md @@ -59,7 +59,7 @@ All tools and resources MUST follow the **Creator Pattern** for dependency injec - **Options Injection Pattern**: Environment-dependent helpers should accept an optional `options` parameter that defaults to `getOptions()`. This allows for explicit dependency injection in tests while maintaining ergonomics via `AsyncLocalStorage` in production. Pure transforms should remain option-agnostic. - **Internal Tools**: `(options = getOptions()): McpTool` -> Returns `[name, schema, handler]`. - **Internal Resources**: `(options = getOptions()): McpResource` -> Returns `[name, uri, config, handler]`. -- **External Tool Plugins**: Authored using the `createMcpTool` helper with an object configuration, exported as `default`. +- **External Tool Plugins**: Authored with `createMcpTool` from `@patternfly/patternfly-mcp/tools`, using an object configuration, exported as `default`. - **Testing**: Creators allow easy mocking: `const tool = usePatternFlyDocsTool(mockOptions)`. ### 2.2 Module Organization and Exports @@ -78,7 +78,7 @@ All tools and resources MUST follow the **Creator Pattern** for dependency injec External tool plugins should follow this basic structure: ```javascript -import { createMcpTool } from '@patternfly/patternfly-mcp'; +import { createMcpTool } from '@patternfly/patternfly-mcp/tools'; export default createMcpTool({ name: 'myTool', @@ -208,7 +208,7 @@ While the codebase emphasizes pragmatism, **public APIs require comprehensive JS /** * Exposed options for CLI use. A focused options interface. * - * Alias of {@link CliOptions} (Internal type). + * Alias of {@link CliOptions} (Internal type). `CliOptions` is deprecated; use {@link PfMcpCliOptions}. */ type PfMcpCliOptions = CliOptions; diff --git a/jest.config.ts b/jest.config.ts index b0e63857..855d7dce 100644 --- a/jest.config.ts +++ b/jest.config.ts @@ -23,7 +23,8 @@ export default { 'src/**/*.ts', '!src/**/.*/**', '!src/cli.ts', - '!src/declarations*' + '!src/declarations*', + '!src/server.workerEntry.ts' ], coverageThreshold: { global: { @@ -62,6 +63,12 @@ export default { ] } }, + { + displayName: 'package', + roots: ['/tests/package'], + testMatch: ['/tests/package/**/*.test.ts'], + ...baseConfig + }, { displayName: 'e2e', roots: ['/tests/e2e'], diff --git a/jest.setupTests.ts b/jest.setupTests.ts index 26ed74ab..2ba2954a 100644 --- a/jest.setupTests.ts +++ b/jest.setupTests.ts @@ -13,6 +13,24 @@ jest.mock('child_process', () => ({ execSync: (...args: unknown[]) => `${JSON.stringify(args)}` })); +/** + * Note: Mock worker_threads to avoid issues with Worker in tests. + * + * Heads-up: If you think you need real worker_threads in a UNIT test, ask yourself: + * "Why are you using real worker threads in a unit test?" + * + * - Unit tests should be fast and deterministic; prefer the default mock. + * - If you truly need real workers, that likely belongs in tests/package or tests/e2e. + * + * Still certain you need real workers in a unit test? Choosing to `jest.unmock` + * signals non-trivial behavior. Your PR/work will be reviewed thoroughly, AND there + * is a higher likelihood you'll be asked to rethink the tests. + * + * @example + * jest.unmock('worker_threads'); + */ +jest.mock('worker_threads'); + /** * Note: Mock pid-port to avoid ES module import issues in Jest * - Returns undefined to simulate port is free (no process found) diff --git a/package.json b/package.json index 2909580a..eb119c3a 100644 --- a/package.json +++ b/package.json @@ -6,12 +6,17 @@ "type": "module", "imports": { "~docsCatalog": "./src/docs.json", - "#toolsHost": "./dist/server.toolsHost.js" + "#toolsHost": "./dist/server.toolsHost.js", + "#workerEntry": "./dist/server.workerEntry.js" }, "exports": { ".": { "types": "./dist/index.d.ts", "default": "./dist/index.js" + }, + "./tools": { + "types": "./dist/tools.d.ts", + "default": "./dist/tools.js" } }, "bin": { @@ -38,10 +43,11 @@ "test:audit-container": "npm run container:build && jest --selectProjects audit:container", "test:ci": "npm test -- --coverage", "test:dev": "npm test -- --watchAll", - "test:integration": "npm run build && NODE_OPTIONS='--experimental-vm-modules' jest --selectProjects e2e", + "test:integration": "npm run build && jest --selectProjects package && NODE_OPTIONS='--experimental-vm-modules' jest --selectProjects e2e", "test:integration-dev": "npm run test:integration -- --watchAll", "test:lint": "eslint .", "test:lint-fix": "eslint . --fix", + "test:package": "npm run build && jest --selectProjects package", "test:spell-docs": "cspell './README.md' './CONTRIBUTING.md' './GOVERNANCE.md' './SECURITY.md' './docs/**/*.md' './guidelines/**/*.md' './docs/**/*.ts' './docs/**/*.js' --config ./cspell.config.json --fail-fast", "test:spell": "cspell './src/**/*.ts' './tests/**/*.ts' --exclude './src/**/*test*' --exclude './tests/**/*test*' --config ./cspell.config.json --fail-fast", "test:types": "tsc --noEmit", diff --git a/src/__tests__/__snapshots__/server.test.ts.snap b/src/__tests__/__snapshots__/server.test.ts.snap index 6ffe41b4..a8a2a8db 100644 --- a/src/__tests__/__snapshots__/server.test.ts.snap +++ b/src/__tests__/__snapshots__/server.test.ts.snap @@ -15,6 +15,12 @@ exports[`runServer should allow server to be stopped, http stop server: diagnost [ "No external tools loaded.", ], + [ + "Registered collection: patternfly-docs", + ], + [ + "Registered collection: patternfly-component-schemas", + ], [ "Registered resource: patternfly-context", ], @@ -77,6 +83,12 @@ exports[`runServer should allow server to be stopped, stdio stop server: diagnos [ "No external tools loaded.", ], + [ + "Registered collection: patternfly-docs", + ], + [ + "Registered collection: patternfly-component-schemas", + ], [ "Registered resource: patternfly-context", ], @@ -139,6 +151,12 @@ exports[`runServer should attempt to run server, create transport, connect, and [ "No external tools loaded.", ], + [ + "Registered collection: patternfly-docs", + ], + [ + "Registered collection: patternfly-component-schemas", + ], [ "Registered resource: patternfly-context", ], @@ -212,6 +230,12 @@ exports[`runServer should attempt to run server, disable SIGINT handler: diagnos [ "No external tools loaded.", ], + [ + "Registered collection: patternfly-docs", + ], + [ + "Registered collection: patternfly-component-schemas", + ], [ "Registered resource: patternfly-context", ], @@ -280,6 +304,12 @@ exports[`runServer should attempt to run server, enable SIGINT handler explicitl [ "No external tools loaded.", ], + [ + "Registered collection: patternfly-docs", + ], + [ + "Registered collection: patternfly-component-schemas", + ], [ "Registered resource: patternfly-context", ], @@ -353,6 +383,12 @@ exports[`runServer should attempt to run server, log warnings for experimental o [ "No external tools loaded.", ], + [ + "Registered collection: patternfly-docs", + ], + [ + "Registered collection: patternfly-component-schemas", + ], [ "Registered resource: patternfly-context", ], @@ -441,6 +477,12 @@ exports[`runServer should attempt to run server, register a tool: diagnostics 1` [ "No external tools loaded.", ], + [ + "Registered collection: patternfly-docs", + ], + [ + "Registered collection: patternfly-component-schemas", + ], [ "Registered resource: patternfly-context", ], @@ -522,6 +564,12 @@ exports[`runServer should attempt to run server, register multiple tools: diagno [ "No external tools loaded.", ], + [ + "Registered collection: patternfly-docs", + ], + [ + "Registered collection: patternfly-component-schemas", + ], [ "Registered resource: patternfly-context", ], @@ -610,6 +658,12 @@ exports[`runServer should attempt to run server, use custom options: diagnostics [ "No external tools loaded.", ], + [ + "Registered collection: patternfly-docs", + ], + [ + "Registered collection: patternfly-component-schemas", + ], [ "Registered resource: patternfly-context", ], @@ -683,6 +737,12 @@ exports[`runServer should attempt to run server, use default tools, http: diagno [ "No external tools loaded.", ], + [ + "Registered collection: patternfly-docs", + ], + [ + "Registered collection: patternfly-component-schemas", + ], [ "Registered resource: patternfly-context", ], @@ -765,6 +825,12 @@ exports[`runServer should attempt to run server, use default tools, stdio: diagn [ "No external tools loaded.", ], + [ + "Registered collection: patternfly-docs", + ], + [ + "Registered collection: patternfly-component-schemas", + ], [ "Registered resource: patternfly-context", ], diff --git a/src/__tests__/server.collections.test.ts b/src/__tests__/server.collections.test.ts index c4fd399f..f0c4d218 100644 --- a/src/__tests__/server.collections.test.ts +++ b/src/__tests__/server.collections.test.ts @@ -1,11 +1,16 @@ -import { composeCollections } from '../server.collections'; import { getOptions, getSessionOptions } from '../options.context'; +import { composeCollections } from '../server.collections'; +import { getHeavyPool } from '../server.workerPool'; jest.mock('../options.context', () => ({ getOptions: jest.fn(), getSessionOptions: jest.fn() })); +jest.mock('../server.workerPool', () => ({ + getHeavyPool: jest.fn() +})); + describe('composeCollections', () => { beforeEach(() => { jest.clearAllMocks(); @@ -28,12 +33,89 @@ describe('composeCollections', () => { expect(callback).toBe(mockHandler); }); + it('should proxy creators that declare parallel execution via runParallel with hash prefix', async () => { + const mockHandler = jest.fn(); + const mockCreator: any = jest.fn(() => [ + 'parallel-collection', + mockHandler, + { runParallel: '#collectionLoremIpsum' } + ]); + + (getOptions as jest.Mock).mockReturnValue({ serverName: 'mcp' }); + (getSessionOptions as jest.Mock).mockReturnValue({ sessionId: 'session-id' }); + + const heavyPool = { runTask: jest.fn().mockResolvedValue({ records: [] }) }; + + (getHeavyPool as jest.Mock).mockReturnValue(heavyPool); + + const result: any = await composeCollections([mockCreator]); + + expect(result.length).toBe(1); + + const [name, handler, config] = result[0](); + + expect(name).toBe('parallel-collection'); + expect(config?._isInternal).toBe(true); + expect(config?.runParallel).toBe('#collectionLoremIpsum'); + + const executionResult = await handler({ inputArg: 'test' }); + + expect(executionResult).toEqual({ records: [] }); + + expect(heavyPool.runTask).toHaveBeenCalledWith({ + moduleSpecifier: '#collectionLoremIpsum', + exportName: 'collectionCallback', + args: { inputArg: 'test' }, + options: expect.any(Object), + session: { sessionId: 'session-id' } + }); + }); + it('should return an empty array when no creators are provided', async () => { const result = await composeCollections([]); expect(result).toEqual([]); }); + it('should wrap creators that declare runSchedule with a deferred task', async () => { + const mockHandler = jest.fn().mockResolvedValue({ records: [{ id: 'row-1' }] }); + const mockCreator: any = jest.fn(() => [ + 'scheduled-collection', + mockHandler, + { runSchedule: { cancelMs: 100, intervalMs: 50 } } + ]); + + const result: any = await composeCollections([mockCreator]); + + expect(result.length).toBe(1); + + const [name, handler, config] = result[0](); + + expect(name).toBe('scheduled-collection'); + expect(config?._isInternal).toBe(true); + expect(handler).not.toBe(mockHandler); + + const executionResult = await handler({ inputArg: 'test' }); + + expect(mockHandler).toHaveBeenCalledWith({ inputArg: 'test' }); + expect(executionResult).toEqual({ records: [{ id: 'row-1' }] }); + }); + + it('should fall back to an empty records result when the deferred task yields undefined', async () => { + const mockHandler = jest.fn().mockResolvedValue(undefined); + const mockCreator: any = jest.fn(() => [ + 'scheduled-empty-collection', + mockHandler, + { runSchedule: { cancelMs: 100, intervalMs: 50 } } + ]); + + const result: any = await composeCollections([mockCreator]); + const [, handler] = result[0](); + const executionResult = await handler(); + + expect(executionResult).toEqual({ records: [] }); + }); + it.each([ { description: 'with custom options', diff --git a/src/__tests__/server.workerPool.test.ts b/src/__tests__/server.workerPool.test.ts new file mode 100644 index 00000000..6aa9b192 --- /dev/null +++ b/src/__tests__/server.workerPool.test.ts @@ -0,0 +1,385 @@ +import { Worker } from 'node:worker_threads'; +import { + buildPersistentPool, + buildTransientPool, + createPoolAbort, + getHeavyPool, + getWorkerScriptPath, + resetWorkerPools, + sendWorkerPoolsShutdown +} from '../server.workerPool'; + +const MockWorker = Worker as jest.MockedClass; + +describe('createPoolAbort', () => { + const payload = { moduleSpecifier: 'spec', args: {} }; + + it.each([ + { + description: 'runTask after abort', + abortBeforeRun: true, + expectedQueueLength: 0 + }, + { + description: 'queued tasks when abort fires', + abortBeforeRun: false, + expectedQueueLength: 1 + }, + { + description: 'in-flight tasks when abort fires', + abortBeforeRun: false, + expectedQueueLength: 1 + } + ])('should reject $description', async ({ abortBeforeRun, expectedQueueLength }) => { + const poolAbort = createPoolAbort(); + const queue: any[] = []; + + if (abortBeforeRun) { + poolAbort.abort(); + } + + const task = poolAbort.runTask(payload, queued => { + queue.push(queued); + }); + + if (!abortBeforeRun) { + poolAbort.abort(); + } + + expect(queue).toHaveLength(expectedQueueLength); + await expect(task).rejects.toThrow('Worker pool shutdown'); + }); + + it('should resolve when the enqueue handler completes the task', async () => { + const poolAbort = createPoolAbort(); + + const task = poolAbort.runTask(payload, queued => { + queued.resolve('ok'); + }); + + await expect(task).resolves.toBe('ok'); + }); + + it('should report aborted state after abort', () => { + const poolAbort = createPoolAbort(); + + expect(poolAbort.isAborted()).toBe(false); + poolAbort.abort(); + expect(poolAbort.isAborted()).toBe(true); + }); +}); + +describe('getWorkerScriptPath', () => { + it('should attempt to return the worker script path', () => { + const workerScriptPath = getWorkerScriptPath(); + + expect(workerScriptPath).toBeDefined(); + }); +}); + +describe('buildPersistentPool', () => { + beforeEach(() => { + jest.clearAllMocks(); + resetWorkerPools(); + }); + + it('should resolve payload on message success', async () => { + const expected = 'task-success-result'; + const emit = { success: true, payload: expected }; + + let messageCallback: any; + const mockOn = jest.fn((mockEvent, mockCallback) => { + if (mockEvent === 'message') { + messageCallback = mockCallback; + } + }); + + MockWorker.mockImplementation((): any => ({ + on: mockOn, + off: jest.fn(), + postMessage: jest.fn(() => { + if (messageCallback) { + messageCallback(emit); + } + }) + })); + + const pool = buildPersistentPool(2); + + const taskPromise = pool.runTask({ + moduleSpecifier: 'data:text/javascript;base64,KCkgPT4gbG9yZW1JcHVzbQ==', + args: { test: true } + }); + + await expect(taskPromise).resolves.toEqual(expected); + }); + + it.each([ + { + description: 'custom error message on message failure', + event: 'message', + emit: { success: false, error: 'Custom execution error' }, + expected: 'Custom execution error' + }, + { + description: 'standard error on worker thread error event', + event: 'error', + emit: new Error('Worker Thread Exception'), + expected: 'Worker Thread Exception' + } + ])('should reject on event, $description', async ({ emit, event, expected }) => { + let targetCallback: any; + const mockOn = jest.fn((mockEvent, mockCallback) => { + if (mockEvent === event) { + targetCallback = mockCallback; + } + }); + + MockWorker.mockImplementation((): any => ({ + on: mockOn, + off: jest.fn(), + postMessage: jest.fn(() => { + if (targetCallback) { + targetCallback(emit); + } + }) + })); + + const pool = buildPersistentPool(2); + + const taskPromise = pool.runTask({ + moduleSpecifier: 'data:text/javascript;base64,KCkgPT4gbG9yZW1JcHVzbQ==', + args: { test: true } + }); + + await expect(taskPromise).rejects.toThrow(expected); + }); + + it('should handle tasks in queue when active workers exceed limit', async () => { + const instances: any[] = []; + + MockWorker.mockImplementation((): any => { + const listeners: Record = {}; + const workerInstance = { + on: jest.fn((event: string, cb: any): any => { + listeners[event] = cb; + + return workerInstance; + }), + off: jest.fn((event: string, _cb: any): any => { + delete listeners[event]; + + return workerInstance; + }), + postMessage: jest.fn(), + // Trigger events manually inside the test + emit: (event: string, value: any) => { + if (listeners[event]) { + listeners[event](value); + } + } + }; + + instances.push(workerInstance); + + return workerInstance as any; + }); + + const pool = buildPersistentPool(2); + + // Initial warm up immediately spawns 2 workers + expect(MockWorker).toHaveBeenCalledTimes(2); + expect(instances).toHaveLength(2); + + // Queue 3 tasks concurrently + const t1 = pool.runTask({ moduleSpecifier: 'spec-1', args: {} }); + const t2 = pool.runTask({ moduleSpecifier: 'spec-2', args: {} }); + const t3 = pool.runTask({ moduleSpecifier: 'spec-3', args: {} }); + + // Resolve the first worker task + instances[0].emit('message', { success: true, payload: 'result-1' }); + + // Resolve the remaining active worker sessions + instances[1].emit('message', { success: true, payload: 'result-2' }); + instances[0].emit('message', { success: true, payload: 'result-3' }); + + // Verify all tasks resolve successfully with their respective values + await expect(t1).resolves.toBe('result-1'); + await expect(t2).resolves.toBe('result-2'); + await expect(t3).resolves.toBe('result-3'); + }); +}); + +describe('buildTransientPool', () => { + beforeEach(() => { + jest.clearAllMocks(); + resetWorkerPools(); + }); + + it('should resolve payload on message success', async () => { + const expected = 'task-success-result'; + const emit = { success: true, payload: expected }; + + const mockOn = jest.fn((mockEvent, mockCallback) => { + if (mockEvent === 'message') { + mockCallback(emit); + } + }); + + MockWorker.mockImplementation((): any => ({ + on: mockOn, + postMessage: jest.fn(), + terminate: jest.fn().mockResolvedValue(0) + })); + + const pool = buildTransientPool(2); + + const taskPromise = pool.runTask({ + moduleSpecifier: 'data:text/javascript;base64,KCkgPT4gbG9yZW1JcHVzbQ==', + args: { test: true } + }); + + await expect(taskPromise).resolves.toEqual(expected); + }); + + it.each([ + { + description: 'custom error message on message failure', + event: 'message', + emit: { success: false, error: 'Custom execution error' }, + expected: 'Custom execution error' + }, + { + description: 'standard error on worker thread error event', + event: 'error', + emit: new Error('Worker Thread Exception'), + expected: 'Worker Thread Exception' + }, + { + description: 'exit code info on unexpected exit event', + event: 'exit', + emit: 1, + expected: 'Transient worker exited unexpectedly with code 1' + } + ])('should reject on event, $description', async ({ emit, event, expected }) => { + const mockOn = jest.fn((mockEvent, mockCallback) => { + if (mockEvent === event) { + mockCallback(emit); + } + }); + + MockWorker.mockImplementation((): any => ({ + on: mockOn, + postMessage: jest.fn(), + terminate: jest.fn().mockResolvedValue(0) + })); + + const pool = buildTransientPool(2); + + const taskPromise = pool.runTask({ + moduleSpecifier: 'data:text/javascript;base64,KCkgPT4gbG9yZW1JcHVzbQ==', + args: { test: true } + }); + + await expect(taskPromise).rejects.toThrow(expected); + }); + + it('should handle tasks in queue when active workers exceed limit', async () => { + const instances: any[] = []; + + MockWorker.mockImplementation((): any => { + const listeners: Record = {}; + const workerInstance = { + on: jest.fn((event: string, cb: any): any => { + listeners[event] = cb; + + return workerInstance; + }), + postMessage: jest.fn(), + terminate: jest.fn().mockResolvedValue(0), + + // Trigger events manually inside the test + emit: (event: string, value: any) => { + if (listeners[event]) { + listeners[event](value); + } + } + }; + + instances.push(workerInstance); + + return workerInstance as any; + }); + + const pool = buildTransientPool(2); + + // Queue 3 tasks concurrently + const t1 = pool.runTask({ moduleSpecifier: 'spec-1', args: {} }); + const t2 = pool.runTask({ moduleSpecifier: 'spec-2', args: {} }); + const t3 = pool.runTask({ moduleSpecifier: 'spec-3', args: {} }); + + // Assert that exactly 2 worker threads are spawned initially, and the 3rd is queued + expect(MockWorker).toHaveBeenCalledTimes(2); + expect(instances).toHaveLength(2); + + // Resolve the first worker task and simulate its exit + instances[0].emit('message', { success: true, payload: 'result-1' }); + instances[0].emit('exit', 0); + + // MockWorker should have been instantiated a third time for the queued task + expect(MockWorker).toHaveBeenCalledTimes(3); + expect(instances).toHaveLength(3); + + // Resolve the remaining active workers + instances[1].emit('message', { success: true, payload: 'result-2' }); + instances[1].emit('exit', 0); + + instances[2].emit('message', { success: true, payload: 'result-3' }); + instances[2].emit('exit', 0); + + // Verify all tasks resolve successfully with their respective values + await expect(t1).resolves.toBe('result-1'); + await expect(t2).resolves.toBe('result-2'); + await expect(t3).resolves.toBe('result-3'); + }); + + it('should reject queued and in-flight tasks on shutdown', async () => { + MockWorker.mockImplementation((): any => ({ + on: jest.fn(), + postMessage: jest.fn(), + terminate: jest.fn().mockResolvedValue(0) + })); + + const pool = buildTransientPool(1); + const activeTask = pool.runTask({ moduleSpecifier: 'spec-active', args: {} }); + const queuedTask = pool.runTask({ moduleSpecifier: 'spec-queued', args: {} }); + + await pool.shutdown(); + + await expect(activeTask).rejects.toThrow('Worker pool shutdown'); + await expect(queuedTask).rejects.toThrow('Worker pool shutdown'); + }); +}); + +describe('worker pool registry', () => { + beforeEach(() => { + resetWorkerPools(); + }); + + it('should return the same heavy pool instance from getHeavyPool', () => { + const first = getHeavyPool(); + const second = getHeavyPool(); + + expect(first).toBe(second); + }); + + it('should shutdown registered pools and clear the registry', async () => { + const pool = getHeavyPool(); + + await sendWorkerPoolsShutdown(); + + const nextPool = getHeavyPool(); + + expect(nextPool).not.toBe(pool); + }); +}); diff --git a/src/__tests__/server.workerRunner.test.ts b/src/__tests__/server.workerRunner.test.ts new file mode 100644 index 00000000..a40aabe6 --- /dev/null +++ b/src/__tests__/server.workerRunner.test.ts @@ -0,0 +1,218 @@ +import { executeTask, keepWorkerAlive, runWorker } from '../server.workerRunner'; + +const mockParentPort = { + postMessage: jest.fn(), + on: jest.fn(), + ref: jest.fn(), + unref: jest.fn() +}; +const mockWorkerData = jest.fn(() => null); + +jest.mock('node:worker_threads', () => ({ + get parentPort() { + return mockParentPort; + }, + get workerData() { + return mockWorkerData(); + } +})); + +jest.mock('../options.context', () => ({ + runWithOptions: jest.fn(), + runWithSession: jest.fn() +})); + +describe('executeTask', () => { + let mockRunWithOptions: any; + let mockRunWithSession: any; + + beforeEach(async () => { + jest.clearAllMocks(); + + const options = await import('../options.context'); + + mockRunWithOptions = jest.mocked(options.runWithOptions).mockImplementation((_options, callback: any) => callback()); + mockRunWithSession = jest.mocked(options.runWithSession).mockImplementation((_options, callback: any) => callback()); + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + it('throws when moduleSpecifier is missing', async () => { + await expect(executeTask({} as any)).rejects.toThrow( + 'No moduleSpecifier specified for worker task.' + ); + }); + + it.each([ + { + description: 'default', + payload: { + exportName: 'default' + } + }, + { + description: 'named', + payload: { + exportName: 'loremIpsum' + } + } + ])('should invoke export payload, $description', async ({ payload }) => { + const mockFunc = jest.fn().mockReturnValue('ok'); + + const mockPayload = { + moduleSpecifier: '/abs/path/to/task.mjs', + args: { a: 1 }, + options: { o: true }, + session: { s: 2 }, + ...payload + }; + + jest.spyOn(global as any, 'Function').mockReturnValue(() => + Promise.resolve({ [payload.exportName || 'default']: mockFunc })); + + const result = await executeTask(mockPayload as any); + + expect(mockFunc).toHaveBeenCalledTimes(1); + expect(mockFunc).toHaveBeenCalledWith(mockPayload.args); + expect(mockRunWithOptions).toHaveBeenCalledWith(mockPayload.options, expect.any(Function)); + expect(mockRunWithSession).toHaveBeenCalledWith(mockPayload.session, expect.any(Function)); + expect(result).toBe('ok'); + }); + + it('should throw if export is not a function', async () => { + jest.spyOn(global as any, 'Function').mockReturnValue(() => + Promise.resolve({ default: 123 })); + + await expect( + executeTask({ moduleSpecifier: '/dolorSit.mjs' } as any) + ).rejects.toThrow("Exported module '/dolorSit.mjs' (export: 'default') must be a function."); + }); +}); + +describe('keepWorkerAlive', () => { + const originalSetTimeout = global.setTimeout; + const originalClearTimeout = global.clearTimeout; + + beforeEach(() => { + jest.clearAllMocks(); + }); + + afterEach(() => { + (global as any).setTimeout = originalSetTimeout; + (global as any).clearTimeout = originalClearTimeout; + }); + + it('should pin parentPort via ref() and return an unref() cleanup handler', () => { + const release = keepWorkerAlive(); + + expect(mockParentPort.ref).toHaveBeenCalledTimes(1); + expect(mockParentPort.unref).not.toHaveBeenCalled(); + + release(); + + expect(mockParentPort.unref).toHaveBeenCalledTimes(1); + }); + + it('should throw by default when parentPort.ref is not a function', () => { + (mockParentPort as any).ref = undefined; + + expect(() => keepWorkerAlive()).toThrow('parentPort.ref is not a function — worker keep-alive failed'); + }); + + it('should fall back to a setTimeout and return a cleanup handler', () => { + (global as any).setTimeout = jest.fn(); + (global as any).clearTimeout = jest.fn(); + const release = keepWorkerAlive({ throwOnParentPortError: false }); + + expect(global.setTimeout).toHaveBeenCalledTimes(1); + expect(global.clearTimeout).not.toHaveBeenCalled(); + + release(); + + expect(global.clearTimeout).toHaveBeenCalledTimes(1); + }); +}); + +describe('runWorker', () => { + beforeEach(async () => { + jest.clearAllMocks(); + + mockWorkerData.mockReturnValue(null); + + const opts = await import('../options.context'); + + jest.mocked(opts.runWithOptions).mockImplementation((_options, callback: any) => callback()); + jest.mocked(opts.runWithSession).mockImplementation((_options, callback: any) => callback()); + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + it('should post success for route A, dynamic import resolves default export', async () => { + jest.spyOn(global as any, 'Function').mockReturnValue(() => + Promise.resolve({ default: () => 'OK' })); + + mockWorkerData.mockReturnValue({ moduleSpecifier: '/any/path.mjs' } as any); + + await runWorker({ throwOnParentPortError: false }); + + expect(mockParentPort.postMessage).toHaveBeenCalledWith({ success: true, payload: 'OK' }); + }); + + it('should post failure for route A, export throws', async () => { + jest.spyOn(global as any, 'Function').mockReturnValue(() => + Promise.resolve({ default: () => { throw new Error('dolor sit'); } })); + + mockWorkerData.mockReturnValue({ moduleSpecifier: '/any/path.mjs' } as any); + + await runWorker({ throwOnParentPortError: false }); + + expect(mockParentPort.postMessage).toHaveBeenCalledWith({ + success: false, + error: expect.objectContaining({ message: 'dolor sit' }) + }); + }); + + it('should post success for route B, incoming task', async () => { + jest.spyOn(global as any, 'Function').mockReturnValue(() => + Promise.resolve({ default: (args: any) => args?.value ?? 123 })); + + let handler: any; + + mockParentPort.on.mockImplementation((_evt, handle) => { + handler = handle; + }); + + runWorker(); + + expect(mockParentPort.on).toHaveBeenCalledWith('message', expect.any(Function)); + + await handler({ moduleSpecifier: '/task.mjs', args: { value: 'OK' } }); + + expect(mockParentPort.postMessage).toHaveBeenCalledWith({ success: true, payload: 'OK' }); + }); + + it('should post failure for route B, export throws', async () => { + jest.spyOn(global as any, 'Function').mockReturnValue(() => + Promise.resolve({ default: () => { throw new Error('lorem ipsum'); } })); + + let handler: any; + + mockParentPort.on.mockImplementation((_evt, handle) => { + handler = handle; + }); + + runWorker(); + expect(mockParentPort.on).toHaveBeenCalledWith('message', expect.any(Function)); + + await handler({ moduleSpecifier: '/task.mjs' }); + + expect(mockParentPort.postMessage).toHaveBeenCalledWith({ + success: false, + error: expect.objectContaining({ message: 'lorem ipsum' }) + }); + }); +}); diff --git a/src/collections.ts b/src/collections.ts index 2d25abde..ce2f1636 100644 --- a/src/collections.ts +++ b/src/collections.ts @@ -39,15 +39,21 @@ interface McpCollectionResult { /** * Standardized Tuple-based Record Source. * - * @note `priority` and `group` are future properties being considered in the related - * collection work as a way to sort and override collections. + * @note **Future**: `priority` and `group` are future properties being considered in the + * related collection work as a way to sort and override collections. + * + * @note **Future**: Review supporting `boolean` variations and async callbacks + * `async (options) => boolean | #${string}` for dynamic configs. * * 0. `name` `{string}`: Unique identifier/name * 1. `handler` `{Function}`: callback function accepting an optional argument * 2. `_config` `{Object}`: Application level record source configuration. Unavailable to * record collection plugins. - * - `_config.runInChildProcess`: Optional callback function to dynamically decide - * if the record source should run in a child process. + * - `_config.runParallel`: Optional internal import specifier (`#specifier`) to run the + * collection handler in a worker thread via the heavy pool. The referenced + * module must export `collectionCallback`. Applied in {@link composeCollections}. + * - `_config.runSchedule`: Optional object to dynamically decide if the record source + * should run in a scheduled interval using {@link DeferTaskOptions} * - `_config.isRequired`: Optional boolean used to control server startup when * collections are required for operation. * - `_config._isInternal`: Optional boolean. Applied internally. Attempting to manually @@ -57,7 +63,8 @@ type McpCollection = [ name: string, handler: (arg?: unknown) => McpCollectionResult | Promise, _config?: { - runInChildProcess?: boolean | ((options?: GlobalOptions) => boolean | Promise); + runParallel?: `#${string}`; + runSchedule?: { cancelMs?: number, intervalMs?: number }; // priority?: number; isRequired?: boolean; // group?: string; @@ -129,8 +136,6 @@ type RegisterOnSettle = (results: RegisterCollectionsResult) => void; * This type encapsulates the outcome of registering collections, grouping the * results into settled, fulfilled, and rejected categories. * - * @typedef {Object} RegisterCollectionsResult - * * @property {RegisterCollectionSettledItem[]} settled - Settled registration results, including * both fulfilled and failed attempts. * @property {McpCollectionResult[]} fulfilled - Successfully registered collections, containing @@ -170,7 +175,7 @@ const registerCollections = async ( onSettle?: RegisterOnSettle, onUpdate?: RegisterOnUpdate, onRequired?: RegisterOnRequired } = {} ): Promise => { - log.debug(`Initiating registration for ${collections.length} collections.`); + log.debug(`Reviewing registration for ${collections.length} collections.`); // Wrapper for each loader; handle incremental updates const registrationPromises = collections.map(async ([name, callback]) => { @@ -229,9 +234,9 @@ const registerCollections = async ( }; if (!res.isSuccess) { - log.error(`Failed to register collection ${item.name}: ${item.reason}`); + log.error(`Failed to register collection "${item.name}": ${item.reason}`); } else { - log.info(`Register collection: ${item.name}`); + log.debug(`Settled collection: ${item.name}`); } return item; diff --git a/src/index.ts b/src/index.ts index 76f2eb62..56801c3c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,24 +1,23 @@ -import { - type CliOptions, - type ExperimentalOptions, - type ProgrammaticOptions +import type { + CliOptions, + ExperimentalOptions, + ProgrammaticOptions } from './options'; import { parseCliOptions, parseProgrammaticOptions } from './options.parser'; import { getSessionOptions, setOptions, runWithSession } from './options.context'; -import { - runServer, - type ServerInstance, - type ServerSettings, - type ServerOnLog, - type ServerOnLogHandler, - type ServerLogEvent, - type ServerStatReport, - type ServerStats, - type ServerGetStats, - type ServerOptions +import type { + ServerInstance, + ServerSettings, + ServerOnLog, + ServerOnLogHandler, + ServerLogEvent, + ServerStatReport, + ServerStats, + ServerGetStats, + ServerOptions } from './server'; import { - createMcpTool, + createMcpTool as createMcpToolFromTools, type ToolCreator, type ToolModule, type ToolConfig, @@ -115,6 +114,13 @@ type PfMcpCliOptions = CliOptions; */ type DeprecatedCliOptions = PfMcpCliOptions; +/** + * `createMcpTool` from the root entry is deprecated for external tool plugins. + * + * @deprecated Import from `@patternfly/patternfly-mcp/tools` instead. + */ +const DeprecatedCreateMcpTool = createMcpToolFromTools; + /** * Exposed options for programmatic use. A limited `DefaultOptions` interface. * @@ -174,7 +180,8 @@ type PfMcpSettings = Pick; * * @example Programmatic: Listening for server stats * import { subscribe, unsubscribe } from 'node:diagnostics_channel'; - * import { start, createMcpTool } from '@patternfly/patternfly-mcp'; + * import { start } from '@patternfly/patternfly-mcp'; + * import { createMcpTool } from '@patternfly/patternfly-mcp/tools'; * * const { stop, isRunning, getStats } = await start(); * const stats = await getStats(); @@ -188,7 +195,8 @@ type PfMcpSettings = Pick; * } * * @example Programmatic: A MCP server with inline tool configuration and JSON inputSchema. - * import { start, createMcpTool } from '@patternfly/patternfly-mcp'; + * import { start } from '@patternfly/patternfly-mcp'; + * import { createMcpTool } from '@patternfly/patternfly-mcp/tools'; * * const myToolModule = createMcpTool({ * name: 'my-tool', @@ -246,6 +254,7 @@ const main = async ( try { // Generate session options const session = getSessionOptions(); + const { runServer } = await import('./server'); // Start the server, apply session values, then apply merged options to ensure stable hashing. return await runWithSession(session, async () => @@ -259,7 +268,7 @@ const main = async ( }; export { - createMcpTool, + DeprecatedCreateMcpTool as createMcpTool, main, main as start, type DeprecatedCliOptions as CliOptions, diff --git a/src/patternFly.getResources.ts b/src/patternFly.getResources.ts index 9963f291..be05c5b4 100644 --- a/src/patternFly.getResources.ts +++ b/src/patternFly.getResources.ts @@ -698,7 +698,7 @@ const setPatternFlyCollection = async ( log.warn('Failed getPatternFlyMcpResources clear.', error); } - log.debug(`Merged ${collection.records.length} records from collection [${name}].`); + log.debug(`Merging collection ${name} records. (${collection.records.length})`); } } catch (error) { log.error(`Failed to update collection [${name}]:`, error); diff --git a/src/server.collections.ts b/src/server.collections.ts index 32540522..31a5015f 100644 --- a/src/server.collections.ts +++ b/src/server.collections.ts @@ -1,20 +1,94 @@ +import { type McpCollection, type McpCollectionCreator, type McpCollectionResult } from './collections'; import { type AppSession, type GlobalOptions } from './options'; import { getOptions, getSessionOptions } from './options.context'; -import { type McpCollectionCreator } from './collections'; +import { getHeavyPool } from './server.workerPool'; +import { deferTask } from './server.task'; +import { formatUnknownError, log } from './logger'; + +/** + * Proxy a collection creator through the global worker thread pool. + * + * @param {McpCollectionCreator} creator - The original creator. + * @param {string} moduleSpecifier - The ESM import specifier to load in the worker. + * @param {GlobalOptions} options - Global options. + * @param {string} exportName - The name of the export to invoke in the worker module. Defaults to 'default'. + * @returns {McpCollectionCreator} The proxied creator function. + */ +const makeParallelProxyCreator = ({ + creator, + moduleSpecifier, + exportName = 'default' +}: { creator: McpCollectionCreator, moduleSpecifier: string, exportName?: string }, +options: GlobalOptions = getOptions()): McpCollectionCreator => () => { + const [name, _callback, config] = creator(options); + + const handler = async (args?: unknown): Promise => { + const currentOptions = getOptions(); + const currentSession = getSessionOptions(); + + return getHeavyPool().runTask({ + moduleSpecifier, + exportName, + args, + options: currentOptions, + session: currentSession + }); + }; + + return config ? [name, handler, config] : [name, handler]; +}; + +/** + * Proxy a collection creator with a deferred task wrapper. + * + * @param {McpCollectionCreator} creator - Original creator. + * @param {CollectionRunSchedule} runSchedule - Schedule config sourced from the collection's + * `_config.runSchedule`. Provides `cancelMs` and `intervalMs` used to build {@link deferTask}. + * @param {GlobalOptions} options - Global options. + * @returns {McpCollectionCreator} The proxied creator function. + */ +const makeScheduledProxyCreator = ({ + creator, + runSchedule +}: { creator: McpCollectionCreator, runSchedule: NonNullable['runSchedule'] }, +options: GlobalOptions = getOptions()): McpCollectionCreator => () => { + const [name, callback, config] = creator(options); + const deferOptions = { + ...(typeof runSchedule?.cancelMs === 'number' ? { cancelMs: runSchedule.cancelMs } : {}), + ...(typeof runSchedule?.intervalMs === 'number' ? { intervalMs: runSchedule.intervalMs } : {}) + }; + + const handler = async (args?: unknown): Promise => { + const task = deferTask(callback, deferOptions)(args); + let response; + + try { + response = await task.start(); + } catch (error) { + log.debug(`Scheduled collection ${name} failed to start: ${formatUnknownError(error)}`); + } + + return response || { records: [] }; + }; + + return config ? [name, handler, config] : [name, handler]; +}; /** * Composes multi-source record collections across process boundaries. * - * @param builtinCreators - * @param {GlobalOptions} _options - Global options. + * @param {McpCollectionCreator[]} builtinCreators - Built-in collection creators. + * @param {GlobalOptions} options - Global options. * @param {AppSession} _session - Session options. - * @returns Promise array of collection creators. + * @returns {Promise} Promise array of collection creators. */ const composeCollections = async ( builtinCreators: McpCollectionCreator[], - _options: GlobalOptions = getOptions(), + options: GlobalOptions = getOptions(), _session: AppSession = getSessionOptions() ): Promise => { + const localCreators: McpCollectionCreator[] = []; + // Wrap built-in creators to enforce trusted _isInternal. Ties into what options, session values are available. const securedBuiltinCreators = builtinCreators.map((creator): McpCollectionCreator => opt => { const [name, callback, config] = creator(opt); @@ -33,9 +107,30 @@ const composeCollections = async ( return []; } - return securedBuiltinCreators; + for (const creator of securedBuiltinCreators) { + const [, , config] = creator(options); + const runHostValue = config?.runParallel; + const runScheduleConfig = config?.runSchedule; + let updatedCreator = creator; + + if (typeof runHostValue === 'string' && runHostValue.startsWith('#')) { + // Use 'collectionCallback' for collection modules that expose a common-named export. + updatedCreator = makeParallelProxyCreator({ creator, moduleSpecifier: runHostValue, exportName: 'collectionCallback' }); + } + + if (typeof runScheduleConfig?.cancelMs === 'number' || typeof runScheduleConfig?.intervalMs === 'number') { + // Layer scheduling so the defer-task guardrails apply to the entire execution, including any worker-pool proxy. + updatedCreator = makeScheduledProxyCreator({ creator: updatedCreator, runSchedule: runScheduleConfig }); + } + + localCreators.push(updatedCreator); + } + + return localCreators; }; export { - composeCollections + composeCollections, + makeParallelProxyCreator, + makeScheduledProxyCreator }; diff --git a/src/server.ts b/src/server.ts index c611821c..7edbb26b 100644 --- a/src/server.ts +++ b/src/server.ts @@ -13,6 +13,7 @@ import { memo } from './server.caching'; import { formatUnknownError, log, type LogEvent } from './logger'; import { createServerLogger } from './server.logger'; import { composeTools, sendToolsHostShutdown } from './server.tools'; +import { sendWorkerPoolsShutdown } from './server.workerPool'; import { composeResources } from './server.resources'; import { type GlobalOptions } from './options'; import { @@ -129,6 +130,8 @@ const registerServerCollections = async (collections: McpCollectionCreator[], op const updatedCollections = collections.map(collectionCreator => { const [name, callback, _config] = collectionCreator(options); + log.info(`Registered collection: ${name}`); + return [ name, async () => runWithSession(session, async () => @@ -149,9 +152,14 @@ const registerServerCollections = async (collections: McpCollectionCreator[], op }); // Update PatternFly collections, see {@link setPatternFlyCollection} - const onUpdate = ({ name, response }: RegisterCollectionItem) => { + const onUpdate = ({ name, response, error }: RegisterCollectionItem) => { if (response) { setPatternFlyCollection(name, response); + log.info(`Update collection: ${name}`); + } + + if (error) { + log.error(`Update collection error "${name}": ${error}`); } }; @@ -360,6 +368,7 @@ const runServer = async (options: ServerOptions = getOptions(), { running = false; await sendToolsHostShutdown(); + await sendWorkerPoolsShutdown(); log.info(`${options.name} closed!\n`); unsubscribeServerLogger?.(); @@ -545,6 +554,7 @@ runServer.memo = memo( export { runServer, + registerServerCollections, registerServerResources, registerServerTools, type ServerInstance, diff --git a/src/server.workerEntry.ts b/src/server.workerEntry.ts new file mode 100644 index 00000000..55228c96 --- /dev/null +++ b/src/server.workerEntry.ts @@ -0,0 +1,3 @@ +import { runWorker } from './server.workerRunner'; + +runWorker(); diff --git a/src/server.workerPool.ts b/src/server.workerPool.ts new file mode 100644 index 00000000..c2d2ff9a --- /dev/null +++ b/src/server.workerPool.ts @@ -0,0 +1,471 @@ +import { Worker } from 'node:worker_threads'; +import { fileURLToPath } from 'node:url'; +import { availableParallelism } from 'node:os'; +import { formatUnknownError } from './logger'; + +/** + * Payload for a task execution, including module details, arguments, and configuration options. + * + * @interface TaskPayload + * + * @property moduleSpecifier Identifier for the module to be imported or executed as part of the + * task. + * @property [exportName] Optional name of the exported function or variable from the specified + * module to be invoked or used. + * @property [args] Optional arguments to be passed to the task or function being executed. + * @property [options] Optional additional options or settings for executing the task. + * @property [session] Optional session-specific data or context associated with the task + * execution. + */ +interface TaskPayload { + moduleSpecifier: string; + exportName?: string; + args?: unknown; + options?: unknown; + session?: unknown; +} + +/** + * IPC payload posted between worker threads and the pool parent. + */ +interface WorkerIpcMessage { + success: boolean; + payload?: unknown; + error?: unknown; +} + +/** + * Throttled worker thread pool for parallel execution. + * + * @interface QueuedTask + * + * @property payload Task payload. + * @property resolve Promise resolve function. + * @property reject Promise reject function. + */ +interface QueuedTask { + payload: TaskPayload; + resolve: (value: unknown) => void; + reject: (reason: unknown) => void; +} + +/** + * Throttled worker thread pool for parallel execution. + * + * @interface WorkerPoolInstance + * + * @property runTask Run task method. + * @property shutdown Best-effort shutdown for queued and active workers. + */ +interface WorkerPoolInstance { + runTask(payload: TaskPayload): Promise; + shutdown(): Promise; +} + +/** + * Registry keys for module-scoped worker pools. + */ +type PoolKind = 'heavy' | 'light'; + +/** + * Registry for worker pools. + */ +const poolRegistry = new Map(); + +/** + * Maximum number of queued tasks to prevent unbounded growth. + * + * @note In the future we can look at adding this to default options. + */ +const MAX_QUEUE_CAP = 50; + +/** + * Resolves the location of the worker entry script safely across bundling and testing frameworks. + */ +const getWorkerScriptPath = (): string => { + try { + return fileURLToPath(import.meta.resolve('#workerEntry')); + } catch { + return new URL('../dist/server.workerEntry.js', import.meta.url).pathname; + } +}; + +/** + * Pool-scoped abort for rejecting queued and in-flight tasks on shutdown. + */ +const createPoolAbort = () => { + const controller = new AbortController(); + const shutdownError = new Error('Worker pool shutdown'); + + const isAborted = (): boolean => controller.signal.aborted; + + const abort = (): void => { + if (!controller.signal.aborted) { + controller.abort(shutdownError); + } + }; + + const runTask = (payload: TaskPayload, enqueue: (task: QueuedTask) => void): Promise => { + if (isAborted()) { + return Promise.reject(shutdownError); + } + + return new Promise((resolve, reject) => { + const onAbort = (): void => { + reject(controller.signal.reason ?? shutdownError); + }; + + controller.signal.addEventListener('abort', onAbort, { once: true }); + + enqueue({ + payload, + resolve: (value: unknown) => { + controller.signal.removeEventListener('abort', onAbort); + resolve(value as T); + }, + reject: (reason: unknown) => { + controller.signal.removeEventListener('abort', onAbort); + reject(reason); + } + }); + }); + }; + + return { isAborted, abort, runTask }; +}; + +/** + * Create a transient worker pool with the specified maximum number of workers. + * Spawns a fresh thread per task, kills it instantly on completion + * + * @note Recommended use is for heavy memory usage, unpredictable processing, and + * long-running scraping cycles. + * + * @private + * + * @param [maxWorkers] -Max number of workers that can run concurrently. Defaults to one + * less than the available parallelism of the system, with a minimum value of 1. + * @returns {WorkerPoolInstance} - An instance of a worker pool, allowing tasks + * to be queued and executed using dedicated transient workers. + */ +const buildTransientPool = (maxWorkers = Math.max(1, availableParallelism() - 1)): WorkerPoolInstance => { + let activeWorkers = 0; + const queue: QueuedTask[] = []; + const transientWorkers = new Set(); + const workerScript = getWorkerScriptPath(); + const poolAbort = createPoolAbort(); + + /** + * Process the next task in the queue if there are available workers and tasks. + */ + const next = (): void => { + if (poolAbort.isAborted() || activeWorkers >= maxWorkers || queue.length === 0) { + return; + } + + const task = queue.shift(); + + if (!task) { + return; + } + + activeWorkers += 1; + spawnTransientWorker(task); + }; + + /** + * Spawn worker, execute a task. + * + * @param task - The task to be executed. + */ + const spawnTransientWorker = (task: QueuedTask): void => { + const { payload, resolve, reject } = task; + let resolved = false; + + try { + // Pass workerData right away to trigger transient execution flow + const worker = new Worker(workerScript, { workerData: payload }); + + transientWorkers.add(worker); + + worker.on('message', (message: WorkerIpcMessage) => { + resolved = true; + if (message && message.success) { + resolve(message.payload); + } else { + reject(new Error(formatUnknownError(message?.error ?? 'Unknown worker error'))); + } + }); + + worker.on('error', err => { + if (!resolved) { + resolved = true; + reject(err); + } + }); + + worker.on('exit', code => { + transientWorkers.delete(worker); + activeWorkers -= 1; + + if (!resolved) { + resolved = true; + reject(new Error(`Transient worker exited unexpectedly with code ${code}`)); + } + + next(); + }); + } catch (error) { + activeWorkers -= 1; + + if (!resolved) { + resolved = true; + reject(error); + } + + next(); + } + }; + + return { + runTask: (payload: TaskPayload): Promise => + poolAbort.runTask(payload, task => { + // Backpressure: reject immediately if queue is full + if (queue.length >= MAX_QUEUE_CAP) { + const err: Error & { code: string } = Object.assign( + new Error(`Worker queue full (${MAX_QUEUE_CAP}).`), + { code: 'ERR_WORKER_QUEUE_FULL' } + ); + + task.reject(err); + + return; + } + queue.push(task); + next(); + }), + + shutdown: async (): Promise => { + poolAbort.abort(); + queue.length = 0; + + await Promise.allSettled([...transientWorkers].map(worker => worker.terminate())); + transientWorkers.clear(); + activeWorkers = 0; + } + }; +}; + +/** + * Create a persistent worker pool with pre-spawned worker threads for handling concurrent tasks. + * Keeps warm threads active, route payloads via IPC messages. + * + * @note Recommended for rapid, light, or frequent computations requiring low-latency invocation. + * + * @private + * + * @param [maxWorkers] -Max number of workers that can run concurrently. Defaults to one + * less than the available parallelism of the system, with a minimum value of 1. + * @returns {WorkerPoolInstance} Object exposing methods to interact with the worker pool, + * including submitting tasks for execution. + */ +const buildPersistentPool = (maxWorkers = Math.max(1, availableParallelism() - 1)): WorkerPoolInstance => { + const queue: QueuedTask[] = []; + const workers: { worker: Worker; active: boolean; reject?: (reason: unknown) => void }[] = []; + const workerScript = getWorkerScriptPath(); + const poolAbort = createPoolAbort(); + + const handleWorkerCrash = (index: number, exitCode?: number) => { + const slot = workers[index]; + + if (!slot) { + return; + } + + slot.worker?.removeAllListeners(); + + if (slot.reject) { + const message = exitCode !== undefined + ? `Persistent worker exited unexpectedly with code ${exitCode}` + : 'Persistent worker thread crashed'; + + slot.reject(new Error(message)); + } else { + slot.active = false; + } + + slot.worker = new Worker(workerScript); + slot.worker.on('error', () => handleWorkerCrash(index)); + slot.worker.on('exit', (code: number) => handleWorkerCrash(index, code)); + next(); + }; + + const next = (): void => { + if (poolAbort.isAborted() || queue.length === 0) { + return; + } + + const idleWorkerSlot = workers.find(worker => !worker.active); + + if (!idleWorkerSlot) { + return; + } + + const task = queue.shift(); + + if (!task) { + return; + } + + idleWorkerSlot.active = true; + const { worker } = idleWorkerSlot; + const { payload, resolve, reject } = task; + + idleWorkerSlot.reject = reject; + + const onMessage = (message: WorkerIpcMessage) => { + cleanup(); + if (message && message.success) { + resolve(message.payload); + } else { + reject(new Error(formatUnknownError(message?.error ?? 'Unknown worker error'))); + } + }; + + const onError = (err: Error) => { + cleanup(); + reject(err); + }; + + const cleanup = () => { + worker.off('message', onMessage); + worker.off('error', onError); + idleWorkerSlot.active = false; + delete idleWorkerSlot.reject; + + next(); + }; + + worker.on('message', onMessage); + worker.on('error', onError); + + // Send the task data via postMessage channel to be captured by persistent listeners + worker.postMessage(payload); + }; + + // Pre-spawn and warm up the permanent thread containers + for (let i = 0; i < maxWorkers; i++) { + const worker = new Worker(workerScript); // Instantiated WITHOUT initial workerData + + workers.push({ worker, active: false }); + worker.on('error', () => handleWorkerCrash(i)); + worker.on('exit', (code: number) => handleWorkerCrash(i, code)); + } + + return { + runTask: (payload: TaskPayload): Promise => + poolAbort.runTask(payload, task => { + // Backpressure: reject immediately if queue is full + if (queue.length >= MAX_QUEUE_CAP) { + const err: Error & { code: string } = Object.assign( + new Error(`Worker queue full (${MAX_QUEUE_CAP}).`), + { code: 'ERR_WORKER_QUEUE_FULL' } + ); + + task.reject(err); + + return; + } + queue.push(task); + next(); + }), + + shutdown: async (): Promise => { + poolAbort.abort(); + queue.length = 0; + + for (const slot of workers) { + slot.worker.removeAllListeners(); + await slot.worker.terminate(); + slot.active = false; + delete slot.reject; + } + } + }; +}; + +/** + * Return the module-scoped pool for the requested kind, creating it on first use. + * + * @private + * @param kind - Registry key for the pool. + * @returns Cached worker pool instance. + */ +const getPool = (kind: PoolKind): WorkerPoolInstance => { + const cached = poolRegistry.get(kind); + + if (cached) { + return cached; + } + + const pool = kind === 'heavy' + ? buildTransientPool(2) + : buildPersistentPool(); + + poolRegistry.set(kind, pool); + + return pool; +}; + +/** + * Transient pool for heavy or costly collection work. + * + * @returns Module-scoped transient worker pool. + */ +const getHeavyPool = (): WorkerPoolInstance => getPool('heavy'); + +/** + * Persistent pool for light, frequent collection work. + * + * @returns Module-scoped persistent worker pool. + */ +const getLightPool = (): WorkerPoolInstance => getPool('light'); + +/** + * Best-effort shutdown for all module-scoped worker pools. + * + * Policy mirrors {@link sendToolsHostShutdown}: + * - Abort queued and in-flight tasks + * - Terminate worker threads + * - Clear the pool registry so a later start gets fresh pools + */ +const sendWorkerPoolsShutdown = async (): Promise => { + await Promise.allSettled( + [...poolRegistry.values()].map(pool => pool.shutdown()) + ); + + poolRegistry.clear(); +}; + +/** + * Clear the module-scoped pool registry. + * + * @private + */ +const resetWorkerPools = (): void => { + poolRegistry.clear(); +}; + +export { + createPoolAbort, + getWorkerScriptPath, + buildTransientPool, + buildPersistentPool, + getHeavyPool, + getLightPool, + resetWorkerPools, + sendWorkerPoolsShutdown, + type TaskPayload, + type QueuedTask, + type WorkerPoolInstance +}; diff --git a/src/server.workerRunner.ts b/src/server.workerRunner.ts new file mode 100644 index 00000000..e3ed9a6e --- /dev/null +++ b/src/server.workerRunner.ts @@ -0,0 +1,177 @@ +import { pathToFileURL } from 'node:url'; +import { parentPort, workerData } from 'node:worker_threads'; +import { runWithOptions, runWithSession } from './options.context'; + +/** + * Data required to define and execute a worker task. + * + * @interface WorkerTaskData + * + * @property moduleSpecifier String that specifies the module to be imported or loaded. + * @property [exportName] Optional string that specifies the name of the export within the module to invoke. + * @property [args] Optional arguments to be passed to the task being executed. + * @property [options] Optional configuration or metadata related to the task execution. + * @property [session] Optional session information or context related to the task environment. + */ +interface WorkerTaskData { + moduleSpecifier: string; + exportName?: string; + args?: unknown; + options?: unknown; + session?: unknown; +} + +/** + * Execute a task defined by the provided payload. + * + * - Dynamically import a module + * - Identify the specified export (default or named) + * - Invokes the callback with the given arguments. + * + * @param {WorkerTaskData} taskPayload - The task payload containing details about the module to load, + * the export to invoke, and arguments to pass to the export. + * @param taskPayload.moduleSpecifier - The path or URL identifying the module to be imported. + * It must be a valid module specifier. + * @param [taskPayload.exportName='default'] - The name of the export to invoke. Defaults to + * 'default' if not specified. + * @param taskPayload.args - Arguments to pass to the exported function when called. + * @param [taskPayload.options] - Configuration options that define specific execution settings. + * @param [taskPayload.session] - Data describing the session context for task isolation. + * @throws {Error} If the `moduleSpecifier` is not provided, or if the specified export is not a function. + * @returns A promise that resolves to the result of the invoked export function. + */ +const executeTask = async ( + { moduleSpecifier, exportName = 'default', args, options, session }: WorkerTaskData +): Promise => { + if (!moduleSpecifier) { + throw new Error('No moduleSpecifier specified for worker task.'); + } + + let resolvedSpec = moduleSpecifier; + + if (resolvedSpec.startsWith('#')) { + resolvedSpec = import.meta.resolve(resolvedSpec); + } else if (!resolvedSpec.startsWith('file://') && !resolvedSpec.startsWith('data:')) { + resolvedSpec = pathToFileURL(resolvedSpec).href; + } + + // Bypass static bundler boundaries cleanly via scoped Function constructor + const dynamicImport = new Function('spec', 'return import(spec)') as (spec: string) => Promise>; + const module = await dynamicImport(resolvedSpec); + + // Map to default fallback hooks if explicit targets are missing + const callback = module[exportName] || (exportName === 'default' ? module.default : undefined); + + if (typeof callback !== 'function') { + throw new Error(`Exported module '${moduleSpecifier}' (export: '${exportName}') must be a function.`); + } + + // Nest execution within both AsyncLocalStorage isolation layouts + return runWithOptions((options as any) || {}, async () => + runWithSession((session as any) || {}, async () => + Promise.resolve(callback(args)))); +}; + +/** + * Make sure a worker thread stays alive by ref'ing the `parentPort` or setting a long-running + * timeout as a fallback mechanism. + * + * @param [options] - Config options. + * @param [options.throwOnParentPortError] - Whether to throw an error if `parentPort.ref` is + * not a function. If `false`, no error is thrown, and the fallback mechanism is used. + * @param [options.timeoutMs] - Milliseconds for the fallback timeout. Defaults to `24` hours. + * @returns Cleanup function that cancels the keep-alive mechanism, either by calling `parentPort.unref` + * or clearing the fallback timer. + * @throws {Error} If `parentPort.ref` is not a function and `throwOnParentPortError` is `true`. + */ +const keepWorkerAlive = ({ throwOnParentPortError = true, timeoutMs = 86_400_000 } = {}): () => void => { + if (parentPort && typeof parentPort.ref === 'function') { + parentPort.ref(); + + return () => { + parentPort?.unref?.(); + }; + } + + if (throwOnParentPortError) { + throw new Error('parentPort.ref is not a function — worker keep-alive failed'); + } + + const safeTimeout = Math.min(timeoutMs, 2_147_483_646); + const timer = setTimeout(() => {}, safeTimeout); + + // By design, do not return the clearTimeout, ensure the timer gets wiped from memory + return () => { + clearTimeout(timer); + }; +}; + +/** + * Route orchestration based on worker thread startup context. + * + * Two distinct routes: + * 1. **Route A: Transient Execution** + * - If `workerData` is provided, the worker immediately processes the task using the + * provided data. + * - Upon task completion, a success or failure message is posted back to the parent thread + * via the `parentPort`. + * + * 2. **Route B: Persistent Execution** + * - If `workerData` is not available, the worker remains active and listens for incoming + * task payloads via `parentPort`. + * - When a task message is received, it processes the task and sends a success or failure + * message back to the parent thread. + * + * Both routes use async and handle errors gracefully to relate results or errors back to the parent. + * + * @param options - Function options. + * @param options.throwOnParentPortError - If true, errors thrown by the parent port will be re-thrown. + */ +const runWorker = ({ throwOnParentPortError }: { throwOnParentPortError?: boolean | undefined } = {}): Promise | void => { + if (workerData) { + /** + * Route A: Transient execution (workerData is loaded immediately) + */ + const clearKeepAlive = keepWorkerAlive({ throwOnParentPortError }); + + return executeTask(workerData as WorkerTaskData) + .then(result => { + parentPort?.postMessage({ success: true, payload: result }); + }) + .catch((error: unknown) => { + parentPort?.postMessage({ + success: false, + error: { + message: error instanceof Error ? error.message : String(error), + stack: error instanceof Error ? error.stack : undefined + } + }); + }) + .finally(() => { + clearKeepAlive(); + }); + } else { + /** + * Route B: Persistent execution (Thread stays open waiting for stream events) + */ + parentPort?.on('message', async (incomingPayload: WorkerTaskData) => { + try { + const result = await executeTask(incomingPayload); + + parentPort?.postMessage({ success: true, payload: result }); + } catch (error: unknown) { + parentPort?.postMessage({ + success: false, + error: { + message: error instanceof Error ? error.message : String(error), + stack: error instanceof Error ? error.stack : undefined + } + }); + } + }); + + return undefined; + } +}; + +export { executeTask, keepWorkerAlive, runWorker, type WorkerTaskData }; diff --git a/src/tools.ts b/src/tools.ts new file mode 100644 index 00000000..534911ea --- /dev/null +++ b/src/tools.ts @@ -0,0 +1,9 @@ +export { + createMcpTool, + type ToolCreator, + type ToolModule, + type ToolConfig, + type ToolMultiConfig, + type ToolExternalOptions, + type ToolInternalOptions +} from './server.toolsUser'; diff --git a/tests/e2e/__fixtures__/tool.echoToolHelper.js b/tests/e2e/__fixtures__/tool.echoToolHelper.js index e06d7b8b..e81cb1d8 100644 --- a/tests/e2e/__fixtures__/tool.echoToolHelper.js +++ b/tests/e2e/__fixtures__/tool.echoToolHelper.js @@ -1,6 +1,6 @@ // Fixture exports a createMcpTool module directly; // eslint-disable-next-line import/no-unresolved -import { createMcpTool } from '@patternfly/patternfly-mcp'; +import { createMcpTool } from '@patternfly/patternfly-mcp/tools'; export default createMcpTool({ name: 'echo_createMcp_tool', diff --git a/tests/e2e/__snapshots__/stdioTransport.test.ts.snap b/tests/e2e/__snapshots__/stdioTransport.test.ts.snap index 50256a48..1e1bc66d 100644 --- a/tests/e2e/__snapshots__/stdioTransport.test.ts.snap +++ b/tests/e2e/__snapshots__/stdioTransport.test.ts.snap @@ -154,9 +154,13 @@ exports[`Logging should allow setting logging options, stderr 1`] = ` ", "[INFO]: No external tools loaded. ", - "[INFO]: Register collection: patternfly-docs + "[INFO]: Registered collection: patternfly-docs ", - "[INFO]: Register collection: patternfly-component-schemas + "[INFO]: Registered collection: patternfly-component-schemas +", + "[INFO]: Update collection: patternfly-component-schemas +", + "[INFO]: Update collection: patternfly-docs ", "[INFO]: Registered resource: patternfly-context ", diff --git a/tests/e2e/httpTransport.test.ts b/tests/e2e/httpTransport.test.ts index ed286918..2e47b3f8 100644 --- a/tests/e2e/httpTransport.test.ts +++ b/tests/e2e/httpTransport.test.ts @@ -2,8 +2,8 @@ * Requires: npm run build prior to running Jest. * - If typings are needed, use public types from dist to avoid type identity mismatches between src and dist */ -// @ts-ignore - dist/index.js isn't necessarily built yet, remember to build before running tests -import { createMcpTool } from '../../dist/index.js'; +// @ts-ignore - dist/ isn't necessarily built yet; run npm run build before e2e tests +import { createMcpTool } from '@patternfly/patternfly-mcp/tools'; import { startServer, type HttpTransportClient, type RpcRequest } from './utils/httpTransportClient'; import { setupFetchMock } from './utils/fetchMock'; diff --git a/tests/package/exports.test.ts b/tests/package/exports.test.ts new file mode 100644 index 00000000..a003e60a --- /dev/null +++ b/tests/package/exports.test.ts @@ -0,0 +1,34 @@ +import { existsSync, readFileSync, readdirSync } from 'node:fs'; +import { join } from 'node:path'; + +describe('package entry bundles', () => { + const distDir = join(process.cwd(), 'dist'); + + beforeAll(() => { + if (!existsSync(distDir)) { + throw new Error('dist/ not found. Run npm run build before build tests.'); + } + }); + + it('should keep worker threads out of index entry chunks', () => { + const indexChunks = readdirSync(distDir).filter(file => /^index.*\.js$/.test(file)); + + expect(indexChunks.length).toBeGreaterThan(0); + + indexChunks.forEach(file => { + const content = readFileSync(join(distDir, file), 'utf8'); + + expect(content).not.toContain('worker_threads'); + }); + }); + + it('should emit a tools entry without worker threads', () => { + const toolsEntry = join(distDir, 'tools.js'); + + expect(existsSync(toolsEntry)).toBe(true); + + const content = readFileSync(toolsEntry, 'utf8'); + + expect(content).not.toContain('worker_threads'); + }); +}); diff --git a/tsconfig.json b/tsconfig.json index 3481e0d6..ab16b5cc 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -24,7 +24,8 @@ "rootDirs": ["./src", "./tests/e2e", "./tests/audit"], "paths": { "#docsCatalog": ["./src/docs.json"], - "#toolsHost": ["./src/server.toolsHost.ts"] + "#toolsHost": ["./src/server.toolsHost.ts"], + "#workerEntry": ["./src/server.workerEntry.ts"] } }, "include": [