diff --git a/doc/idle-sleep-safety.md b/doc/idle-sleep-safety.md new file mode 100644 index 0000000000..61f0977f04 --- /dev/null +++ b/doc/idle-sleep-safety.md @@ -0,0 +1,98 @@ +# Automatic idle sleep safety + +`GET /api/instance/task-drain?idleSleepSafety=1&ownerId=` adds an +instance-wide `idleSleepSafety` report for instance administrators and signed +Cloud control callers. Ordinary task drains keep their existing behavior. +Company members and agents cannot read this report. + +## Hosting protocol + +1. Hold external admission at the hosting layer. Queue user requests for wake or + abort sleep when activity arrives. Do not send traffic into an instance that + is being stopped. +2. Send `POST /api/instance/task-drain` with + `{ "purpose": "idle", "ttlMs": 120000 }`. Idle holds require a TTL from 5 to + 300 seconds. Save the returned `ownerId` and `expiresAt`. An existing drain + returns 409. An unsupported purpose must leave automatic sleep disabled. +3. Poll the report with that exact owner. Only + `{ "version": 1, "backgroundWork": "none" }` permits sleep. Require the same + `ownerId`, `draining: true`, zero run/wake counters, and enough remaining TTL + for the provider stop. `present`, `unknown`, missing fields, or errors must + keep the instance awake. Do not interpret `quiescent` alone as authorization. +4. Revalidate the same hold immediately before stopping the instance. Keep + external admission closed through the provider stop. Never use an old report + after a restart, expired hold, changed owner, or failed validation. +5. If sleep is abandoned, send `DELETE /api/instance/task-drain?ownerId=`. + A missing or stale owner cannot release another controller's idle hold. The + TTL clears an abandoned hold if the controller loses its response. + +An idle hold rejects new tenant HTTP requests and WebSocket upgrades with 503 +and `Retry-After: 1`. Health checks and authenticated task-drain control remain +available. Ordinary deploy drains do not enable this HTTP gate. The host must +handle retry/wake at its ingress before enabling automatic sleep. +Health and control mutations remain tracked through completion, including +signed bootstrap writes. Only the task-drain read skips request tracking. +Its authentication finishes under a separate work token before entering the +report, so it counts concurrent writes without blocking on its own scan. + +## Accepted work and cleanup + +HTTP admission is installed before body parsing, authentication, webhooks and +MCP ingress. The Express adapter tracks handler promises in nested routers and +error middleware. Sending a response or losing the client does not complete an +async handler. A callback-only handler that abandons its response without ending it remains a +sleep blocker. An async route that settles after a disconnect releases its +request token; normal cancellation does not permanently block sleep. New routes must be registered before this adapter is installed. +Fire-and-forget work must have durable queue state or use `trackIdleWork`; a +response is never a substitute for tracking that work. +Accepted live-event WebSocket authentication remains counted until its promise +settles, including after the raw socket closes. Detached startup reconciliation +for built-in agents, managed homes and persisted runtime services also remains +counted after readiness until its writes settle. + +Scheduler work already in flight remains counted until its promise settles. +Idle holds pause new scheduler admissions. Database backup promises remain +counted through success or failure. Configured periodic backups also block +sleep until a host owns a durable wake schedule for them. A generation counter invalidates a +scan if tracked work both starts and finishes during inspection. Reports remain +unknown until startup recovery and HTTP tracking are installed. + +The report inspects accounting and sandbox-cleanup spools directly. Any entry, +including malformed JSON, temporary files and failed write probes, blocks sleep. +Read errors and invalid spool paths return unknown. Recovery readers that skip +bad records cannot certify an empty spool. + +An orphan cleanup token starts before its spool write and remains held through +buffering, queue splices, flush retries and buffer overflow. It ends only after +the database owns cleanup. Queued usage receipts stay counted until capture +settles. If receipt capture and the database failure fence both fail, a token +continues to block sleep until the failure fence is durably written. A query +that updates zero rows does not clear that debt. Unrecoverable cleanup requires +operator reconciliation; this API does not discard it to save compute. + +## Persisted work and limits + +The report checks all companies in one bounded read-only database transaction. +Queued and orphaned runs, timers, retries, accounting debt, unfinished issues, +cleanup, active routines, integrations, external API credentials, and plugin +work block sleep. Completed ordinary run history does not. Less common work +sources conservatively block on any retained records. Active issue watchdogs +block sleep even when their watched issue is complete and no review has started. +Missing migrations, malformed configuration and database errors produce unknown +without exposing SQL or tenant data. + +Every enabled plugin blocks sleep. A version label does not prove that arbitrary +worker code has no background activity. No plugin approval or configuration +bypass is supported. Scheduled and externally triggered work needs a durable +host wake mechanism before this restriction can be relaxed. + +This is a single-process admission protocol. A restart clears the owner. The +host must serialize sleep with deploy/wake operations and bind its final check +to the instance it stops. An unexpected restart after the final check can still +race an external provider stop; a process-local report cannot fence that remote +operation by itself. Multiple replicas require shared fencing. The protocol +does not make existing in-memory-only debt survive an unplanned process crash. + +Related public PR #13413 adds runtime-service controller requirements and uses +the same `purpose: "idle"` / `ownerId` vocabulary. This report conservatively +blocks all retained runtime-service state instead of approving such services. diff --git a/packages/shared/src/validators/instance.ts b/packages/shared/src/validators/instance.ts index ee5a3fd3fa..b1117dbaf4 100644 --- a/packages/shared/src/validators/instance.ts +++ b/packages/shared/src/validators/instance.ts @@ -129,8 +129,11 @@ export const patchInstanceSettingsSchema = z.object({ export const MAX_TASK_DRAIN_TTL_MS = 24 * 60 * 60 * 1000; export const startTaskDrainRequestSchema = z.object({ + purpose: z.literal("idle").optional(), ttlMs: z.number().int().positive().max(MAX_TASK_DRAIN_TTL_MS).nullable().optional(), -}).strict(); +}).strict().refine(value => value.purpose !== "idle" || + (typeof value.ttlMs === "number" && value.ttlMs >= 5_000 && value.ttlMs <= 300_000), + { message: "Idle holds require ttlMs between 5000 and 300000", path: ["ttlMs"] }); export type InstanceGeneralSettings = z.infer; // The patch schema removes each default so an absent key stays absent. Declare diff --git a/server/src/__tests__/cloud-runtime-identity.test.ts b/server/src/__tests__/cloud-runtime-identity.test.ts index 08749398ea..be32714d0a 100644 --- a/server/src/__tests__/cloud-runtime-identity.test.ts +++ b/server/src/__tests__/cloud-runtime-identity.test.ts @@ -18,6 +18,8 @@ import { import { routineWebhookUrl } from "../services/routines.js"; import { paperclipCloudConnectorEnrollmentStatus } from "../services/paperclip-cloud-connector-enrollment.js"; import { cloudRuntimeIdentityMiddleware } from "../middleware/cloud-runtime-identity.js"; +import { idleAdmissionMiddleware, trackIdleRequestHandlers } from "../middleware/idle-admission.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; import { healthRoutes } from "../routes/health.js"; import { getEmbeddedPostgresTestSupport, @@ -186,6 +188,37 @@ describeEmbeddedPostgres("Cloud runtime identity", () => { }); }); + it.each([false, true])("counts exempt bootstrap writes until they settle (failure=%s)", async (fail) => { + let enterWrite!: () => void, finishWrite!: () => void; + const entered = new Promise(resolve => { enterWrite = resolve; }); + const pending = new Promise(resolve => { finishWrite = resolve; }); + const transaction = db.transaction.bind(db); + vi.spyOn(db, "transaction").mockImplementationOnce(async (...args) => { + enterWrite(); + await pending; + if (fail) throw new Error("fixture bootstrap write failed"); + return transaction(...args); + }); + const app = express(); + app.use(idleAdmissionMiddleware); + app.use(cloudRuntimeIdentityMiddleware(db)); + app.get("/api/health", (_req, res) => res.sendStatus(204)); + trackIdleRequestHandlers(app); + const requestTime = Math.floor(Date.now() / 1000); + const accepted = request(app).get("/api/health") + .set("x-paperclip-cloud-runtime-identity", assertion({ claims: { iat: requestTime, exp: requestTime + 300 } })) + .then(response => response); + try { + await entered; + expect(idleWorkSnapshot().active).toBe(1); + } finally { + finishWrite(); + const response = await accepted; + expect(response.status).toBe(fail ? 401 : 204); + } + expect(idleWorkSnapshot().active).toBe(0); + }); + it("keeps probes idle until a verified durable claim and stays active across restart", async () => { const env = { ...process.env, PAPERCLIP_CLOUD_WARM_STANDBY: "1" }; const isStandby = await createCloudWarmStandby(db, env); diff --git a/server/src/__tests__/environment-runtime.test.ts b/server/src/__tests__/environment-runtime.test.ts index 14a0327f11..216e213666 100644 --- a/server/src/__tests__/environment-runtime.test.ts +++ b/server/src/__tests__/environment-runtime.test.ts @@ -1,3 +1,4 @@ +import { idleWorkSnapshot } from "../services/task-admission.js"; import { createHash, randomUUID } from "node:crypto"; import { mkdir, mkdtemp, readdir, rm, writeFile } from "node:fs/promises"; import os from "node:os"; @@ -2372,6 +2373,11 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { }); it("buffers the orphan in-process when the durable spool write also fails", async () => { + const idleBefore = idleWorkSnapshot().active; + let releaseFlush!: () => void; + let enteredFlush!: () => void; + const flushGate = new Promise(resolve => { releaseFlush = resolve; }); + const flushEntered = new Promise(resolve => { enteredFlush = resolve; }); const { companyId, environment, runId } = await seedEnvironment({ driver: "sandbox", name: "Foreign-bound Fake Sandbox Cleanup Spool Unwritable", @@ -2413,7 +2419,8 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { if (databaseDown) { return Promise.reject(new Error("pending-cleanup write failed; database down")); } - return real.insertPendingCleanupLease(input); + enteredFlush(); + return flushGate.then(() => real.insertPendingCleanupLease(input)); }, }; }); @@ -2454,11 +2461,18 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { "sandbox_orphan_cleanup_write_failed", ); expect(fallbackLog?.[0]).toMatchObject({ persisted: false, buffered: true }); + expect(idleWorkSnapshot().active).toBe(idleBefore + 1); // The same runtime still keeps the orphan in-process, so a flush after the // database recovers lands the durable row. databaseDown = false; - const flushed = await runtime.flushDeferredOrphanCleanups(); + const flushing = runtime.flushDeferredOrphanCleanups(); + await flushEntered; + // The batch was spliced out of the queue, but the DB does not own it yet. + expect(idleWorkSnapshot().active).toBe(idleBefore + 1); + releaseFlush(); + const flushed = await flushing; + expect(idleWorkSnapshot().active).toBe(idleBefore); expect(flushed).toEqual({ recovered: 1, pending: 0 }); const rows = await db .select() @@ -2467,6 +2481,7 @@ describeEmbeddedPostgres("environmentRuntimeService", () => { expect(rows).toHaveLength(1); expect(rows[0]?.status).toBe("pending_cleanup"); } finally { + releaseFlush(); logSpy.mockRestore(); factorySpy.mockRestore(); destroySpy.mockRestore(); diff --git a/server/src/__tests__/idle-accounting-work.test.ts b/server/src/__tests__/idle-accounting-work.test.ts new file mode 100644 index 0000000000..31cde3373d --- /dev/null +++ b/server/src/__tests__/idle-accounting-work.test.ts @@ -0,0 +1,76 @@ +import { randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { agents, companies, createDb, heartbeatRuns, type Db } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { createRunUsageRecorder } from "../services/usage-receipts.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("idle accounting work", () => { + let database: Awaited>; + let db: ReturnType; + const directories: string[] = []; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("idle-accounting-"); + db = createDb(database.connectionString); + }, 60_000); + afterEach(async () => { + await db.execute(sql`TRUNCATE companies CASCADE`); + for (const directory of directories.splice(0)) await fs.rm(directory, { force: true, recursive: true }); + }); + afterAll(async () => { await database?.cleanup(); }); + + async function recorder() { + const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Accounting idle test", issuePrefix: "ACCT" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Test", role: "engineer", status: "idle", adapterType: "process" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", status: "running" }); + const directory = await fs.mkdtemp(path.join(os.tmpdir(), "idle-accounting-")); + directories.push(directory); + let offline = false; + const proxy = new Proxy(db, { get(target, property) { + if (property === "transaction" && offline) return () => Promise.reject(new Error("Database unavailable")); + const value = Reflect.get(target, property); + return typeof value === "function" ? value.bind(target) : value; + } }) as Db; + const recorder = await createRunUsageRecorder(proxy, { companyId, runId, adapterType: "process" }, directory); + return { recorder, directory, runId, setOffline: (value: boolean) => { offline = value; } }; + } + + it("counts queued receipts until capture and accounting writes settle", async () => { + const before = idleWorkSnapshot().active; + const { recorder: usage } = await recorder(); + const first = usage.capture({ complete: false, usage: { inputTokens: 1, outputTokens: 1 } }); + const second = usage.capture({ complete: true, usage: { inputTokens: 2, outputTokens: 2 } }); + expect(idleWorkSnapshot().active).toBe(before + 2); + await Promise.all([first, second]); + await usage.persistFailure(); + expect(idleWorkSnapshot().active).toBe(before); + }); + + it("keeps unpersisted capture failure as a blocker until its failure fence reaches the database", async () => { + const before = idleWorkSnapshot().active; + const { recorder: usage, directory, runId, setOffline } = await recorder(); + await fs.rm(directory, { recursive: true }); + await fs.writeFile(directory, "unwritable spool"); + setOffline(true); + await expect(usage.capture({ complete: true, usage: { inputTokens: 1, outputTokens: 1 } })).rejects.toThrow(); + await expect(usage.persistFailure()).rejects.toThrow(); + expect(idleWorkSnapshot().active).toBe(before + 1); + setOffline(false); + // A successful query which updates zero rows is not a durable fence. + await db.update(heartbeatRuns).set({ costAccountedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + await usage.persistFailure(); + expect(idleWorkSnapshot().active).toBe(before + 1); + await db.update(heartbeatRuns).set({ costAccountedAt: null }).where(eq(heartbeatRuns.id, runId)); + await usage.persistFailure(); + expect(idleWorkSnapshot().active).toBe(before); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(run.costAccountingPending).toBe(true); + expect(run.usageJson?.accountingCaptureFailed).toBe(true); + }); +}); diff --git a/server/src/__tests__/idle-admission.test.ts b/server/src/__tests__/idle-admission.test.ts new file mode 100644 index 0000000000..81f268a1ff --- /dev/null +++ b/server/src/__tests__/idle-admission.test.ts @@ -0,0 +1,103 @@ +import express from "express"; +import request from "supertest"; +import { request as httpRequest } from "node:http"; +import { afterEach, describe, expect, it } from "vitest"; +import { idleAdmissionMiddleware, trackIdleRequestHandlers } from "../middleware/idle-admission.js"; +import { idleWorkSnapshot, startTaskDrain, stopTaskDrain, readTaskDrain, beginIdleTrackedWork } from "../services/task-admission.js"; + +function deferred() { + let resolve!: () => void; + const promise = new Promise(done => { resolve = done; }); + return { promise, resolve }; +} +function app() { const app = express(); app.use(idleAdmissionMiddleware); return app; } +afterEach(() => stopTaskDrain()); + +describe("idle admission", () => { + it("preserves legacy drains, but blocks all new work during an owned idle hold", async () => { + const server = app(); + server.all("/{*path}", (_req, res) => res.sendStatus(204)); + trackIdleRequestHandlers(server); + startTaskDrain({ ttlMs: 60_000 }); + await request(server).post("/api/issues").expect(204); + startTaskDrain({ purpose: "idle", ttlMs: 60_000 }); + for (const path of ["/api/issues", "/api/chat-webhooks/test", "/mcp", "/api/auth/callback"]) { + await request(server).get(path).expect(503).expect("Retry-After", "1"); + await request(server).post(path).expect(503); + } + await request(server).get("/api/health").expect(204); + await request(server).get("/api/instance/task-drain/").expect(204); + await request(server).post("/api/instance/task-drain").expect(204); + stopTaskDrain(); + await request(server).post("/api/issues").expect(204); + expect(idleWorkSnapshot().active).toBe(0); + }); + + it("counts a nested async handler after it sends a response", async () => { + const server = app(); + const router = express.Router(); + const done = deferred(); + router.post("/work", async (_req, res) => { res.sendStatus(202); await done.promise; }); + server.use("/api", router); + trackIdleRequestHandlers(server); + await request(server).post("/api/work").expect(202); + expect(idleWorkSnapshot().active).toBe(1); + startTaskDrain({ purpose: "idle", ttlMs: 60_000 }); + await request(server).post("/api/work").expect(503); + done.resolve(); + await new Promise(resolve => setImmediate(resolve)); + expect(idleWorkSnapshot().active).toBe(0); + }); + + it("does not mistake a client disconnect for completed accepted work", async () => { + const server = app(); + const entered = deferred(), done = deferred(), closed = deferred(); + server.post("/work", async (_req, res) => { res.once("close", closed.resolve); entered.resolve(); await done.promise; /* client has left */ }); + trackIdleRequestHandlers(server); + const listener = server.listen(0, "127.0.0.1"); + await new Promise(resolve => listener.once("listening", resolve)); + const address = listener.address() as { port: number }; + const client = httpRequest({ host: "127.0.0.1", port: address.port, path: "/work", method: "POST" }); + client.on("error", () => undefined); + client.end(); + try { + await entered.promise; + client.destroy(); + await closed.promise; + startTaskDrain({ purpose: "idle", ttlMs: 60_000 }); + expect(idleWorkSnapshot().active).toBe(1); + done.resolve(); + await new Promise(resolve => setImmediate(resolve)); + expect(idleWorkSnapshot().active).toBe(0); + } finally { + done.resolve(); client.destroy(); + await new Promise(resolve => listener.close(() => resolve())); + } + }); + + it("settles errors through async Express error middleware", async () => { + const server = app(), done = deferred(); + server.get("/work", async () => { throw new Error("test"); }); + server.use(async (_err: Error, _req: express.Request, res: express.Response, _next: express.NextFunction) => { + res.sendStatus(500); await done.promise; + }); + trackIdleRequestHandlers(server); + await request(server).get("/work").expect(500); + expect(idleWorkSnapshot().active).toBe(1); + done.resolve(); await new Promise(resolve => setImmediate(resolve)); + expect(idleWorkSnapshot().active).toBe(0); + }); + + it("uses unique owners even for simultaneous starts and expires the hold", () => { + const first = startTaskDrain({ purpose: "idle", ttlMs: 60_000 }); + const second = startTaskDrain({ purpose: "idle", ttlMs: 60_000 }); + expect(second.ownerId).not.toBe(first.ownerId); + expect(readTaskDrain(second.expiresAt!)).toBeNull(); + }); + + it("records even work that starts and finishes between observations", () => { + const before = idleWorkSnapshot(); + const done = beginIdleTrackedWork(); done(); done(); + expect(idleWorkSnapshot()).toEqual({ active: before.active, generation: before.generation + 2 }); + }); +}); diff --git a/server/src/__tests__/idle-local-work.test.ts b/server/src/__tests__/idle-local-work.test.ts new file mode 100644 index 0000000000..51e9dae011 --- /dev/null +++ b/server/src/__tests__/idle-local-work.test.ts @@ -0,0 +1,51 @@ +import { promises as fs } from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { inspectIdleSpool, readIdleLocalWork, markIdleIngressTracked, markIdleStartupComplete, registerIdleSpoolDirectory } from "../services/idle-local-work.js"; + +let root: string; +beforeEach(async () => { root = await fs.mkdtemp(path.join(os.tmpdir(), "idle-spool-")); }); +afterEach(async () => { vi.restoreAllMocks(); vi.unstubAllEnvs(); await fs.rm(root, { recursive: true, force: true }); }); +describe("idle cleanup inspection", () => { + it("refuses to authorize sleep until startup and ingress tracking are ready", async () => { + expect(await readIdleLocalWork()).toBe("unknown"); + }); + it("accepts absent and empty directories", async () => { + expect(await inspectIdleSpool(path.join(root, "absent"))).toBe("none"); + expect(await inspectIdleSpool(root)).toBe("none"); + }); + it.each(["receipt.json", "malformed.json", "partial.tmp", "write.probe", "unexpected"])("retains every spool entry: %s", async name => { + await fs.writeFile(path.join(root, name), "not valid JSON"); + expect(await inspectIdleSpool(root)).toBe("present"); + await fs.rm(path.join(root, name)); + expect(await inspectIdleSpool(root)).toBe("none"); + }); + it("fails closed on a non-directory, dangling symlink, or read failure", async () => { + const file = path.join(root, "file"), link = path.join(root, "link"); + await fs.writeFile(file, ""); + await fs.symlink(path.join(root, "missing"), link); + expect(await inspectIdleSpool(file)).toBe("unknown"); + expect(await inspectIdleSpool(path.join(file, "nested"))).toBe("unknown"); + expect(await inspectIdleSpool(link)).toBe("unknown"); + vi.spyOn(fs, "opendir").mockRejectedValue(Object.assign(new Error("private path"), { code: "EACCES" })); + expect(await inspectIdleSpool(root)).toBe("unknown"); + }); + it("checks custom spool directories as well as the default paths", async () => { + vi.stubEnv("PAPERCLIP_HOME", root); + vi.stubEnv("SANDBOX_ORPHAN_CLEANUP_SPOOL_DIR", path.join(root, "orphans")); + const custom = path.join(root, "custom-accounting"); + registerIdleSpoolDirectory(custom); + markIdleIngressTracked(); markIdleStartupComplete({ scheduledBackups: false }); + expect(await readIdleLocalWork()).toBe("none"); + markIdleStartupComplete({ scheduledBackups: true }); + expect(await readIdleLocalWork()).toBe("present"); + markIdleStartupComplete({ scheduledBackups: false }); + await fs.mkdir(custom); + await fs.writeFile(path.join(custom, "receipt.tmp"), "incomplete"); + expect(await readIdleLocalWork()).toBe("present"); + await fs.rm(custom, { recursive: true }); + expect(await readIdleLocalWork()).toBe("none"); + }); + +}); diff --git a/server/src/__tests__/idle-sleep-safety.test.ts b/server/src/__tests__/idle-sleep-safety.test.ts new file mode 100644 index 0000000000..1579247a49 --- /dev/null +++ b/server/src/__tests__/idle-sleep-safety.test.ts @@ -0,0 +1,253 @@ +import { beginIdleTrackedWork } from "../services/task-admission.js"; +import { randomUUID } from "node:crypto"; +import { eq, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + agentInstructionWorkingCopies, agents, agentApiKeys, agentWakeupRequests, companies, companySecretProposals, createDb, + adapterAuthSessions, environments, environmentLeases, executionWorkspaces, + heartbeatRuns, issues, issueWatchdogs, projects, routines, type Db, +} from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js"; + +const now = Date.UTC(2020, 0, 1); +const held = (): IdleSleepDrainStatus => ({ + draining: true, startedAt: new Date(now - 1000), expiresAt: new Date(now + 60_000), + activeRuns: 0, pendingWakes: 0, +}); +const unknown = { version: 1, backgroundWork: "unknown" }; +const present = { version: 1, backgroundWork: "present" }; +const none = { version: 1, backgroundWork: "none" }; +const ownerId = "d0b833f4-4098-42de-8420-1907f3aa4895"; +const owned = () => ({ ...held(), ownerId }); +const emptyLocal = async () => "none" as const; + + +describe("idle sleep safety failure boundaries", () => { + it.each([ + { draining: false }, { activeRuns: 1 }, { pendingWakes: 1 }, + { startedAt: null }, { expiresAt: new Date(now) }, + ])("does not query durable work without a quiet admission hold: %j", async (change) => { + const transaction = vi.fn(); + expect(await readIdleSleepSafety({ transaction } as unknown as Db, () => ({ ...held(), ...change }), () => now)) + .toEqual(unknown); + expect(transaction).not.toHaveBeenCalled(); + }); + + it("fails closed when the database is unavailable", async () => { + const db = { transaction: vi.fn().mockRejectedValue(new Error("private connection detail")) } as unknown as Db; + expect(await readIdleSleepSafety(db, held, () => now)).toEqual(unknown); + }); + + it.each([ + { draining: false }, { startedAt: new Date(now) }, { expiresAt: new Date(now) }, + { activeRuns: 1 }, { pendingWakes: 1 }, + ])("rejects a report when the hold changes during the durable scan: %j", async (change) => { + const getStatus = vi.fn().mockReturnValueOnce(held()).mockReturnValue({ ...held(), ...change }); + const execute = vi.fn().mockResolvedValue([{ blocked: true }]); + const transaction = vi.fn(async (fn: (tx: unknown) => Promise) => fn({ execute })); + expect(await readIdleSleepSafety({ transaction } as unknown as Db, getStatus, () => now)).toEqual(unknown); + expect(transaction).toHaveBeenCalledWith(expect.any(Function), { isolationLevel: "repeatable read", accessMode: "read only" }); + }); + + it("reports persisted work when the admission hold remains unchanged", async () => { + const execute = vi.fn().mockResolvedValue([{ blocked: true }]); + const transaction = vi.fn(async (fn: (tx: unknown) => Promise) => fn({ execute })); + expect(await readIdleSleepSafety({ transaction } as unknown as Db, held, () => now)).toEqual(present); + }); +}); + +describe("idle sleep admission and local work", () => { + const emptyDb = () => ({ transaction: async (fn: (tx: unknown) => Promise) => fn({ execute: async () => [{ blocked: false }] }) }) as unknown as Db; + it("requires the exact current idle owner and a bounded expiry", async () => { + expect(await readIdleSleepSafety(emptyDb(), held, () => now, undefined, emptyLocal)).toEqual(unknown); + expect(await readIdleSleepSafety(emptyDb(), owned, () => now, "old-owner", emptyLocal)).toEqual(unknown); + expect(await readIdleSleepSafety(emptyDb(), () => ({ ...owned(), expiresAt: null }), () => now, ownerId, emptyLocal)).toEqual(unknown); + expect(await readIdleSleepSafety(emptyDb(), owned, () => now, ownerId, emptyLocal)).toEqual(none); + }); + it.each(["present", "unknown"] as const)("retains local work reported as %s", async state => { + expect(await readIdleSleepSafety(emptyDb(), owned, () => now, ownerId, async () => state)) + .toEqual({ version: 1, backgroundWork: state }); + }); + it("refuses sleep while accepted work remains in flight", async () => { + const done = beginIdleTrackedWork(); + try { expect(await readIdleSleepSafety(emptyDb(), owned, () => now, ownerId, emptyLocal)).toEqual(unknown); } + finally { done(); } + expect(await readIdleSleepSafety(emptyDb(), owned, () => now, ownerId, emptyLocal)).toEqual(none); + }); + it("invalidates a scan even when concurrent work finishes before the final check", async () => { + const inspect = async () => { const done = beginIdleTrackedWork(); done(); return "none" as const; }; + expect(await readIdleSleepSafety(emptyDb(), owned, () => now, ownerId, inspect)).toEqual(unknown); + }); + it("rechecks owner identity after disk inspection, including same-millisecond replacement", async () => { + let status = owned(); + const inspect = async () => { status = { ...status, ownerId: "replacement" }; return "none" as const; }; + expect(await readIdleSleepSafety(emptyDb(), () => status, () => now, ownerId, inspect)).toEqual(unknown); + }); + it("rechecks expiry after disk inspection and hides inspection errors", async () => { + let clock = now; + expect(await readIdleSleepSafety(emptyDb(), owned, () => clock, ownerId, async () => { + clock += 60_000; return "none"; + })).toEqual(unknown); + expect(await readIdleSleepSafety(emptyDb(), owned, () => now, ownerId, async () => { throw new Error("private spool path"); })).toEqual(unknown); + }); +}); + +const support = await getEmbeddedPostgresTestSupport(); +if (!support.supported) console.warn(`Skipping idle sleep Postgres tests: ${support.reason}`); +(support.supported ? describe : describe.skip)("idle sleep durable work", () => { + let db: ReturnType; + let database: Awaited>; + + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("idle-sleep-safety-"); + db = createDb(database.connectionString); + }, 60_000); + afterEach(async () => { await db.execute(sql`TRUNCATE companies, plugins, environments CASCADE`); }); + afterAll(async () => { await database?.cleanup(); }); + + const read = () => readIdleSleepSafety(db, owned, () => now, ownerId, emptyLocal); + async function seed() { + const companyId = randomUUID(); + const agentId = randomUUID(); + const runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Idle test", issuePrefix: "IDLE" }); + await db.insert(agents).values({ id: agentId, companyId, name: "On-demand agent", role: "engineer", status: "idle", adapterType: "process" }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", status: "succeeded" }); + return { companyId, agentId, runId }; + } + + it("keeps saved agent files awake until deferred cleanup completes", async () => { + const { companyId, agentId, runId } = await seed(); + await db.insert(agentInstructionWorkingCopies).values({ + runId, companyId, agentId, responsibleUserId: "test-user", entryFile: "AGENTS.md", + baseHash: "test-hash", localRoot: "/tmp/idle-copy", executionRoot: "/tmp/idle-copy", + location: "local", state: "saved", processStoppedAt: new Date(now), + receipt: { schema: "paperclip.agent-files.v1", cleanupPending: true }, + }); + expect(await read()).toEqual(present); + await db.update(agentInstructionWorkingCopies).set({ receipt: { schema: "paperclip.agent-files.v1" } }) + .where(eq(agentInstructionWorkingCopies.runId, runId)); + expect(await read()).toEqual(none); + }); + + it("authorizes an empty database and completed history under a quiet owned hold", async () => { + expect(await read()).toEqual(none); + const { companyId } = await seed(); + await db.insert(issues).values({ companyId, title: "Finished work", status: "done" }); + await db.insert(routines).values({ companyId, title: "Paused schedule", status: "paused" }); + await db.insert(environmentLeases).values({ companyId, status: "released" }); + expect(await read()).toEqual(none); + }); + + it("blocks a saved watchdog on a completed issue before its first review starts", async () => { + const { companyId, agentId } = await seed(); + const [issue] = await db.insert(issues).values({ companyId, title: "Finished work", status: "done" }).returning(); + expect(await read()).toEqual(none); + const [watchdog] = await db.insert(issueWatchdogs).values({ + companyId, issueId: issue!.id, watchdogAgentId: agentId, status: "active", + }).returning(); + expect(watchdog!.watchdogIssueId).toBeNull(); + expect(watchdog!.lastTriggeredAt).toBeNull(); + expect(await read()).toEqual(present); + await db.update(issueWatchdogs).set({ status: "disabled" }).where(eq(issueWatchdogs.id, watchdog!.id)); + expect(await read()).toEqual(none); + }); + + it.each([ + ["queued run", { status: "queued" }], + ["orphan running run", { status: "running" }], + ["future retry", { scheduledRetryAt: new Date(now + 86_400_000) }], + ["unsettled accounting", { costAccountingPending: true }], + ] as const)("blocks %s even when process counters are zero", async (_name, change) => { + const { runId } = await seed(); + await db.update(heartbeatRuns).set(change).where(eq(heartbeatRuns.id, runId)); + expect(await read()).toEqual(present); + }); + + it("reports an enabled heartbeat timer and permits sleep after it is disabled", async () => { + const { agentId } = await seed(); + await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: true, intervalSec: 86_400 } } }).where(eq(agents.id, agentId)); + expect(await read()).toEqual(present); + await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: false, intervalSec: 86_400 } } }).where(eq(agents.id, agentId)); + expect(await read()).toEqual(none); + }); + + it("blocks a durable deferred wake", async () => { + const { companyId, agentId } = await seed(); + await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "assignment", status: "deferred_issue_execution" }); + expect(await read()).toEqual(present); + }); + + it("blocks an active routine without waiting for its next due time", async () => { + const { companyId } = await seed(); + await db.insert(routines).values({ companyId, title: "Tomorrow", status: "active" }); + expect(await read()).toEqual(present); + }); + + it("blocks a remote resource pending cleanup", async () => { + const { companyId } = await seed(); + await db.insert(environmentLeases).values({ companyId, status: "pending_cleanup" }); + expect(await read()).toEqual(present); + }); + + it("reports a login sandbox awaiting cleanup", async () => { + const { companyId } = await seed(); + const [environment] = await db.insert(environments).values({ name: "Login fixture" }).returning(); + await db.insert(adapterAuthSessions).values({ + companyId, environmentId: environment!.id, adapterType: "codex_local", + startedByUserId: "fixture-user", publicSessionId: "fixture-session", status: "cleanup_pending", + }); + expect(await read()).toEqual(present); + }); + + it("reports a pending secret proposal with future expiry", async () => { + const { companyId, agentId, runId } = await seed(); + await db.insert(companySecretProposals).values({ + companyId, kind: "secret", proposedName: "Fixture", proposedKey: "FIXTURE", + justification: "Test pending proposal", proposedByAgentId: agentId, originRunId: runId, + expiresAt: new Date(now + 86_400_000), + }); + expect(await read()).toEqual(present); + }); + + it("reports retained execution workspaces that may need delayed cleanup", async () => { + const { companyId } = await seed(); + const [project] = await db.insert(projects).values({ companyId, name: "Fixture project" }).returning(); + await db.insert(executionWorkspaces).values({ + companyId, projectId: project!.id, name: "Fixture workspace", mode: "isolated", + strategyType: "git_worktree", status: "closed", cleanupEligibleAt: new Date(now + 86_400_000), + }); + expect(await read()).toEqual(present); + }); + + it("blocks externally usable agent keys until revoked", async () => { + const { companyId, agentId } = await seed(); + const [key] = await db.insert(agentApiKeys).values({ companyId, agentId, name: "Fixture", keyHash: "fixture-not-a-key" }).returning(); + expect(await read()).toEqual(present); + await db.update(agentApiKeys).set({ revokedAt: new Date(now) }).where(eq(agentApiKeys.id, key!.id)); + expect(await read()).toEqual(none); + }); + + it("checks every company in the instance", async () => { + await seed(); + const otherId = randomUUID(); + await db.insert(companies).values({ id: otherId, name: "Other company", issuePrefix: "OTHER" }); + await db.insert(issues).values({ companyId: otherId, title: "Pending work", status: "todo" }); + expect(await read()).toEqual(present); + }); + + it("blocks enabled plugins even without declared jobs or webhooks", async () => { + await db.execute(sql`INSERT INTO plugins (plugin_key, package_name, version, manifest_json) + VALUES ('demo.on-demand', 'demo-plugin', '1.0.0', '{}'::jsonb)`); + expect(await read()).toEqual(present); + await db.execute(sql`UPDATE plugins SET status = 'disabled'`); + expect(await read()).toEqual(none); + }); + + 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); } + finally { await db.execute(sql`ALTER TABLE heartbeat_runs RENAME COLUMN hidden_pending TO cost_accounting_pending`); } + }); +}); diff --git a/server/src/__tests__/instance-settings-routes.test.ts b/server/src/__tests__/instance-settings-routes.test.ts index 68b5fe7d40..2483cd4a3e 100644 --- a/server/src/__tests__/instance-settings-routes.test.ts +++ b/server/src/__tests__/instance-settings-routes.test.ts @@ -29,8 +29,10 @@ const mockCompanyService = vi.hoisted(() => ({ })); const mockLogActivity = vi.hoisted(() => vi.fn()); const mockPublishActivity = vi.hoisted(() => vi.fn()); +const mockReadIdleSleepSafety = vi.hoisted(() => vi.fn()); function registerModuleMocks() { + vi.doMock("../services/idle-sleep-safety.js", () => ({ readIdleSleepSafety: mockReadIdleSleepSafety })); vi.doMock("../services/index.js", () => ({ companyService: () => mockCompanyService, heartbeatService: () => mockHeartbeatService, @@ -89,6 +91,7 @@ describe("instance settings routes", () => { beforeEach(() => { vi.clearAllMocks(); + mockReadIdleSleepSafety.mockReset(); // vi.clearAllMocks() clears recorded calls only; it does not remove a // mockImplementation a prior test installed. Reinstall the default here // so a stateful implementation from one test can never leak into the @@ -960,6 +963,54 @@ describe("instance settings routes", () => { expect(res.status).toBe(200); expect(res.body).toEqual(idleStatus); + expect(mockReadIdleSleepSafety).not.toHaveBeenCalled(); + }); + + it("acquires an owned bounded idle hold and forwards its owner for inspection", async () => { + const ownerId = "d0b833f4-4098-42de-8420-1907f3aa4895"; + mockHeartbeatService.computeTaskDrain.mockReturnValue({ ownerId, startedAt: new Date(), expiresAt: new Date(Date.now() + 60_000) }); + 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); + }); + + it.each([{}, { ttlMs: null }, { ttlMs: 1 }, { ttlMs: 300_001 }])("rejects an unbounded or invalid idle TTL: %j", async body => { + await request(createApp(adminActor)).post("/api/instance/task-drain").send({ ...body, purpose: "idle" }).expect(400); + expect(mockHeartbeatService.applyTaskDrain).not.toHaveBeenCalled(); + }); + + it("rejects replacement and ownerless or stale release of an idle hold", async () => { + mockHeartbeatService.getTaskDrainStatus.mockReturnValue({ ...idleStatus, draining: true, ownerId: "current-owner" }); + const app = createApp(adminActor); + await request(app).post("/api/instance/task-drain").send({}).expect(409); + await request(app).delete("/api/instance/task-drain").expect(409); + await request(app).delete("/api/instance/task-drain?ownerId=previous-owner").expect(409); + expect(mockHeartbeatService.applyTaskDrain).not.toHaveBeenCalled(); + expect(mockHeartbeatService.stopTaskDrain).not.toHaveBeenCalled(); + await request(app).delete("/api/instance/task-drain?ownerId=current-owner").expect(200); + expect(mockHeartbeatService.stopTaskDrain).toHaveBeenCalledOnce(); + }); + + it("returns the opt-in instance-wide safety report to an instance admin", async () => { + mockHeartbeatService.getTaskDrainStatus.mockReturnValue(idleStatus); + const report = { version: 1, backgroundWork: "unknown" }; + mockReadIdleSleepSafety.mockResolvedValue(report); + 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.mock.calls[0][1]()).toEqual(idleStatus); + }); + + it.each([ + ["company member", nonAdminActor], + ["agent", { type: "agent", agentId: "agent-1", companyId: "company-1" }], + ["anonymous caller", { type: "none" }], + ])("does not expose the instance-wide work report to a %s", async (_name, actor) => { + const res = await request(createApp(actor)).get("/api/instance/task-drain?idleSleepSafety=1"); + expect(res.status).toBe(403); + expect(mockReadIdleSleepSafety).not.toHaveBeenCalled(); }); it("writes an activity record for every company, then applies the same drain values, in one transaction", async () => { diff --git a/server/src/__tests__/live-events-ws.test.ts b/server/src/__tests__/live-events-ws.test.ts index 55e78d8656..91e6d9cac9 100644 --- a/server/src/__tests__/live-events-ws.test.ts +++ b/server/src/__tests__/live-events-ws.test.ts @@ -4,6 +4,8 @@ import type { Duplex } from "node:stream"; import { beforeEach, describe, expect, it, vi } from "vitest"; import { setupLiveEventsWebSocketServer } from "../realtime/live-events-ws.js"; import { logger } from "../middleware/logger.js"; +import { idleWorkSnapshot, startTaskDrain, stopTaskDrain } from "../services/task-admission.js"; +import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; vi.mock("../middleware/logger.js", () => ({ logger: { @@ -67,6 +69,43 @@ describe("setupLiveEventsWebSocketServer", () => { vi.clearAllMocks(); }); + it.each([false, true])("counts accepted upgrade authentication after disconnect (failure=%s)", async (fail) => { + const server = new EventEmitter(); + const socket = new FakeUpgradeSocket(); + let finishAuth!: () => void; + const pending = new Promise(resolve => { finishAuth = resolve; }); + setupLiveEventsWebSocketServer(server as never, {} as never, { + deploymentMode: "authenticated", + resolveCloudActor: async () => { + await pending; + if (fail) throw new Error("fixture authentication write failed"); + return { userId: "fixture-user", companyIds: ["company-1"] }; + }, + }); + server.emit("upgrade", createUpgradeRequest(), socket as unknown as Duplex, Buffer.alloc(0)); + const hold = startTaskDrain({ purpose: "idle", ttlMs: 60_000 }); + const emptyDb = { transaction: async (run: (tx: unknown) => Promise) => + run({ execute: async () => [{ blocked: false }] }) }; + const report = () => readIdleSleepSafety(emptyDb as never, + () => ({ ...hold, draining: true, activeRuns: 0, pendingWakes: 0 }), Date.now, hold.ownerId, async () => "none"); + try { + socket.destroy(); + await flushPromises(); + expect(idleWorkSnapshot().active).toBe(1); + expect((await report()).backgroundWork).toBe("unknown"); + finishAuth(); + await flushPromises(); + expect(idleWorkSnapshot().active).toBe(0); + expect((await report()).backgroundWork).toBe("none"); + expect(socket.endedChunks).toEqual([]); + } finally { + finishAuth(); + await flushPromises(); + stopTaskDrain(); + server.emit("close"); + } + }); + it("does not write a rejection response after the raw upgrade socket is already closed", async () => { const server = new EventEmitter(); setupLiveEventsWebSocketServer(server as never, {} as never, { deploymentMode: "authenticated" }); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 4de339d035..7b89d44dd9 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -1,3 +1,7 @@ +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; +import { reconcileBuiltInAgentsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup } from "../services/index.js"; +import { runDatabaseBackup } from "@paperclipai/db"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { chmodSync, mkdirSync, mkdtempSync, readFileSync, rmSync, statSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; @@ -437,6 +441,65 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each(([ + ["built-in agents", reconcileBuiltInAgentsOnStartup], + ["managed homes", reconcileCodexLocalManagedHomesOnStartup], + ["runtime services", reconcilePersistedRuntimeServicesOnStartup], + ] as const).flatMap(([name, reconcile]) => [false, true].map(fail => ({ name, reconcile, fail })))) + ("counts detached $name startup writes after readiness (failure=$fail)", async ({ reconcile, fail }) => { + let finishStartup!: () => void; + const pending = new Promise(resolve => { finishStartup = resolve; }); + vi.mocked(reconcile).mockImplementationOnce(async () => { + await pending; + if (fail) throw new Error("fixture startup write failed"); + return { reconciled: 0, seeded: 0, failed: 0 } as never; + }); + const emptyDb = { transaction: async (run: (tx: unknown) => Promise) => + run({ execute: async () => [{ blocked: false }] }) }; + const hold = { ownerId: "fixture-owner", draining: true, activeRuns: 0, pendingWakes: 0, + startedAt: new Date(), expiresAt: new Date(Date.now() + 60_000) }; + const report = () => readIdleSleepSafety(emptyDb as never, () => hold, Date.now, hold.ownerId, async () => "none"); + try { + await startServer(); + await new Promise(resolve => setImmediate(resolve)); + expect(idleWorkSnapshot().active).toBe(1); + expect((await report()).backgroundWork).toBe("unknown"); + finishStartup(); + await new Promise(resolve => setImmediate(resolve)); + expect(idleWorkSnapshot().active).toBe(0); + expect((await report()).backgroundWork).toBe("none"); + } finally { + finishStartup(); + await new Promise(resolve => setImmediate(resolve)); + } + }); + + it.each([false, true])("counts an accepted backup until completion, including failure=%s", async fails => { + await startServer(); + await new Promise(resolve => setImmediate(resolve)); + const before = idleWorkSnapshot().active; + let resolveBackup!: (value: never) => void; + let rejectBackup!: (error: Error) => void; + vi.mocked(runDatabaseBackup).mockImplementationOnce(() => new Promise((resolve, reject) => { + resolveBackup = resolve; rejectBackup = reject; + })); + const options = createAppMock.mock.calls[0]?.[1] as unknown as { + databaseBackupService: { runManualBackup(): Promise }; + }; + const backup = options.databaseBackupService.runManualBackup(); + await Promise.resolve(); + expect(idleWorkSnapshot().active).toBe(before + 1); + if (fails) { + const rejection = expect(backup).rejects.toThrow("backup failed"); + rejectBackup(new Error("backup failed")); + await rejection; + } else { + resolveBackup({ backupFile: "backup.sql.gz", sizeBytes: 1, prunedCount: 0 } as never); + await backup; + } + expect(idleWorkSnapshot().active).toBe(before); + }); + it("starts without PAPERCLIP_DECISION_SIGNING_SECRET by generating a persisted key", async () => { const originalHome = process.env.PAPERCLIP_HOME; const originalInstanceId = process.env.PAPERCLIP_INSTANCE_ID; diff --git a/server/src/app.ts b/server/src/app.ts index 1b615c8f37..6c9dd7e32e 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -1,3 +1,5 @@ +import { idleAdmissionMiddleware, trackIdleRequestHandlers } from "./middleware/idle-admission.js"; +import { isIdleTaskDrainActive, trackIdleWork } from "./services/task-admission.js"; import { customerSuccessRoutes } from "./routes/customer-success.js"; import { cloudWarmStandbyMiddleware } from "./middleware/cloud-warm-standby.js"; import type { CloudWarmStandby } from "./services/cloud-warm-standby.js"; @@ -520,6 +522,7 @@ export async function createApp( }, ) { const app = express(); + app.use(idleAdmissionMiddleware); const staticUi = express.Router(); app.locals.paperclipDb = db; const isWarmStandby = opts.cloudWarmStandby ?? (() => false); @@ -629,7 +632,7 @@ export async function createApp( // mutation guard. The Chat SDK adapter verifies the provider signature // before Paperclip persists or acts on any event. const emailChannels = emailChannelService(db, { - isBackgroundWorkEnabled: () => !isWarmStandby(), + isBackgroundWorkEnabled: () => !isWarmStandby() && !isIdleTaskDrainActive(), heartbeat: connectionIntentHeartbeat, storage: opts.storageService, publicBaseUrl: opts.chatWebhookPublicBaseUrl ?? opts.authPublicBaseUrl, @@ -841,7 +844,7 @@ export async function createApp( const jobStore = pluginJobStore(db); const lifecycle = pluginLifecycleManager(db, { workerManager }); const scheduler = createPluginJobScheduler({ - isBackgroundWorkEnabled: () => !isWarmStandby(), + isBackgroundWorkEnabled: () => !isWarmStandby() && !isIdleTaskDrainActive(), db, jobStore, workerManager, @@ -1000,14 +1003,14 @@ export async function createApp( if (publicMcpOAuth) { const dispatch = createMcpApiDispatch(api); publicMcpEvents = createPublicMcpEvents(db, publicMcpOAuth, dispatch, { - isBackgroundWorkEnabled: () => !isWarmStandby(), + isBackgroundWorkEnabled: () => !isWarmStandby() && !isIdleTaskDrainActive(), }); publicMcpEvents.start(); const transfers = createPublicMcpTransfers(db, publicMcpOAuth, dispatch, opts.storageService); publicMcpIngress.use(transfers.router); publicMcpIngress.use(publicMcpIngressRoutes(publicMcpOAuth, createPublicMcpExecutor(db, publicMcpOAuth, dispatch, transfers), publicMcpEvents)); const dotOAuth = createPublicMcpOAuth(db, { ...publicMcpOAuth.config, resource: publicMcpOAuth.config.origin + "/mcp/runner" }); - dotMcpEvents = createPublicMcpEvents(db, dotOAuth, dispatch, { enableDotRunner: true, isBackgroundWorkEnabled: () => !isWarmStandby() }); + dotMcpEvents = createPublicMcpEvents(db, dotOAuth, dispatch, { enableDotRunner: true, isBackgroundWorkEnabled: () => !isWarmStandby() && !isIdleTaskDrainActive() }); dotMcpEvents.start(); publicMcpIngress.use(publicMcpIngressRoutes(dotOAuth, createPublicMcpExecutor(db, dotOAuth, dispatch), dotMcpEvents, createDotRunnerMcpTools(db))); api.use(publicMcpManagementRoutes(publicMcpOAuth, dotOAuth)); @@ -1194,7 +1197,7 @@ export async function createApp( } }; const flushPendingFeedbackExports = async () => { - if (feedbackExportShuttingDown || isWarmStandby()) return; + if (feedbackExportShuttingDown || isWarmStandby() || isIdleTaskDrainActive()) return; try { await opts.feedbackExportService?.flushPendingFeedbackTraces(); } catch (err) { @@ -1243,25 +1246,25 @@ export async function createApp( }); const unsubscribeChatPublicationSignals = subscribeAllCompanyLiveEvents( (event) => { - if (!isWarmStandby() && isChatPublicationCommitSignal(event)) + if (!isWarmStandby() && !isIdleTaskDrainActive() && isChatPublicationCommitSignal(event)) chatReconciliation.notifyPublications(); }, ); let chatPublicationTimer: ReturnType | null = setInterval( () => { - if (!isWarmStandby()) chatReconciliation.reconcile(); + if (!isWarmStandby() && !isIdleTaskDrainActive()) chatReconciliation.reconcile(); }, CHAT_PUBLICATION_FLUSH_INTERVAL_MS, ); chatPublicationTimer.unref?.(); - if (!isWarmStandby()) chatReconciliation.reconcile(); + if (!isWarmStandby() && !isIdleTaskDrainActive()) chatReconciliation.reconcile(); // Abandoned chunked-import spool sweep: hourly (plus once at startup), // deleting spool dirs whose transfer saw no activity for 24h and cancelling // their still-open ledger runs. Same setInterval + unref + shutdown-clear // shape as the feedback export flush above. const importTransferSpoolRoot = resolveDefaultImportTransferSpoolRoot(); const sweepImportTransferSpools = () => { - if (isWarmStandby()) return; + if (isWarmStandby() || isIdleTaskDrainActive()) return; sweepAbandonedImportTransferSpools(db, importTransferSpoolRoot) .then((result) => { if (result.swept > 0) { @@ -1276,11 +1279,11 @@ export async function createApp( }); }; const browserUseTimer = setInterval(() => { - if (isWarmStandby()) return; + if (isWarmStandby() || isIdleTaskDrainActive()) return; void browserUse.sweep().catch(() => logger.warn("Browser Use reconciliation failed; retrying.")); }, 3000); browserUseTimer.unref?.(); - if (!isWarmStandby()) void browserUse.sweep().catch(() => logger.warn("Browser Use startup reconciliation failed; retrying.")); + if (!isWarmStandby() && !isIdleTaskDrainActive()) void browserUse.sweep().catch(() => logger.warn("Browser Use startup reconciliation failed; retrying.")); let importTransferSweepTimer: ReturnType | null = setInterval( sweepImportTransferSpools, @@ -1369,7 +1372,7 @@ export async function createApp( .catch((err) => { logger.error({ err }, "Failed to load ready plugins on startup"); }); - app.locals.bundledPluginsStartup = bundledPluginsStartup; + app.locals.bundledPluginsStartup = trackIdleWork(bundledPluginsStartup); // The shutdown hook runs at most once. It caches the in-flight promise, so a // second caller (for example the `exit` handler) awaits the same completion // instead of starting a second teardown. @@ -1427,5 +1430,6 @@ export async function createApp( void flushPluginLogBuffer(); }); + trackIdleRequestHandlers(app); return app; } diff --git a/server/src/index.ts b/server/src/index.ts index 6412a8b35b..0c79e723e6 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,5 @@ +import { isIdleTaskDrainActive, beginIdleTrackedWork, trackIdleWork } from "./services/task-admission.js"; +import { markIdleStartupComplete } from "./services/idle-local-work.js"; import { cloudWarmStandbyServerOptions } from "./middleware/cloud-warm-standby.js"; import { createCloudWarmStandby } from "./services/cloud-warm-standby.js"; import { subscribeAllCompanyLiveEvents } from "./services/live-events.js"; @@ -830,6 +832,7 @@ async function startServerWithDatabaseTeardown( } databaseBackupInFlight = true; + const finishIdleBackup = beginIdleTrackedWork(); const startedAt = new Date(); const startedAtMs = Date.now(); const label = trigger === "scheduled" ? "Automatic" : "Manual"; @@ -873,6 +876,7 @@ async function startServerWithDatabaseTeardown( throw err; } finally { databaseBackupInFlight = false; + finishIdleBackup(); } }; const pluginWorkerManager = createPluginWorkerManager(); @@ -928,7 +932,7 @@ async function startServerWithDatabaseTeardown( managedPluginAutoInstall, }); // Upgrade admission runs before every WebSocket listener, outside Express. - const server = createServer(cloudWarmStandbyServerOptions(isWarmStandby), app as unknown as RequestListener); + const server = createServer(cloudWarmStandbyServerOptions(() => isWarmStandby() || isIdleTaskDrainActive()), app as unknown as RequestListener); // Increase keep-alive timeouts to safely outlive default idle timeouts // of common reverse proxies and load balancers (like AWS ALB, Nginx, or Traefik). @@ -1021,7 +1025,9 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "startup reconciliation of managed runtime control operations failed"); } - void reconcilePersistedRuntimeServicesOnStartup(db as any) + // These jobs need not delay readiness, but their writes must finish before + // an idle report can authorize sleep. Count the whole detached promise. + void trackIdleWork(reconcilePersistedRuntimeServicesOnStartup(db as any) .then((result) => { if ( result.reconciled > 0 @@ -1046,12 +1052,12 @@ async function startServerWithDatabaseTeardown( }) .catch((err) => { logger.error({ err }, "startup reconciliation of persisted runtime services failed"); - }); + })); // Backfill auth.json into any already-isolated codex_local managed home that // was created by the #8272 isolation guard before the Phase 1 seeding fix. // Idempotent; the Phase 1 execute-time seeding covers new strandings. - void reconcileCodexLocalManagedHomesOnStartup(db) + void trackIdleWork(reconcileCodexLocalManagedHomesOnStartup(db) .then((result) => { if (result.seeded > 0 || result.failed > 0) { logger.warn( @@ -1068,9 +1074,9 @@ async function startServerWithDatabaseTeardown( }) .catch((err) => { logger.error({ err }, "startup reconciliation of codex_local managed homes failed"); - }); + })); - void reconcileBuiltInAgentsOnStartup(db as any) + void trackIdleWork(reconcileBuiltInAgentsOnStartup(db as any) .then((result) => { if ( result.reconciled > 0 @@ -1087,7 +1093,7 @@ async function startServerWithDatabaseTeardown( }) .catch((err) => { logger.error({ err }, "startup reconciliation of built-in agents failed"); - }); + })); // Force the instance onto the Kubernetes sandbox provider when configured via // env (PAPERCLIP_EXECUTION_MODE=kubernetes). Runs BEFORE the heartbeat resumes @@ -1148,11 +1154,13 @@ async function startServerWithDatabaseTeardown( let heartbeatSchedulerInterval: ReturnType | null = null; const heartbeatSchedulerInFlight = new Set>(); const trackHeartbeatSchedulerWork = (work: Promise) => { + const finishIdleWork = beginIdleTrackedWork(); let tracked: Promise; tracked = Promise.resolve(work) .then(() => undefined, () => undefined) .finally(() => { heartbeatSchedulerInFlight.delete(tracked); + finishIdleWork(); }); heartbeatSchedulerInFlight.add(tracked); }; @@ -1171,7 +1179,7 @@ async function startServerWithDatabaseTeardown( ["local_ai_login_cleanup", () => localAiLoginService(db).reapExpired()], ] as const; const sweepExecutionControl = () => { - if (heartbeatSchedulerStopped || isWarmStandby()) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped || isWarmStandby()) return; // Independent durable queues must not block one another. Each queue remains // single-flight; a later sweep observes committed transitions from its peers. for (const [queue, work] of executionControlSweeps) { @@ -1187,7 +1195,7 @@ async function startServerWithDatabaseTeardown( sweepExecutionControl(); const startHeartbeatSchedulerInterval = (callback: () => void) => { heartbeatSchedulerInterval = setInterval(() => { - if (!isWarmStandby()) callback(); + if (!isWarmStandby() && !isIdleTaskDrainActive()) callback(); }, config.heartbeatSchedulerIntervalMs); heartbeatSchedulerInterval?.unref?.(); }; @@ -1196,7 +1204,7 @@ async function startServerWithDatabaseTeardown( enabled: async () => (await instanceSettingsService(db).getExperimental()).enableExternalObjects === true, }); const scheduleExternalObjectRefreshSweep = (now = new Date()) => { - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(externalObjects .refreshDueObjectsForActiveCompanies(50, now) .then((result) => { @@ -1236,7 +1244,7 @@ async function startServerWithDatabaseTeardown( // Activity publication happens after the status transaction commits. This is // a best-effort fast path; the durable outbox and sweeps remain authoritative. const unsubscribeChatCompletions = subscribeAllCompanyLiveEvents(event => { - if (heartbeatSchedulerStopped || event.type !== "activity.logged" || + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped || event.type !== "activity.logged" || event.payload.action !== "issue.updated" || typeof event.payload.entityId !== "string") return; trackHeartbeatSchedulerWork(chatCompletionDeliveries.sweepPending({ companyId: event.companyId, taskId: event.payload.entityId }) .catch(err => logger.error({ err }, "post-commit chat completion delivery failed"))); @@ -1259,7 +1267,7 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "environment lease cleanup sweep failed"); }); const scheduleEnvironmentLeaseCleanupSweep = () => { - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(runEnvironmentLeaseCleanupSweep(ENVIRONMENT_LEASE_CLEANUP_SWEEP_BACKOFF_MS)); }; const githubConnectionEvents = githubConnectionEventService(db as any, { @@ -1273,7 +1281,7 @@ async function startServerWithDatabaseTeardown( ?? null, }); const scheduleGitHubConnectionEventPoll = () => { - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(githubConnectionEvents.pollOnce() .then((result) => { if (result.leased > 0 || result.failed > 0) { @@ -1285,7 +1293,7 @@ async function startServerWithDatabaseTeardown( })); }; const scheduleGitHubConnectionContinuitySweep = () => { - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(tools.sweepGitHubConnectionContinuity() .then((result) => { if (result.due > 0 || result.failed > 0) { @@ -1335,7 +1343,7 @@ async function startServerWithDatabaseTeardown( workspaceReaperCooldownDays: config.workspaceReaperCooldownDays, }); const scheduleMergedPullRequestConfirmationSweep = () => { - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(mergedPullRequestConfirmations .sweepMergedPullRequestConfirmations() .then((result) => { @@ -1353,7 +1361,7 @@ async function startServerWithDatabaseTeardown( let lastTerminalWorkspaceSkipLogAt = 0; const terminalWorkspaceSkipLogIntervalMs = 10 * 60 * 1000; const scheduleTerminalWorkspaceSweep = () => { - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(terminalWorkspaces .sweepTerminalWorkspaces() .then((result) => { @@ -1404,7 +1412,7 @@ async function startServerWithDatabaseTeardown( } }; const scheduleAdapterLoginReaperSweep = () => { - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(adapterLoginReaper .sweep() .then(logAdapterLoginReaperResult) @@ -1430,7 +1438,7 @@ async function startServerWithDatabaseTeardown( } }; const scheduleSetupTokenReaperSweep = () => { - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(setupTokenReaper .sweep() .then(logSetupTokenReaperResult) @@ -1649,7 +1657,7 @@ async function startServerWithDatabaseTeardown( // can then wait through an already-running suppression check before it // captures the authoritative set of running heartbeat rows. trackHeartbeatSchedulerWork((async () => { - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(decisionExecutor.sweepExpired().catch((err: unknown) => { logger.error({ err }, "decision expiry sweep failed"); })); @@ -1677,10 +1685,10 @@ async function startServerWithDatabaseTeardown( })); } - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; scheduleExternalObjectRefreshSweep(new Date()); - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; scheduleMergedPullRequestConfirmationSweep(); scheduleGitHubConnectionEventPoll(); scheduleGitHubConnectionContinuitySweep(); @@ -1689,7 +1697,7 @@ async function startServerWithDatabaseTeardown( scheduleSetupTokenReaperSweep(); scheduleEnvironmentLeaseCleanupSweep(); - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(routines .tickScheduledTriggers(new Date()) .then((result) => { @@ -1701,7 +1709,7 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "routine scheduler tick failed"); })); - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork((async () => { const experimental = await instanceSettingsService(db).getExperimental(); if (experimental.enableStatusCards !== true) return; @@ -1730,7 +1738,7 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "status-card scheduler tick failed"); })); - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(environmentCustomImages .cleanupExpiredSetupSessions() .then((result) => { @@ -1742,7 +1750,7 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "environment customImage setup cleanup failed"); })); - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(tools .sweepConnectionHealth() .then((swept) => { @@ -1780,7 +1788,7 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "periodic question-response delivery sweep failed"); })); - if (heartbeatSchedulerStopped) return; + if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) { // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. @@ -1865,7 +1873,7 @@ async function startServerWithDatabaseTeardown( "Automatic database backups enabled", ); setInterval(() => { - if (isWarmStandby()) return; + if (isWarmStandby() || isIdleTaskDrainActive()) return; void runServerDatabaseBackup("scheduled").catch(() => { // runServerDatabaseBackup already logs the failure with context. }); @@ -1890,6 +1898,7 @@ async function startServerWithDatabaseTeardown( throw err; } + markIdleStartupComplete({ scheduledBackups: config.databaseBackupEnabled }); setStartupRecoveryPhase("ready"); logger.info(`Server startup recovery complete on ${config.host}:${listenPort}`); void systemdNotify(["--ready", `--status=Listening on ${config.host}:${listenPort}`]).then((notified) => { diff --git a/server/src/middleware/auth.ts b/server/src/middleware/auth.ts index 8bc814f44f..18fb7ddd2f 100644 --- a/server/src/middleware/auth.ts +++ b/server/src/middleware/auth.ts @@ -28,6 +28,7 @@ import { logger } from "./logger.js"; import { captureRunIdentity } from "../services/run-identity.js"; import { boardAuthService } from "../services/board-auth.js"; import { retryIdempotentDatabaseOperation } from "../database-retry.js"; +import { beginIdleTrackedWork } from "../services/task-admission.js"; export { isTransientDbConnectionError, @@ -226,7 +227,7 @@ const publicMcpGatewayProtocolPath = /^\/mcp\/gateways\/gw_[a-f0-9]{32}\/?$/i; export function actorMiddleware(db: Db, opts: ActorMiddlewareOptions): RequestHandler { const boardAuth = boardAuthService(db); - return async (req, _res, next) => { + const authenticate: RequestHandler = async (req, _res, next) => { req.actor = opts.deploymentMode === "local_trusted" ? { @@ -501,6 +502,22 @@ export function actorMiddleware(db: Db, opts: ActorMiddlewareOptions): RequestHa next(); }; + return async (req, res, next) => { + // Health and task-drain requests bypass tenant admission, but resolving + // their actor can still write users, companies, memberships and key usage. + // Finish all authentication before entering the next handler: the report + // must count concurrent authentication, without counting its own auth. + const finish = beginIdleTrackedWork(); + let continueRequest: (() => void) | undefined; + try { + await authenticate(req, res, (error?: unknown) => { + continueRequest = () => next(error); + }); + } finally { + finish(); + } + continueRequest?.(); + }; } /** diff --git a/server/src/middleware/cloud-tenant-actor.test.ts b/server/src/middleware/cloud-tenant-actor.test.ts index f8514d4f1a..1e090fc5e3 100644 --- a/server/src/middleware/cloud-tenant-actor.test.ts +++ b/server/src/middleware/cloud-tenant-actor.test.ts @@ -1,9 +1,13 @@ import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import type { Request } from "express"; +import express, { type Request } from "express"; +import request from "supertest"; import { and, eq } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { authUsers, companies, companyMemberships, instanceSettings, instanceUserRoles } from "@paperclipai/db"; -import { cloudActorHeaderSourceFromHeaders, resolveCloudTenantActor } from "./auth.js"; +import { actorMiddleware, cloudActorHeaderSourceFromHeaders, resolveCloudTenantActor } from "./auth.js"; +import { idleAdmissionMiddleware, trackIdleRequestHandlers } from "./idle-admission.js"; +import { idleWorkSnapshot, startTaskDrain, stopTaskDrain } from "../services/task-admission.js"; +import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; // Minimal fake Drizzle Db: records every table passed to .insert() / .delete() and // supports the chained call shapes used by resolveCloudTenantActor (values / @@ -18,6 +22,7 @@ function createFakeDb(options: { membershipQueryRows?: Array<{ companyId: string; membershipRole: string | null; status: string }>; settingsRow?: Record | null; selectThrows?: boolean; + beforeWrite?: () => Promise; } = {}) { const membershipRow = options.membershipRow ?? { companyId: "company-x", membershipRole: "owner", status: "active" }; @@ -41,8 +46,9 @@ function createFakeDb(options: { chain.onConflictDoUpdate = () => chain; chain.onConflictDoNothing = () => chain; chain.where = () => chain; - chain.returning = async () => [membershipRow]; - chain.then = (resolve: (v: unknown) => unknown) => Promise.resolve(undefined).then(resolve); + chain.returning = async () => { await options.beforeWrite?.(); return [membershipRow]; }; + chain.then = (resolve: (v: unknown) => unknown, reject?: (error: unknown) => unknown) => + Promise.resolve().then(options.beforeWrite).then(resolve, reject); const db = { insert: (table: unknown) => { insertedTables.push(table); @@ -122,6 +128,7 @@ describe("resolveCloudTenantActor (shared-pool hardening)", () => { process.env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN = "test-server-token"; }); afterEach(() => { + stopTaskDrain(); delete process.env.PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN; delete process.env.PAPERCLIP_MANAGED_CONFIG; }); @@ -159,6 +166,52 @@ describe("resolveCloudTenantActor (shared-pool hardening)", () => { expect(insertedTables).toContain(companyMemberships); }); + it.each(["/api/health", "/api/instance/task-drain"])("counts unfinished authentication on exempt %s requests", async (path) => { + let enterWrite!: () => void, finishWrite!: () => void; + const entered = new Promise(resolve => { enterWrite = resolve; }); + const pending = new Promise(resolve => { finishWrite = resolve; }); + const { db, insertedTables } = createFakeDb({ beforeWrite: async () => { enterWrite(); await pending; } }); + const server = express(); + server.use(idleAdmissionMiddleware); + server.use(actorMiddleware(db, { deploymentMode: "authenticated", resolveSession: async () => null })); + const hold = startTaskDrain({ purpose: "idle", ttlMs: 60_000 }); + const emptyDb = { transaction: async (run: (tx: unknown) => Promise) => + run({ execute: async () => [{ blocked: false }] }) } as unknown as Db; + server.get("/api/health", (_req, res) => res.sendStatus(204)); + server.get("/api/instance/task-drain", async (_req, res) => { + res.json(await readIdleSleepSafety(emptyDb, () => ({ ...hold, draining: true, activeRuns: 0, pendingWakes: 0 }), + Date.now, hold.ownerId, async () => "none")); + }); + trackIdleRequestHandlers(server); + const accepted = request(server).get(path).set(VALID_HEADERS).then(response => response); + try { + await entered; + expect(idleWorkSnapshot().active).toBeGreaterThan(0); + const during = await request(server).get("/api/instance/task-drain").expect(200); + expect(during.body.backgroundWork).toBe("unknown"); + } finally { + finishWrite(); + await accepted; + } + expect(insertedTables).toEqual(expect.arrayContaining([authUsers, companies, companyMemberships])); + expect(idleWorkSnapshot().active).toBe(0); + const after = await request(server).get("/api/instance/task-drain").expect(200); + expect(after.body.backgroundWork).toBe("none"); + }); + + it("releases failed control authentication and preserves Express error handling", async () => { + const { db } = createFakeDb({ beforeWrite: async () => { throw new Error("fixture write failed"); } }); + const server = express(); + server.use(idleAdmissionMiddleware); + server.use(actorMiddleware(db, { deploymentMode: "authenticated", resolveSession: async () => null })); + server.get("/api/health", (_req, res) => res.sendStatus(204)); + server.use((_err: Error, _req: express.Request, res: express.Response, _next: express.NextFunction) => res.sendStatus(500)); + trackIdleRequestHandlers(server); + await request(server).get("/api/health").set(VALID_HEADERS).expect(500); + expect(idleWorkSnapshot().active).toBe(0); + await request(server).get("/api/health").expect(204); + }); + it("resyncs an A to B to A context transition inside the debounce window", async () => { const { db, insertedTables } = createFakeDb(); const contextA = VALID_HEADERS; diff --git a/server/src/middleware/idle-admission.ts b/server/src/middleware/idle-admission.ts new file mode 100644 index 0000000000..c9de5f03ec --- /dev/null +++ b/server/src/middleware/idle-admission.ts @@ -0,0 +1,99 @@ +import { markIdleIngressTracked } from "../services/idle-local-work.js"; +import type { Application, Request, RequestHandler } from "express"; +import { beginIdleTrackedWork, isIdleTaskDrainActive } from "../services/task-admission.js"; + +type RequestWork = { pending: number; ended: boolean; closed: boolean; asyncRoute: boolean; routeSequence: number; active: boolean; finish: () => void }; +const requests = new WeakMap(); +function settle(work: RequestWork) { + if (work.active && (work.ended || (work.closed && work.asyncRoute)) && work.pending === 0) { + work.active = false; + work.finish(); + } +} + +// Only the exact control endpoint and health probes bypass admission. The +// control route still authenticates each operation. GET is otherwise tracked: +// OAuth callbacks and tool streams can create work too. +// Reachability does not exempt health/bootstrap or control mutations from +// tracking. Only the task-drain read skips request tracking to avoid counting +// its own scan; actorMiddleware counts its preceding authentication separately. +function isControlRequest(req: Request) { + const path = req.path.replace(/\/$/, ""); + return (path === "/api/instance/task-drain" && ["GET", "POST", "DELETE"].includes(req.method)) || + (path === "/api/health" && ["GET", "HEAD"].includes(req.method)); +} + +/** Install before body parsers, auth, webhooks and tool ingress. */ +export const idleAdmissionMiddleware: RequestHandler = (req, res, next) => { + const controlRequest = isControlRequest(req); + if (!controlRequest && isIdleTaskDrainActive()) { + res.set("Retry-After", "1").status(503).json({ error: "instance_preparing_to_sleep" }); + return; + } + if (controlRequest && req.method === "GET" && req.path.replace(/\/$/, "") === "/api/instance/task-drain") { + next(); + return; + } + const work: RequestWork = { pending: 0, ended: false, closed: false, asyncRoute: false, routeSequence: 0, active: true, finish: beginIdleTrackedWork() }; + requests.set(req, work); + res.once("close", () => { work.closed = true; settle(work); }); + // close/aborted are NOT completion: an async handler can still be committing + // a mutation after its client leaves. Wait for the application's end and all + // returned handler promises. Callback-only handlers abandoned without end + // stay blockers; an async route proves completion when its promise settles. + const end = res.end; + res.end = function (this: typeof res, ...args: Parameters) { + try { return end.apply(this, args); } + finally { work.ended = true; settle(work); } + } as typeof end; + next(); +}; + +// Express 5 exposes its router stack. Walk it once after route registration, +// preserving router objects and error-handler arity. Waiting only for finish +// loses async work after res.json(), including the disconnected-client case. +// Keep this adapter isolated and exercise it against real Express in tests. +type Layer = { handle: Function & { stack?: Layer[] }; route?: { stack: Layer[] } }; +export function trackIdleRequestHandlers(app: Application): void { + const seen = new Set(); + const visit = (stack: Layer[], route = false) => { + for (const layer of stack) { + if (seen.has(layer)) continue; + seen.add(layer); + if (layer.route) { visit(layer.route.stack, true); continue; } + if (layer.handle.stack) { visit(layer.handle.stack); continue; } + const original = layer.handle; + if (original === idleAdmissionMiddleware) continue; + const invoke = (req: Request, args: unknown[]) => { + const work = requests.get(req); + if (!work) return original(...args); + if (!work.active) { work.active = true; work.finish = beginIdleTrackedWork(); } + work.pending++; + const routeSequence = route ? ++work.routeSequence : undefined; + if (route) work.asyncRoute = false; + let calledNext = false; + const originalNext = args[args.length - 1] as (...args: unknown[]) => unknown; + args[args.length - 1] = (...nextArgs: unknown[]) => { calledNext = true; return originalNext(...nextArgs); }; + const finish = () => { work.pending--; settle(work); }; + try { + const result = original(...args); + if (result && typeof result.then === "function") { + return Promise.resolve(result).finally(() => { + // An async validation middleware which calls next() does not + // prove completion of a later callback-only route handler. + if (route && !calledNext && work.routeSequence === routeSequence) work.asyncRoute = true; + finish(); + }); + } + finish(); + return result; + } catch (error) { finish(); throw error; } + }; + layer.handle = original.length === 4 + ? function (error: unknown, req: Request, res: unknown, next: unknown) { return invoke(req, [error, req, res, next]); } + : function (req: Request, res: unknown, next: unknown) { return invoke(req, [req, res, next]); }; + } + }; + visit(app.router.stack as Layer[]); + markIdleIngressTracked(); +} diff --git a/server/src/realtime/live-events-ws.ts b/server/src/realtime/live-events-ws.ts index b9989cd2fa..67e836a0cf 100644 --- a/server/src/realtime/live-events-ws.ts +++ b/server/src/realtime/live-events-ws.ts @@ -9,6 +9,7 @@ import type { DeploymentMode, LiveEvent } from "@paperclipai/shared"; import type { BetterAuthSessionResult } from "../auth/better-auth.js"; import { logger } from "../middleware/logger.js"; import { subscribeCompanyLiveEvents } from "../services/live-events.js"; +import { trackIdleWork } from "../services/task-admission.js"; interface WsSocket { readyState: number; @@ -384,7 +385,9 @@ export function setupLiveEventsWebSocketServer( return; } - void authorizeUpgrade(db, req, companyId, url, { + // Upgrade admission precedes this async authentication. An idle hold or + // socket close must not make its accepted database writes disappear. + void trackIdleWork(authorizeUpgrade(db, req, companyId, url, { deploymentMode: opts.deploymentMode, resolveSessionFromHeaders: opts.resolveSessionFromHeaders, resolveCloudActor: opts.resolveCloudActor, @@ -411,7 +414,7 @@ export function setupLiveEventsWebSocketServer( .catch((err) => { logger.error({ err, path: req.url }, "failed websocket upgrade authorization"); rejectUpgrade(socket, "500 Internal Server Error", "upgrade failed"); - }); + })); }); return wss; diff --git a/server/src/routes/instance-settings.ts b/server/src/routes/instance-settings.ts index 6763a3ab2c..1e1fd26c41 100644 --- a/server/src/routes/instance-settings.ts +++ b/server/src/routes/instance-settings.ts @@ -1,3 +1,4 @@ +import { trackIdleWork } from "../services/task-admission.js"; import { Router, type Request } from "express"; import { companies, type Db } from "@paperclipai/db"; import { @@ -6,13 +7,14 @@ import { patchInstanceGeneralSettingsSchema, startTaskDrainRequestSchema, } from "@paperclipai/shared"; -import { forbidden } from "../errors.js"; +import { conflict, forbidden } from "../errors.js"; import { cloudTenantPrimaryCompanyId, getCloudStackContext, isCloudManagedInstance, } from "../services/cloud-instance.js"; import { getHiddenSettings } from "../services/settings-visibility.js"; +import { readIdleSleepSafety } from "../services/idle-sleep-safety.js"; import { validate } from "../middleware/validate.js"; import { logger } from "../middleware/logger.js"; import { @@ -106,7 +108,7 @@ function assertCanManageInstanceSettings(req: Request) { let taskDrainTransitionQueue: Promise = Promise.resolve(); function withTaskDrainTransition(run: () => Promise): Promise { - const turn = taskDrainTransitionQueue.then(run); + const turn = trackIdleWork(taskDrainTransitionQueue.then(run)); // Normalize to a settled void promise for the next caller in line, so a // rejected transition (a failed audit write, for example) cannot wedge // every later transition behind it. @@ -297,6 +299,15 @@ export function instanceSettingsRoutes(db: Db) { router.get("/instance/task-drain", async (req, res) => { assertBoardOrgAccess(req); + if (req.query.idleSleepSafety === "1") { + // The report covers every company in this process. Ordinary company + // 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); + res.json({ ...heartbeat.getTaskDrainStatus(), idleSleepSafety }); + return; + } res.json(heartbeat.getTaskDrainStatus()); }); @@ -315,7 +326,12 @@ export function instanceSettingsRoutes(db: Db) { // startedAt reflects the moment this request actually took effect, // not the moment it arrived and was queued behind another transition. const drain = await withTaskDrainTransition(async () => { - const computed = heartbeat.computeTaskDrain({ ttlMs }); + const prior = heartbeat.getTaskDrainStatus(); + if (prior?.ownerId || (req.body.purpose === "idle" && prior?.draining)) { + throw conflict("Another task drain is already active"); + } + const computed = heartbeat.computeTaskDrain({ ttlMs, + ...(req.body.purpose === "idle" ? { purpose: "idle" as const } : {}) }); // One transaction for every company's audit row, so a write that // succeeds for one company and fails for another never leaves a // partial activity history behind — either every company gets the @@ -339,6 +355,7 @@ export function instanceSettingsRoutes(db: Db) { details: { startedAt: computed.startedAt, expiresAt: computed.expiresAt, + ...(computed.ownerId ? { purpose: "idle", ownerId: computed.ownerId } : {}), }, }, postCommitActivityPublications), ), @@ -369,6 +386,10 @@ export function instanceSettingsRoutes(db: Db) { // queued transition. const wasActive = await withTaskDrainTransition(async () => { const priorStatus = heartbeat.getTaskDrainStatus(); + if ((priorStatus.ownerId || req.query.ownerId !== undefined) && + (typeof req.query.ownerId !== "string" || req.query.ownerId !== priorStatus.ownerId)) { + throw conflict("Task drain ownership changed"); + } // Read wasActive once, here, and use this same value for the audit // detail and the response body below. A TTL that expires between two // separate reads would otherwise make the two values disagree. diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index b913ba6dbd..81b0b5e2c4 100644 --- a/server/src/routes/openapi.ts +++ b/server/src/routes/openapi.ts @@ -6611,7 +6611,8 @@ registry.registerPath({ tags: ["instance"], summary: "Get the task-drain status for this process only; quiescent counts in-process work, and a process restart clears it even when the database still holds running rows", - responses: { 200: r.ok(), 401: r.unauthorized }, + request: { query: z.object({ ownerId: z.string().uuid().optional(), idleSleepSafety: z.literal("1").optional().describe("Instance admins can request a conservative durable-work report while admission and ingress are held. Unknown or present work must prevent automatic sleep.") }) }, + responses: { 200: r.ok(), 401: r.unauthorized, 403: r.forbidden }, }); registry.registerPath({ @@ -6626,6 +6627,7 @@ registry.registerPath({ 400: r.badRequest, 401: r.unauthorized, 403: r.forbidden, + 409: r.conflict, }, }); @@ -6634,7 +6636,8 @@ registry.registerPath({ path: "/api/instance/task-drain", tags: ["instance"], summary: "End a task drain and restore run admission", - responses: { 200: r.ok(), 401: r.unauthorized, 403: r.forbidden }, + request: { query: z.object({ ownerId: z.string().uuid().optional() }) }, + responses: { 200: r.ok(), 401: r.unauthorized, 403: r.forbidden, 409: r.conflict }, }); registry.registerPath({ diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 07ba10ebe4..d28e072a2e 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -1,3 +1,4 @@ +import { beginIdleTrackedWork } from "./task-admission.js"; import { hasStopOnlyCleanup, prepareSandboxStopAndRetain, readStopOnlyCleanup, settleStopOnlyCleanup, stopOnlyCleanupKey } from "./sandbox-stop-and-retain.js"; import { readEnvironmentCreationCleanupError } from "@paperclipai/plugin-sdk"; import { preserveEnvironmentSyncOutErrorDiagnostic } from "./environment-sync-out-error.js"; @@ -1327,6 +1328,16 @@ function createSandboxEnvironmentDriver( // write, so a flushed row and a synchronous row are indistinguishable to the // sweep. const deferredOrphanCleanups: DeferredOrphanCleanupRecord[] = []; + // Keep the token through queue splices, failed flushes and buffer overflow. + // Release only when the database durably owns cleanup of this exact orphan. + const orphanIdleWork = new Map void>(); + const orphanIdleKey = (record: DeferredOrphanCleanupRecord) => JSON.stringify([ + record.companyId, record.provider, record.providerLeaseId ?? record, + ]); + const trackOrphanIdleWork = (record: DeferredOrphanCleanupRecord) => { + const key = orphanIdleKey(record); + if (!orphanIdleWork.has(key)) orphanIdleWork.set(key, beginIdleTrackedWork()); + }; const DEFERRED_ORPHAN_CLEANUP_FAILURE_REASON = "acquire_rejected_teardown_failed"; // The in-process buffer alone loses an orphan on a restart. So the driver also @@ -1345,6 +1356,7 @@ function createSandboxEnvironmentDriver( // Add one orphan record to the in-process buffer. Report `false` only when the // buffer is full, so the caller keeps the error log as the last durable handle. const enqueueDeferredOrphanCleanup = (record: DeferredOrphanCleanupRecord): boolean => { + trackOrphanIdleWork(record); // Dedup by the provider lease id, so a repeated failure for the same orphan // never buffers it twice. Each acquire mints a unique provider lease id, so // two distinct orphans never collide. Skip the dedup for a null lease id, @@ -1406,6 +1418,9 @@ function createSandboxEnvironmentDriver( // leaves the spooled copy for a later flush, which re-inserts a duplicate // the idempotent teardown handles. This order never loses the orphan. await orphanCleanupSpool.remove(record); + const idleKey = orphanIdleKey(record); + orphanIdleWork.get(idleKey)?.(); + orphanIdleWork.delete(idleKey); recovered += 1; } catch { // The database is still down. Re-queue the record for a later flush, @@ -1544,6 +1559,7 @@ function createSandboxEnvironmentDriver( cause: unknown, cleanupWriteError: unknown, ): Promise => { + trackOrphanIdleWork(record); const persisted = await orphanCleanupSpool.append(record); const buffered = enqueueDeferredOrphanCleanup(record); logger.error( diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e0bb61500e..d89725adf5 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -677,6 +677,7 @@ import { } from "./effective-run-config-fingerprints.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; import { serverVersion } from "../version.js"; +import { computeTaskDrain, applyTaskDrain, startTaskDrain, stopTaskDrain, readTaskDrain } from "./task-admission.js"; const MAX_LIVE_LOG_CHUNK_BYTES = 8 * 1024; @@ -1352,61 +1353,8 @@ const nativeSessionResumeDispatchTimers = new Map< string, ReturnType >(); -// Task drain: an operator-controlled hold on new run admission, so a caller -// can wait for active work to finish before it stops the process. The state -// lives in process memory only — a process restart clears it — and it sits at -// module scope like activeRunExecutions above, so both the pure -// resolveHeartbeatSchedulingSuppression() check and every heartbeatService() -// instance see the same drain. -let taskDrainState: { startedAt: Date; expiresAt: Date | null } | null = null; - -function readTaskDrain( - now: Date, -): { startedAt: Date; expiresAt: Date | null } | null { - if ( - taskDrainState && - taskDrainState.expiresAt !== null && - taskDrainState.expiresAt.getTime() <= now.getTime() - ) { - taskDrainState = null; - } - return taskDrainState; -} - -/** Compute the drain a start call would apply, without changing state. */ -export function computeTaskDrain(opts: { ttlMs?: number | null } = {}): { - startedAt: Date; - expiresAt: Date | null; -} { - const startedAt = new Date(); - const ttlMs = opts.ttlMs ?? null; - const expiresAt = - ttlMs === null ? null : new Date(startedAt.getTime() + ttlMs); - return { startedAt, expiresAt }; -} - -/** Assign the given drain as the current task-drain state. */ -export function applyTaskDrain(drain: { - startedAt: Date; - expiresAt: Date | null; -}): void { - taskDrainState = drain; -} - -export function startTaskDrain(opts: { ttlMs?: number | null } = {}): { - startedAt: Date; - expiresAt: Date | null; -} { - const drain = computeTaskDrain(opts); - applyTaskDrain(drain); - return drain; -} - -export function stopTaskDrain(): { wasActive: boolean } { - const wasActive = readTaskDrain(new Date()) !== null; - taskDrainState = null; - return { wasActive }; -} +// Shared with HTTP admission so an idle hold fences work before inspection. +export { computeTaskDrain, applyTaskDrain, startTaskDrain, stopTaskDrain } from "./task-admission.js"; /** * Report the task-drain state for this process only. `activeRuns` and @@ -1421,11 +1369,13 @@ export function getTaskDrainStatus(): { activeRuns: number; pendingWakes: number; quiescent: boolean; + ownerId?: string; } { const state = readTaskDrain(new Date()); const activeRuns = activeRunExecutionPromises.size; const pendingWakes = activeWakeupPromises.size; return { + ...(state?.ownerId ? { ownerId: state.ownerId } : {}), draining: state !== null, startedAt: state?.startedAt ?? null, expiresAt: state?.expiresAt ?? null, diff --git a/server/src/services/idle-local-work.ts b/server/src/services/idle-local-work.ts new file mode 100644 index 0000000000..1095c298f8 --- /dev/null +++ b/server/src/services/idle-local-work.ts @@ -0,0 +1,43 @@ +import { promises as fs } from "node:fs"; +import path from "node:path"; +import { resolvePaperclipInstanceRoot } from "../home-paths.js"; +import { idleWorkSnapshot } from "./task-admission.js"; + +export type IdleLocalWork = "none" | "present" | "unknown"; +const spoolDirectories = new Set(); +export function registerIdleSpoolDirectory(directory: string): void { spoolDirectories.add(path.resolve(directory)); } +export const idleAccountingSpoolPath = () => path.join(resolvePaperclipInstanceRoot(), "accounting-receipts"); +export const idleOrphanSpoolPath = () => process.env.SANDBOX_ORPHAN_CLEANUP_SPOOL_DIR ?? + path.join(resolvePaperclipInstanceRoot(), "data", "sandbox-orphan-cleanup"); +let startupComplete = false; +let ingressTracked = false; +let scheduledBackups = false; +export function markIdleIngressTracked() { ingressTracked = true; } +export function markIdleStartupComplete(options: { scheduledBackups: boolean }) { + scheduledBackups = options.scheduledBackups; + startupComplete = true; +} + +/** Any entry, including malformed JSON, a temp file or a failed probe, is + * evidence to retain. Recovery readers intentionally skip some such files; + * they cannot certify that stopping the process is safe. */ +export async function inspectIdleSpool(directory: string): Promise { + try { + const info = await fs.lstat(directory); + if (!info.isDirectory() || info.isSymbolicLink()) return "unknown"; + const dir = await fs.opendir(directory); + try { return (await dir.read()) === null ? "none" : "present"; } + finally { await dir.close(); } + } catch (error) { + return (error as NodeJS.ErrnoException).code === "ENOENT" ? "none" : "unknown"; + } +} + +export async function readIdleLocalWork(): Promise { + if (!startupComplete || !ingressTracked) return "unknown"; + if (scheduledBackups || idleWorkSnapshot().active !== 0) return "present"; + const directories = new Set([...spoolDirectories, idleAccountingSpoolPath(), idleOrphanSpoolPath()]); + const results = await Promise.all([...directories].map(inspectIdleSpool)); + if (results.includes("unknown")) return "unknown"; + return results.includes("present") ? "present" : "none"; +} diff --git a/server/src/services/idle-sleep-safety.ts b/server/src/services/idle-sleep-safety.ts new file mode 100644 index 0000000000..574b57c11f --- /dev/null +++ b/server/src/services/idle-sleep-safety.ts @@ -0,0 +1,125 @@ +import { idleWorkSnapshot } from "./task-admission.js"; +import { readIdleLocalWork, type IdleLocalWork } from "./idle-local-work.js"; +import type { Db } from "@paperclipai/db"; +import { sql } from "drizzle-orm"; + +export interface IdleSleepDrainStatus { + ownerId?: string; + draining: boolean; + startedAt: Date | null; + expiresAt: Date | null; + activeRuns: number; + pendingWakes: number; +} + +export interface IdleSleepSafety { + version: 1; + backgroundWork: "none" | "present" | "unknown"; +} + +// 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 +// have a durable wake path. In particular, an empty in-memory run set is not +// evidence that a restarted process has no persisted work to recover. +const WORK_CHECKS = [ + `SELECT 1 FROM agents WHERE status = 'running' OR + (status NOT IN ('terminated', 'paused', 'pending_approval') AND + runtime_config->'heartbeat'->>'enabled' = 'true' AND + COALESCE((runtime_config->'heartbeat'->>'intervalSec')::numeric, 0) > 0)`, + `SELECT 1 FROM heartbeat_runs WHERE + status NOT IN ('succeeded', 'failed', 'cancelled', 'timed_out', 'interrupted') + OR scheduled_retry_at IS NOT NULL OR cost_accounting_pending`, + `SELECT 1 FROM agent_instruction_working_copies WHERE + state NOT IN ('saved', 'unchanged', 'resolved', 'unavailable', 'conflict') + OR process_stopped_at IS NULL OR next_attempt_at IS NOT NULL + OR receipt ? 'baseline' OR receipt->>'cleanupPending' = 'true'`, + `SELECT 1 FROM budget_reservations WHERE state NOT IN ('settled', 'released')`, + // Persistent machine credentials can receive work without a human opening + // the instance. They need an ingress wake contract before we can sleep. + `SELECT 1 FROM agent_api_keys WHERE revoked_at IS NULL`, + `SELECT 1 FROM board_api_keys WHERE revoked_at IS NULL + AND (expires_at IS NULL OR expires_at > now())`, + `SELECT 1 FROM agent_wakeup_requests WHERE + status NOT IN ('completed', 'failed', 'cancelled', 'skipped', 'timed_out')`, + `SELECT 1 FROM issues WHERE status NOT IN ('done', 'cancelled')`, + // A completed issue can still need its first watchdog review, including a + // retry after immediate evaluation failed before creating a review/run. + `SELECT 1 FROM issue_watchdogs WHERE status <> 'disabled'`, + `SELECT 1 FROM workspace_operations WHERE status NOT IN ('succeeded', 'failed', 'cancelled')`, + `SELECT 1 FROM environment_leases WHERE status NOT IN ('released', 'expired')`, + `SELECT 1 FROM routines WHERE status NOT IN ('paused', 'archived')`, + `SELECT 1 FROM routine_runs WHERE status NOT IN ('completed', 'failed', 'skipped', 'cancelled')`, + `SELECT 1 FROM plugin_jobs WHERE status <> 'paused'`, + `SELECT 1 FROM plugin_job_runs WHERE status NOT IN ('succeeded', 'failed', 'cancelled')`, + `SELECT 1 FROM plugin_webhook_deliveries WHERE status NOT IN ('succeeded', 'failed')`, + `SELECT 1 FROM tool_connections WHERE status <> 'archived'`, + `SELECT 1 FROM issue_recovery_actions WHERE status NOT IN ('resolved', 'cancelled')`, + `SELECT 1 FROM status_cards WHERE archived_at IS NULL`, + `SELECT 1 FROM external_objects WHERE NOT is_terminal OR next_refresh_at IS NOT NULL`, + // These less common work sources fail closed on any retained state. Their + // terminal-state exceptions can be added with tests for the owning service. + ...[ + "chat_endpoints", "chat_deliveries", "chat_publications", "chat_actions", + "chat_completion_deliveries", "connection_event_deliveries", + "connection_intent_deliveries", "tool_action_requests", "tool_action_deliveries", + "issue_question_response_deliveries", "workspace_runtime_services", + "execution_workspace_runtime_leases", "pipeline_automation_executions", + "native_run_finalizations", "company_transfer_runs", "decisions", "decision_effect_executions", + "decision_archive_notification_outbox", "browser_use_sessions", "browser_use_runs", + "browser_use_browsers", "environment_custom_image_setup_sessions", "feedback_exports", + "adapter_auth_sessions", "company_secret_proposals", "execution_workspaces", + "mcp_oauth_grants", "mcp_mutation_receipts", "mcp_event_subscriptions", + "mcp_event_deliveries", "mcp_attachment_uploads", "dot_agent_bindings", + "dot_runner_assignments", "dot_runner_operations", "dot_mailbox_items", + ].map((table) => `SELECT 1 FROM ${table}`), +] as const; + +function sameQuietHold(before: IdleSleepDrainStatus, after: IdleSleepDrainStatus, now: number): boolean { + return before.ownerId === after.ownerId && before.draining && after.draining && before.activeRuns === 0 && before.pendingWakes === 0 + && after.activeRuns === 0 && after.pendingWakes === 0 + && before.startedAt !== null && after.startedAt?.getTime() === before.startedAt.getTime() + && (after.expiresAt === null || after.expiresAt.getTime() > now); +} + +/** Inventory persisted work after the task-drain admission hold lands. + * Only a caller holding the exact idle owner may receive an empty report. + */ +export async function readIdleSleepSafety( + db: Db, + getDrainStatus: () => IdleSleepDrainStatus, + now: () => number = Date.now, + ownerId?: string, + inspectLocalWork: () => Promise = readIdleLocalWork, +): Promise { + const unknown: IdleSleepSafety = { version: 1, backgroundWork: "unknown" }; + const before = getDrainStatus(); + const localBefore = idleWorkSnapshot(); + if (!sameQuietHold(before, before, now())) return unknown; + try { + 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')`); + 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" }); + if (blocked === undefined || !sameQuietHold(before, getDrainStatus(), now())) return unknown; + if (blocked) return { version: 1, backgroundWork: "present" }; + if (!ownerId || ownerId !== before.ownerId || before.expiresAt === null || localBefore.active !== 0) return unknown; + const local = await inspectLocalWork(); + const after = getDrainStatus(); + const localAfter = idleWorkSnapshot(); + if (!sameQuietHold(before, after, now()) || localAfter.active !== 0 || + localBefore.generation !== localAfter.generation) return unknown; + return { version: 1, backgroundWork: local }; + + } catch { + // Missing migrations, unknown state, malformed configuration, timeouts and + // database failures never authorize stopping the application. Do not leak + // SQL, configuration or tenant data through this control response. + return unknown; + } +} diff --git a/server/src/services/sandbox-orphan-cleanup-spool.ts b/server/src/services/sandbox-orphan-cleanup-spool.ts index a1729735ed..a6f97b8e8f 100644 --- a/server/src/services/sandbox-orphan-cleanup-spool.ts +++ b/server/src/services/sandbox-orphan-cleanup-spool.ts @@ -1,7 +1,7 @@ import { promises as fs } from "node:fs"; import path from "node:path"; import { createHash } from "node:crypto"; -import { resolvePaperclipInstanceRoot } from "../home-paths.js"; +import { idleOrphanSpoolPath, registerIdleSpoolDirectory } from "./idle-local-work.js"; /** * One orphan pending-cleanup record. The acquire builds it when the conditional @@ -72,13 +72,6 @@ const RECORD_FILE_SUFFIX = ".json"; // without a torn write. let tempFileCounter = 0; -function resolveDefaultSpoolDir(): string { - return ( - process.env.SANDBOX_ORPHAN_CLEANUP_SPOOL_DIR ?? - path.resolve(resolvePaperclipInstanceRoot(), "data", "sandbox-orphan-cleanup") - ); -} - // Build the stable file name for one record. Key on the provider lease id, the // value the teardown needs, so a repeated append for the same orphan reuses the // same file. A null provider lease id carries no identity, so fall back to a @@ -147,7 +140,8 @@ async function writeFileDurable(absPath: string, data: string): Promise { * directory for a test. */ export function createSandboxOrphanCleanupSpool(spoolDir?: string): SandboxOrphanCleanupSpool { - const dir = spoolDir ?? resolveDefaultSpoolDir(); + const dir = spoolDir ?? idleOrphanSpoolPath(); + registerIdleSpoolDirectory(dir); return { async append(record) { diff --git a/server/src/services/task-admission.ts b/server/src/services/task-admission.ts new file mode 100644 index 0000000000..a72fd752e7 --- /dev/null +++ b/server/src/services/task-admission.ts @@ -0,0 +1,50 @@ +import { randomUUID } from "node:crypto"; + +export type TaskDrainState = { startedAt: Date; expiresAt: Date | null; ownerId?: string }; +let taskDrainState: TaskDrainState | null = null; + +export function readTaskDrain(now: Date): TaskDrainState | null { + if (taskDrainState?.expiresAt && taskDrainState.expiresAt.getTime() <= now.getTime()) taskDrainState = null; + return taskDrainState; +} + +export function computeTaskDrain(opts: { ttlMs?: number | null; purpose?: "idle" } = {}): TaskDrainState { + const startedAt = new Date(); + return { startedAt, expiresAt: opts.ttlMs == null ? null : new Date(startedAt.getTime() + opts.ttlMs), + ...(opts.purpose === "idle" ? { ownerId: randomUUID() } : {}) }; +} + +export function applyTaskDrain(drain: TaskDrainState): void { taskDrainState = drain; } +export function startTaskDrain(opts: { ttlMs?: number | null; purpose?: "idle" } = {}): TaskDrainState { + const drain = computeTaskDrain(opts); + applyTaskDrain(drain); + return drain; +} +export function stopTaskDrain(): { wasActive: boolean } { + const wasActive = readTaskDrain(new Date()) !== null; + taskDrainState = null; + return { wasActive }; +} +export function isIdleTaskDrainActive(): boolean { return Boolean(readTaskDrain(new Date())?.ownerId); } + +// Tokens count work until its actual promise settles, including after HTTP +// disconnects and after a cleanup batch has been removed from its input queue. +// The generation also catches work which starts AND finishes during a scan. +let generation = 0; +let active = 0; +export function beginIdleTrackedWork(): () => void { + generation++; + active++; + let finished = false; + return () => { + if (finished) return; + finished = true; + active--; + generation++; + }; +} +export function trackIdleWork(work: Promise): Promise { + const finish = beginIdleTrackedWork(); + return work.finally(finish); +} +export function idleWorkSnapshot() { return { generation, active }; } diff --git a/server/src/services/usage-receipts.ts b/server/src/services/usage-receipts.ts index 3606cd3a0c..8233828757 100644 --- a/server/src/services/usage-receipts.ts +++ b/server/src/services/usage-receipts.ts @@ -1,3 +1,4 @@ +import { beginIdleTrackedWork } from "./task-admission.js"; import { priceAnthropicReceipt } from "./anthropic-pricing.js"; import { priceCodexReceipt } from "./codex-pricing.js"; import { randomUUID } from "node:crypto"; @@ -8,7 +9,7 @@ import { and, eq, isNull, sql } from "drizzle-orm"; import { heartbeatRuns, runUsageReceipts, type Db } from "@paperclipai/db"; import type { AdapterExecutionResult, AdapterUsageCheckpoint } from "@paperclipai/adapter-utils"; import { pricingProvenanceSchema, usdToUnits } from "@paperclipai/shared"; -import { resolvePaperclipInstanceRoot } from "../home-paths.js"; +import { idleAccountingSpoolPath, registerIdleSpoolDirectory } from "./idle-local-work.js"; import { conflict, notFound } from "../errors.js"; import { logger } from "../middleware/logger.js"; import { receiptFingerprint } from "./receipt-fingerprint.js"; @@ -37,9 +38,10 @@ const envelopeSchema = z.object({ adapterType: z.string().max(100), receipt: checkpointSchema, }); export type UsageReceiptEnvelope = z.infer; -export const usageReceiptSpoolPath = () => path.join(resolvePaperclipInstanceRoot(), "accounting-receipts"); +export const usageReceiptSpoolPath = idleAccountingSpoolPath; export async function spoolUsageReceipt(envelope: UsageReceiptEnvelope, directory = usageReceiptSpoolPath()) { + registerIdleSpoolDirectory(directory); const parsed = envelopeSchema.parse(envelope); await ensureDurableDirectory(directory); const target = path.join(directory, `${parsed.id}.json`); @@ -185,6 +187,7 @@ export async function recoverPendingRunUsageReceipts(db: Db, input: { companyId: } export async function createRunUsageRecorder(db: Db, input: { companyId: string; runId: string; adapterType: string }, directory = usageReceiptSpoolPath()) { + registerIdleSpoolDirectory(directory); const sourceId = randomUUID(); // Verify durable storage before starting paid work, including a directory // sync. The database binding fences checkpoints from replaced controllers. @@ -211,26 +214,31 @@ export async function createRunUsageRecorder(db: Db, input: { companyId: string; } let chain = Promise.resolve(); let captureFailed = bound[0].usageJson?.accountingCaptureFailed === true; + let finishFailedCapture: (() => void) | undefined; async function persistFailure() { if (!captureFailed) return; - await withAccountingTransaction(db, input.companyId, async tx => { - await tx.update(heartbeatRuns).set({ + const persisted = await withAccountingTransaction(db, input.companyId, async tx => { + const rows = await tx.update(heartbeatRuns).set({ costAccountingPending: true, accountingLastError: "usage_capture_failed", usageJson: sql`coalesce(${heartbeatRuns.usageJson}, '{}'::jsonb) || '{"accountingCaptureFailed":true,"accountingReceiptReady":false}'::jsonb`, }).where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId), - isNull(heartbeatRuns.costAccountedAt), sql`${heartbeatRuns.usageJson}->>'accountingReceiptSourceId' = ${sourceId}`)); + isNull(heartbeatRuns.costAccountedAt), sql`${heartbeatRuns.usageJson}->>'accountingReceiptSourceId' = ${sourceId}`)).returning({ id: heartbeatRuns.id }); + return rows.length === 1; }); + if (persisted) { finishFailedCapture?.(); finishFailedCapture = undefined; } } function enqueue(action: () => Promise) { + const finishQueuedCapture = beginIdleTrackedWork(); const result = chain.then(action).catch(async error => { captureFailed = true; + finishFailedCapture ??= beginIdleTrackedWork(); // Persist before exposing rejection: the adapter's flush may throw before // complete(), and failure finalization must not settle an older receipt. try { await persistFailure(); } catch (failure) { logger.error({ err: failure, runId: input.runId }, "Usage capture failure fence requires retry before finalization"); } throw error; }); - chain = result.then(() => undefined, () => undefined); + chain = result.then(() => undefined, () => undefined).finally(finishQueuedCapture); return result; } async function capture(raw: AdapterUsageCheckpoint) {