From 1206373a955b938b1623a6dc7a87a313eef83d21 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Fri, 6 Nov 2026 10:35:57 -0800 Subject: [PATCH] fix(environments): close Apex findings on managed plugin environments (#19526) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A managed instance gets its instance-level environments from the `PAPERCLIP_MANAGED_CONFIG` document, and #19508 added plugin environment entries to it > - Greptile's Apex review of #19508 arrived after it merged, with three findings > - A credential inside an array nested in an array passed the secret check and the response redaction > - Recovery of an archived environment listened on one worker handle, which a restart replaces, and its attempts were not tracked through idle checks and shutdown > - This pull request closes all three findings > - The benefit is that credentials never pass the check, an environment comes back after any worker restart, and shutdown never cuts off a recovery write ## Linked Issues or Issue Description Refs #19508 (the Apex findings this pull request fixes) **Subsystem affected** Server environments: the managed-config parser, the environment routes' response floor, the managed environment bootstrap, and the plugin worker manager. **Problem or motivation** Greptile's Apex review of #19508 found three problems. (1) `findSecretLikeConfigKey` and `redactSecretLikeConfigKeys` skip array elements that are arrays, so `{"groups":[[{"apiToken":"..."}]]}` passes the boot check and is echoed to restricted readers. (2) Recovery subscribes to `ready` on one worker handle. A restart such as a settings save replaces that handle, so the new worker's `ready` never reaches the listener, and the environment stays archived. (3) A recovery attempt is fire-and-forget. The worker manager's work token ends when the worker replies, but the database write continues, so an idle check or the database shutdown can overlap it. **Proposed solution** Walk arrays at any depth in both functions. Add a manager-wide `onWorkerReady` subscription and use it for recovery. Run every recovery attempt through a controller that tracks it as idle work and that shutdown stops and awaits before the database closes. **Alternatives considered** Re-subscribing on each new handle after a restart. That needs a hook in every restart path. The manager-wide subscription covers all of them in one place. **Roadmap alignment** This only fixes code that #19508 added or touched. It adds no product surface. ## What Changed - `server/src/services/managed-config.ts`: `findSecretLikeConfigKey` walks arrays at any depth (new `findSecretLikeConfigKeyInValue`). This covers sandbox `config` and plugin `driverConfig`. - `server/src/routes/environments.ts`: `redactSecretLikeConfigKeys` walks arrays at any depth (new `redactSecretLikeConfigValue`). - `server/src/services/plugin-worker-manager.ts`: new optional `onWorkerReady(listener)` on `PluginWorkerManager`. It fires on every worker `ready`, including a replacement handle from a restart. A listener error is logged and does not affect the worker. - `server/src/services/managed-environments.ts`: - Sandbox and plugin recovery subscribe through `subscribeWorkerReady`. It prefers `onWorkerReady` and falls back to the handle when the manager has no such method. - New `createManagedEnvironmentRecovery()` controller. Each attempt runs inside `trackIdleWork` for its whole duration. `stop()` removes every `ready` subscription, refuses new attempts, and awaits unfinished ones. - `server/src/index.ts`: creates the controller, passes it to `applyManagedEnvironments`, and awaits `stop()` during shutdown, after the delivery workers stop and before the application teardown and database close. ## Verification - `cd server && npx vitest run src/__tests__/managed-config.test.ts`: a secret-like key in an array nested in an array refuses boot, for plugin `driverConfig` and sandbox `config`. - `npx vitest run src/__tests__/environment-routes.test.ts`: a nested-array credential is removed from the response, and other nested values stay. - `npx vitest run src/services/managed-environments-plugin-entries.test.ts`: - With `onWorkerReady`, recovery never consults the handle. It ignores another plugin's `ready`, re-ensures on this plugin's `ready`, and then unsubscribes. - An attempt counts in `idleWorkSnapshot()` until its ensure settles. `stop()` unsubscribes, waits for the unfinished attempt, and later `ready` events start nothing. - `npx vitest run src/services/managed-environments.test.ts src/__tests__/managed-plugin-environments.test.ts`: pass. - `pnpm --filter @paperclipai/server typecheck` passes. - I did not run the full `pnpm test:run`, `pnpm -r typecheck` or `pnpm build`. CI covers them. ## Risks - The redaction now also walks nested arrays in sandbox config responses. Only credential-shaped keys are removed; other values are unchanged. - The boot check now refuses a document that hides a secret-like key in a nested array. A valid document has no such key. - `onWorkerReady` is optional, so other `PluginWorkerManager` implementations and test fakes keep working. - Shutdown now waits for an unfinished recovery attempt. An attempt is one worker call plus one short transaction. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, through Claude Code with tool use (file edits, shell, test runs). ## 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 --- .../src/__tests__/environment-routes.test.ts | 17 +++ server/src/__tests__/managed-config.test.ts | 7 + .../server-startup-feedback-export.test.ts | 15 +- server/src/index.ts | 14 ++ server/src/routes/environments.ts | 17 ++- server/src/services/index.ts | 2 +- server/src/services/managed-config.ts | 23 +-- ...anaged-environments-plugin-entries.test.ts | 74 +++++++++- server/src/services/managed-environments.ts | 134 +++++++++++++----- server/src/services/plugin-worker-manager.ts | 17 +++ 10 files changed, 266 insertions(+), 54 deletions(-) diff --git a/server/src/__tests__/environment-routes.test.ts b/server/src/__tests__/environment-routes.test.ts index 0274007ac2d..826e7517858 100644 --- a/server/src/__tests__/environment-routes.test.ts +++ b/server/src/__tests__/environment-routes.test.ts @@ -577,6 +577,23 @@ describe("environment routes", () => { expect(JSON.stringify(res.body)).not.toContain("config-credential-must-never-echo"); }); + it("redacts credential-shaped keys inside arrays nested in arrays", async () => { + mockEnvironmentService.getById.mockResolvedValue({ + ...createPlatformSandboxEnvironment(), + config: { + provider: "daytona", + groups: [[{ name: "a", apiToken: "nested-credential-must-never-echo" }], ["plain"]], + }, + }); + const app = createApp(ownerAdminActor); + + const res = await request(app).get("/api/environments/env-managed-1"); + + expect(res.status).toBe(200); + expect(res.body.config).toEqual({ provider: "daytona", groups: [[{ name: "a" }], ["plain"]] }); + expect(JSON.stringify(res.body)).not.toContain("nested-credential-must-never-echo"); + }); + it("keeps blanking env vars on legacy kubernetes-marker rows", async () => { // Pre-generalization builds may have written platform values into // these rows' env vars, so the legacy floor stays absolute there. diff --git a/server/src/__tests__/managed-config.test.ts b/server/src/__tests__/managed-config.test.ts index e01bdbb31d0..e160ff37f98 100644 --- a/server/src/__tests__/managed-config.test.ts +++ b/server/src/__tests__/managed-config.test.ts @@ -411,6 +411,11 @@ describe("parseManagedConfigEnv environments section", () => { ), ), ).toThrow(/adapters\[0\].clientSecret/); + expect(() => + parseManagedConfigEnv( + envWith(validDoc({ environments: [entry({ config: { pools: [[], [{ privateKey: { password: "s" } }]] } })] })), + ), + ).toThrow(/pools\[1\]\[0\].privateKey.password/); }); }); @@ -482,6 +487,8 @@ describe("parseManagedConfigEnv plugin environment entries", () => { .toThrow(/"environments\[0\].driverConfig" key "apiToken" looks secret-bearing/); expect(() => parseManagedConfigEnv(docWith([remoteHost({ driverConfig: { auth: [{ password: "p" }] } })]))) .toThrow(/auth\[0\].password/); + expect(() => parseManagedConfigEnv(docWith([remoteHost({ driverConfig: { groups: [[{ apiToken: "t" }]] } })]))) + .toThrow(/groups\[0\]\[0\].apiToken/); }); it("refuses unknown keys, including sandbox keys on a plugin entry and the reverse", () => { diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 92a080eb594..57b0de55862 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -37,6 +37,10 @@ vi.mock("../services/adapter-settings.js", async () => ({ initializeAdapterSettings: vi.fn(async () => null), })); vi.mock("../services/cloud-runtime-config.js", () => ({ initializeCloudRuntimeConfig: cloudRuntimeStartupMocks.initialize })); +const managedEnvironmentMocks = vi.hoisted(() => { + const recovery = { run: vi.fn(), onStop: vi.fn(), stop: vi.fn(async () => {}) }; + return { recovery, apply: vi.fn(async () => null), factory: vi.fn(() => recovery) }; +}); const cloudMigrationMocks = vi.hoisted(() => ({ migrate: vi.fn(async () => null), qualify: vi.fn(async () => null) })); vi.mock("../services/cloud-configuration-migration.js", () => ({ migrateCloudConfiguration: cloudMigrationMocks.migrate })); vi.mock("../services/cloud-configuration-qualification.js", () => ({ qualifyCloudConfigurationForCurrentImage: cloudMigrationMocks.qualify })); @@ -482,7 +486,8 @@ vi.mock("../services/index.js", () => ({ })), feedbackService: feedbackServiceFactoryMock, bootstrapExecutionPolicyFromEnv: vi.fn(async () => null), - applyManagedEnvironments: vi.fn(async () => null), + applyManagedEnvironments: managedEnvironmentMocks.apply, + createManagedEnvironmentRecovery: managedEnvironmentMocks.factory, environmentCustomImageService: environmentCustomImagesServiceFactoryMock, executionWorkspaceService: executionWorkspaceServiceFactoryMock, externalObjectService: externalObjectsServiceFactoryMock, @@ -1369,6 +1374,14 @@ describe("startServer feedback export wiring", () => { expect(createAppMock).not.toHaveBeenCalled(); }); + it("stops managed environment recovery when a later startup step fails", async () => { + managedEnvironmentMocks.recovery.stop.mockClear(); + managedEnvironmentMocks.apply.mockRejectedValueOnce(new Error("managed environments refused")); + await expect(startServer()).rejects.toThrow("managed environments refused"); + expect(managedEnvironmentMocks.factory).toHaveBeenCalled(); + expect(managedEnvironmentMocks.recovery.stop).toHaveBeenCalledTimes(1); + }); + it("passes the feedback export service into createApp so pending traces flush in runtime", async () => { const started = await startServer(); diff --git a/server/src/index.ts b/server/src/index.ts index 0ba89a9fd11..20c9b06c937 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -137,6 +137,7 @@ import { cloudActorHeaderSourceFromHeaders, resolveCloudTenantActor } from "./mi import { feedbackService, applyManagedEnvironments, + createManagedEnvironmentRecovery, backfillPrincipalAccessCompatibility, backfillLegacyToolOAuthTokens, bootstrapExecutionPolicyFromEnv, @@ -1175,6 +1176,17 @@ async function startServerWithDatabaseTeardown( throw err; } + // Post-boot recovery of managed environments (re-ensuring a row when its + // plugin worker becomes ready). Shutdown stops it before the database closes, + // and so does the startup-failure teardown: a later startup step that throws + // must not leave an attempt writing through a closed database. + const managedEnvironmentRecovery = createManagedEnvironmentRecovery(); + const closeDatabaseAfterStartupFailure = startupDatabase.close; + startupDatabase.close = async () => { + await managedEnvironmentRecovery.stop(); + await closeDatabaseAfterStartupFailure?.(); + }; + // Ensure the sandbox and plugin environments declared in the managed-config // document (`environments` section) before the heartbeat resumes queued runs. The // document already parsed fail-closed above; the ensure step itself is @@ -1190,6 +1202,7 @@ async function startServerWithDatabaseTeardown( locals?: { bundledPluginsStartup?: Promise; managedBundledPluginKeys?: readonly string[] }; }).locals; const managedEnvironmentsResult = await applyManagedEnvironments(db as any, managedConfig, { + recovery: managedEnvironmentRecovery, pluginsReady: appLocals?.bundledPluginsStartup, autoInstalledPluginKeys: appLocals?.managedBundledPluginKeys, workerManager: pluginWorkerManager, @@ -2043,6 +2056,7 @@ async function startServerWithDatabaseTeardown( await runtimeResponseDelivery?.stop(); await deploymentHints?.stop(); await deliveryWork.stop(); + await managedEnvironmentRecovery.stop(); // Whatever the drain did not finalize (timed-out runs, the hot-restart // skip path) still has a local-only tail when the in-flight run-log diff --git a/server/src/routes/environments.ts b/server/src/routes/environments.ts index 09b2023b7e7..3cafcc7f7fa 100644 --- a/server/src/routes/environments.ts +++ b/server/src/routes/environments.ts @@ -93,19 +93,18 @@ function redactSecretLikeConfigKeys(value: Record): Record = {}; for (const [key, child] of Object.entries(value)) { if (SECRET_LIKE_CONFIG_KEY_PATTERN.test(key)) continue; - if (isPlainRecord(child)) { - result[key] = redactSecretLikeConfigKeys(child); - } else if (Array.isArray(child)) { - result[key] = child.map((element) => - isPlainRecord(element) ? redactSecretLikeConfigKeys(element) : element, - ); - } else { - result[key] = child; - } + result[key] = redactSecretLikeConfigValue(child); } return result; } +/** Objects and arrays at any depth, including arrays nested in arrays. */ +function redactSecretLikeConfigValue(value: unknown): unknown { + if (isPlainRecord(value)) return redactSecretLikeConfigKeys(value); + if (Array.isArray(value)) return value.map(redactSecretLikeConfigValue); + return value; +} + /** * Floor view of a platform-provisioned environment on a cloud-managed * instance: credential-shaped config keys are never echoed — to ANY actor, diff --git a/server/src/services/index.ts b/server/src/services/index.ts index dedc461b788..3fc81402416 100644 --- a/server/src/services/index.ts +++ b/server/src/services/index.ts @@ -145,7 +145,7 @@ export { type ManagedInstanceConfig, } from "./managed-config.js"; export { bootstrapExecutionPolicyFromEnv } from "./execution-policy-bootstrap.js"; -export { applyManagedEnvironments } from "./managed-environments.js"; +export { applyManagedEnvironments, createManagedEnvironmentRecovery } from "./managed-environments.js"; export { buildExportFidelityReport, collectExportFidelityCounts } from "./export-fidelity.js"; export { companyPortabilityService } from "./company-portability.js"; export { teamsCatalogService } from "./teams-catalog.js"; diff --git a/server/src/services/managed-config.ts b/server/src/services/managed-config.ts index bea4e28ab6f..f73d0b16e78 100644 --- a/server/src/services/managed-config.ts +++ b/server/src/services/managed-config.ts @@ -146,16 +146,19 @@ function findSecretLikeConfigKey(value: Record, path: string): for (const [key, child] of Object.entries(value)) { const childPath = path.length > 0 ? `${path}.${key}` : key; if (SECRET_LIKE_CONFIG_KEY_PATTERN.test(key)) return childPath; - if (isPlainObject(child)) { - const nested = findSecretLikeConfigKey(child, childPath); - if (nested) return nested; - } else if (Array.isArray(child)) { - for (const [index, element] of child.entries()) { - if (!isPlainObject(element)) continue; - const nested = findSecretLikeConfigKey(element, `${childPath}[${index}]`); - if (nested) return nested; - } - } + const nested = findSecretLikeConfigKeyInValue(child, childPath); + if (nested) return nested; + } + return null; +} + +/** Objects and arrays at any depth, including arrays nested in arrays. */ +function findSecretLikeConfigKeyInValue(value: unknown, path: string): string | null { + if (isPlainObject(value)) return findSecretLikeConfigKey(value, path); + if (!Array.isArray(value)) return null; + for (const [index, element] of value.entries()) { + const nested = findSecretLikeConfigKeyInValue(element, `${path}[${index}]`); + if (nested) return nested; } return null; } diff --git a/server/src/services/managed-environments-plugin-entries.test.ts b/server/src/services/managed-environments-plugin-entries.test.ts index e9241b6728f..7f2386c959a 100644 --- a/server/src/services/managed-environments-plugin-entries.test.ts +++ b/server/src/services/managed-environments-plugin-entries.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from "vitest"; import { unprocessable } from "../errors.js"; -import { applyManagedEnvironments } from "./managed-environments.js"; +import { applyManagedEnvironments, createManagedEnvironmentRecovery } from "./managed-environments.js"; +import { idleWorkSnapshot } from "./task-admission.js"; import { PLUGIN_KEY, environmentsSeam, @@ -316,6 +317,77 @@ describe("applyManagedEnvironments plugin entries", () => { expect(manager.getWorker).not.toHaveBeenCalled(); }); + it("keeps waiting through a restart that replaces the worker handle", async () => { + // Manager-wide `ready` subscription: the worker handle is never consulted, + // so a replacement handle's `ready` still reaches the listener. + const readyListeners = new Set<(pluginId: string) => void>(); + const manager = { + isRunning: vi.fn(() => false), + call: vi.fn(), + getWorker: vi.fn(() => undefined), + onWorkerReady: vi.fn((listener: (pluginId: string) => void) => { + readyListeners.add(listener); + return () => { readyListeners.delete(listener); }; + }), + }; + const environments = environmentsSeam(); + await applyManagedEnvironments(noDb, configWith([pluginEntry()]), { + env: {}, + workerManager: manager, + environments, + instanceSettings: instanceSettingsSeam(), + resolveEnvironmentPlugin: readyPlugin(), + autoInstalledPluginKeys: [PLUGIN_KEY], + }); + expect(manager.getWorker).not.toHaveBeenCalled(); + + manager.isRunning.mockReturnValue(true); + for (const listener of [...readyListeners]) listener("another-plugin"); + await tick(); + expect(environments.ensureManagedPluginEnvironment).not.toHaveBeenCalled(); + for (const listener of [...readyListeners]) listener("remote-host-plugin"); + await tick(); + expect(environments.ensureManagedPluginEnvironment).toHaveBeenCalledTimes(1); + expect(readyListeners.size).toBe(0); + }); + + it("tracks recovery attempts as idle work and lets shutdown await and unsubscribe them", async () => { + const { handle, fireReady } = recoveryHandle(); + let finishEnsure!: () => void; + const ensureManagedPluginEnvironment = vi.fn(() => new Promise((resolve) => { + finishEnsure = () => resolve({ environment: pluginEnvironmentRow(), action: "updated" }); + })); + const recovery = createManagedEnvironmentRecovery(); + const manager = workerManager(false, handle); + await applyManagedEnvironments(noDb, configWith([pluginEntry()]), { + env: {}, + workerManager: manager, + environments: environmentsSeam({ ensureManagedPluginEnvironment: ensureManagedPluginEnvironment as never }), + instanceSettings: instanceSettingsSeam(), + resolveEnvironmentPlugin: readyPlugin(), + autoInstalledPluginKeys: [PLUGIN_KEY], + recovery, + }); + const idleBefore = idleWorkSnapshot().active; + manager.isRunning.mockReturnValue(true); + fireReady(); + expect(idleWorkSnapshot().active).toBe(idleBefore + 1); + + let stopped = false; + const stopping = recovery.stop().then(() => { stopped = true; }); + await tick(); + expect(stopped).toBe(false); + expect(handle.off).toHaveBeenCalled(); + finishEnsure(); + await stopping; + expect(stopped).toBe(true); + expect(idleWorkSnapshot().active).toBe(idleBefore); + + fireReady(); + await tick(); + expect(ensureManagedPluginEnvironment).toHaveBeenCalledTimes(1); + }); + it("treats a missing plugin, a plugin without the driver, or a worker manager without call as unavailable", async () => { for (const [resolveEnvironmentPlugin, manager] of [ [vi.fn(async () => null), workerManager()], diff --git a/server/src/services/managed-environments.ts b/server/src/services/managed-environments.ts index 36d9c515c3a..febcbb03dbc 100644 --- a/server/src/services/managed-environments.ts +++ b/server/src/services/managed-environments.ts @@ -63,10 +63,50 @@ import type { ManagedResourceStockStatus } from "./managed-resource-drift.js"; import { parseExecutionPolicyBootstrapEnv } from "./execution-policy-bootstrap.js"; import { resolvePluginSandboxProviderDriverByKey } from "./plugin-environment-driver.js"; import { pluginRegistryService } from "./plugin-registry.js"; +import { trackIdleWork } from "./task-admission.js"; import { resolveBundledCatalogRoot, resolveBundledPluginInstalls } from "./bundled-plugins.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; import type { ManagedPluginEnvironmentReconcileAction } from "./managed-plugin-environments.js"; +/** + * Lifecycle of the recovery work `applyManagedEnvironments` leaves behind + * after boot (re-ensuring an archived environment when its plugin worker + * becomes ready). + */ +export interface ManagedEnvironmentRecovery { + /** Run one attempt as idle-tracked work, unless stopped. */ + run(work: () => Promise): void; + /** Register an unsubscribe function for `stop()`. Calls it at once when stopped. */ + onStop(unsubscribe: () => void): void; + /** Unsubscribe every listener, refuse new attempts, and await unfinished ones. */ + stop(): Promise; +} + +export function createManagedEnvironmentRecovery(): ManagedEnvironmentRecovery { + const inFlight = new Set>(); + const unsubscribers = new Set<() => void>(); + let stopped = false; + return { + run(work) { + if (stopped) return; + const attempt: Promise = trackIdleWork(work()) + .catch(() => undefined) + .finally(() => { inFlight.delete(attempt); }); + inFlight.add(attempt); + }, + onStop(unsubscribe) { + if (stopped) unsubscribe(); + else unsubscribers.add(unsubscribe); + }, + async stop() { + stopped = true; + for (const unsubscribe of unsubscribers) unsubscribe(); + unsubscribers.clear(); + while (inFlight.size > 0) await Promise.allSettled([...inFlight]); + }, + }; +} + export interface ApplyManagedEnvironmentsOptions { env?: Record; /** @@ -90,7 +130,7 @@ export interface ApplyManagedEnvironmentsOptions { * Plugin entries also need `call`, to run the plugin's own config * validation; without it every plugin entry is treated as unavailable. */ - workerManager?: Pick & Partial> & { + workerManager?: Pick & Partial> & { getWorker(pluginId: string): | { on(event: "ready", listener: (payload: { pluginId: string }) => void): void; @@ -134,6 +174,13 @@ export interface ApplyManagedEnvironmentsOptions { * startup. When omitted, the list is resolved here the same way. */ autoInstalledPluginKeys?: readonly string[]; + /** + * Owns the post-boot recovery attempts and their `ready` subscriptions: + * each attempt counts as idle-tracked work, and `stop()` (called on + * shutdown before the database closes) unsubscribes and awaits unfinished + * attempts. Defaults to a controller nobody stops (tests). + */ + recovery?: ManagedEnvironmentRecovery; } export interface ManagedPluginEnvironmentReconciliationOutcome { @@ -341,27 +388,50 @@ export async function applyManagedEnvironments( }; // Recovery path for a `ready` plugin whose worker was down at check time: - // that shape is usually a crash in restart-backoff, and the manager's - // respawn re-emits `ready` on the same handle. A one-shot listener re-runs - // the idempotent ensure so the recovered provider becomes selectable again - // without waiting for the next boot (this pass has already archived the - // row). The post-subscribe `isRunning` re-check closes the race where the - // worker recovered between the gate check and the subscription. No handle - // means no respawn is coming (activation never started a worker), so the - // degraded-until-next-boot posture stands. - const scheduleRecoveryReactivation = (pluginId: string, reactivateEntry: () => Promise): void => { + // that shape is usually a crash in restart-backoff, and the manager + // re-emits `ready` when it respawns or replaces the worker. A one-shot + // listener re-runs the idempotent ensure so the recovered provider becomes + // selectable again without waiting for the next boot (this pass has already + // archived the row). The post-subscribe `isRunning` re-check closes the race + // where the worker recovered between the gate check and the subscription. + // Each attempt runs through `recovery`, so idle checks count it and shutdown + // awaits it. + const recovery = opts.recovery ?? createManagedEnvironmentRecovery(); + + /** + * Call `listener` on each `ready` of the plugin's worker. Prefers the + * manager-wide subscription, which survives a restart replacing the worker + * handle (for example a settings save that restarts the worker). Returns + * the unsubscribe function, or null when there is no way to hear a `ready` + * (no manager-wide subscription and no handle: activation never started a + * worker, so no respawn is coming). + */ + const subscribeWorkerReady = (pluginId: string, listener: () => void): (() => void) | null => { const manager = opts.workerManager; + if (manager?.onWorkerReady) { + return manager.onWorkerReady((readyPluginId) => { + if (readyPluginId === pluginId) listener(); + }); + } const handle = manager?.getWorker(pluginId); - if (!handle) return; + if (!handle) return null; + handle.on("ready", listener); + return () => handle.off("ready", listener); + }; + + const scheduleRecoveryReactivation = (pluginId: string, reactivateEntry: () => Promise): void => { let reactivated = false; + let unsubscribe: (() => void) | null = null; const reactivate = (): void => { if (reactivated) return; reactivated = true; - handle.off("ready", reactivate); - void reactivateEntry(); + unsubscribe?.(); + recovery.run(reactivateEntry); }; - handle.on("ready", reactivate); - if (manager?.isRunning(pluginId)) reactivate(); + unsubscribe = subscribeWorkerReady(pluginId, reactivate); + if (!unsubscribe) return; + recovery.onStop(unsubscribe); + if (opts.workerManager?.isRunning(pluginId)) reactivate(); }; const reactivateSandboxEntry = async (spec: ManagedInstanceConfig["environments"][number]): Promise => { @@ -525,37 +595,35 @@ export async function applyManagedEnvironments( * worker that crashes during an attempt (for example inside * `environmentValidateConfig`) leaves the listener armed, so its next * respawn retries. A `ready` that arrives during an attempt queues one more - * attempt. No handle means no respawn is coming (activation never started a - * worker), so the degraded-until-next-boot posture stands. + * attempt. The subscription survives a restart that replaces the worker + * handle (see `subscribeWorkerReady`). */ const schedulePluginRecovery = (spec: PluginSpec, pluginId: string, workerAccess: PluginWorkerAccess): void => { - const manager = opts.workerManager; - const handle = manager?.getWorker(pluginId); - if (!handle) return; const logBase = { name: spec.name, pluginKey: spec.pluginKey, driverKey: spec.driverKey }; let settled = false; let inFlight = false; let queued = false; + let unsubscribe: (() => void) | null = null; const settle = (): void => { settled = true; - handle.off("ready", attempt); + unsubscribe?.(); }; - function attempt(): void { + const attempt = (): void => { if (settled) return; if (inFlight) { queued = true; return; } inFlight = true; - void ensurePluginEntry(spec, workerAccess) - .then((result) => { + recovery.run(async () => { + try { + const result = await ensurePluginEntry(spec, workerAccess); settle(); logger.info( { ...logBase, environmentId: result.environment.id, action: result.action }, "managed plugin environment reactivated after plugin worker recovery", ); - }) - .catch((err: unknown) => { + } catch (err) { if (isPermanentConfigRejection(err)) { settle(); logger.error( @@ -568,17 +636,19 @@ export async function applyManagedEnvironments( "failed to reactivate managed plugin environment; retrying when the plugin worker is ready again", ); } - }) - .finally(() => { + } finally { inFlight = false; if (queued && !settled) { queued = false; attempt(); } - }); - } - handle.on("ready", attempt); - if (manager?.isRunning(pluginId)) attempt(); + } + }); + }; + unsubscribe = subscribeWorkerReady(pluginId, attempt); + if (!unsubscribe) return; + recovery.onStop(unsubscribe); + if (opts.workerManager?.isRunning(pluginId)) attempt(); }; for (const spec of pluginSpecs) { diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index 40ce4870c78..723eb327485 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -832,6 +832,12 @@ export interface PluginWorkerManager { retireStartup?(contextId: string): Promise; /** Subscribe after a worker successfully applies live company configuration. */ onConfigApplied(listener: () => void): () => void; + /** + * Subscribe to every worker `ready`: first start, crash respawn, and a + * replacement handle from a restart. Unlike `getWorker(id).on("ready")`, it + * survives the handle being replaced. Returns the unsubscribe function. + */ + onWorkerReady?(listener: (pluginId: string) => void): () => void; inspectIdleSleep?(hold: { ownerId: string; expiresAt: number }): Promise<{ backgroundWork: "none" | "present" | "unknown"; pluginIds: string[]; catalogGeneration?: string; @@ -4044,6 +4050,7 @@ export function createPluginWorkerManager( }; const startupWorkers = () => new Set([...workers.values(), ...retiredStartupWorkers]); const configApplied = new Set<() => void>(); + const workerReady = new Set<(pluginId: string) => void>(); let activeInspections = 0; /** Per-plugin startup locks to prevent concurrent spawn races. */ const startupLocks = new Map>(); @@ -4063,6 +4070,10 @@ export function createPluginWorkerManager( configApplied.add(listener); return () => { configApplied.delete(listener); }; }, + onWorkerReady(listener) { + workerReady.add(listener); + return () => { workerReady.delete(listener); }; + }, async inspectIdleSleep(hold) { if (startupLocks.size) return { backgroundWork: "unknown", pluginIds: [] }; const snapshot = [...workers.entries()]; @@ -4113,6 +4124,12 @@ export function createPluginWorkerManager( idleWorkerManifests.set(handle, pluginIdleManifestGeneration(options.manifest)); if (existing) retainStartupWorker(existing); workers.set(pluginId, handle); + handle.on("ready", (payload) => { + for (const listener of [...workerReady]) { + try { listener(payload.pluginId); } + catch (err) { log.warn({ err, pluginId: payload.pluginId }, "worker ready listener failed"); } + } + }); // Subscribe to crash/ready events for live event forwarding if (managerOptions?.onWorkerEvent) {