From 3367b75ccce34d02f355cda1f1ed3fe0b34cf93d Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Thu, 8 Oct 2026 08:35:45 -0700 Subject: [PATCH] fix: fence accepted work and cleanup before idle sleep (#15522) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A host can stop an idle instance to reduce unused compute. > - Zero active runs do not prove that requests, saved work, or cleanup are complete. > - A client can disconnect while its request still writes, and failed cleanup can remain only in memory or on disk. > - This pull request holds new admission and checks accepted work, cleanup, and persisted work under an owned drain. > - The host gets an empty report only while those checks remain valid. ## Linked Issues or Issue Description Refs #13413. That companion PR uses the same owned-hold vocabulary for runtime services. This PR covers HTTP requests, scheduler work, accounting, cleanup, and the instance work inventory. It does not include the preview gateway changes. **What existing behavior does this improve?** The instance task-drain API reports process counters. It does not prove that a host can safely stop the instance for idle sleep. **Current behavior** A quiet run set can coexist with an unfinished request, cleanup after a completed run, future work, or a failed accounting write. **Proposed behavior** Provide a bounded owned idle hold. Block new ingress, track accepted handler promises, inspect durable and local work, and return `none` only when the same hold stays quiet through the checks. Keep normal deployment drains compatible. **Reason and benefit** Hosts can identify eligible idle instances without treating a disconnect or failed cleanup write as completed work. ## What Changed - Add `purpose: "idle"`, a bounded TTL, unique owners, and owner-checked release to task drain. Existing holds cannot be replaced by another API request. - Gate HTTP ingress before parsers, auth, webhooks, and MCP. Gate new WebSocket upgrades. Track async handlers in nested Express routers and error middleware until they settle, even after the response or client disconnect. - Count accepted live-event WebSocket authentication through settlement, even after disconnects. Count detached built-in agent, managed-home and runtime-service startup reconciliation after readiness. - Keep health and control mutations tracked. Count control-request authentication separately from the read-only report, including concurrent user/company/membership writes. - Pause new scheduler admissions during idle holds. Count work already in flight, including database backups, and reject scans whose work generation changes. Periodic backups block sleep without a host wake schedule. - Inspect accounting and orphan-cleanup spool directories without skipping temporary or malformed entries. Retain orphan tokens through queue splices, flush failures, and buffer overflow. Keep failed usage capture counted until its database failure fence is written. - Check persisted work across companies in a bounded read-only transaction. Include deferred agent-file cleanup and saved watchdogs whose watched issues are complete. Enabled plugins and unsupported retained work remain blockers. - Require the exact idle owner and completed startup tracking before returning an empty report. Keep reports free of tenant details. - Document the hosting protocol, retry behavior, conservative blockers, and the remaining external provider-stop race. ## Verification - Current head: `8d9a599b732c2047b8c671d4799065d7c90a3567`, rebased on master `941a3fa991aeb97eb1ac390c65b7973b5f6de1ad`. Both heartbeat helper extractions are preserved. GitHub confirms no merge conflicts. - Focused heartbeat renderer/run-log, drain, control-auth, admission, route and PostgreSQL inventory checks: 234 tests passed in 10 suites after the rebase. - The POST task-drain contract includes the expected `409` conflict response. The 84 OpenAPI and instance-settings route tests and server typecheck passed after that final documentation fix. - Final accepted-upgrade/startup regression run: 104 tests passed in five suites, including success and failure after readiness or disconnect. Final server typecheck also passed. - `pnpm -r typecheck`: passed. - `pnpm build`: passed. - `pnpm install --frozen-lockfile` and the PR diff secret scan passed. No dependency or lockfile changes are added by this PR. - The prior verification covered HTTP disconnects and early responses, async error handlers, concurrent authentication writes, signed bootstrap, saved watchdogs, backup promises, orphan cleanup, spools and failed accounting fences. Those tests passed. - The last full local `pnpm test:run` stopped in the general-server phase with 16,658 passing tests and five skill/connector fixture failures caused by an ancestor workspace skill directory. That full local run preceded this rebase and has not been repeated for the import conflict. Full current-head CI passed: 53 successful checks and two conditional skips, with no failures. - All eight review findings are fixed and their threads resolved, including accepted upgrade authentication, detached startup writes and the POST conflict contract. Current-head Apex review is 5/5 with no new findings; all eight review threads remain resolved. - No live provider stop or production change was performed. ## Risks - The hosting controller must use the owned protocol and hold external admission through its final validation and provider stop. A legacy drain cannot authorize idle sleep. During an idle hold, new requests receive 503 with `Retry-After: 1`; the host must handle queueing or retry before enabling this path. - This is a single-process protocol. An unexpected restart after the last validation can race an external provider stop. The host must serialize deploy/wake/sleep operations and bind the validation to the instance it stops. Multiple replicas need shared fencing. - Some retained state conservatively prevents sleep, including every enabled plugin. This PR does not promise that every inactive instance becomes eligible. - The HTTP adapter uses Express 5 router layers. Real Express tests cover nested routes, errors and disconnects. New routes must register before tracking is installed. Detached work must have durable state or explicit work tracking. - Unrecoverable in-memory cleanup debt keeps the instance awake until reconciliation. This change does not make such debt survive an unplanned process crash. No schema migration or provider configuration change is included. ## Model Used OpenAI GPT-6 via Codex, with reasoning, repository inspection, tool use and code execution. The exact deployment variant/model ID and context window are not exposed in this session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run focused tests locally and they pass; the full local fixture limitation and clean full CI result are documented above - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- doc/idle-sleep-safety.md | 98 +++++++ packages/shared/src/validators/instance.ts | 5 +- .../__tests__/cloud-runtime-identity.test.ts | 33 +++ .../src/__tests__/environment-runtime.test.ts | 19 +- .../__tests__/idle-accounting-work.test.ts | 76 ++++++ server/src/__tests__/idle-admission.test.ts | 103 +++++++ server/src/__tests__/idle-local-work.test.ts | 51 ++++ .../src/__tests__/idle-sleep-safety.test.ts | 253 ++++++++++++++++++ .../instance-settings-routes.test.ts | 51 ++++ server/src/__tests__/live-events-ws.test.ts | 39 +++ .../server-startup-feedback-export.test.ts | 63 +++++ server/src/app.ts | 28 +- server/src/index.ts | 63 +++-- server/src/middleware/auth.ts | 19 +- .../src/middleware/cloud-tenant-actor.test.ts | 61 ++++- server/src/middleware/idle-admission.ts | 99 +++++++ server/src/realtime/live-events-ws.ts | 7 +- server/src/routes/instance-settings.ts | 27 +- server/src/routes/openapi.ts | 7 +- server/src/services/environment-runtime.ts | 16 ++ server/src/services/heartbeat.ts | 60 +---- server/src/services/idle-local-work.ts | 43 +++ server/src/services/idle-sleep-safety.ts | 125 +++++++++ .../services/sandbox-orphan-cleanup-spool.ts | 12 +- server/src/services/task-admission.ts | 50 ++++ server/src/services/usage-receipts.ts | 20 +- 26 files changed, 1304 insertions(+), 124 deletions(-) create mode 100644 doc/idle-sleep-safety.md create mode 100644 server/src/__tests__/idle-accounting-work.test.ts create mode 100644 server/src/__tests__/idle-admission.test.ts create mode 100644 server/src/__tests__/idle-local-work.test.ts create mode 100644 server/src/__tests__/idle-sleep-safety.test.ts create mode 100644 server/src/middleware/idle-admission.ts create mode 100644 server/src/services/idle-local-work.ts create mode 100644 server/src/services/idle-sleep-safety.ts create mode 100644 server/src/services/task-admission.ts 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) {