From 3d1d5294d994c2c787cabb28b48c6c8de40a7490 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Thu, 8 Oct 2026 12:57:32 -0700 Subject: [PATCH] Drain idle plugin workers before automatic sleep (#15599) ## Thinking Path > - Paperclip runs autonomous work through agents, schedules, and plugins. > - Automatic idle sleep must preserve accepted work and unfinished cleanup. > - Every enabled plugin currently blocks sleep, including unused providers. > - An enabled flag or package version cannot prove that a worker is idle. > - This change adds a live worker drain under the existing owned hold. > - Idle workers can allow sleep while unknown work stays protected. ## Linked Issues or Issue Description Refs #15522. Related: #15391 adds plugin readiness for agent admission; this change concerns instance sleep and does not replace that contract. **Subsystem affected** Plugin worker lifecycle and automatic idle sleep. **Problem or motivation** A workspace with no pending work cannot sleep when any plugin is enabled. Removing that check alone would lose accepted RPCs, background tasks, or cleanup after a caller timeout. **Proposed solution** Require a live `onIdleDrain` handshake from each worker. Close admission in both processes for the exact owner and expiry. Count accepted work until completion. Continue checking durable work separately. ## What Changed - Add bounded worker holds, exact-owner release, automatic expiry, and an abort signal for plugin-owned background work. - Count host and worker requests through their real completion receipts. Keep timed-out work counted. Check active notifications and terminal routes. - Accept enabled plugins only when their current workers provide matching runtime receipts. Missing workers, old SDKs, crashes, invalid replies, and unknown cleanup still prevent sleep. - Let unused Daytona workers opt in. Once a worker contacts the provider, it remains a blocker for that process lifetime. This restriction avoids treating its existing timeout and terminal-close behavior as a cleanup receipt. - Document the plugin author contract. No schema or user-facing API is added. ## Verification - All hosted CI checks passed on 6ffb682894, including all server, runner, workspace, build, typecheck, and end-to-end checks. One unrelated OpenCode transport test hit a five-second timeout and passed on rerun. - `pnpm -r typecheck` passed. - `pnpm build` passed. - Focused Vitest: 293 tests passed across worker admission, real child-process RPC, durable idle checks, and Daytona. - Adjacent worker manager and duplex tests: 119 passed. - Real child-process tests cover late worker completion, a host write after caller timeout, stale owners, closed admission, and release. - SDK tests cover hook failures, busy workers, expiry without another request, and late acknowledgements after expiry. - Initial long local run: 17,202 passed, 10 failed. Five failures involved parent-directory skill fixtures; four of those were already reproduced and passed in an isolated worktree. One HTTP test had a socket hang-up; all 66 tests in that file passed on rerun. The other four used new tests with runtime modules loaded before the follow-up edits; all pass in the fresh focused run above. A clean build and full-suite rerun at the final commit are running in an isolated worktree. No full-suite pass is claimed. - Follow-up: 44 focused tests passed, including repeated owned holds without an intervening normal request. Worker status changes now invalidate an in-progress scan. - Apex follow-up: 13 idle tests, 119 adjacent worker tests, 71 instance-settings route tests, and server typecheck passed. A real queued notification completes its write while preparation is in flight; synchronous and async session callback failures are logged. - Diff whitespace and added-line secret/private-reference scans passed. ## Risks - Plugins are trusted code. A hook must account for work outside SDK RPCs and keep it quiescent until its signal aborts. There is no package-name or manifest-only exemption. - This first change targets unused workers. Used Daytona workers remain awake until remote cleanup receipts are complete. Restarting does not bypass durable lease and recovery checks. - Unknown completion remains a blocker, including after a worker crash. This can retain cost but cannot authorize sleep from a timeout alone. - Enabled jobs, schedules, credentials, integrations, and other durable work remain blockers. Durable wake ownership is separate follow-up work. - Rollback restores the blanket plugin blocker. No data migration is required. ## Model Used OpenAI Codex, GPT-6. Used repository inspection, reasoning, code editing, and command execution. The runtime did not report an exact model revision or context-window size. ## 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 linked existing public issues or described the issue in this PR - [x] I have not referenced internal or instance-local issues or links - [x] My branch name describes the change and contains no internal identifiers - [x] Focused local tests pass; full CI is green (long local rerun status is recorded above) - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation - [x] I have considered and documented risks - [x] All Paperclip CI gates are green - [x] Greptile Apex is 5/5 on 6ffb682894 with no open findings - [x] I will address all review comments before requesting merge --------- Co-authored-by: Paperclip --- doc/plugins/PLUGIN_AUTHORING_GUIDE.md | 32 +++++ .../daytona/src/plugin.test.ts | 15 +++ .../sandbox-providers/daytona/src/plugin.ts | 11 ++ packages/plugins/sdk/src/define-plugin.ts | 11 ++ packages/plugins/sdk/src/idle-drain.test.ts | 55 ++++++++ packages/plugins/sdk/src/idle-drain.ts | 67 ++++++++++ packages/plugins/sdk/src/protocol.ts | 4 + packages/plugins/sdk/src/worker-rpc-host.ts | 51 ++++++-- .../__tests__/fixtures/plugin-worker-idle.cjs | 27 ++++ .../src/__tests__/idle-sleep-safety.test.ts | 16 +++ .../instance-settings-routes.test.ts | 33 ++++- .../src/__tests__/plugin-idle-manager.test.ts | 98 +++++++++++++++ server/src/__tests__/plugin-idle-rpc.test.ts | 89 +++++++++++++ server/src/app.ts | 2 +- server/src/routes/instance-settings.ts | 7 +- server/src/services/idle-sleep-safety.ts | 21 +++- server/src/services/plugin-worker-manager.ts | 117 +++++++++++++++++- 17 files changed, 636 insertions(+), 20 deletions(-) create mode 100644 packages/plugins/sdk/src/idle-drain.test.ts create mode 100644 packages/plugins/sdk/src/idle-drain.ts create mode 100644 server/src/__tests__/fixtures/plugin-worker-idle.cjs create mode 100644 server/src/__tests__/plugin-idle-manager.test.ts create mode 100644 server/src/__tests__/plugin-idle-rpc.test.ts diff --git a/doc/plugins/PLUGIN_AUTHORING_GUIDE.md b/doc/plugins/PLUGIN_AUTHORING_GUIDE.md index a3728720e1c..4030c0104d9 100644 --- a/doc/plugins/PLUGIN_AUTHORING_GUIDE.md +++ b/doc/plugins/PLUGIN_AUTHORING_GUIDE.md @@ -25,6 +25,38 @@ It is intentionally narrower than [PLUGIN_SPEC.md](./PLUGIN_SPEC.md). The spec i building custom versions. - `ctx.assets` is not supported in the current runtime. +## Idle sleep + +An enabled worker blocks automatic idle sleep unless it implements +`onIdleDrain(signal): Promise<"none" | "present" | "unknown">`. This is a live +runtime handshake, not a manifest or version exemption. Older SDKs remain +blockers. Manual shutdown keeps its existing behavior. + +The host checks the exact owned, expiring task-drain hold. It closes new worker +admission and checks outstanding RPCs in both directions and live terminal +routes. The SDK also counts accepted handlers and notifications until they +actually finish. Caller timeouts do not count as completion. An unconfirmed +write, missing worker, crash, invalid reply, or expired hold prevents sleep. + +The hook must return `none` only when plugin-owned timers, sockets, detached +operations, and cleanup are settled and cannot start work until `signal` +aborts. Do not cancel useful work to make a worker idle. Return `present` while +work remains and `unknown` when completion cannot be proved. The SDK closes +ordinary RPC admission while the hook runs and until the matching hold releases +or expires. Release aborts the signal; expiry does so without another request. +Hooks that suspend autonomous work must arrange its safe resumption on abort. + +The handshake does not exempt database work. Enabled jobs, deliveries, +schedules, external ingress, and other persisted work still block sleep until +they have their own safe wake contract. An installed plugin without a matching +live worker receipt also blocks sleep. + +Daytona currently opts in only before its worker has contacted the provider. +After provider access it remains a blocker for that process lifetime, even if +the visible operation finishes. Its timeout and terminal cleanup paths need +stronger completion receipts before that restriction can be relaxed. Restarting +does not bypass the separate persisted lease and recovery checks. + ## Durable resource lifecycle inbox Plugins with `events.subscribe` can read durable, company-scoped resource hooks diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts index ac3a7c1dd0a..6022d9b3fe7 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts @@ -40,6 +40,21 @@ import { environmentCreationCleanupErrorData, readEnvironmentCreationCleanupErro import manifest from "./manifest.js"; import { parseTarVerboseListingLine, splitLinkEntryOnce } from "./file-sync.js"; +describe("Daytona idle eligibility", () => { + it("permits a fresh unused provider but remains conservative after provider access", async () => { + __resetDaytonaSandboxHandleCacheForTest(); + const signal = new AbortController().signal; + expect(await plugin.definition.onIdleDrain!(signal)).toBe("none"); + mockGet.mockRejectedValueOnce(new Error("provider unreachable")); + await plugin.definition.onEnvironmentProbe!({ + driverKey: "daytona", companyId: "fixture-company", environmentId: "fixture-environment", + config: { apiKey: "fixture-not-a-key" }, + }).catch(() => undefined); + expect(await plugin.definition.onIdleDrain!(signal)).toBe("present"); + __resetDaytonaSandboxHandleCacheForTest(); + }); +}); + function createMockSandbox(overrides: { id?: string; name?: string; diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index 4dfa65c9d59..9c085d6a696 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -107,6 +107,10 @@ let timingNow: () => number = () => Date.now(); // read the tracer through `getPluginTracer()`. Before `setup` runs (or in a // test) the tracer is a no-op, so a span never throws. let pluginContext: PluginContext | null = null; +// Provider deadlines and terminal close failures do not prove remote work +// stopped. Until those paths provide complete cleanup receipts, only a worker +// which has never contacted the provider can opt into automatic idle sleep. +let providerUsed = false; /** * Return the plugin tracer. It is the injected `ctx.tracer` after `setup`, or a @@ -313,6 +317,7 @@ function resolveApiKey(config: DaytonaDriverConfig): string { } function createDaytonaClient(config: DaytonaDriverConfig): Daytona { + providerUsed = true; const clientConfig: DaytonaConfig = { apiKey: resolveApiKey(config), }; @@ -1570,6 +1575,7 @@ const sandboxHandleSessionStore = (() => { * Not used in production. */ export function __resetDaytonaSandboxHandleCacheForTest(): void { + providerUsed = false; sandboxHandleCache.reset(); sandboxHandleTeardownGates.reset(); sandboxHandleActivityGates.reset(); @@ -2281,6 +2287,11 @@ const plugin = definePlugin({ return { status: "ok", message: "Daytona sandbox provider plugin healthy" }; }, + async onIdleDrain() { + return providerUsed || daytonaLoginPtyByRoute.size > 0 || daytonaDuplexChannelByRoute.size > 0 + ? "present" : "none"; + }, + async onEnvironmentValidateConfig( params: PluginEnvironmentValidateConfigParams, ): Promise { diff --git a/packages/plugins/sdk/src/define-plugin.ts b/packages/plugins/sdk/src/define-plugin.ts index 947a67cbb55..12b763957c8 100644 --- a/packages/plugins/sdk/src/define-plugin.ts +++ b/packages/plugins/sdk/src/define-plugin.ts @@ -293,6 +293,17 @@ export interface PluginDefinition { */ onShutdown?(): Promise; + /** Prove plugin-owned background work is quiescent for idle sleep. + * The SDK first closes admission and checks accepted RPCs and notifications. + * Return `none` only when timers, sockets, detached operations and cleanup + * are settled and cannot start work until signal aborts. Do not cancel useful + * work to satisfy this check. Missing hooks, failures and uncertainty block + * sleep. The signal aborts on exact-owner release or bounded hold expiry. + * Plugins with autonomous work must keep returning `present` unless they + * can safely suspend and resume that work under this contract. + */ + onIdleDrain?(signal: AbortSignal): Promise<"none" | "present" | "unknown">; + /** * Called to validate the current plugin configuration. * diff --git a/packages/plugins/sdk/src/idle-drain.test.ts b/packages/plugins/sdk/src/idle-drain.test.ts new file mode 100644 index 00000000000..cc59f7881e7 --- /dev/null +++ b/packages/plugins/sdk/src/idle-drain.test.ts @@ -0,0 +1,55 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createPluginIdleDrain } from "./idle-drain.js"; + +describe("plugin idle admission", () => { + afterEach(() => vi.useRealTimers()); + const hold = () => ({ ownerId: "idle-owner", expiresAt: Date.now() + 10_000 }); + + it("keeps accepted work counted until actual settlement", async () => { + const gate = createPluginIdleDrain(); + const done = gate.begin(); + const inspect = vi.fn(async () => "none" as const); + expect(await gate.prepare(hold(), inspect)).toBe("present"); + expect(inspect).not.toHaveBeenCalled(); + done(); done(); + expect(await gate.prepare(hold(), inspect)).toBe("none"); + expect(() => gate.begin()).toThrow("held"); + gate.close(); + }); + + it("rejects legacy hooks, invalid deadlines, busy replies and hook errors", async () => { + const gate = createPluginIdleDrain(); + expect(await gate.prepare(hold())).toBe("present"); + expect(await gate.prepare({ ...hold(), expiresAt: Infinity }, async () => "none")).toBe("unknown"); + expect(await gate.prepare(hold(), async () => "present")).toBe("present"); + expect(await gate.prepare(hold(), async () => { throw new Error("private detail"); })).toBe("unknown"); + gate.begin()(); + }); + + it("releases only the matching owner and resumes on expiry without another request", async () => { + vi.useFakeTimers(); + const gate = createPluginIdleDrain(); + const lease = hold(); + let signal!: AbortSignal; + expect(await gate.prepare(lease, async (value) => { signal = value; return "none"; })).toBe("none"); + gate.release("stale-owner"); + expect(signal.aborted).toBe(false); + expect(await gate.prepare({ ...lease, ownerId: "other-owner" }, async () => "none")).toBe("unknown"); + await vi.advanceTimersByTimeAsync(10_000); + expect(signal.aborted).toBe(true); + gate.begin()(); + }); + + it("holds admission while the hook runs and rejects a late acknowledgement", async () => { + vi.useFakeTimers(); + const gate = createPluginIdleDrain(); + let resolve!: (state: "none") => void; + const pending = gate.prepare(hold(), () => new Promise((done) => { resolve = done; })); + expect(() => gate.begin()).toThrow("held"); + await vi.advanceTimersByTimeAsync(10_000); + expect(await gate.prepare(hold(), async () => "none")).toBe("present"); + resolve("none"); + expect(await pending).toBe("unknown"); + gate.begin()(); + }); +}); diff --git a/packages/plugins/sdk/src/idle-drain.ts b/packages/plugins/sdk/src/idle-drain.ts new file mode 100644 index 00000000000..ae6e4c6be46 --- /dev/null +++ b/packages/plugins/sdk/src/idle-drain.ts @@ -0,0 +1,67 @@ +export type PluginIdleHold = { ownerId: string; expiresAt: number }; +export type PluginIdleWork = "none" | "present" | "unknown"; + +/** Worker admission stays closed for the exact bounded hold. A caller timeout + * never releases a token: only the actual operation's settlement does. */ +export function createPluginIdleDrain() { + let active = 0; + let preparing = 0; + let hold: (PluginIdleHold & { controller: AbortController; ready: boolean }) | null = null; + let timer: ReturnType | undefined; + + function release(ownerId: string): void { + if (hold?.ownerId !== ownerId) return; + const prior = hold; + hold = null; + clearTimeout(timer); + prior.controller.abort(); + } + + function current() { + if (hold && hold.expiresAt <= Date.now()) release(hold.ownerId); + return hold; + } + + function begin(): () => void { + if (current()) throw new Error("Plugin is held for idle sleep"); + active++; + let done = false; + return () => { if (!done) { done = true; active--; } }; + } + + async function prepare( + requested: PluginIdleHold, + inspect?: (signal: AbortSignal) => Promise, + ): Promise { + if (!requested || typeof requested.ownerId !== "string" || !requested.ownerId || + !Number.isFinite(requested.expiresAt) || requested.expiresAt <= Date.now() || + requested.expiresAt > Date.now() + 300_000) return "unknown"; + if (!inspect) return "present"; + const prior = current(); + if (prior) return prior.ownerId === requested.ownerId && prior.expiresAt === requested.expiresAt && prior.ready + ? "none" : "unknown"; + if (active !== 0 || preparing !== 0) return "present"; + const candidate = { ...requested, controller: new AbortController(), ready: false }; + hold = candidate; + timer = setTimeout(() => release(candidate.ownerId), requested.expiresAt - Date.now()); + timer.unref?.(); + preparing++; + try { + const result = await inspect(candidate.controller.signal); + if (current() !== candidate || active !== 0) return "unknown"; + if (result !== "none") { + release(candidate.ownerId); + return result === "present" ? "present" : "unknown"; + } + candidate.ready = true; + return "none"; + } catch { + if (hold === candidate) release(candidate.ownerId); + return "unknown"; + } finally { + preparing--; + } + } + + return { begin, prepare, release, close: () => { if (hold) release(hold.ownerId); } }; +} diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index 76f0cd42284..7b2af372750 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -1330,6 +1330,8 @@ export interface HostToWorkerMethods { health: [params: Record, result: PluginHealthDiagnostics]; /** @see PLUGIN_SPEC.md §12.5 */ shutdown: [params: Record, result: void]; + prepareIdleSleep: [params: { ownerId: string; expiresAt: number }, result: { ownerId: string; expiresAt: number; backgroundWork: "none" | "present" | "unknown" }]; + releaseIdleSleep: [params: { ownerId: string }, result: void]; /** @see PLUGIN_SPEC.md §13.3 */ validateConfig: [params: ValidateConfigParams, result: PluginConfigValidationResult]; /** @see PLUGIN_SPEC.md §13.4 */ @@ -1471,6 +1473,8 @@ export const HOST_TO_WORKER_REQUIRED_METHODS: readonly HostToWorkerMethodName[] /** Optional methods the worker MAY implement. */ export const HOST_TO_WORKER_OPTIONAL_METHODS: readonly HostToWorkerMethodName[] = [ + "prepareIdleSleep", + "releaseIdleSleep", "validateConfig", "configChanged", "onEvent", diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index 786e71954fb..d6781c1b6db 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -1,6 +1,7 @@ import type { AiConnectionRouterRequest, AiConnectionRouterResult } from "@paperclipai/shared"; import { environmentCreationCleanupErrorData } from "./environment-creation-cleanup.js"; import { environmentSyncErrorData } from "./environment-sync-error.js"; +import { createPluginIdleDrain } from "./idle-drain.js"; /** * Worker-side RPC host — runs inside the child process spawned by the host. * @@ -335,6 +336,8 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost let running = true; let initialized = false; + const idleDrain = createPluginIdleDrain(); + const unsettledHostCalls = new Map void>(); let manifest: PaperclipPluginManifestV1 | null = null; let currentConfig: Record = {}; // The company whose config was last applied via configChanged. Used to fail @@ -397,6 +400,9 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost nextOutboundId = 1; } const id = nextOutboundId++; + // Keep the token after an RPC timeout. Only an actual host response + // proves that a host-side write has finished. + unsettledHostCalls.set(id, idleDrain.begin()); const timeout = timeoutMs ?? rpcTimeoutMs; let settled = false; @@ -439,6 +445,8 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost }; sendMessage(request); } catch (err) { + unsettledHostCalls.get(id)?.(); + unsettledHostCalls.delete(id); settle(reject, err instanceof Error ? err : new Error(String(err))); } }); @@ -1563,8 +1571,11 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost */ async function handleHostRequest(request: JsonRpcRequest): Promise { const { id, method, params } = request; - + let done: (() => void) | undefined; try { + if (method !== "prepareIdleSleep" && method !== "releaseIdleSleep" && method !== "shutdown") { + done = idleDrain.begin(); + } const invoke = () => dispatchMethod(method, params); const result = request.paperclipInvocation ? await invocationContextStorage.run(request.paperclipInvocation, invoke) @@ -1584,6 +1595,8 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost method === "environmentAcquireLease" || method === "environmentDestroyLease" ? environmentCreationCleanupErrorData(err) : method === "environmentSyncOut" ? environmentSyncErrorData(err) : undefined)); + } finally { + done?.(); } } @@ -1592,6 +1605,16 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost */ async function dispatchMethod(method: string, params: unknown): Promise { switch (method) { + case "prepareIdleSleep": { + const hold = params as { ownerId: string; expiresAt: number }; + const backgroundWork = !initialized || sessionEventCallbacks.size > 0 + ? "present" + : await idleDrain.prepare(hold, plugin.definition.onIdleDrain?.bind(plugin.definition)); + return { ...hold, backgroundWork }; + } + case "releaseIdleSleep": + idleDrain.release((params as { ownerId: string }).ownerId); + return; case "initialize": return handleInitialize(params as InitializeParams); @@ -1746,6 +1769,7 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost if (plugin.definition.onConfigChanged) supportedMethods.push("configChanged"); if (plugin.definition.onHealth) supportedMethods.push("health"); if (plugin.definition.onShutdown) supportedMethods.push("shutdown"); + if (plugin.definition.onIdleDrain) supportedMethods.push("prepareIdleSleep", "releaseIdleSleep"); if (plugin.definition.onApiRequest) supportedMethods.push("handleApiRequest"); if (plugin.definition.onDetectExternalObjects) supportedMethods.push("detectExternalObjects"); if (plugin.definition.onRouteAiConnection) supportedMethods.push("routeAiConnection"); @@ -2228,6 +2252,9 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost const id = response.id; if (id === null || id === undefined) return; + unsettledHostCalls.get(id)?.(); + unsettledHostCalls.delete(id); + const pending = pendingRequests.get(id); if (!pending) return; @@ -2284,16 +2311,23 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost } else if (isJsonRpcNotification(message)) { // Dispatch host→worker push notifications const notif = message as JsonRpcNotification & { method: string; params?: unknown }; - const runNotification = (fn: () => void | Promise) => { - if (notif.paperclipInvocation) { - return invocationContextStorage.run(notif.paperclipInvocation, fn); - } - return fn(); + const runNotification = async (fn: () => void | Promise) => { + const done = idleDrain.begin(); + try { + await (notif.paperclipInvocation ? invocationContextStorage.run(notif.paperclipInvocation, fn) : fn()); + } finally { done(); } }; - if (notif.method === "agents.sessions.event" && notif.params) { + if (notif.method === "releaseIdleSleep" && notif.params) { + idleDrain.release((notif.params as { ownerId: string }).ownerId); + } else if (notif.method === "agents.sessions.event" && notif.params) { const event = notif.params as AgentSessionEvent; const cb = sessionEventCallbacks.get(event.sessionId); - if (cb) cb(event); + if (cb) void runNotification(() => cb(event)).catch((err) => { + notifyHost("log", { + level: "error", + message: `Failed to handle session event notification: ${err instanceof Error ? err.message : String(err)}`, + }); + }); } else if (notif.method === "onEvent" && notif.params) { // Plugin event bus notifications — dispatch to registered event handlers Promise.resolve(runNotification(() => handleOnEvent(notif.params as OnEventParams))).catch((err) => { @@ -2312,6 +2346,7 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost function cleanup(): void { running = false; + idleDrain.close(); // Close readline if (readline) { diff --git a/server/src/__tests__/fixtures/plugin-worker-idle.cjs b/server/src/__tests__/fixtures/plugin-worker-idle.cjs new file mode 100644 index 00000000000..4b989b447ea --- /dev/null +++ b/server/src/__tests__/fixtures/plugin-worker-idle.cjs @@ -0,0 +1,27 @@ +// Use the real SDK so host tests cross a process and the NDJSON boundary. +void Promise.all([ + import("../../../../packages/plugins/sdk/src/define-plugin.ts"), + import("../../../../packages/plugins/sdk/src/worker-rpc-host.ts"), +]).then(([{ definePlugin }, { startWorkerRpcHost }]) => { + let context; + startWorkerRpcHost({ plugin: definePlugin({ + async setup(ctx) { + context = ctx; + ctx.events.on("fixture.write", async () => { + await ctx.state.set({ scopeKind: "instance", stateKey: "event" }, { value: 1 }); + }); + }, + async onIdleDrain() { return "none"; }, + async onHealth() { + if (process.env.IDLE_TEST_WRITE === "1") { + await context.state.set({ scopeKind: "instance", stateKey: "fixture" }, { value: 1 }); + } + return { status: "ok" }; + }, + async onEnvironmentProbe(params) { + if (params.config.crash) process.exit(1); + await new Promise((resolve) => setTimeout(resolve, params.config.delayMs)); + return { ok: true, summary: "fixture complete" }; + }, + }) }); +}); diff --git a/server/src/__tests__/idle-sleep-safety.test.ts b/server/src/__tests__/idle-sleep-safety.test.ts index 1579247a494..832673aca58 100644 --- a/server/src/__tests__/idle-sleep-safety.test.ts +++ b/server/src/__tests__/idle-sleep-safety.test.ts @@ -245,6 +245,22 @@ if (!support.supported) console.warn(`Skipping idle sleep Postgres tests: ${supp expect(await read()).toEqual(none); }); + it("accepts only exact worker drain receipts and still checks durable work", async () => { + const [{ id }] = await db.execute<{ id: string }>(sql`INSERT INTO plugins (plugin_key, package_name, version, manifest_json) + VALUES ('demo.idle', 'demo-idle', '1.0.0', '{}'::jsonb) RETURNING id`); + const inspect = vi.fn(async () => ({ backgroundWork: "none" as const, pluginIds: [id!] })); + const scan = () => readIdleSleepSafety(db, owned, () => now, ownerId, emptyLocal, inspect); + expect(await scan()).toEqual(none); + expect(inspect).toHaveBeenCalledWith({ ownerId, expiresAt: owned().expiresAt!.getTime() }); + await db.execute(sql`INSERT INTO plugins (plugin_key, package_name, version, manifest_json) + VALUES ('demo.unknown', 'demo-unknown', '1.0.0', '{}'::jsonb)`); + expect(await scan()).toEqual(present); + await db.execute(sql`UPDATE plugins SET status = 'disabled' WHERE plugin_key = 'demo.unknown'`); + const { companyId } = await seed(); + await db.insert(issues).values({ companyId, title: "Still pending", status: "todo" }); + expect(await scan()).toEqual(present); + }); + it("fails closed when the installed schema is older than the report", async () => { await db.execute(sql`ALTER TABLE heartbeat_runs RENAME COLUMN cost_accounting_pending TO hidden_pending`); try { expect(await read()).toEqual(unknown); } diff --git a/server/src/__tests__/instance-settings-routes.test.ts b/server/src/__tests__/instance-settings-routes.test.ts index 2483cd4a3e1..ffee8171aaf 100644 --- a/server/src/__tests__/instance-settings-routes.test.ts +++ b/server/src/__tests__/instance-settings-routes.test.ts @@ -2,6 +2,7 @@ import express from "express"; import request from "supertest"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { hoistModuleGraph } from "./helpers/hoist-module-graph.js"; +import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; const mockInstanceSettingsService = vi.hoisted(() => ({ get: vi.fn(), @@ -76,7 +77,7 @@ describe("instance settings routes", () => { return { errorHandler, instanceSettingsRoutes }; }); - function createApp(actor: any) { + function createApp(actor: any, pluginWorkers?: PluginWorkerManager) { const { errorHandler, instanceSettingsRoutes } = routeModules.value; const app = express(); app.use(express.json()); @@ -84,7 +85,7 @@ describe("instance settings routes", () => { req.actor = actor; next(); }); - app.use("/api", instanceSettingsRoutes(mockDb as any)); + app.use("/api", instanceSettingsRoutes(mockDb as any, pluginWorkers)); app.use(errorHandler); return app; } @@ -972,7 +973,7 @@ describe("instance settings routes", () => { await request(createApp(adminActor)).post("/api/instance/task-drain").send({ purpose: "idle", ttlMs: 60_000 }).expect(200); expect(mockHeartbeatService.computeTaskDrain).toHaveBeenCalledWith({ purpose: "idle", ttlMs: 60_000 }); await request(createApp(adminActor)).get(`/api/instance/task-drain?idleSleepSafety=1&ownerId=${ownerId}`).expect(200); - expect(mockReadIdleSleepSafety).toHaveBeenLastCalledWith(mockDb, expect.any(Function), Date.now, ownerId); + expect(mockReadIdleSleepSafety).toHaveBeenLastCalledWith(mockDb, expect.any(Function), Date.now, ownerId, undefined, undefined); }); it.each([{}, { ttlMs: null }, { ttlMs: 1 }, { ttlMs: 300_001 }])("rejects an unbounded or invalid idle TTL: %j", async body => { @@ -999,10 +1000,34 @@ describe("instance settings routes", () => { const res = await request(createApp(adminActor)).get("/api/instance/task-drain?idleSleepSafety=1"); expect(res.status).toBe(200); expect(res.body).toEqual({ ...idleStatus, idleSleepSafety: report }); - expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, undefined); + expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now, undefined, undefined, undefined); expect(mockReadIdleSleepSafety.mock.calls[0][1]()).toEqual(idleStatus); }); + it("uses the live worker manager for inspection and releases only the owned hold", async () => { + const inspect = vi.fn(async () => ({ backgroundWork: "none", pluginIds: [] })); + const release = vi.fn(); + const workers = { + inspectIdleSleep: inspect, + releaseIdleSleep: release, + } as unknown as PluginWorkerManager; + const ownerId = "fixture-idle-owner"; + const lease = { ownerId, expiresAt: Date.now() + 60_000 }; + mockHeartbeatService.getTaskDrainStatus.mockReturnValue({ ...idleStatus, draining: true, ownerId }); + mockReadIdleSleepSafety.mockImplementation(async (_db, _status, _now, owner, _local, inspectPlugins) => { + expect(owner).toBe(ownerId); + return { version: 1, backgroundWork: (await inspectPlugins(lease)).backgroundWork }; + }); + const app = createApp(adminActor, workers); + await request(app).get(`/api/instance/task-drain?idleSleepSafety=1&ownerId=${ownerId}`).expect(200); + expect(inspect).toHaveBeenCalledWith(lease); + expect(inspect.mock.contexts[0]).toBe(workers); + await request(app).delete("/api/instance/task-drain?ownerId=stale-owner").expect(409); + expect(release).not.toHaveBeenCalled(); + await request(app).delete(`/api/instance/task-drain?ownerId=${ownerId}`).expect(200); + expect(release).toHaveBeenCalledOnce(); + }); + it.each([ ["company member", nonAdminActor], ["agent", { type: "agent", agentId: "agent-1", companyId: "company-1" }], diff --git a/server/src/__tests__/plugin-idle-manager.test.ts b/server/src/__tests__/plugin-idle-manager.test.ts new file mode 100644 index 00000000000..1ea93bc172b --- /dev/null +++ b/server/src/__tests__/plugin-idle-manager.test.ts @@ -0,0 +1,98 @@ +import { fileURLToPath } from "node:url"; +import { createRequire } from "node:module"; +import { setTimeout as delay } from "node:timers/promises"; +import { afterEach, describe, expect, it } from "vitest"; +import { createPluginWorkerHandle } from "../services/plugin-worker-manager.js"; +import { startTaskDrain, stopTaskDrain } from "../services/task-admission.js"; + +const manifest = { id: "test.idle", apiVersion: 1 as const, version: "1.0.0", displayName: "Idle fixture", + description: "Idle fixture", author: "Paperclip", categories: ["automation" as const], capabilities: [], entrypoints: { worker: "worker.cjs" } }; +function worker(write = false, hostHandler = async () => undefined) { + return createPluginWorkerHandle("test.idle", { + entrypointPath: fileURLToPath(new URL("./fixtures/plugin-worker-idle.cjs", import.meta.url)), + execArgv: ["--import", createRequire(import.meta.url).resolve("tsx")], + manifest, config: {}, instanceInfo: { instanceId: "fixture", hostVersion: "1.0.0" }, apiVersion: 1, + env: { IDLE_TEST_WRITE: write ? "1" : "0" }, + hostHandlers: { "state.set": hostHandler, "events.subscribe": async () => ({}) }, + }); +} +function hold() { + const lease = startTaskDrain({ purpose: "idle", ttlMs: 30_000 }); + return { ownerId: lease.ownerId!, expiresAt: lease.expiresAt!.getTime() }; +} + +describe("plugin manager idle receipts", () => { + afterEach(() => stopTaskDrain()); + it("retains timed-out worker operations and admits work after the owned hold releases", async () => { + const handle = worker(); + await handle.start(); + try { + await expect(handle.call("environmentProbe", { driverKey: "fixture", companyId: "fixture", environmentId: "fixture", config: { delayMs: 200 } }, 10)).rejects.toThrow("timed out"); + const lease = hold(); + expect(await handle.prepareIdleSleep!(lease)).toBe("present"); + await expect.poll(() => handle.prepareIdleSleep!(lease)).toBe("none"); + await expect(handle.call("health", {})).rejects.toThrow("held"); + expect(await handle.prepareIdleSleep!({ ...lease, ownerId: "stale" })).toBe("unknown"); + stopTaskDrain(); handle.releaseIdleSleep!(); + expect(await handle.call("health", {})).toMatchObject({ status: "ok" }); + expect(await handle.prepareIdleSleep!(hold())).toBe("none"); + stopTaskDrain(); + // No intervening request or explicit worker release: the next controller + // attempt must discard the worker manager's previous ownership record. + expect(await handle.prepareIdleSleep!(hold())).toBe("none"); + } finally { stopTaskDrain(); await handle.stop(); } + }); + + it("keeps a host write counted after the caller stops waiting", async () => { + let finish!: () => void; + const handle = worker(true, () => new Promise((resolve) => { finish = resolve; })); + await handle.start(); + try { + await expect(handle.call("health", {}, 30)).rejects.toThrow("timed out"); + const lease = hold(); + expect(await handle.prepareIdleSleep!(lease)).toBe("present"); + finish(); + await expect.poll(() => handle.prepareIdleSleep!(lease)).toBe("none"); + } finally { stopTaskDrain(); await handle.stop(); } + }); + + it("lets a queued notification finish its write while idle preparation is in flight", async () => { + let finish!: () => void; + let writes = 0; + const handle = worker(false, () => new Promise((resolve) => { + writes++; + finish = resolve; + })); + await handle.start(); + try { + // Finish setup and its subscription receipt before queuing the event. + await handle.call("health", {}); + handle.notify("onEvent", { event: { eventType: "fixture.write" } }); + // No await: the host has set its hold before it can read the event's + // worker-to-host write. The worker must still be allowed to complete it. + const lease = hold(); + const preparation = handle.prepareIdleSleep!(lease); + expect(await preparation).not.toBe("none"); + await expect.poll(() => writes).toBe(1); + expect(await handle.prepareIdleSleep!(lease)).toBe("present"); + finish(); + await expect.poll(() => handle.prepareIdleSleep!(lease)).toBe("none"); + } finally { finish?.(); stopTaskDrain(); await handle.stop(); } + }); + + it("resumes crash recovery after an idle hold without discarding the restart", async () => { + const handle = worker(); + await handle.start(); + handle.on("crash", () => { hold(); }); + try { + await expect(handle.call("environmentProbe", { driverKey: "fixture", companyId: "fixture", environmentId: "fixture", config: { crash: true } })).rejects.toThrow(); + await delay(1_600); + expect(handle.status).toBe("backoff"); + expect(handle.diagnostics().nextRestartAt).not.toBeNull(); + stopTaskDrain(); + await expect.poll(() => handle.status, { timeout: 5_000 }).toBe("running"); + // Recovery restores service, but lost completion receipts remain unsafe. + expect(await handle.prepareIdleSleep!(hold())).toBe("unknown"); + } finally { stopTaskDrain(); await handle.stop(); } + }); +}); diff --git a/server/src/__tests__/plugin-idle-rpc.test.ts b/server/src/__tests__/plugin-idle-rpc.test.ts new file mode 100644 index 00000000000..c1e626c1b88 --- /dev/null +++ b/server/src/__tests__/plugin-idle-rpc.test.ts @@ -0,0 +1,89 @@ +import { PassThrough } from "node:stream"; +import { describe, expect, it } from "vitest"; +import { definePlugin, type PluginDefinition } from "../../../packages/plugins/sdk/src/define-plugin.js"; +import { startWorkerRpcHost } from "../../../packages/plugins/sdk/src/worker-rpc-host.js"; + +function fixture(definition: PluginDefinition, rpcTimeoutMs = 30_000) { + const stdin = new PassThrough(); + const stdout = new PassThrough(); + const worker = startWorkerRpcHost({ plugin: definePlugin(definition), stdin, stdout, rpcTimeoutMs }); + const messages: any[] = []; + stdout.on("data", (chunk) => { for (const line of String(chunk).trim().split("\n")) messages.push(JSON.parse(line)); }); + let id = 0; + async function call(method: string, params: unknown = {}) { + const callId = ++id; + stdin.write(JSON.stringify({ jsonrpc: "2.0", id: callId, method, params }) + "\n"); + await expect.poll(() => messages.find((message) => message.id === callId && !message.method)).toBeDefined(); + return messages.find((message) => message.id === callId && !message.method); + } + const initialize = () => call("initialize", { manifest: { id: "test.idle", apiVersion: 1 }, config: {}, apiVersion: 1 }); + return { call, initialize, messages, stdin, close: () => worker.stop() }; +} + +describe("plugin idle RPC", () => { + const hold = () => ({ ownerId: "idle-owner", expiresAt: Date.now() + 30_000 }); + + it("seals an idle worker, fences stale releases, and restores normal calls", async () => { + const f = fixture({ async setup() {}, async onIdleDrain() { return "none"; } }); + try { + expect((await f.initialize()).result.supportedMethods).toContain("prepareIdleSleep"); + const lease = hold(); + expect((await f.call("prepareIdleSleep", lease)).result).toEqual({ ...lease, backgroundWork: "none" }); + expect((await f.call("health")).error).toBeDefined(); + await f.call("releaseIdleSleep", { ownerId: "stale" }); + expect((await f.call("health")).error).toBeDefined(); + await f.call("releaseIdleSleep", lease); + expect((await f.call("health")).result.status).toBe("ok"); + } finally { f.close(); } + }); + + it("does not mistake a caller timeout for a completed worker handler", async () => { + let finish!: () => void; + const f = fixture({ async setup() {}, async onIdleDrain() { return "none"; }, + async onHealth() { await new Promise((resolve) => { finish = resolve; }); return { status: "ok" }; } }); + try { + await f.initialize(); + const accepted = f.call("health"); + await expect.poll(() => finish).toBeDefined(); + expect((await f.call("prepareIdleSleep", hold())).result.backgroundWork).toBe("present"); + finish(); await accepted; + expect((await f.call("prepareIdleSleep", hold())).result.backgroundWork).toBe("none"); + } finally { f.close(); } + }); + + it("retains timed-out worker-to-host writes until the host's late receipt", async () => { + let write!: () => Promise; + const f = fixture({ async setup(ctx) { write = () => ctx.state.set({ scopeKind: "instance", stateKey: "fixture" }, { value: 1 }); }, + async onIdleDrain() { return "none"; } }, 10); + try { + await f.initialize(); + await expect(write()).rejects.toThrow("timed out"); + const request = f.messages.find((message) => message.method === "state.set"); + expect((await f.call("prepareIdleSleep", hold())).result.backgroundWork).toBe("present"); + f.stdin.write(JSON.stringify({ jsonrpc: "2.0", id: request.id, result: null }) + "\n"); + expect((await f.call("prepareIdleSleep", hold())).result.backgroundWork).toBe("none"); + } finally { f.close(); } + }); + + it.each([false, true])("logs session callback failures (async: %s)", async (asynchronous) => { + let subscribe!: () => Promise; + const f = fixture({ async setup(ctx) { + subscribe = () => ctx.agents.sessions.sendMessage("fixture-session", "fixture-company", { + prompt: "fixture", + onEvent: asynchronous + ? async () => { throw new Error("fixture callback failure"); } + : () => { throw new Error("fixture callback failure"); }, + }); + } }); + try { + await f.initialize(); + const subscription = subscribe(); + const request = f.messages.find((message) => message.method === "agents.sessions.sendMessage"); + f.stdin.write(JSON.stringify({ jsonrpc: "2.0", id: request.id, result: {} }) + "\n"); + await subscription; + f.stdin.write(JSON.stringify({ jsonrpc: "2.0", method: "agents.sessions.event", params: { sessionId: "fixture-session" } }) + "\n"); + await expect.poll(() => f.messages.find((message) => message.method === "log" && + message.params.level === "error" && message.params.message.includes("fixture callback failure"))).toBeDefined(); + } finally { f.close(); } + }); +}); diff --git a/server/src/app.ts b/server/src/app.ts index 6c9dd7e32ec..7cdebae7f1d 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -834,7 +834,7 @@ export async function createApp( api.use(resourceMembershipRoutes(db)); api.use(primaryAgentRoutes(db)); api.use(inboxDismissalRoutes(db)); - api.use(instanceSettingsRoutes(db)); + api.use(instanceSettingsRoutes(db, workerManager)); if (opts.databaseBackupService) { api.use(instanceDatabaseBackupRoutes(opts.databaseBackupService)); } diff --git a/server/src/routes/instance-settings.ts b/server/src/routes/instance-settings.ts index 1e1fd26c41b..ef578bb7fda 100644 --- a/server/src/routes/instance-settings.ts +++ b/server/src/routes/instance-settings.ts @@ -15,6 +15,7 @@ import { } from "../services/cloud-instance.js"; import { getHiddenSettings } from "../services/settings-visibility.js"; import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; +import type { PluginWorkerManager } from "../services/plugin-worker-manager.js"; import { validate } from "../middleware/validate.js"; import { logger } from "../middleware/logger.js"; import { @@ -119,7 +120,7 @@ function withTaskDrainTransition(run: () => Promise): Promise { return turn; } -export function instanceSettingsRoutes(db: Db) { +export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManager) { const router = Router(); const svc = instanceSettingsService(db); const environments = environmentService(db); @@ -304,7 +305,8 @@ export function instanceSettingsRoutes(db: Db) { // members may read process counters, but not instance-wide work state. assertCanManageInstanceSettings(req); const idleSleepSafety = await readIdleSleepSafety(db, () => heartbeat.getTaskDrainStatus(), Date.now, - typeof req.query.ownerId === "string" ? req.query.ownerId : undefined); + typeof req.query.ownerId === "string" ? req.query.ownerId : undefined, undefined, + pluginWorkers?.inspectIdleSleep?.bind(pluginWorkers)); res.json({ ...heartbeat.getTaskDrainStatus(), idleSleepSafety }); return; } @@ -418,6 +420,7 @@ export function instanceSettingsRoutes(db: Db) { ), ); heartbeat.stopTaskDrain(); + pluginWorkers?.releaseIdleSleep?.(); // See the POST handler above for why a publish failure here is // swallowed instead of failing the route: the audit record already // committed, so a publish failure here must not undo a drain-stop diff --git a/server/src/services/idle-sleep-safety.ts b/server/src/services/idle-sleep-safety.ts index 574b57c11f9..ad24e3037d1 100644 --- a/server/src/services/idle-sleep-safety.ts +++ b/server/src/services/idle-sleep-safety.ts @@ -17,6 +17,11 @@ export interface IdleSleepSafety { backgroundWork: "none" | "present" | "unknown"; } +export type InspectIdlePlugins = (hold: { ownerId: string; expiresAt: number }) => Promise<{ + backgroundWork: "none" | "present" | "unknown"; + pluginIds: string[]; +}>; + // These are deliberately conservative. Completed agent runs and workspace // operations are history, not reasons to keep an otherwise idle instance up. // Other background features remain awake until their work and inbound events @@ -91,18 +96,28 @@ export async function readIdleSleepSafety( now: () => number = Date.now, ownerId?: string, inspectLocalWork: () => Promise = readIdleLocalWork, + inspectPlugins?: InspectIdlePlugins, ): Promise { const unknown: IdleSleepSafety = { version: 1, backgroundWork: "unknown" }; const before = getDrainStatus(); const localBefore = idleWorkSnapshot(); if (!sameQuietHold(before, before, now())) return unknown; try { + let quietPlugins: string[] = []; + if (inspectPlugins) { + if (!ownerId || ownerId !== before.ownerId || before.expiresAt === null || localBefore.active !== 0) return unknown; + const pluginWork = await inspectPlugins({ ownerId, expiresAt: before.expiresAt.getTime() }); + if (pluginWork.backgroundWork !== "none") return { version: 1, backgroundWork: pluginWork.backgroundWork }; + quietPlugins = pluginWork.pluginIds; + } const blocked = await db.transaction(async (tx) => { await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); const checks = WORK_CHECKS.map((query) => sql`EXISTS (${sql.raw(query)})`); - // Version labels and manifest declarations do not prove arbitrary worker - // code has no background activity. No plugin approvals are supported. - checks.push(sql`EXISTS (SELECT 1 FROM plugins WHERE status <> 'disabled')`); + // Only the current worker's owned runtime drain can exempt a plugin. + // Missing workers, old SDKs, and newly installed plugins still block. + checks.push(quietPlugins.length === 0 + ? sql`EXISTS (SELECT 1 FROM plugins WHERE status <> 'disabled')` + : sql`EXISTS (SELECT 1 FROM plugins WHERE status <> 'disabled' AND id NOT IN (${sql.join(quietPlugins.map((id) => sql`${id}::uuid`), sql`, `)}))`); const rows = await tx.execute<{ blocked: boolean }>(sql`SELECT ${sql.join(checks, sql` OR `)} AS blocked`); return rows.length === 1 && typeof rows[0]?.blocked === "boolean" ? rows[0].blocked : undefined; }, { isolationLevel: "repeatable read", accessMode: "read only" }); diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index af6721f3d4b..2d3f961d530 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -20,6 +20,7 @@ import { fork, type ChildProcess } from "node:child_process"; import { randomUUID } from "node:crypto"; +import { beginIdleTrackedWork, readTaskDrain, trackIdleWork } from "./task-admission.js"; import { EventEmitter } from "node:events"; import { createInterface, type Interface as ReadlineInterface } from "node:readline"; import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; @@ -676,6 +677,8 @@ interface ExecuteLogRoute { * with exponential backoff automatically when `autoRestart` is enabled. */ export interface PluginWorkerHandle { + prepareIdleSleep?(hold: { ownerId: string; expiresAt: number }): Promise<"none" | "present" | "unknown">; + releaseIdleSleep?(): void; /** The plugin ID this worker serves. */ readonly pluginId: string; @@ -793,6 +796,10 @@ export interface WorkerDiagnostics { * for starting/stopping all workers and routing RPC calls. */ export interface PluginWorkerManager { + inspectIdleSleep?(hold: { ownerId: string; expiresAt: number }): Promise<{ + backgroundWork: "none" | "present" | "unknown"; pluginIds: string[]; + }>; + releaseIdleSleep?(): void; /** * Register and start a worker for a plugin. * @@ -888,6 +895,30 @@ export function createPluginWorkerHandle( // Pending RPC requests awaiting a response const pendingRequests = new Map(); + // Unlike RPC promises, these survive timeouts. A late response is the + // completion receipt. A worker crash cannot manufacture one. + const unsettledCalls = new Map void>(); + let activeHostHandlers = 0; + let idleHold: { ownerId: string; expiresAt: number; ready: boolean } | null = null; + + function releaseIdleSleep(): void { + const prior = idleHold; + idleHold = null; + if (prior && childProcess?.stdin?.writable) { + try { + sendMessage({ jsonrpc: JSONRPC_VERSION, method: "releaseIdleSleep", params: { ownerId: prior.ownerId } }); + } catch { /* Worker-side expiry still bounds a lost release. */ } + } + } + + function idleSleepHeld(): boolean { + if (!idleHold) return false; + const drain = readTaskDrain(new Date()); + if (drain?.ownerId !== idleHold.ownerId || drain.expiresAt?.getTime() !== idleHold.expiresAt) { + releaseIdleSleep(); + } + return idleHold !== null; + } let nextRequestId = 1; const activeInvocations = new Map(); // Host-owned execute routes, keyed by the host-issued invocation id. Only an @@ -1020,6 +1051,8 @@ export function createPluginWorkerHandle( function setStatus(newStatus: WorkerStatus): void { const prev = status; if (prev === newStatus) return; + const done = beginIdleTrackedWork(); + done(); status = newStatus; log.debug({ from: prev, to: newStatus }, "worker status change"); emitter.emit("status", { pluginId, status: newStatus, previousStatus: prev }); @@ -1093,6 +1126,8 @@ export function createPluginWorkerHandle( return; } + unsettledCalls.get(id)?.(); + unsettledCalls.delete(id); const pending = pendingRequests.get(id); if (!pending) { log.warn({ id }, "received response for unknown request id"); @@ -2695,6 +2730,23 @@ export function createPluginWorkerHandle( * Handle a JSON-RPC request from the worker (worker→host call). */ async function handleWorkerRequest(request: JsonRpcRequest): Promise { + const done = beginIdleTrackedWork(); + activeHostHandlers++; + try { + // A queued notification may already be running in the worker while its + // prepare request is in flight. Let its writes finish until that worker + // has positively acknowledged that all accepted handlers have drained. + if (idleSleepHeld() && idleHold?.ready) { + try { + sendMessage(createErrorResponse(request.id, PLUGIN_RPC_ERROR_CODES.WORKER_UNAVAILABLE, "Plugin is held for idle sleep")); + } catch { /* Worker may have exited after sending its request. */ } + return; + } + await dispatchWorkerRequest(request); + } finally { activeHostHandlers--; done(); } + } + + async function dispatchWorkerRequest(request: JsonRpcRequest): Promise { const method = request.method as WorkerToHostMethodName; const handler = options.hostHandlers[method] as | ((params: unknown, context?: WorkerHostCallContext) => Promise) @@ -3072,6 +3124,12 @@ export function createPluginWorkerHandle( backoffTimer = setTimeout(async () => { backoffTimer = null; nextRestartAt = null; + // A temporary idle hold must not consume the crash-recovery attempt. + // Keep the bounded backoff until admission reopens. + if (readTaskDrain(new Date())?.ownerId) { + scheduleRestart(); + return; + } try { await startInternal(); } catch (err) { @@ -3293,6 +3351,11 @@ export function createPluginWorkerHandle( executeLogSink?: ExecuteLogSink, ): Promise { const rpcPromise = new Promise((resolve, reject) => { + const idleControl = method === "prepareIdleSleep" || method === "releaseIdleSleep"; + if (!idleControl && method !== "shutdown" && idleSleepHeld()) { + reject(new Error("Plugin is held for idle sleep")); + return; + } if (!childProcess?.stdin?.writable) { reject( new Error( @@ -3303,6 +3366,7 @@ export function createPluginWorkerHandle( } const id = nextRequestId++; + if (!idleControl) unsettledCalls.set(id, beginIdleTrackedWork()); const timeout = resolveRpcCallTimeoutMs(timeoutMs, rpcTimeoutMs); const invocationScope = deriveInvocationScope(method, params); const invocation = invocationScope ? registerInvocation(invocationScope) : null; @@ -3368,6 +3432,8 @@ export function createPluginWorkerHandle( } catch (err) { clearTimeout(timer); pendingRequests.delete(id); + unsettledCalls.get(id)?.(); + unsettledCalls.delete(id); clearInvocation(invocation); clearExecuteRoute(invocation?.id); reject( @@ -3394,6 +3460,38 @@ export function createPluginWorkerHandle( // ----------------------------------------------------------------------- const handle: PluginWorkerHandle = { + releaseIdleSleep, + async prepareIdleSleep(hold) { + const drain = readTaskDrain(new Date()); + if (drain?.ownerId !== hold.ownerId || drain.expiresAt?.getTime() !== hold.expiresAt || + hold.expiresAt <= Date.now()) return "unknown"; + // Clear a previous hold even when this idle worker received no ordinary + // call between two controller attempts. + idleSleepHeld(); + if (status !== "running" || totalCrashes > 0) return "unknown"; + if (!supportedMethods.includes("prepareIdleSleep")) return "present"; + if (unsettledCalls.size || activeHostHandlers || loginPtyRoutesByHostRouteId.size || + liveDuplexRoutes.size || openingDuplexRoutes.size || terminalDuplexRoutes.size) return "present"; + if (idleHold && (idleHold.ownerId !== hold.ownerId || idleHold.expiresAt !== hold.expiresAt)) return "unknown"; + const candidate = idleHold ??= { ...hold, ready: false }; + try { + const result = await callInternal("prepareIdleSleep", hold, Math.min(5_000, hold.expiresAt - Date.now())); + if (!idleSleepHeld() || idleHold !== candidate || status !== "running" || unsettledCalls.size || activeHostHandlers || + result.ownerId !== hold.ownerId || result.expiresAt !== hold.expiresAt) { + if (idleHold === candidate) releaseIdleSleep(); + return "unknown"; + } + if (result.backgroundWork === "none") { + candidate.ready = true; + return "none"; + } + releaseIdleSleep(); + return result.backgroundWork === "present" ? "present" : "unknown"; + } catch { + if (idleHold === candidate) releaseIdleSleep(); + return "unknown"; + } + }, get pluginId() { return pluginId; }, @@ -3459,6 +3557,7 @@ export function createPluginWorkerHandle( notify(method: string, params: unknown) { if (status !== "running") return; + if (idleSleepHeld()) throw new Error("Plugin is held for idle sleep"); const invocationScope = deriveInvocationScope(method, params); // Notifications have no response to settle on, so the invocation scope // is GC'd by TTL. Call-path invocations are registered without a TTL and @@ -3637,6 +3736,20 @@ export function createPluginWorkerManager( ); return { + async inspectIdleSleep(hold) { + if (startupLocks.size) return { backgroundWork: "unknown", pluginIds: [] }; + const snapshot = [...workers.entries()]; + const results = await Promise.all(snapshot.map(async ([, worker]) => + worker.prepareIdleSleep ? worker.prepareIdleSleep(hold) : "unknown")); + if (startupLocks.size || snapshot.length !== workers.size || snapshot.some(([id, worker]) => workers.get(id) !== worker || worker.status !== "running")) { + return { backgroundWork: "unknown", pluginIds: [] }; + } + const backgroundWork = results.includes("present") ? "present" : results.includes("unknown") ? "unknown" : "none"; + return { backgroundWork, pluginIds: backgroundWork === "none" ? snapshot.map(([id]) => id) : [] }; + }, + releaseIdleSleep() { + for (const worker of workers.values()) worker.releaseIdleSleep?.(); + }, async startWorker( pluginId: string, options: WorkerStartOptions, @@ -3690,7 +3803,7 @@ export function createPluginWorkerManager( log.info({ pluginId }, "starting plugin worker"); // Set the lock before awaiting start() to prevent concurrent spawns - const startPromise = handle.start().then(() => handle).finally(() => { + const startPromise = trackIdleWork(handle.start()).then(() => handle).finally(() => { startupLocks.delete(pluginId); }); startupLocks.set(pluginId, startPromise); @@ -3706,7 +3819,7 @@ export function createPluginWorkerManager( } log.info({ pluginId }, "stopping plugin worker"); - await handle.stop(); + await trackIdleWork(handle.stop()); workers.delete(pluginId); },