mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-21 00:05:49 +02:00
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 on6ffb682894, 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 on6ffb682894with no open findings - [x] I will address all review comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip authored and GitHub committed
1 parent
4fe45bf8fe
commit
3d1d5294d9
17 files changed
+636
-20
No files matched your search
@@ -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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<PluginEnvironmentValidationResult> {
|
||||
|
||||
@@ -293,6 +293,17 @@ export interface PluginDefinition {
|
||||
*/
|
||||
onShutdown?(): Promise<void>;
|
||||
|
||||
/** 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.
|
||||
*
|
||||
|
||||
@@ -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()();
|
||||
});
|
||||
});
|
||||
@@ -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<typeof setTimeout> | 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<PluginIdleWork>,
|
||||
): Promise<PluginIdleWork> {
|
||||
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); } };
|
||||
}
|
||||
@@ -1330,6 +1330,8 @@ export interface HostToWorkerMethods {
|
||||
health: [params: Record<string, never>, result: PluginHealthDiagnostics];
|
||||
/** @see PLUGIN_SPEC.md §12.5 */
|
||||
shutdown: [params: Record<string, never>, 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",
|
||||
|
||||
@@ -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<string | number, () => void>();
|
||||
let manifest: PaperclipPluginManifestV1 | null = null;
|
||||
let currentConfig: Record<string, unknown> = {};
|
||||
// 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<void> {
|
||||
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<unknown> {
|
||||
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<void>) => {
|
||||
if (notif.paperclipInvocation) {
|
||||
return invocationContextStorage.run(notif.paperclipInvocation, fn);
|
||||
}
|
||||
return fn();
|
||||
const runNotification = async (fn: () => void | Promise<void>) => {
|
||||
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) {
|
||||
|
||||
@@ -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" };
|
||||
},
|
||||
}) });
|
||||
});
|
||||
@@ -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); }
|
||||
|
||||
@@ -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" }],
|
||||
|
||||
@@ -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<void>((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<void>((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(); }
|
||||
});
|
||||
});
|
||||
@@ -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<void>((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<unknown>;
|
||||
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<unknown>;
|
||||
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(); }
|
||||
});
|
||||
});
|
||||
+1
-1
@@ -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));
|
||||
}
|
||||
|
||||
@@ -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<T>(run: () => Promise<T>): Promise<T> {
|
||||
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
|
||||
|
||||
@@ -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<IdleLocalWork> = readIdleLocalWork,
|
||||
inspectPlugins?: InspectIdlePlugins,
|
||||
): Promise<IdleSleepSafety> {
|
||||
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" });
|
||||
|
||||
@@ -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<string | number, PendingRequest>();
|
||||
// Unlike RPC promises, these survive timeouts. A late response is the
|
||||
// completion receipt. A worker crash cannot manufacture one.
|
||||
const unsettledCalls = new Map<string | number, () => 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<string, ActiveInvocation>();
|
||||
// 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<void> {
|
||||
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<void> {
|
||||
const method = request.method as WorkerToHostMethodName;
|
||||
const handler = options.hostHandlers[method] as
|
||||
| ((params: unknown, context?: WorkerHostCallContext) => Promise<unknown>)
|
||||
@@ -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<HostToWorkerMethods[M][1]> {
|
||||
const rpcPromise = new Promise<HostToWorkerMethods[M][1]>((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);
|
||||
},
|
||||
|
||||
|
||||
Reference in new issue
Block a user