Recover managed bundled plugin workers on demand (#10429)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Managed deployments can auto-provision bundled sandbox-provider plugins so cloud or remote execution environments appear in the board UI > - In a multi-service deployment, several server processes can share one database and boot concurrently > - A sibling process can create a bundled plugin row while the web process sees it before it reaches `ready` > - The web process correctly avoids clobbering the existing row, but its startup `loadAll()` can miss the plugin and never start that worker locally > - The environments capabilities route then filters out the sandbox provider because the plugin is ready in the database but not running in the web process > - This pull request adds a narrow managed-bundle recovery path that lazily starts the missing worker when the capabilities route sees a ready managed bundled plugin > - The benefit is that the sandbox provider becomes visible after the install finishes, without requiring a web-process restart ## Linked Issues or Issue Description - No public GitHub issue found for this exact deployment race. - Related broad plugin runtime context: Refs #432. Bug description: - What happened: in a managed multi-service deployment with shared database state and bundled plugin auto-install enabled, the API-serving process can skip a plugin row while it is still `installed`, run startup plugin loading before that row becomes `ready`, and then permanently omit the sandbox provider from environment capabilities. - Expected behavior: once the managed bundled plugin row reaches `ready`, the API-serving process should be able to start the plugin worker and include its sandbox provider without a restart. - Steps to reproduce: boot a web process and a sibling worker process concurrently; have the sibling create the bundled plugin row and transition it to `ready` after the web process has already skipped auto-install and run `loadAll()`. - Deployment mode: managed multi-service deployment with shared database state and `plugins.autoInstall` configured. ## What Changed - Added a managed bundled plugin worker recovery helper that single-flights lazy `loadSingle()` starts and only allows configured managed bundled plugin keys. - Passed the managed recovery hook into the environments capabilities route. - Updated `listReadyPluginEnvironmentDrivers()` to attempt bounded recovery for ready managed bundled plugins whose worker is missing in the current process, and only for plugins that actually declare a `sandbox_provider` environment driver. - Made request-time recovery use `loadSingle(id, { markErrorOnFailure: false })` so a local activation failure in one process never transitions the shared plugin row to `error` (a sibling process may be running the plugin successfully). - When error writes are suppressed and activation fails after the worker was spawned, the loader now tears down the partially-registered local runtime (scheduler registration, event subscriptions, agent tools, worker process) instead of leaving a half-activated worker lingering; the teardown steps are factored out of `unloadSingle()` into a shared helper. - A failed recovery attempt now discards the crashed/stopped handle it left registered in the worker manager (a worker that dies during initialize is killed without a scheduled restart), so later capability requests can retry recovery instead of being blocked by the handle-presence gate until a process restart. Handles in starting/running/backoff states are left to the worker manager's own lifecycle; recovery only ever starts when no handle existed, so no pre-existing worker can be affected. - Added a regression test suite covering the installed-to-ready race, allowlist behavior, the driver-kind gate, existing worker handles, concurrent single-flight recovery, bounded slow recovery attempts, suppressed shared error-state writes, partial-runtime teardown on late activation failure, and retry after a dead handle is discarded. ## Verification - `pnpm vitest run src/__tests__/plugin-environment-driver-ready-recovery.test.ts` (in `server/`) passed: 10 tests. - `pnpm --filter @paperclipai/server typecheck` passed. ## Risks - Low risk for self-hosted single-process deployments because lazy recovery is only wired when managed plugin auto-install config is present; with no managed config the capabilities route takes the exact pre-change code path. - The capabilities route can wait briefly while attempting recovery; the attempt is bounded and defaults to 2 seconds. - Failed recovery keeps the prior behavior of omitting the provider until a later successful worker start, and now also cleans up any partially-started local worker so retries begin from a clean slate. ## Model Used - Initial implementation: OpenAI GPT-5 via Codex local coding agent, with repository tool use and command execution. - Review-feedback follow-ups (driver-kind gate, partial-runtime teardown, expanded regression tests): Claude Fable 5 (claude-fable-5) via Claude Code, with repository tool use and command execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin Foley
GitHub
parent
d6e235cbcf
commit
78f8c6c3d4
|
|
@ -0,0 +1,515 @@
|
|||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { PaperclipPluginManifestV1 } from "@paperclipai/shared";
|
||||
import { createManagedBundledPluginWorkerRecovery } from "../app.js";
|
||||
import { listReadyPluginEnvironmentDrivers } from "../services/plugin-environment-driver.js";
|
||||
import { pluginLoader, type PluginLoader } from "../services/plugin-loader.js";
|
||||
import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
|
||||
|
||||
const mockRegistry = vi.hoisted(() => ({
|
||||
getById: vi.fn(),
|
||||
list: vi.fn(),
|
||||
listConfigs: vi.fn(),
|
||||
update: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("../services/plugin-registry.js", () => ({
|
||||
pluginRegistryService: () => mockRegistry,
|
||||
}));
|
||||
|
||||
const PLUGIN_ID = "plugin-daytona";
|
||||
const PLUGIN_KEY = "paperclip.daytona-sandbox-provider";
|
||||
|
||||
const manifest: PaperclipPluginManifestV1 = {
|
||||
id: PLUGIN_KEY,
|
||||
apiVersion: 1,
|
||||
version: "1.0.0",
|
||||
displayName: "Daytona Sandbox Provider",
|
||||
description: "Provides Daytona-backed sandboxes.",
|
||||
author: "Paperclip",
|
||||
categories: ["automation"],
|
||||
capabilities: ["environment.drivers.register"],
|
||||
entrypoints: { worker: "dist/worker.js" },
|
||||
environmentDrivers: [
|
||||
{
|
||||
driverKey: "daytona",
|
||||
kind: "sandbox_provider",
|
||||
displayName: "Daytona",
|
||||
description: "Daytona sandbox provider",
|
||||
configSchema: { type: "object", properties: {} },
|
||||
},
|
||||
],
|
||||
};
|
||||
|
||||
function createPlugin(status: string, options: { id?: string; pluginKey?: string } = {}) {
|
||||
return {
|
||||
id: options.id ?? PLUGIN_ID,
|
||||
pluginKey: options.pluginKey ?? PLUGIN_KEY,
|
||||
status,
|
||||
manifestJson: manifest,
|
||||
};
|
||||
}
|
||||
|
||||
function createWorkerManager(options: {
|
||||
running?: boolean;
|
||||
hasHandle?: boolean;
|
||||
} = {}) {
|
||||
let running = options.running ?? false;
|
||||
const workerManager = {
|
||||
isRunning: vi.fn(() => running),
|
||||
getWorker: vi.fn(() => (options.hasHandle ? { status: "backoff" } : undefined)),
|
||||
} as unknown as PluginWorkerManager;
|
||||
return {
|
||||
workerManager,
|
||||
markRunning: () => {
|
||||
running = true;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function createDeferred<T = void>() {
|
||||
let resolve!: (value?: T | PromiseLike<T>) => void;
|
||||
let reject!: (reason?: unknown) => void;
|
||||
const promise = new Promise<T>((resolvePromise, rejectPromise) => {
|
||||
resolve = resolvePromise;
|
||||
reject = rejectPromise;
|
||||
});
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
||||
describe("listReadyPluginEnvironmentDrivers worker recovery", () => {
|
||||
beforeEach(() => {
|
||||
mockRegistry.getById.mockReset();
|
||||
mockRegistry.list.mockReset();
|
||||
mockRegistry.listConfigs.mockReset();
|
||||
mockRegistry.update.mockReset();
|
||||
});
|
||||
|
||||
it("recovers a managed bundled provider that was installed by a sibling process after this process skipped it", async () => {
|
||||
const plugin = createPlugin("installed");
|
||||
mockRegistry.list.mockImplementation(async () => [plugin]);
|
||||
const worker = createWorkerManager();
|
||||
const startWorker = vi.fn(async () => {
|
||||
worker.markRunning();
|
||||
return true;
|
||||
});
|
||||
|
||||
await expect(
|
||||
listReadyPluginEnvironmentDrivers({
|
||||
db: {} as never,
|
||||
workerManager: worker.workerManager,
|
||||
recoverMissingWorker: {
|
||||
pluginKeys: [PLUGIN_KEY],
|
||||
startWorker,
|
||||
},
|
||||
}),
|
||||
).resolves.toEqual([]);
|
||||
expect(startWorker).not.toHaveBeenCalled();
|
||||
|
||||
plugin.status = "ready";
|
||||
const drivers = await listReadyPluginEnvironmentDrivers({
|
||||
db: {} as never,
|
||||
workerManager: worker.workerManager,
|
||||
recoverMissingWorker: {
|
||||
pluginKeys: [PLUGIN_KEY],
|
||||
startWorker,
|
||||
},
|
||||
});
|
||||
|
||||
expect(startWorker).toHaveBeenCalledWith({
|
||||
id: PLUGIN_ID,
|
||||
pluginKey: PLUGIN_KEY,
|
||||
});
|
||||
expect(drivers).toEqual([
|
||||
expect.objectContaining({
|
||||
pluginId: PLUGIN_ID,
|
||||
pluginKey: PLUGIN_KEY,
|
||||
driverKey: "daytona",
|
||||
displayName: "Daytona",
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("does not lazy-start ready plugins outside the managed bundled allowlist", async () => {
|
||||
const plugin = createPlugin("ready");
|
||||
mockRegistry.list.mockResolvedValue([plugin]);
|
||||
const worker = createWorkerManager();
|
||||
const startWorker = vi.fn(async () => {
|
||||
worker.markRunning();
|
||||
return true;
|
||||
});
|
||||
|
||||
const drivers = await listReadyPluginEnvironmentDrivers({
|
||||
db: {} as never,
|
||||
workerManager: worker.workerManager,
|
||||
recoverMissingWorker: {
|
||||
pluginKeys: ["paperclip.kubernetes-sandbox-provider"],
|
||||
startWorker,
|
||||
},
|
||||
});
|
||||
|
||||
expect(startWorker).not.toHaveBeenCalled();
|
||||
expect(drivers).toEqual([]);
|
||||
});
|
||||
|
||||
it("does not lazy-start managed bundled plugins without sandbox provider drivers", async () => {
|
||||
const plugin = {
|
||||
...createPlugin("ready"),
|
||||
manifestJson: {
|
||||
...manifest,
|
||||
capabilities: [],
|
||||
environmentDrivers: undefined,
|
||||
},
|
||||
};
|
||||
mockRegistry.list.mockResolvedValue([plugin]);
|
||||
const worker = createWorkerManager();
|
||||
const startWorker = vi.fn(async () => {
|
||||
worker.markRunning();
|
||||
return true;
|
||||
});
|
||||
|
||||
const drivers = await listReadyPluginEnvironmentDrivers({
|
||||
db: {} as never,
|
||||
workerManager: worker.workerManager,
|
||||
recoverMissingWorker: {
|
||||
pluginKeys: [PLUGIN_KEY],
|
||||
startWorker,
|
||||
},
|
||||
});
|
||||
|
||||
expect(startWorker).not.toHaveBeenCalled();
|
||||
expect(drivers).toEqual([]);
|
||||
});
|
||||
|
||||
it("leaves existing worker-manager recovery handles alone", async () => {
|
||||
const plugin = createPlugin("ready");
|
||||
mockRegistry.list.mockResolvedValue([plugin]);
|
||||
const worker = createWorkerManager({ hasHandle: true });
|
||||
const startWorker = vi.fn(async () => true);
|
||||
|
||||
const drivers = await listReadyPluginEnvironmentDrivers({
|
||||
db: {} as never,
|
||||
workerManager: worker.workerManager,
|
||||
recoverMissingWorker: {
|
||||
pluginKeys: [PLUGIN_KEY],
|
||||
startWorker,
|
||||
},
|
||||
});
|
||||
|
||||
expect(startWorker).not.toHaveBeenCalled();
|
||||
expect(drivers).toEqual([]);
|
||||
});
|
||||
|
||||
it("single-flights concurrent managed bundled recovery starts", async () => {
|
||||
const plugin = createPlugin("ready");
|
||||
mockRegistry.list.mockResolvedValue([plugin]);
|
||||
const worker = createWorkerManager();
|
||||
const loadStarted = createDeferred();
|
||||
const releaseLoad = createDeferred();
|
||||
const loadSingle = vi.fn(async () => {
|
||||
loadStarted.resolve();
|
||||
await releaseLoad.promise;
|
||||
worker.markRunning();
|
||||
return { success: true };
|
||||
});
|
||||
const startWorker = createManagedBundledPluginWorkerRecovery({
|
||||
managedBundledPluginKeys: [PLUGIN_KEY],
|
||||
workerManager: worker.workerManager,
|
||||
getLoader: () => ({ loadSingle }) as Pick<PluginLoader, "loadSingle">,
|
||||
});
|
||||
|
||||
const first = listReadyPluginEnvironmentDrivers({
|
||||
db: {} as never,
|
||||
workerManager: worker.workerManager,
|
||||
recoverMissingWorker: {
|
||||
pluginKeys: [PLUGIN_KEY],
|
||||
startWorker,
|
||||
},
|
||||
});
|
||||
await loadStarted.promise;
|
||||
const second = listReadyPluginEnvironmentDrivers({
|
||||
db: {} as never,
|
||||
workerManager: worker.workerManager,
|
||||
recoverMissingWorker: {
|
||||
pluginKeys: [PLUGIN_KEY],
|
||||
startWorker,
|
||||
},
|
||||
});
|
||||
|
||||
releaseLoad.resolve();
|
||||
const [firstDrivers, secondDrivers] = await Promise.all([first, second]);
|
||||
|
||||
expect(loadSingle).toHaveBeenCalledTimes(1);
|
||||
expect(loadSingle).toHaveBeenCalledWith(PLUGIN_ID, {
|
||||
markErrorOnFailure: false,
|
||||
});
|
||||
expect(firstDrivers).toEqual([
|
||||
expect.objectContaining({
|
||||
pluginId: PLUGIN_ID,
|
||||
driverKey: "daytona",
|
||||
}),
|
||||
]);
|
||||
expect(secondDrivers).toEqual(firstDrivers);
|
||||
});
|
||||
|
||||
it("keeps request-time recovery failures process-local instead of marking shared plugin state errored", async () => {
|
||||
const plugin = createPlugin("ready");
|
||||
mockRegistry.list.mockResolvedValue([plugin]);
|
||||
const worker = createWorkerManager();
|
||||
const loadSingle = vi.fn(async () => ({
|
||||
plugin,
|
||||
success: false,
|
||||
registered: { worker: false, eventSubscriptions: 0, jobs: 0, webhooks: 0, tools: 0 },
|
||||
error: "local worker spawn failed",
|
||||
}));
|
||||
const startWorker = createManagedBundledPluginWorkerRecovery({
|
||||
managedBundledPluginKeys: [PLUGIN_KEY],
|
||||
workerManager: worker.workerManager,
|
||||
getLoader: () => ({ loadSingle }) as Pick<PluginLoader, "loadSingle">,
|
||||
});
|
||||
|
||||
const drivers = await listReadyPluginEnvironmentDrivers({
|
||||
db: {} as never,
|
||||
workerManager: worker.workerManager,
|
||||
recoverMissingWorker: {
|
||||
pluginKeys: [PLUGIN_KEY],
|
||||
startWorker,
|
||||
},
|
||||
});
|
||||
|
||||
expect(loadSingle).toHaveBeenCalledWith(PLUGIN_ID, {
|
||||
markErrorOnFailure: false,
|
||||
});
|
||||
expect(drivers).toEqual([]);
|
||||
});
|
||||
|
||||
it("discards the dead handle a failed recovery attempt leaves behind so later requests retry", async () => {
|
||||
const plugin = createPlugin("ready");
|
||||
mockRegistry.list.mockResolvedValue([plugin]);
|
||||
|
||||
// Simulates the initialize-failure path: startWorker registers a handle,
|
||||
// the worker dies during initialize, and no restart is scheduled — the
|
||||
// crashed handle stays registered in the worker manager.
|
||||
let handle: { status: string } | undefined;
|
||||
let running = false;
|
||||
const stopWorker = vi.fn(async () => {
|
||||
handle = undefined;
|
||||
});
|
||||
const workerManager = {
|
||||
isRunning: vi.fn(() => running),
|
||||
getWorker: vi.fn(() => handle),
|
||||
stopWorker,
|
||||
} as unknown as PluginWorkerManager;
|
||||
|
||||
const loadSingle = vi.fn<PluginLoader["loadSingle"]>()
|
||||
.mockImplementationOnce(async () => {
|
||||
handle = { status: "crashed" };
|
||||
throw new Error("Worker initialize failed");
|
||||
})
|
||||
.mockImplementationOnce(async () => {
|
||||
handle = { status: "running" };
|
||||
running = true;
|
||||
return { plugin, success: true } as never;
|
||||
});
|
||||
|
||||
const startWorker = createManagedBundledPluginWorkerRecovery({
|
||||
managedBundledPluginKeys: [PLUGIN_KEY],
|
||||
workerManager,
|
||||
getLoader: () => ({ loadSingle }) as Pick<PluginLoader, "loadSingle">,
|
||||
});
|
||||
|
||||
const firstDrivers = await listReadyPluginEnvironmentDrivers({
|
||||
db: {} as never,
|
||||
workerManager,
|
||||
recoverMissingWorker: {
|
||||
pluginKeys: [PLUGIN_KEY],
|
||||
startWorker,
|
||||
},
|
||||
});
|
||||
|
||||
expect(firstDrivers).toEqual([]);
|
||||
expect(stopWorker).toHaveBeenCalledWith(PLUGIN_ID);
|
||||
expect(handle).toBeUndefined();
|
||||
|
||||
const secondDrivers = await listReadyPluginEnvironmentDrivers({
|
||||
db: {} as never,
|
||||
workerManager,
|
||||
recoverMissingWorker: {
|
||||
pluginKeys: [PLUGIN_KEY],
|
||||
startWorker,
|
||||
},
|
||||
});
|
||||
|
||||
expect(loadSingle).toHaveBeenCalledTimes(2);
|
||||
expect(secondDrivers).toEqual([
|
||||
expect.objectContaining({
|
||||
pluginId: PLUGIN_ID,
|
||||
driverKey: "daytona",
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("lets callers suppress shared error-state writes on activation failure", async () => {
|
||||
const plugin = {
|
||||
...createPlugin("ready"),
|
||||
packageName: "@paperclipai/missing-sandbox-provider",
|
||||
packagePath: null,
|
||||
version: "1.0.0",
|
||||
};
|
||||
mockRegistry.getById.mockResolvedValue(plugin);
|
||||
const lifecycleManager = {
|
||||
markError: vi.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
const loader = pluginLoader(
|
||||
{} as never,
|
||||
{
|
||||
enableLocalFilesystem: false,
|
||||
enableNpmDiscovery: false,
|
||||
localPluginDir: "__missing_plugin_dir__",
|
||||
},
|
||||
{
|
||||
lifecycleManager,
|
||||
workerManager: {},
|
||||
eventBus: {},
|
||||
jobScheduler: {},
|
||||
jobStore: {},
|
||||
toolDispatcher: {},
|
||||
buildHostHandlers: vi.fn(),
|
||||
instanceInfo: {
|
||||
instanceId: "test",
|
||||
hostVersion: "0.0.0",
|
||||
},
|
||||
} as never,
|
||||
);
|
||||
|
||||
await expect(loader.loadSingle(PLUGIN_ID)).resolves.toMatchObject({
|
||||
success: false,
|
||||
error: expect.stringContaining("Package root not found"),
|
||||
});
|
||||
expect(lifecycleManager.markError).toHaveBeenCalledTimes(1);
|
||||
|
||||
lifecycleManager.markError.mockClear();
|
||||
await expect(loader.loadSingle(PLUGIN_ID, { markErrorOnFailure: false })).resolves.toMatchObject({
|
||||
success: false,
|
||||
error: expect.stringContaining("Package root not found"),
|
||||
});
|
||||
expect(lifecycleManager.markError).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("tears down partially-activated runtime state when error writes are suppressed", async () => {
|
||||
const fixtureDir = fs.mkdtempSync(path.join(os.tmpdir(), "plugin-recovery-fixture-"));
|
||||
try {
|
||||
fs.writeFileSync(
|
||||
path.join(fixtureDir, "package.json"),
|
||||
JSON.stringify({
|
||||
name: "@paperclipai/daytona-sandbox-provider",
|
||||
version: "1.0.0",
|
||||
paperclipPlugin: { manifest: "manifest.mjs" },
|
||||
}),
|
||||
);
|
||||
fs.writeFileSync(
|
||||
path.join(fixtureDir, "manifest.mjs"),
|
||||
`export default ${JSON.stringify(manifest)};`,
|
||||
);
|
||||
fs.mkdirSync(path.join(fixtureDir, "dist"));
|
||||
fs.writeFileSync(path.join(fixtureDir, "dist", "worker.js"), "");
|
||||
|
||||
const plugin = {
|
||||
...createPlugin("ready"),
|
||||
packageName: "@paperclipai/daytona-sandbox-provider",
|
||||
packagePath: fixtureDir,
|
||||
version: "1.0.0",
|
||||
};
|
||||
mockRegistry.getById.mockResolvedValue(plugin);
|
||||
mockRegistry.listConfigs.mockResolvedValue([]);
|
||||
mockRegistry.update.mockResolvedValue(undefined);
|
||||
|
||||
const lifecycleManager = { markError: vi.fn().mockResolvedValue(undefined) };
|
||||
const workerManager = {
|
||||
startWorker: vi.fn().mockResolvedValue(undefined),
|
||||
isRunning: vi.fn(() => true),
|
||||
stopWorker: vi.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
const eventBus = {
|
||||
forPlugin: vi.fn(() => {
|
||||
throw new Error("event bus offline");
|
||||
}),
|
||||
clearPlugin: vi.fn(),
|
||||
subscriptionCount: vi.fn(() => 0),
|
||||
};
|
||||
const jobScheduler = { unregisterPlugin: vi.fn().mockResolvedValue(undefined) };
|
||||
const toolDispatcher = { unregisterPluginTools: vi.fn(), registerPluginTools: vi.fn() };
|
||||
|
||||
const loader = pluginLoader(
|
||||
{} as never,
|
||||
{
|
||||
enableLocalFilesystem: false,
|
||||
enableNpmDiscovery: false,
|
||||
localPluginDir: "__missing_plugin_dir__",
|
||||
},
|
||||
{
|
||||
lifecycleManager,
|
||||
workerManager,
|
||||
eventBus,
|
||||
jobScheduler,
|
||||
jobStore: {},
|
||||
toolDispatcher,
|
||||
buildHostHandlers: vi.fn(() => ({})),
|
||||
instanceInfo: {
|
||||
instanceId: "test",
|
||||
hostVersion: "0.0.0",
|
||||
},
|
||||
} as never,
|
||||
);
|
||||
|
||||
await expect(loader.loadSingle(PLUGIN_ID, { markErrorOnFailure: false })).resolves.toMatchObject({
|
||||
success: false,
|
||||
error: expect.stringContaining("event bus offline"),
|
||||
});
|
||||
|
||||
expect(lifecycleManager.markError).not.toHaveBeenCalled();
|
||||
expect(jobScheduler.unregisterPlugin).toHaveBeenCalledWith(PLUGIN_ID);
|
||||
expect(eventBus.clearPlugin).toHaveBeenCalledWith(PLUGIN_KEY);
|
||||
expect(toolDispatcher.unregisterPluginTools).toHaveBeenCalledWith(PLUGIN_KEY);
|
||||
expect(workerManager.stopWorker).toHaveBeenCalledWith(PLUGIN_ID);
|
||||
} finally {
|
||||
fs.rmSync(fixtureDir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("bounds slow managed bundled recovery attempts without serial waits", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const plugins = [
|
||||
createPlugin("ready"),
|
||||
createPlugin("ready", {
|
||||
id: "plugin-modal",
|
||||
pluginKey: "paperclip.modal-sandbox-provider",
|
||||
}),
|
||||
];
|
||||
mockRegistry.list.mockResolvedValue(plugins);
|
||||
const worker = createWorkerManager();
|
||||
const startWorker = vi.fn(() => new Promise<boolean>(() => {}));
|
||||
|
||||
const driversPromise = listReadyPluginEnvironmentDrivers({
|
||||
db: {} as never,
|
||||
workerManager: worker.workerManager,
|
||||
recoverMissingWorker: {
|
||||
pluginKeys: [PLUGIN_KEY, "paperclip.modal-sandbox-provider"],
|
||||
startWorker,
|
||||
timeoutMs: 25,
|
||||
},
|
||||
});
|
||||
await Promise.resolve();
|
||||
|
||||
expect(startWorker).toHaveBeenCalledTimes(2);
|
||||
await vi.advanceTimersByTimeAsync(25);
|
||||
|
||||
await expect(driversPromise).resolves.toEqual([]);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
@ -64,7 +64,7 @@ import { pluginUiStaticRoutes } from "./routes/plugin-ui-static.js";
|
|||
import { readBrandedStaticIndexHtml } from "./static-index-html.js";
|
||||
import { applyUiBranding } from "./ui-branding.js";
|
||||
import { logger } from "./middleware/logger.js";
|
||||
import { DEFAULT_LOCAL_PLUGIN_DIR, pluginLoader } from "./services/plugin-loader.js";
|
||||
import { DEFAULT_LOCAL_PLUGIN_DIR, pluginLoader, type PluginLoader } from "./services/plugin-loader.js";
|
||||
import {
|
||||
SELF_HOSTED_AUTO_INSTALL_KEYS,
|
||||
ensureBundledPlugins,
|
||||
|
|
@ -148,6 +148,87 @@ export function shouldEnablePrivateHostnameGuard(opts: {
|
|||
);
|
||||
}
|
||||
|
||||
export function createManagedBundledPluginWorkerRecovery(input: {
|
||||
managedBundledPluginKeys: readonly string[];
|
||||
workerManager: Pick<PluginWorkerManager, "getWorker" | "isRunning" | "stopWorker">;
|
||||
getLoader: () => Pick<PluginLoader, "loadSingle"> | null;
|
||||
}): (plugin: { id: string; pluginKey: string }) => Promise<boolean> {
|
||||
const recoverablePluginKeys = new Set(input.managedBundledPluginKeys);
|
||||
const inFlightStarts = new Map<string, Promise<boolean>>();
|
||||
|
||||
// A failed attempt can leave behind the dead handle it registered (e.g. the
|
||||
// worker process died during initialize, which kills the process without
|
||||
// scheduling a restart). No pre-existing handle survives to a recovery
|
||||
// attempt — recovery only starts when getWorker() was empty — so discarding
|
||||
// the dead handle lets a later capability request retry instead of being
|
||||
// blocked by the handle-presence gate until the process restarts. Handles
|
||||
// in starting/running/backoff states belong to the worker manager's own
|
||||
// lifecycle and are left alone.
|
||||
const discardDeadRecoveryHandle = async (plugin: { id: string; pluginKey: string }) => {
|
||||
const handle = input.workerManager.getWorker(plugin.id);
|
||||
if (!handle || (handle.status !== "crashed" && handle.status !== "stopped")) return;
|
||||
try {
|
||||
await input.workerManager.stopWorker(plugin.id);
|
||||
} catch (err) {
|
||||
logger.warn(
|
||||
{
|
||||
pluginId: plugin.id,
|
||||
pluginKey: plugin.pluginKey,
|
||||
err: err instanceof Error ? err.message : String(err),
|
||||
},
|
||||
"failed to discard dead plugin worker handle after recovery failure",
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
return async (plugin) => {
|
||||
if (!recoverablePluginKeys.has(plugin.pluginKey)) return false;
|
||||
|
||||
const inFlight = inFlightStarts.get(plugin.id);
|
||||
if (inFlight) return inFlight;
|
||||
|
||||
const startPromise = (async () => {
|
||||
if (input.workerManager.getWorker(plugin.id)) {
|
||||
return input.workerManager.isRunning(plugin.id);
|
||||
}
|
||||
|
||||
const loader = input.getLoader();
|
||||
if (!loader) return false;
|
||||
|
||||
try {
|
||||
const result = await loader.loadSingle(plugin.id, {
|
||||
markErrorOnFailure: false,
|
||||
});
|
||||
if (result.success === true || input.workerManager.isRunning(plugin.id)) {
|
||||
return true;
|
||||
}
|
||||
await discardDeadRecoveryHandle(plugin);
|
||||
return false;
|
||||
} catch (err) {
|
||||
logger.warn(
|
||||
{
|
||||
pluginId: plugin.id,
|
||||
pluginKey: plugin.pluginKey,
|
||||
err: err instanceof Error ? err.message : String(err),
|
||||
},
|
||||
"managed bundled plugin lazy worker recovery failed",
|
||||
);
|
||||
await discardDeadRecoveryHandle(plugin);
|
||||
throw err;
|
||||
}
|
||||
})();
|
||||
|
||||
inFlightStarts.set(plugin.id, startPromise);
|
||||
try {
|
||||
return await startPromise;
|
||||
} finally {
|
||||
if (inFlightStarts.get(plugin.id) === startPromise) {
|
||||
inFlightStarts.delete(plugin.id);
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
export async function createApp(
|
||||
db: Db,
|
||||
opts: {
|
||||
|
|
@ -238,6 +319,34 @@ export async function createApp(
|
|||
|
||||
const hostServicesDisposers = new Map<string, () => void>();
|
||||
const workerManager = opts.pluginWorkerManager ?? createPluginWorkerManager();
|
||||
const managedAutoInstallKeys = opts.managedPluginAutoInstall ?? null;
|
||||
const bundledCatalogRoot =
|
||||
opts.bundledPluginCatalogRoot ?? resolveBundledCatalogRoot(process.env);
|
||||
const bundledPluginInstalls = resolveBundledPluginInstalls(
|
||||
managedAutoInstallKeys ?? SELF_HOSTED_AUTO_INSTALL_KEYS,
|
||||
{
|
||||
catalogRoot: bundledCatalogRoot,
|
||||
env: process.env,
|
||||
enforceCatalogRoot: managedAutoInstallKeys !== null,
|
||||
},
|
||||
);
|
||||
const managedBundledPluginKeys =
|
||||
managedAutoInstallKeys !== null
|
||||
? bundledPluginInstalls.map((install) => install.pluginKey)
|
||||
: [];
|
||||
let runtimePluginLoader: Pick<PluginLoader, "loadSingle"> | null = null;
|
||||
// A sibling process can install a managed bundled plugin while this process
|
||||
// skips the mid-install row, then finish the row after this process's
|
||||
// loadAll() pass. The capabilities route may recover only those managed
|
||||
// bundles by starting their ready-but-unstarted worker lazily.
|
||||
const recoverManagedBundledPluginWorker =
|
||||
managedAutoInstallKeys !== null
|
||||
? createManagedBundledPluginWorkerRecovery({
|
||||
managedBundledPluginKeys,
|
||||
workerManager,
|
||||
getLoader: () => runtimePluginLoader,
|
||||
})
|
||||
: undefined;
|
||||
|
||||
// Mount API routes
|
||||
const api = Router();
|
||||
|
|
@ -271,7 +380,15 @@ export async function createApp(
|
|||
api.use(fileResourceRoutes(db));
|
||||
api.use(routineRoutes(db, { pluginWorkerManager: workerManager }));
|
||||
api.use(pipelineRoutes(db));
|
||||
api.use(environmentRoutes(db, { pluginWorkerManager: workerManager }));
|
||||
api.use(environmentRoutes(db, {
|
||||
pluginWorkerManager: workerManager,
|
||||
recoverMissingPluginWorker: recoverManagedBundledPluginWorker
|
||||
? {
|
||||
pluginKeys: managedBundledPluginKeys,
|
||||
startWorker: recoverManagedBundledPluginWorker,
|
||||
}
|
||||
: undefined,
|
||||
}));
|
||||
api.use(executionWorkspaceRoutes(db, { pluginWorkerManager: workerManager }));
|
||||
api.use(goalRoutes(db));
|
||||
api.use(boardChatRoutes(db, { deploymentMode: opts.deploymentMode }));
|
||||
|
|
@ -380,6 +497,7 @@ export async function createApp(
|
|||
},
|
||||
},
|
||||
);
|
||||
runtimePluginLoader = loader;
|
||||
api.use(
|
||||
toolGatewayRoutes(db, toolGateway),
|
||||
);
|
||||
|
|
@ -570,21 +688,10 @@ export async function createApp(
|
|||
// drive the key list from the control plane; self-hosted instances keep
|
||||
// the pre-existing behavior of ensuring only the kubernetes bundle.
|
||||
//
|
||||
// Resolution below is deliberately synchronous and NOT fail-safe: an
|
||||
// Resolution is deliberately synchronous and NOT fail-safe: an
|
||||
// unknown key or a path escaping the bundled catalog root throws out of
|
||||
// createApp so a managed instance refuses to start (positive allowlist,
|
||||
// fail closed).
|
||||
const managedAutoInstallKeys = opts.managedPluginAutoInstall ?? null;
|
||||
const bundledCatalogRoot =
|
||||
opts.bundledPluginCatalogRoot ?? resolveBundledCatalogRoot(process.env);
|
||||
const bundledPluginInstalls = resolveBundledPluginInstalls(
|
||||
managedAutoInstallKeys ?? SELF_HOSTED_AUTO_INSTALL_KEYS,
|
||||
{
|
||||
catalogRoot: bundledCatalogRoot,
|
||||
env: process.env,
|
||||
enforceCatalogRoot: managedAutoInstallKeys !== null,
|
||||
},
|
||||
);
|
||||
// SAFETY: installation is fully fail-safe. Any failure
|
||||
// (missing bundle, install error, load error) is caught, logged, and
|
||||
// swallowed per plugin so the server ALWAYS finishes booting. A degraded
|
||||
|
|
|
|||
|
|
@ -47,7 +47,10 @@ import {
|
|||
} from "../services/environment-config.js";
|
||||
import { probeEnvironment } from "../services/environment-probe.js";
|
||||
import { secretService } from "../services/secrets.js";
|
||||
import { listReadyPluginEnvironmentDrivers } from "../services/plugin-environment-driver.js";
|
||||
import {
|
||||
listReadyPluginEnvironmentDrivers,
|
||||
type ReadyPluginWorkerRecovery,
|
||||
} from "../services/plugin-environment-driver.js";
|
||||
import { getConfiguredSecretProvider } from "../secrets/configured-provider.js";
|
||||
import { assertBoardOrgAccess, getActorInfo } from "./authz.js";
|
||||
import type { PluginWorkerManager } from "../services/plugin-worker-manager.js";
|
||||
|
|
@ -261,7 +264,10 @@ function assertNoClientPlatformProvisionedMarkers(metadata: unknown): void {
|
|||
|
||||
export function environmentRoutes(
|
||||
db: Db,
|
||||
options: { pluginWorkerManager?: PluginWorkerManager } = {},
|
||||
options: {
|
||||
pluginWorkerManager?: PluginWorkerManager;
|
||||
recoverMissingPluginWorker?: ReadyPluginWorkerRecovery;
|
||||
} = {},
|
||||
) {
|
||||
const router = Router();
|
||||
const svc = environmentService(db);
|
||||
|
|
@ -602,6 +608,7 @@ export function environmentRoutes(
|
|||
const pluginDrivers = await listReadyPluginEnvironmentDrivers({
|
||||
db,
|
||||
workerManager: options.pluginWorkerManager,
|
||||
recoverMissingWorker: options.recoverMissingPluginWorker,
|
||||
});
|
||||
res.json(getEnvironmentCapabilities(
|
||||
AGENT_ADAPTER_TYPES,
|
||||
|
|
|
|||
|
|
@ -30,10 +30,49 @@ import {
|
|||
import { pluginRegistryService } from "./plugin-registry.js";
|
||||
import type { PluginWorkerManager } from "./plugin-worker-manager.js";
|
||||
|
||||
export interface ReadyPluginWorkerRecovery {
|
||||
pluginKeys: readonly string[];
|
||||
startWorker(plugin: { id: string; pluginKey: string }): Promise<boolean>;
|
||||
timeoutMs?: number;
|
||||
}
|
||||
|
||||
export interface ReadyPluginEnvironmentDriver {
|
||||
pluginId: string;
|
||||
pluginKey: string;
|
||||
driverKey: string;
|
||||
displayName: string;
|
||||
description?: string;
|
||||
configSchema: PluginEnvironmentDriverDeclaration["configSchema"];
|
||||
supportsReusableLeases?: PluginEnvironmentDriverDeclaration["supportsReusableLeases"];
|
||||
supportsInteractiveSetup?: PluginEnvironmentDriverDeclaration["supportsInteractiveSetup"];
|
||||
interactiveSetupConnectionTypes?: PluginEnvironmentDriverDeclaration["interactiveSetupConnectionTypes"];
|
||||
supportsTemplateCapture?: PluginEnvironmentDriverDeclaration["supportsTemplateCapture"];
|
||||
templateRefKind?: PluginEnvironmentDriverDeclaration["templateRefKind"];
|
||||
templateConfigBinding?: PluginEnvironmentDriverDeclaration["templateConfigBinding"];
|
||||
supportsTemplateDelete?: PluginEnvironmentDriverDeclaration["supportsTemplateDelete"];
|
||||
}
|
||||
|
||||
export function pluginDriverProviderKey(config: Pick<PluginEnvironmentConfig, "pluginKey" | "driverKey">): string {
|
||||
return `${config.pluginKey}:${config.driverKey}`;
|
||||
}
|
||||
|
||||
const DEFAULT_READY_PLUGIN_WORKER_RECOVERY_TIMEOUT_MS = 2_000;
|
||||
|
||||
async function resolveWithTimeout<T>(promise: Promise<T>, timeoutMs: number, timeoutValue: T): Promise<T> {
|
||||
if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) return await promise;
|
||||
let timeout: ReturnType<typeof setTimeout> | null = null;
|
||||
try {
|
||||
return await Promise.race([
|
||||
promise,
|
||||
new Promise<T>((resolve) => {
|
||||
timeout = setTimeout(() => resolve(timeoutValue), timeoutMs);
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
if (timeout) clearTimeout(timeout);
|
||||
}
|
||||
}
|
||||
|
||||
export async function resolvePluginEnvironmentDriver(input: {
|
||||
db: Db;
|
||||
workerManager: PluginWorkerManager;
|
||||
|
|
@ -94,30 +133,67 @@ export async function resolvePluginSandboxProviderDriverByKey(input: {
|
|||
export async function listReadyPluginEnvironmentDrivers(input: {
|
||||
db: Db;
|
||||
workerManager?: PluginWorkerManager;
|
||||
recoverMissingWorker?: ReadyPluginWorkerRecovery;
|
||||
}) {
|
||||
if (!input.workerManager) return [];
|
||||
const pluginRegistry = pluginRegistryService(input.db);
|
||||
const plugins = await pluginRegistry.list();
|
||||
return plugins.flatMap((plugin) => {
|
||||
if (plugin.status !== "ready" || !input.workerManager?.isRunning(plugin.id)) return [];
|
||||
return (plugin.manifestJson.environmentDrivers ?? [])
|
||||
.filter((driver) => driver.kind === "sandbox_provider")
|
||||
.map((driver) => ({
|
||||
pluginId: plugin.id,
|
||||
const recoverablePluginKeys = new Set(input.recoverMissingWorker?.pluginKeys ?? []);
|
||||
const readyPlugins = plugins.filter((plugin) => plugin.status === "ready");
|
||||
const recoveryAttempts: Promise<boolean>[] = [];
|
||||
|
||||
for (const plugin of readyPlugins) {
|
||||
const hasSandboxProviderDriver = plugin.manifestJson.environmentDrivers?.some(
|
||||
(driver) => driver.kind === "sandbox_provider",
|
||||
) ?? false;
|
||||
const canRecover =
|
||||
hasSandboxProviderDriver
|
||||
&& !input.workerManager.isRunning(plugin.id)
|
||||
&& recoverablePluginKeys.has(plugin.pluginKey)
|
||||
&& !input.workerManager.getWorker(plugin.id);
|
||||
if (!canRecover || !input.recoverMissingWorker) continue;
|
||||
const timeoutMs =
|
||||
input.recoverMissingWorker.timeoutMs ?? DEFAULT_READY_PLUGIN_WORKER_RECOVERY_TIMEOUT_MS;
|
||||
recoveryAttempts.push(resolveWithTimeout(
|
||||
input.recoverMissingWorker.startWorker({
|
||||
id: plugin.id,
|
||||
pluginKey: plugin.pluginKey,
|
||||
driverKey: driver.driverKey,
|
||||
displayName: driver.displayName,
|
||||
description: driver.description,
|
||||
configSchema: driver.configSchema,
|
||||
supportsReusableLeases: driver.supportsReusableLeases,
|
||||
supportsInteractiveSetup: driver.supportsInteractiveSetup,
|
||||
interactiveSetupConnectionTypes: driver.interactiveSetupConnectionTypes,
|
||||
supportsTemplateCapture: driver.supportsTemplateCapture,
|
||||
templateRefKind: driver.templateRefKind,
|
||||
templateConfigBinding: driver.templateConfigBinding,
|
||||
supportsTemplateDelete: driver.supportsTemplateDelete,
|
||||
}));
|
||||
});
|
||||
}).catch(() => false),
|
||||
timeoutMs,
|
||||
false,
|
||||
));
|
||||
}
|
||||
|
||||
if (recoveryAttempts.length > 0) {
|
||||
await Promise.all(recoveryAttempts);
|
||||
}
|
||||
|
||||
const rows: ReadyPluginEnvironmentDriver[] = [];
|
||||
for (const plugin of readyPlugins) {
|
||||
if (!input.workerManager.isRunning(plugin.id)) {
|
||||
continue;
|
||||
}
|
||||
rows.push(
|
||||
...(plugin.manifestJson.environmentDrivers ?? [])
|
||||
.filter((driver) => driver.kind === "sandbox_provider")
|
||||
.map((driver) => ({
|
||||
pluginId: plugin.id,
|
||||
pluginKey: plugin.pluginKey,
|
||||
driverKey: driver.driverKey,
|
||||
displayName: driver.displayName,
|
||||
description: driver.description,
|
||||
configSchema: driver.configSchema,
|
||||
supportsReusableLeases: driver.supportsReusableLeases,
|
||||
supportsInteractiveSetup: driver.supportsInteractiveSetup,
|
||||
interactiveSetupConnectionTypes: driver.interactiveSetupConnectionTypes,
|
||||
supportsTemplateCapture: driver.supportsTemplateCapture,
|
||||
templateRefKind: driver.templateRefKind,
|
||||
templateConfigBinding: driver.templateConfigBinding,
|
||||
supportsTemplateDelete: driver.supportsTemplateDelete,
|
||||
})),
|
||||
);
|
||||
}
|
||||
return rows;
|
||||
}
|
||||
|
||||
export async function validatePluginSandboxProviderConfig(input: {
|
||||
|
|
|
|||
|
|
@ -360,6 +360,20 @@ export interface PluginLoadResult {
|
|||
};
|
||||
}
|
||||
|
||||
export interface PluginLoadSingleOptions {
|
||||
/**
|
||||
* When false, activation failures are reported to the caller but do not
|
||||
* transition the shared plugin row to `error`. Partially-registered local
|
||||
* runtime state (worker process, scheduler/tool registrations) is still
|
||||
* torn down in this process.
|
||||
*/
|
||||
markErrorOnFailure?: boolean;
|
||||
}
|
||||
|
||||
interface PluginActivateOptions {
|
||||
markErrorOnFailure: boolean;
|
||||
}
|
||||
|
||||
/**
|
||||
* Result of activating all ready plugins at server startup.
|
||||
*/
|
||||
|
|
@ -540,7 +554,7 @@ export interface PluginLoader {
|
|||
*
|
||||
* @see PLUGIN_SPEC.md §8.3 — Install Process
|
||||
*/
|
||||
loadSingle(pluginId: string): Promise<PluginLoadResult>;
|
||||
loadSingle(pluginId: string, options?: PluginLoadSingleOptions): Promise<PluginLoadResult>;
|
||||
|
||||
/**
|
||||
* Deactivate a single plugin — stop its worker and unregister all
|
||||
|
|
@ -1937,9 +1951,10 @@ export function pluginLoader(
|
|||
* capabilities (tools, jobs, etc.).
|
||||
*
|
||||
* @param pluginId - The UUID of the plugin to load.
|
||||
* @param options - Optional activation behavior overrides.
|
||||
* @returns A promise that resolves with the result of the activation.
|
||||
*/
|
||||
async loadSingle(pluginId: string): Promise<PluginLoadResult> {
|
||||
async loadSingle(pluginId: string, options: PluginLoadSingleOptions = {}): Promise<PluginLoadResult> {
|
||||
if (!runtimeServices) {
|
||||
throw new Error(
|
||||
"Cannot loadSingle: no PluginRuntimeServices provided. " +
|
||||
|
|
@ -1975,7 +1990,9 @@ export function pluginLoader(
|
|||
);
|
||||
}
|
||||
|
||||
return activatePlugin(plugin);
|
||||
return activatePlugin(plugin, {
|
||||
markErrorOnFailure: options.markErrorOnFailure ?? true,
|
||||
});
|
||||
},
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
|
|
@ -1994,40 +2011,7 @@ export function pluginLoader(
|
|||
"plugin-loader: unloading single plugin",
|
||||
);
|
||||
|
||||
const {
|
||||
workerManager,
|
||||
eventBus,
|
||||
jobScheduler,
|
||||
toolDispatcher,
|
||||
} = runtimeServices;
|
||||
|
||||
// 1. Unregister from job scheduler (cancels in-flight runs)
|
||||
try {
|
||||
await jobScheduler.unregisterPlugin(pluginId);
|
||||
} catch (err) {
|
||||
log.warn(
|
||||
{ pluginId, err: err instanceof Error ? err.message : String(err) },
|
||||
"plugin-loader: failed to unregister from job scheduler (best-effort)",
|
||||
);
|
||||
}
|
||||
|
||||
// 2. Clear event subscriptions
|
||||
eventBus.clearPlugin(pluginKey);
|
||||
|
||||
// 3. Unregister agent tools
|
||||
toolDispatcher.unregisterPluginTools(pluginKey);
|
||||
|
||||
// 4. Stop the worker process
|
||||
try {
|
||||
if (workerManager.isRunning(pluginId)) {
|
||||
await workerManager.stopWorker(pluginId);
|
||||
}
|
||||
} catch (err) {
|
||||
log.warn(
|
||||
{ pluginId, err: err instanceof Error ? err.message : String(err) },
|
||||
"plugin-loader: failed to stop worker during unload (best-effort)",
|
||||
);
|
||||
}
|
||||
await teardownPluginRuntime(pluginId, pluginKey);
|
||||
|
||||
log.info(
|
||||
{ pluginId, pluginKey },
|
||||
|
|
@ -2060,6 +2044,58 @@ export function pluginLoader(
|
|||
},
|
||||
};
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Internal: teardownPluginRuntime — shared by unloadSingle and activation
|
||||
// failure cleanup
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Tear down a plugin's runtime state in this process: scheduler
|
||||
* registration, event subscriptions, agent tools, and the worker process.
|
||||
* Does not touch the plugin's database row.
|
||||
*/
|
||||
async function teardownPluginRuntime(
|
||||
pluginId: string,
|
||||
pluginKey: string,
|
||||
): Promise<void> {
|
||||
if (!runtimeServices) return;
|
||||
|
||||
const {
|
||||
workerManager,
|
||||
eventBus,
|
||||
jobScheduler,
|
||||
toolDispatcher,
|
||||
} = runtimeServices;
|
||||
|
||||
// 1. Unregister from job scheduler (cancels in-flight runs)
|
||||
try {
|
||||
await jobScheduler.unregisterPlugin(pluginId);
|
||||
} catch (err) {
|
||||
log.warn(
|
||||
{ pluginId, err: err instanceof Error ? err.message : String(err) },
|
||||
"plugin-loader: failed to unregister from job scheduler (best-effort)",
|
||||
);
|
||||
}
|
||||
|
||||
// 2. Clear event subscriptions
|
||||
eventBus.clearPlugin(pluginKey);
|
||||
|
||||
// 3. Unregister agent tools
|
||||
toolDispatcher.unregisterPluginTools(pluginKey);
|
||||
|
||||
// 4. Stop the worker process
|
||||
try {
|
||||
if (workerManager.isRunning(pluginId)) {
|
||||
await workerManager.stopWorker(pluginId);
|
||||
}
|
||||
} catch (err) {
|
||||
log.warn(
|
||||
{ pluginId, err: err instanceof Error ? err.message : String(err) },
|
||||
"plugin-loader: failed to stop worker during unload (best-effort)",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Internal: activatePlugin — shared logic for loadAll and loadSingle
|
||||
// -------------------------------------------------------------------------
|
||||
|
|
@ -2069,10 +2105,13 @@ export function pluginLoader(
|
|||
* sync jobs, register tools.
|
||||
*
|
||||
* This is the core orchestration logic shared by `loadAll()` and `loadSingle()`.
|
||||
* Failures are caught and reported in the result; the plugin is marked as
|
||||
* `error` in the database when activation fails.
|
||||
* Failures are caught and reported in the result. By default the plugin is
|
||||
* marked as `error` in the database when activation fails.
|
||||
*/
|
||||
async function activatePlugin(plugin: PluginRecord): Promise<PluginLoadResult> {
|
||||
async function activatePlugin(
|
||||
plugin: PluginRecord,
|
||||
options: PluginActivateOptions = { markErrorOnFailure: true },
|
||||
): Promise<PluginLoadResult> {
|
||||
const pluginId = plugin.id;
|
||||
const pluginKey = plugin.pluginKey;
|
||||
let activePlugin = plugin;
|
||||
|
|
@ -2351,18 +2390,37 @@ export function pluginLoader(
|
|||
"plugin-loader: failed to activate plugin",
|
||||
);
|
||||
|
||||
// Mark the plugin as errored in the database so it is not retried
|
||||
// automatically on next startup without operator intervention.
|
||||
try {
|
||||
await lifecycleManager.markError(pluginId, `Activation failed: ${errorMessage}`);
|
||||
} catch (markErr) {
|
||||
log.error(
|
||||
{
|
||||
pluginId,
|
||||
err: markErr instanceof Error ? markErr.message : String(markErr),
|
||||
},
|
||||
"plugin-loader: failed to mark plugin as error after activation failure",
|
||||
);
|
||||
if (options.markErrorOnFailure) {
|
||||
// Mark the plugin as errored in the database so it is not retried
|
||||
// automatically on next startup without operator intervention.
|
||||
// markError also deactivates the plugin runtime in this process.
|
||||
try {
|
||||
await lifecycleManager.markError(pluginId, `Activation failed: ${errorMessage}`);
|
||||
} catch (markErr) {
|
||||
log.error(
|
||||
{
|
||||
pluginId,
|
||||
err: markErr instanceof Error ? markErr.message : String(markErr),
|
||||
},
|
||||
"plugin-loader: failed to mark plugin as error after activation failure",
|
||||
);
|
||||
}
|
||||
} else if (registered.worker) {
|
||||
// The shared plugin row stays untouched, but this process spawned a
|
||||
// worker before the failure — tear down the partially-registered
|
||||
// runtime so a half-activated plugin does not linger locally.
|
||||
try {
|
||||
await teardownPluginRuntime(pluginId, pluginKey);
|
||||
} catch (cleanupErr) {
|
||||
log.warn(
|
||||
{
|
||||
pluginId,
|
||||
pluginKey,
|
||||
err: cleanupErr instanceof Error ? cleanupErr.message : String(cleanupErr),
|
||||
},
|
||||
"plugin-loader: failed to tear down partially-activated plugin runtime",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
|
|
|
|||
Loading…
Reference in New Issue