feat(server): expose owned sleep blocker diagnostics

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-11-02 11:23:11 -06:00
1 parent 0748b90062
commit 4c4e4d3135
11 files changed
+805 -126

No files matched your search

@@ -3,6 +3,8 @@ import os from "node:os";
import path from "node:path";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { inspectIdleSpool, inspectIdleQuotaRecovery, idleQuotaRecoveryPath, readIdleLocalWork, markIdleIngressTracked, markIdleStartupComplete, registerIdleSpoolDirectory, registerIdleLocalWorkProbe } from "../services/idle-local-work.js";
import { idleAccountingSpoolPath } from "../services/idle-local-work.js";
import { createIdleSleepDiagnostics } from "../services/idle-sleep-diagnostics.js";
let root: string;
beforeEach(async () => { root = await fs.mkdtemp(path.join(os.tmpdir(), "idle-spool-")); });
@@ -66,6 +68,33 @@ describe("idle cleanup inspection", () => {
expect(await readIdleLocalWork()).toBe("none");
});
it("collects simultaneous local blockers and errors without reading entries", async () => {
vi.stubEnv("PAPERCLIP_HOME", root);
const orphan = path.join(root, "orphan-spool");
vi.stubEnv("SANDBOX_ORPHAN_CLEANUP_SPOOL_DIR", orphan);
markIdleIngressTracked(); markIdleStartupComplete({ scheduledBackups: true });
const accounting = idleAccountingSpoolPath();
await fs.mkdir(accounting, { recursive: true });
await fs.mkdir(orphan);
await fs.writeFile(path.join(accounting, "private-receipt.json"), "malformed");
await fs.writeFile(path.join(orphan, "partial.tmp"), "private resource");
const unregister = registerIdleLocalWorkProbe(() => { throw new Error("private recovery failure"); });
const diagnostic = createIdleSleepDiagnostics();
const read = vi.spyOn(fs, "readFile");
try {
expect(await readIdleLocalWork({ diagnostic: diagnostic.record })).toBe("unknown");
expect(diagnostic.report().checks).toEqual(expect.arrayContaining([
{ reason: "local.automatic_backups", state: "present" },
{ reason: "local.registered_probes", state: "unknown" },
{ reason: "local.accounting_spool", state: "present" },
{ reason: "local.orphan_spool", state: "present" },
]));
expect(diagnostic.report().complete).toBe(false);
expect(JSON.stringify(diagnostic.report())).not.toMatch(/private|partial.tmp|orphan-spool/);
expect(read).not.toHaveBeenCalled();
} finally { unregister(); markIdleStartupComplete({ scheduledBackups: false }); }
});
});
describe("idle credential recovery inspection", () => {
@@ -0,0 +1,128 @@
import { randomUUID } from "node:crypto";
import { eq, sql } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
import { agentApiKeys, agents, budgetPolicies, chatDeliveries, chatEndpoints, companies,
companyTransferRuns, connectionEventDeliveries, createDb, heartbeatRuns, issues,
routineTriggers, routines, toolApplications, toolConnections, type Db } from "@paperclipai/db";
import { readIdleSleepSafety } from "../services/idle-sleep-safety.js";
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
const now = Date.UTC(2026, 10, 2);
const ownerId = randomUUID();
const hold = () => ({ ownerId, draining: true, startedAt: new Date(now - 1000),
expiresAt: new Date(now + 60_000), activeRuns: 0, pendingWakes: 0 });
const localNone = async () => "none" as const;
describe("existing-instance sleep blocker diagnostics", () => {
let db: Db;
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
beforeAll(async () => {
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_IDLE_SLEEP_DIAGNOSTICS_TEST_DATABASE_URL");
if (!external) temporary = await startEmbeddedPostgresTestDatabase("idle-sleep-diagnostics-");
db = createDb(external ?? temporary!.connectionString, { maxConnections: 4 });
}, 60_000);
afterEach(async () => { await db.execute(sql`TRUNCATE companies, plugins, environments CASCADE`); });
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
const read = (scheduled = false, monitors = false) => readIdleSleepSafety(db, hold, () => now, ownerId,
localNone, undefined, undefined, scheduled, monitors, { collectDiagnostics: true });
const blockers = (report: Awaited<ReturnType<typeof read>>) => report.diagnostics?.checks
.filter(check => check.state === "present").map(check => check.reason);
async function seed() {
const companyId = randomUUID(), agentId = randomUUID(), runId = randomUUID();
await db.insert(companies).values({ id: companyId, name: "Existing inventory fixture", issuePrefix: `DI${companyId.slice(0, 6)}` });
await db.insert(agents).values({ id: agentId, companyId, name: "On-demand worker", role: "engineer", adapterType: "process", status: "idle" });
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand",
status: "succeeded", finishedAt: new Date(now - 1000) });
return { companyId, agentId, runId };
}
async function chatHistory(f: Awaited<ReturnType<typeof seed>>) {
const [application] = await db.insert(toolApplications).values({ companyId: f.companyId, name: "Slack fixture", type: "rest_api" }).returning();
const [connection] = await db.insert(toolConnections).values({ companyId: f.companyId, applicationId: application!.id,
name: "Unconfigured Slack", uid: randomUUID(), connectionPurpose: "channel", transport: "chat_sdk", status: "draft", enabled: false }).returning();
const [endpoint] = await db.insert(chatEndpoints).values({ companyId: f.companyId, connectionId: connection!.id,
provider: "slack", publicId: randomUUID(), assignedAgentId: f.agentId, status: "draft", setup: { step: "provider_setup" } }).returning();
const [delivery] = await db.insert(chatDeliveries).values({ companyId: f.companyId, endpointId: endpoint!.id,
providerEventId: randomUUID(), deduplicationKey: randomUUID(), eventKind: "message", normalizedEvent: {},
state: "processed", processedAt: new Date(now - 1000) }).returning();
return delivery!;
}
it("inspects every current predicate while retaining settled ordinary history", async () => {
const f = await seed();
const report = await read();
expect(report.backgroundWork).toBe("none");
expect(report.diagnostics?.complete).toBe(true);
expect(blockers(report)).toEqual([]);
expect(report.diagnostics?.checks.filter(check => !check.reason.includes(".") ||
["agents.execution_or_timer", "agents.lifecycle", "agents.company_hold", "heartbeat_runs.budget_cancellation"].includes(check.reason)))
.toHaveLength(72);
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId))).toHaveLength(1);
});
it("reports history, credentials, debt, running work and future configuration simultaneously", async () => {
const f = await seed();
const history = await chatHistory(f);
await db.insert(agentApiKeys).values({ companyId: f.companyId, agentId: f.agentId, name: "Saved machine access", keyHash: randomUUID() });
await db.insert(companyTransferRuns).values({ companyId: f.companyId, direction: "export", status: "completed",
actorKey: "fixture-board", containerRef: {}, idempotencyKey: randomUUID(), finishedAt: new Date(now - 2000) });
await db.insert(budgetPolicies).values({ companyId: f.companyId, scopeType: "company", scopeId: f.companyId,
windowKind: "calendar_month_utc", enforcementVersion: 2, enforcementDeliveredVersion: 1 });
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, invocationSource: "on_demand", status: "queued" });
await db.insert(routines).values({ companyId: f.companyId, title: "Configured future duty", status: "active", assigneeAgentId: f.agentId });
const report = await read();
expect(report.backgroundWork).toBe("present");
expect(report.diagnostics?.complete).toBe(true);
expect(blockers(report)).toEqual(["agent_api_keys", "budget_policies", "chat_deliveries", "company_transfer_runs", "heartbeat_runs", "routines"]);
expect(await db.select().from(chatDeliveries).where(eq(chatDeliveries.id, history.id)))
.toMatchObject([{ state: "processed", processedAt: history.processedAt }]);
expect(await db.select().from(agentApiKeys)).toHaveLength(1);
expect(await db.select().from(companyTransferRuns)).toMatchObject([{ status: "completed" }]);
});
it.each(["processed", "retry"] as const)("identifies retained %s chat rows without changing domain policy", async state => {
const delivery = await chatHistory(await seed());
await db.update(chatDeliveries).set({ state }).where(eq(chatDeliveries.id, delivery.id));
expect(blockers(await read())).toEqual(["chat_deliveries"]);
expect(await db.select().from(chatDeliveries)).toMatchObject([{ id: delivery.id, state }]);
});
it("distinguishes processed connector receipts from pending application without clearing history", async () => {
const f = await seed();
const [receipt] = await db.insert(connectionEventDeliveries).values({ companyId: f.companyId, provider: "github",
providerDeliveryId: randomUUID(), event: "issues", status: "processed", processedAt: new Date(now - 1000) }).returning();
expect(blockers(await read())).toEqual([]);
await db.update(connectionEventDeliveries).set({ status: "received", processedAt: null }).where(eq(connectionEventDeliveries.id, receipt!.id));
expect(blockers(await read())).toEqual(["connection_event_deliveries"]);
expect(await db.select().from(connectionEventDeliveries)).toHaveLength(1);
});
it("collects schedule and unrelated history reasons from the same negotiated inspection", async () => {
const f = await seed();
await chatHistory(f);
const [routine] = await db.insert(routines).values({ companyId: f.companyId, title: "Future routine", status: "active", assigneeAgentId: f.agentId }).returning();
const [trigger] = await db.insert(routineTriggers).values({ companyId: f.companyId, routineId: routine!.id, kind: "schedule",
cronExpression: "*/15 * * * *", timezone: "UTC", nextRunAt: new Date(now + 600_000), enabled: true, archived: false, setupPending: false }).returning();
await db.insert(issues).values({ companyId: f.companyId, title: "Passive backlog", status: "backlog" });
expect(blockers(await read())).toEqual(["chat_deliveries", "issues", "routines"]);
expect(blockers(await read(true, true))).toEqual(["chat_deliveries"]);
await db.update(routineTriggers).set({ nextRunAt: new Date(now - 1000) }).where(eq(routineTriggers.id, trigger!.id));
const due = await read(true, true);
expect(blockers(due)).toEqual(["chat_deliveries", "routines.schedule"]);
expect(await db.select().from(routineTriggers)).toMatchObject([{ nextRunAt: new Date(now - 1000) }]);
});
it("fails closed on a real malformed SQL configuration without leaking values", async () => {
const f = await seed();
await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: true, intervalSec: "private-bad-interval" } } })
.where(eq(agents.id, f.agentId));
const report = await read();
expect(report.backgroundWork).toBe("unknown");
expect(report.diagnostics?.complete).toBe(false);
expect(report.diagnostics?.checks).toContainEqual({ reason: "durable.snapshot", state: "unknown" });
expect(JSON.stringify(report)).not.toMatch(/private-bad-interval|numeric|SELECT|runtime_config/);
});
});
@@ -0,0 +1,164 @@
import { PgDialect } from "drizzle-orm/pg-core";
import { readFileSync } from "node:fs";
import type { SQL } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import { describe, expect, it, vi } from "vitest";
import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js";
import { beginIdleTrackedWork } from "../services/task-admission.js";
const now = Date.UTC(2026, 10, 2);
const ownerId = "inventory-owned-hold";
const hold = (): IdleSleepDrainStatus => ({ ownerId, draining: true, startedAt: new Date(now - 1000),
expiresAt: new Date(now + 60_000), activeRuns: 0, pendingWakes: 0 });
const localNone = async () => "none" as const;
const collect = { collectDiagnostics: true } as const;
function database(present: string[] = [], change?: (row: Record<string, unknown>) => unknown) {
let keys: string[] = [];
const execute = vi.fn(async (statement: SQL) => {
const query = new PgDialect().sqlToQuery(statement).sql;
if (query.startsWith("SET LOCAL")) return [];
if (query.endsWith(" AS blocked")) return [{ blocked: present.length > 0 }];
keys = [...query.matchAll(/ AS "([^"]+)"(?=, EXISTS |$)/g)].map(match => match[1]!);
const row = Object.fromEntries(keys.map(key => [key, present.includes(key)]));
return change ? change(row) : [row];
});
const transaction = vi.fn(async (run: (tx: unknown) => Promise<unknown>) => run({ execute }));
return { db: { transaction } as unknown as Db, transaction, execute, keys: () => keys };
}
describe("owned idle sleep diagnostics", () => {
it("reports simultaneous blockers from every predicate in one safe snapshot", async () => {
const f = database(["agent_api_keys", "heartbeat_runs", "chat_deliveries", "muse_mailbox_items"]);
const result = await readIdleSleepSafety(f.db, hold, () => now, ownerId, localNone,
undefined, undefined, false, false, collect);
expect(result.backgroundWork).toBe("present");
expect(result.diagnostics?.complete).toBe(true);
expect(result.diagnostics?.checks.filter(check => check.state === "present").map(check => check.reason))
.toEqual(["agent_api_keys", "chat_deliveries", "heartbeat_runs", "muse_mailbox_items"]);
expect(f.keys()).toHaveLength(72);
expect(new Set(f.keys()).size).toBe(f.keys().length);
const inventory = readFileSync(new URL("../../../doc/plans/2026-11-02-sleep-blocker-inventory.md", import.meta.url), "utf8");
const documented = [...inventory.matchAll(/^\| `([a-z0-9_.]+)` \|/gm)].map(match => match[1]);
expect(f.keys().filter(key => !documented.includes(key))).toEqual([]);
expect(f.transaction).toHaveBeenCalledWith(expect.any(Function), { isolationLevel: "repeatable read", accessMode: "read only" });
expect(f.execute).toHaveBeenCalledTimes(2);
});
it("can report each registered predicate independently, including every retained-table family", async () => {
const active: string[] = [];
const f = database(active);
const read = () => readIdleSleepSafety(f.db, hold, () => now, ownerId, localNone,
undefined, undefined, false, false, collect);
await read();
for (const key of f.keys()) {
active.splice(0, active.length, key);
const result = await read();
expect(result.backgroundWork, key).toBe("present");
expect(result.diagnostics?.checks.filter(check => check.state === "present"), key)
.toEqual([{ reason: key, state: "present" }]);
}
});
it("preserves ordinary local and plugin short circuits and response shape", async () => {
const f = database();
const plugins = vi.fn(async () => ({ backgroundWork: "present" as const, pluginIds: [] }));
expect(await readIdleSleepSafety(f.db, hold, () => now, ownerId, async () => "present", plugins))
.toEqual({ version: 1, backgroundWork: "present" });
expect(plugins).not.toHaveBeenCalled();
expect(f.transaction).not.toHaveBeenCalled();
expect(await readIdleSleepSafety(f.db, hold, () => now, ownerId, localNone, plugins))
.toEqual({ version: 1, backgroundWork: "present" });
expect(f.transaction).not.toHaveBeenCalled();
expect(await readIdleSleepSafety(f.db, hold, () => now, ownerId, localNone))
.toEqual({ version: 1, backgroundWork: "none" });
});
it("continues safe inspection after local and plugin blockers without starting a backup", async () => {
const f = database(["company_secret_proposals"]);
const backup = vi.fn(async () => true);
const plugins = vi.fn(async () => ({ backgroundWork: "present" as const, pluginIds: [] }));
const result = await readIdleSleepSafety(f.db, hold, () => now, ownerId, async record => {
record?.("local.accounting_spool", "present");
record?.("local.orphan_spool", "present");
return "present";
}, plugins, backup, false, false, collect);
expect(result.backgroundWork).toBe("present");
expect(result.diagnostics?.checks).toEqual(expect.arrayContaining([
{ reason: "company_secret_proposals", state: "present" },
{ reason: "local.accounting_spool", state: "present" },
{ reason: "local.orphan_spool", state: "present" },
{ reason: "plugins.runtime_drain", state: "present" },
{ reason: "backup.checkpoint", state: "not_checked" },
]));
expect(backup).not.toHaveBeenCalled();
});
it("does not run checkpoint actions or treat unperformed actions as quiet", async () => {
const f = database();
const backup = vi.fn(async () => true);
const result = await readIdleSleepSafety(f.db, hold, () => now, ownerId, localNone,
undefined, backup, false, false, collect);
expect(result.backgroundWork).toBe("unknown");
expect(result.diagnostics?.complete).toBe(true);
expect(result.diagnostics?.checks).toContainEqual({ reason: "backup.checkpoint", state: "not_checked" });
expect(backup).not.toHaveBeenCalled();
});
it.each([undefined, "wrong-owner"])("never opens the DB for diagnostic owner %s", async owner => {
const f = database();
const result = await readIdleSleepSafety(f.db, hold, () => now, owner, localNone,
undefined, undefined, false, false, collect);
expect(result.backgroundWork).toBe("unknown");
expect(result.diagnostics?.complete).toBe(false);
expect(f.transaction).not.toHaveBeenCalled();
});
it("skips unsafe plugin RPCs while accepted local work is active", async () => {
const f = database();
const plugins = vi.fn();
const finish = beginIdleTrackedWork();
try {
const result = await readIdleSleepSafety(f.db, hold, () => now, ownerId, async () => "present",
plugins, undefined, false, false, collect);
expect(result.backgroundWork).toBe("unknown");
expect(result.diagnostics?.complete).toBe(false);
expect(result.diagnostics?.checks).toContainEqual({ reason: "plugins.runtime_drain", state: "not_checked" });
expect(plugins).not.toHaveBeenCalled();
} finally { finish(); }
});
it.each(["owner", "expiry", "generation"])("invalidates the observation after changed %s", async changed => {
const f = database();
let current = hold();
let clock = now;
let reads = 0;
const result = await readIdleSleepSafety(f.db, () => current, () => clock, ownerId, async () => {
if (++reads === 2) {
if (changed === "owner") current = { ...current, ownerId: "replacement" };
if (changed === "expiry") clock += 60_001;
if (changed === "generation") { const finish = beginIdleTrackedWork(); finish(); }
}
return "none";
}, undefined, undefined, false, false, collect);
expect(result.backgroundWork).toBe("unknown");
expect(result.diagnostics?.complete).toBe(false);
});
it.each([
(row: Record<string, unknown>) => [{ ...row, heartbeat_runs: "false" }],
(row: Record<string, unknown>) => [{ ...row, extra_private_field: "sensitive" }],
(row: Record<string, unknown>) => { delete row.heartbeat_runs; return [row]; },
() => [],
(row: Record<string, unknown>) => [row, row],
() => { throw new Error("private SQL, credential, spool path"); },
])("fails closed and redacts malformed/unavailable durable reads", async change => {
const f = database([], change);
const result = await readIdleSleepSafety(f.db, hold, () => now, ownerId, localNone,
undefined, undefined, false, false, collect);
expect(result.backgroundWork).toBe("unknown");
expect(result.diagnostics?.complete).toBe(false);
expect(result.diagnostics?.checks).toContainEqual({ reason: "durable.snapshot", state: "unknown" });
expect(JSON.stringify(result)).not.toMatch(/sensitive|private SQL|credential,|extra_private/);
});
});
@@ -10,6 +10,7 @@ import {
projects, projectWorkspaces, routines, routineTriggers, toolApplications, toolConnections, workspaceOperations, executionWorkspaceRuntimeLeases, type Db,
} from "@paperclipai/db";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js";
const now = Date.UTC(2020, 0, 1);
@@ -162,18 +163,19 @@ describe("idle sleep admission and local work", () => {
});
});
const support = await getEmbeddedPostgresTestSupport();
const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_IDLE_SLEEP_SAFETY_TEST_DATABASE_URL");
const support = externalDatabaseUrl ? { supported: true } : 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<typeof createDb>;
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
beforeAll(async () => {
database = await startEmbeddedPostgresTestDatabase("idle-sleep-safety-");
db = createDb(database.connectionString);
if (!externalDatabaseUrl) database = await startEmbeddedPostgresTestDatabase("idle-sleep-safety-");
db = createDb(externalDatabaseUrl ?? database.connectionString, { maxConnections: 4 });
}, 60_000);
afterEach(async () => { await db.execute(sql`TRUNCATE companies, plugins, environments CASCADE`); });
afterAll(async () => { await database?.cleanup(); });
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
const read = () => readIdleSleepSafety(db, owned, () => now, ownerId, emptyLocal);
const readScheduled = () => readIdleSleepSafety(db, owned, () => now, ownerId, emptyLocal, undefined, undefined, true);
@@ -1132,6 +1132,28 @@ describe("instance settings routes", () => {
expect(mockReadIdleSleepSafety.mock.calls[0][1]()).toEqual(idleStatus);
});
it("collects fixed-key diagnostics only through the privileged safety report", async () => {
mockHeartbeatService.getTaskDrainStatus.mockReturnValue(idleStatus);
mockReadIdleSleepSafety.mockResolvedValue({ version: 1, backgroundWork: "present",
diagnostics: { version: 1, complete: true, checks: [{ reason: "agent_api_keys", state: "present" }] } });
await request(createApp(adminActor)).get("/api/instance/task-drain?idleSleepDiagnostics=1").expect(200);
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
const response = await request(createApp(adminActor))
.get("/api/instance/task-drain?idleSleepSafety=1&idleSleepDiagnostics=1&ownerId=fixture-owner").expect(200);
expect(response.body.idleSleepSafety.diagnostics.checks).toEqual([{ reason: "agent_api_keys", state: "present" }]);
expect(mockReadIdleSleepSafety).toHaveBeenCalledWith(mockDb, expect.any(Function), Date.now,
"fixture-owner", expect.any(Function), undefined, undefined, false, false, { collectDiagnostics: true });
mockReadIdleSleepSafety.mockClear();
for (const actor of [nonAdminActor, { type: "agent", agentId: "agent-1", companyId: "company-1", source: "api_key" }]) {
await request(createApp(actor))
.get("/api/instance/task-drain?idleSleepSafety=1&idleSleepDiagnostics=1&ownerId=fixture-owner").expect(403);
}
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
await request(createApp(adminActor))
.get("/api/instance/task-drain?idleSleepSafety=1&idleSleepDiagnostics=1&scheduledWake=2&ownerId=fixture-owner").expect(403);
expect(mockReadIdleSleepSafety).not.toHaveBeenCalled();
});
it.each(["1", "2"])("reserves scheduling capability %s for verified Cloud control callers", async version => {
mockHeartbeatService.getTaskDrainStatus.mockReturnValue(idleStatus);
for (const source of ["session", "cloud_tenant", "local_implicit"]) {
+3 -2
View File
@@ -334,8 +334,9 @@ export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManag
}
const idleSleepSafety = await readIdleSleepSafety(db, () => heartbeat.getTaskDrainStatus(), Date.now,
typeof req.query.ownerId === "string" ? req.query.ownerId : undefined,
() => readIdleLocalWork({ backupCheckpoint: Boolean(prepareIdleDatabaseBackup) }),
pluginWorkers?.inspectIdleSleep?.bind(pluginWorkers), prepareIdleDatabaseBackup, scheduledRoutines, scheduledIssueMonitors);
diagnostic => readIdleLocalWork({ backupCheckpoint: Boolean(prepareIdleDatabaseBackup), diagnostic }),
pluginWorkers?.inspectIdleSleep?.bind(pluginWorkers), prepareIdleDatabaseBackup, scheduledRoutines, scheduledIssueMonitors,
...(req.query.idleSleepDiagnostics === "1" ? [{ collectDiagnostics: true }] as const : [] as const));
res.json({ ...heartbeat.getTaskDrainStatus(), idleSleepSafety });
return;
}
+41 -11
View File
@@ -2,6 +2,7 @@ import { promises as fs } from "node:fs";
import path from "node:path";
import { resolvePaperclipInstanceRoot } from "../home-paths.js";
import { idleWorkSnapshot } from "./task-admission.js";
import type { IdleSleepDiagnosticSink } from "./idle-sleep-diagnostics.js";
export type IdleLocalWork = "none" | "present" | "unknown";
const spoolDirectories = new Set<string>();
@@ -61,21 +62,50 @@ export async function inspectIdleQuotaRecovery(directory: string): Promise<IdleL
}
}
export async function readIdleLocalWork(options: { backupCheckpoint?: boolean } = {}): Promise<IdleLocalWork> {
if (!startupComplete || !ingressTracked) return "unknown";
export async function readIdleLocalWork(options: { backupCheckpoint?: boolean; diagnostic?: IdleSleepDiagnosticSink } = {}): Promise<IdleLocalWork> {
const record = options.diagnostic;
const states: IdleLocalWork[] = [];
record?.("local.startup", startupComplete ? "none" : "unknown");
record?.("local.ingress_tracking", ingressTracked ? "none" : "unknown");
if (!startupComplete || !ingressTracked) {
if (!record) return "unknown";
states.push("unknown");
}
// This exemption belongs only to a scan which will finish a new backup
// under the same admission hold before it can return an empty report.
if ((scheduledBackups && !options.backupCheckpoint) || idleWorkSnapshot().active !== 0) return "present";
const backup = scheduledBackups && !options.backupCheckpoint;
const active = idleWorkSnapshot().active !== 0;
record?.("local.automatic_backups", backup ? "present" : "none");
record?.("local.tracked_work", active ? "present" : "none");
if (backup || active) {
if (!record) return "present";
states.push("present");
}
let probes: IdleLocalWork;
try {
const local = [...localWorkProbes].map(probe => probe());
if (local.includes("unknown")) return "unknown";
if (local.includes("present")) return "present";
} catch { return "unknown"; }
const directories = new Set([...spoolDirectories, idleAccountingSpoolPath(), idleOrphanSpoolPath()]);
probes = local.some(state => state !== "none" && state !== "present") ? "unknown"
: local.includes("present") ? "present" : "none";
} catch { probes = "unknown"; }
record?.("local.registered_probes", probes);
if (probes !== "none" && !record) return probes;
states.push(probes);
const accounting = idleAccountingSpoolPath();
const orphan = idleOrphanSpoolPath();
const directories = new Set([...spoolDirectories, accounting, orphan]);
const results = await Promise.all([
...[...directories].map(inspectIdleSpool),
inspectIdleQuotaRecovery(idleQuotaRecoveryPath()),
...[...directories].map(async directory => {
const state = await inspectIdleSpool(directory);
record?.(directory === accounting ? "local.accounting_spool"
: directory === orphan ? "local.orphan_spool" : "local.registered_spools", state);
return state;
}),
inspectIdleQuotaRecovery(idleQuotaRecoveryPath()).then(state => {
record?.("local.quota_recovery", state);
return state;
}),
]);
if (results.includes("unknown")) return "unknown";
return results.includes("present") ? "present" : "none";
states.push(...results);
if (states.includes("unknown")) return "unknown";
return states.includes("present") ? "present" : "none";
}
@@ -0,0 +1,46 @@
export type IdleSleepCheckState = "none" | "present" | "unknown" | "not_checked";
export type IdleSleepDiagnosticSink = (reason: string, state: IdleSleepCheckState) => void;
export interface IdleSleepDiagnostics {
version: 1;
complete: boolean;
checks: { reason: string; state: IdleSleepCheckState }[];
}
export const IDLE_LOCAL_DIAGNOSTIC_REASONS = [
"local.startup", "local.ingress_tracking", "local.automatic_backups", "local.tracked_work",
"local.registered_probes", "local.accounting_spool", "local.orphan_spool",
"local.registered_spools", "local.quota_recovery",
] as const;
export function createIdleSleepDiagnostics() {
const states = new Map<string, IdleSleepCheckState>();
const allowed = new Set<string>([
...IDLE_LOCAL_DIAGNOSTIC_REASONS,
"inspection.hold", "inspection.generation", "durable.snapshot", "local.inspection",
"plugins.runtime_drain", "routines.schedule", "issues.schedule", "backup.checkpoint",
]);
const rank = { none: 0, not_checked: 1, present: 2, unknown: 3 } as const;
const record: IdleSleepDiagnosticSink = (reason, state) => {
if (!allowed.has(reason) || !Object.hasOwn(rank, state)) throw new Error("Invalid idle diagnostic check");
const previous = states.get(reason);
if (previous === undefined || rank[state] > rank[previous]) states.set(reason, state);
};
return {
record,
expect(reasons: string[]) {
for (const reason of reasons) {
if (allowed.has(reason)) throw new Error("Duplicate idle diagnostic reason");
allowed.add(reason);
}
},
report(): IdleSleepDiagnostics {
return {
version: 1,
complete: [...states].every(([reason, state]) => reason === "backup.checkpoint"
|| (state !== "unknown" && state !== "not_checked")),
checks: [...states].sort(([a], [b]) => a.localeCompare(b)).map(([reason, state]) => ({ reason, state })),
};
},
};
}
+183 -107
View File
@@ -4,7 +4,8 @@ import { readIdleRoutineSchedule, type ScheduledWake } from "./idle-scheduled-wo
import { combineIdleSchedules, readIdleIssueSchedule } from "./idle-issue-schedule.js";
import { connectionDeliveryWaitingForResponse, finishedIdleRun as finishedRun, finishedIdleWake, finishedFailedIdleLease as finishedFailedLease, passiveBoardRecovery } from "./idle-work-proof.js";
import type { Db } from "@paperclipai/db";
import { sql } from "drizzle-orm";
import { sql, type SQLWrapper } from "drizzle-orm";
import { createIdleSleepDiagnostics, type IdleSleepDiagnosticSink, type IdleSleepDiagnostics } from "./idle-sleep-diagnostics.js";
import { verifiedResourceCleanup } from "./cleanup-verification-proof.js";
import { companyHoldMismatch } from "../modules/agent-lifecycle/index.js";
import { pendingBudgetRunCancellationPredicate } from "./heartbeat/budget-cancellation.js";
@@ -23,6 +24,7 @@ export interface IdleSleepSafety {
version: 1 | 2;
backgroundWork: "none" | "present" | "unknown";
scheduledWake?: ScheduledWake;
diagnostics?: IdleSleepDiagnostics;
}
export type InspectIdlePlugins = (hold: { ownerId: string; expiresAt: number }) => Promise<{
@@ -40,50 +42,52 @@ export type InspectIdlePlugins = (hold: { ownerId: string; expiresAt: number })
// 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 workCheck = (reason: string, query: string | SQLWrapper) => ({ reason, query: typeof query === "string" ? sql.raw(query) : query });
const WORK_CHECKS = [
`SELECT 1 FROM agents WHERE status = 'running' OR
workCheck("agents.execution_or_timer", `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
COALESCE((runtime_config->'heartbeat'->>'intervalSec')::numeric, 0) > 0)`),
workCheck("heartbeat_runs", `SELECT 1 FROM heartbeat_runs WHERE
status NOT IN ('succeeded', 'failed', 'cancelled', 'timed_out', 'interrupted')
OR (scheduled_retry_at IS NOT NULL
AND (finished_at IS NULL OR scheduled_retry_at > finished_at))
OR cost_accounting_pending`,
`SELECT 1 FROM agent_instruction_working_copies WHERE
OR cost_accounting_pending`),
workCheck("agent_instruction_working_copies", `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 sandbox_creation_operations WHERE state = 'pending'`,
`SELECT 1 FROM budget_reservations WHERE state NOT IN ('settled', 'released')`,
`SELECT 1 FROM budget_policies WHERE enforcement_version > enforcement_delivered_version`,
`SELECT 1 FROM agents WHERE lifecycle_state IN
('preparing', 'verifying', 'pausing', 'resuming', 'terminating', 'cleaning_up')`,
`SELECT 1 FROM decision_invocations WHERE status = 'running'`,
`SELECT 1 FROM prompt_injection_invocations WHERE status = 'running'`,
OR receipt ? 'baseline' OR receipt->>'cleanupPending' = 'true'`),
workCheck("sandbox_creation_operations", `SELECT 1 FROM sandbox_creation_operations WHERE state = 'pending'`),
workCheck("budget_reservations", `SELECT 1 FROM budget_reservations WHERE state NOT IN ('settled', 'released')`),
workCheck("budget_policies", `SELECT 1 FROM budget_policies WHERE enforcement_version > enforcement_delivered_version`),
workCheck("agents.lifecycle", `SELECT 1 FROM agents WHERE lifecycle_state IN
('preparing', 'verifying', 'pausing', 'resuming', 'terminating', 'cleaning_up')`),
workCheck("decision_invocations", `SELECT 1 FROM decision_invocations WHERE status = 'running'`),
workCheck("prompt_injection_invocations", `SELECT 1 FROM prompt_injection_invocations WHERE status = 'running'`),
// 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())`,
workCheck("agent_api_keys", `SELECT 1 FROM agent_api_keys WHERE revoked_at IS NULL`),
workCheck("board_api_keys", `SELECT 1 FROM board_api_keys WHERE revoked_at IS NULL
AND (expires_at IS NULL OR expires_at > now())`),
// Coalescing consumes a request into another run. Retain it as work until
// both the request and that exact company/agent's run have finished.
`SELECT 1 FROM agent_wakeup_requests w WHERE
workCheck("agent_wakeup_requests", `SELECT 1 FROM agent_wakeup_requests w WHERE
w.status NOT IN ('completed', 'failed', 'cancelled', 'skipped', 'timed_out', 'coalesced')
OR (w.status = 'coalesced' AND (w.finished_at IS NULL OR NOT EXISTS (
SELECT 1 FROM heartbeat_runs r WHERE r.id = w.run_id
AND r.company_id = w.company_id AND r.agent_id = w.agent_id
AND ${finishedRun("r")})))`,
`SELECT 1 FROM issues WHERE status NOT IN ('done', 'cancelled')`,
AND ${finishedRun("r")})))`),
workCheck("issues", `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')`,
workCheck("issue_watchdogs", `SELECT 1 FROM issue_watchdogs WHERE status <> 'disabled'`),
workCheck("workspace_operations", `SELECT 1 FROM workspace_operations WHERE status NOT IN ('succeeded', 'failed', 'cancelled')`),
// A local directory has no independent worker. Keep unfamiliar providers,
// runtime/command configuration, reopen fences and due cleanup awake. Runs,
// operations, runtime services, retained leases and backup work are checked
// separately; inspecting the row never deletes or shortens retained data.
`SELECT 1 FROM execution_workspaces w WHERE (
workCheck("execution_workspaces", `SELECT 1 FROM execution_workspaces w WHERE (
w.provider_type IN ('local_fs', 'git_worktree')
AND w.strategy_type IN ('project_primary', 'git_worktree')
AND w.mode IN ('inherit', 'shared_workspace', 'isolated_workspace', 'operator_branch', 'reuse_existing', 'agent_default')
@@ -107,16 +111,16 @@ const WORK_CHECKS = [
AND COALESCE(w.metadata->'config'->>'runtimeProvisionCommand', '') = ''
AND COALESCE(w.metadata->'config'->>'teardownCommand', '') = ''
AND COALESCE(w.metadata->'config'->>'cleanupCommand', '') = ''))
) IS NOT TRUE`,
`SELECT 1 FROM environment_leases l WHERE status NOT IN ('released', 'expired')
AND (${finishedFailedLease("l")}) IS NOT TRUE`,
`SELECT 1 FROM environment_custom_image_setup_sessions s WHERE
) IS NOT TRUE`),
workCheck("environment_leases", `SELECT 1 FROM environment_leases l WHERE status NOT IN ('released', 'expired')
AND (${finishedFailedLease("l")}) IS NOT TRUE`),
workCheck("environment_custom_image_setup_sessions", `SELECT 1 FROM environment_custom_image_setup_sessions s WHERE
(s.status IN ('promoted', 'cancelled', 'timed_out', 'failed') AND s.finished_at IS NOT NULL
AND s.connection_secret_ref IS NULL AND (${verifiedResourceCleanup('s', 'image_setup')})) IS NOT TRUE`,
AND s.connection_secret_ref IS NULL AND (${verifiedResourceCleanup('s', 'image_setup')})) IS NOT TRUE`),
// An unconfigured Slack draft has no provider identity or inbound runtime.
// Setup still wakes through normal browser admission. Retain every enabled,
// previously active, partially connected, foreign, or unsupported endpoint.
`SELECT 1 FROM chat_endpoints endpoint WHERE (endpoint.provider = 'slack'
workCheck("chat_endpoints", `SELECT 1 FROM chat_endpoints endpoint WHERE (endpoint.provider = 'slack'
AND endpoint.status = 'draft' AND endpoint.setup->>'step' = 'provider_setup'
AND (endpoint.setup - ARRAY['step', 'command']) = '{}'::jsonb
AND (endpoint.setup->'command' IS NULL OR endpoint.setup->'command' = 'null'::jsonb
@@ -129,33 +133,33 @@ const WORK_CHECKS = [
AND EXISTS (SELECT 1 FROM tool_connections draft_connection
WHERE draft_connection.id = endpoint.connection_id AND draft_connection.company_id = endpoint.company_id
AND draft_connection.connection_purpose = 'channel' AND draft_connection.transport = 'chat_sdk'
AND draft_connection.status = 'draft' AND draft_connection.enabled = false)) IS NOT TRUE`,
`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')`,
AND draft_connection.status = 'draft' AND draft_connection.enabled = false)) IS NOT TRUE`),
workCheck("routines", `SELECT 1 FROM routines WHERE status NOT IN ('paused', 'archived')`),
workCheck("routine_runs", `SELECT 1 FROM routine_runs WHERE status NOT IN ('completed', 'failed', 'skipped', 'cancelled')`),
workCheck("plugin_jobs", `SELECT 1 FROM plugin_jobs WHERE status <> 'paused'`),
workCheck("plugin_job_runs", `SELECT 1 FROM plugin_job_runs WHERE status NOT IN ('succeeded', 'failed', 'cancelled')`),
workCheck("plugin_webhook_deliveries", `SELECT 1 FROM plugin_webhook_deliveries WHERE status NOT IN ('succeeded', 'failed')`),
// Saved outbound MCP configuration and disabled drafts have no independent
// listener. Keep local processes, channels and GitHub webhook bindings awake;
// accepted calls, credential recovery and delivery work have separate fences.
`SELECT 1 FROM tool_connections c WHERE c.status <> 'archived' AND (
workCheck("tool_connections", `SELECT 1 FROM tool_connections c WHERE c.status <> 'archived' AND (
(c.status = 'draft' AND c.enabled = false)
OR (c.status = 'active' AND c.connection_purpose = 'ai' AND c.transport = 'runtime_auth')
OR (c.status = 'active' AND c.connection_purpose = 'tool' AND c.transport = 'mcp_remote'
AND NOT EXISTS (SELECT 1 FROM connection_grants g WHERE g.connection_id = c.id
AND (g.company_id <> c.company_id OR g.provider_tenant->'github' IS NOT NULL)))
) IS NOT TRUE`,
) IS NOT TRUE`),
// An explicit board escalation has no automatic retry. Retain the visible
// action and all execution/cleanup checks while the board decides what to do.
`SELECT 1 FROM issue_recovery_actions a WHERE a.status NOT IN ('resolved', 'cancelled')
AND (${passiveBoardRecovery('a')}) IS NOT TRUE`,
`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`,
workCheck("issue_recovery_actions", `SELECT 1 FROM issue_recovery_actions a WHERE a.status NOT IN ('resolved', 'cancelled')
AND (${passiveBoardRecovery('a')}) IS NOT TRUE`),
workCheck("status_cards", `SELECT 1 FROM status_cards WHERE archived_at IS NULL`),
workCheck("external_objects", `SELECT 1 FROM external_objects WHERE NOT is_terminal OR next_refresh_at IS NOT NULL`),
// Login history keeps its provider reference after teardown. A terminal
// label alone is insufficient: require a finished, unclaimed session and
// positive cleanup evidence for its resource in the same company/environment.
// Cast only UUID-shaped session references, leaving both lease indexes usable.
`SELECT 1 FROM adapter_auth_sessions s WHERE
workCheck("adapter_auth_sessions", `SELECT 1 FROM adapter_auth_sessions s WHERE
s.status NOT IN ('authenticated', 'completed', 'failed', 'timed_out', 'cancelled')
OR s.finished_at IS NULL OR s.promotion_expires_at IS NOT NULL
OR (s.provider_lease_id IS NOT NULL AND NOT EXISTS (
@@ -174,11 +178,11 @@ const WORK_CHECKS = [
WHEN s.provider_lease_id ~* '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'
THEN s.provider_lease_id::uuid END)
AND ((l.status IN ('released', 'expired') OR (${finishedFailedLease("l")})) AND l.cleanup_status = 'success'
AND l.released_at IS NOT NULL) IS NOT TRUE))`,
AND l.released_at IS NOT NULL) IS NOT TRUE))`),
// Answer delivery history is not work once its acknowledged handoff has
// finished. Keep pending deliveries, unfinished target runs, and missing or
// unfinished fallback wake receipts awake. Other work checks still apply.
`SELECT 1 FROM issue_question_response_deliveries d WHERE
workCheck("issue_question_response_deliveries", `SELECT 1 FROM issue_question_response_deliveries d WHERE
d.status NOT IN ('delivered', 'fallback_queued', 'failed')
OR (d.status IN ('delivered', 'fallback_queued') AND d.acknowledged_at IS NULL)
OR (d.status = 'delivered' AND d.target_run_id IS NULL)
@@ -195,36 +199,36 @@ const WORK_CHECKS = [
OR (w.status = 'coalesced' AND EXISTS (
SELECT 1 FROM heartbeat_runs r WHERE r.id = w.run_id
AND r.company_id = w.company_id AND r.agent_id = w.agent_id
AND ${finishedRun("r")})))))`,
AND ${finishedRun("r")})))))`),
// Cloud owns inbound event durability. A processed receipt is idempotency
// history; any downstream run, wake or delivery is still checked separately.
`SELECT 1 FROM connection_event_deliveries WHERE status <> 'processed' OR processed_at IS NULL`,
workCheck("connection_event_deliveries", `SELECT 1 FROM connection_event_deliveries WHERE status <> 'processed' OR processed_at IS NULL`),
// A delivered outbox row is history after its exact accepted wake finishes.
// A system-retired connection card may also be parked behind a still-pending
// response. Keep the outbox intact: the normal reply path resumes delivery.
`SELECT 1 FROM connection_intent_deliveries d WHERE
workCheck("connection_intent_deliveries", `SELECT 1 FROM connection_intent_deliveries d WHERE
(d.delivered_at IS NULL AND (${connectionDeliveryWaitingForResponse('d')}) IS NOT TRUE)
OR EXISTS (SELECT 1 FROM agent_wakeup_requests connection_wake
WHERE connection_wake.company_id = d.company_id AND connection_wake.idempotency_key IN (
'connection-intent:' || d.interaction_id::text || ':accepted',
'connection-intent:' || d.interaction_id::text || ':rejected',
'connection-intent:' || d.interaction_id::text || ':expired')
AND (${finishedIdleWake('connection_wake')}) IS NOT TRUE)`,
AND (${finishedIdleWake('connection_wake')}) IS NOT TRUE)`),
// Saved private feedback and acknowledged uploads have no delivery job.
// Pending/failed exports and unfamiliar states retain the instance. A sent
// label requires its completion stamp and a cleared failure reason.
`SELECT 1 FROM feedback_exports WHERE (status = 'local_only'
OR (status = 'sent' AND exported_at IS NOT NULL AND failure_reason IS NULL)) IS NOT TRUE`,
workCheck("feedback_exports", `SELECT 1 FROM feedback_exports WHERE (status = 'local_only'
OR (status = 'sent' AND exported_at IS NOT NULL AND failure_reason IS NULL)) IS NOT TRUE`),
// Skill package conversion and retention are optional maintenance. Legacy
// contents remain authoritative until verified conversion; retries may wait
// for ingress, including when external package storage is disabled.
`SELECT 1 FROM project_skill_sources b JOIN companies c ON c.id = b.company_id
WHERE c.status = 'active' AND b.active AND b.source_id IS NULL`,
`SELECT 1 FROM company_skill_sources s JOIN companies c ON c.id = s.company_id
workCheck("project_skill_sources", `SELECT 1 FROM project_skill_sources b JOIN companies c ON c.id = b.company_id
WHERE c.status = 'active' AND b.active AND b.source_id IS NULL`),
workCheck("company_skill_sources", `SELECT 1 FROM company_skill_sources s JOIN companies c ON c.id = s.company_id
WHERE c.status = 'active' AND s.enabled AND s.automatic_updates
AND (s.access_scope = 'company' OR EXISTS (
SELECT 1 FROM project_skill_sources b WHERE b.source_id = s.id AND b.company_id = s.company_id AND b.active))`,
`SELECT 1 FROM native_run_finalizations f WHERE (
SELECT 1 FROM project_skill_sources b WHERE b.source_id = s.id AND b.company_id = s.company_id AND b.active))`),
workCheck("native_run_finalizations", `SELECT 1 FROM native_run_finalizations f WHERE (
f.phase = 'committed' AND f.lease_owner IS NULL AND f.lease_expires_at IS NULL
AND f.next_attempt_at IS NULL AND f.control_deadline_at IS NULL
AND f.recovery_state IS NULL AND f.recovery_request_id IS NULL
@@ -258,7 +262,7 @@ const WORK_CHECKS = [
AND (effect.company_id <> f.company_id OR effect.issue_id <> f.issue_id
OR effect.delivery_state <> 'delivered' OR effect.delivered_at IS NULL
OR effect.next_attempt_at IS NOT NULL))
) IS NOT TRUE`,
) IS NOT TRUE`),
// These less common work sources fail closed on any retained state. Their
// terminal-state exceptions can be added with tests for the owning service.
...[
@@ -275,7 +279,7 @@ const WORK_CHECKS = [
"dot_runner_assignments", "dot_runner_operations", "dot_mailbox_items",
"muse_agent_bindings", "muse_credentials", "muse_runner_assignments",
"muse_runner_operations", "muse_mailbox_items", "muse_input_deliveries", "external_agent_holds",
].map((table) => `SELECT 1 FROM ${table}`),
].map((table) => workCheck(table, `SELECT 1 FROM ${table}`)),
] as const;
function sameQuietHold(before: IdleSleepDrainStatus, after: IdleSleepDrainStatus, now: number): boolean {
@@ -288,94 +292,166 @@ function sameQuietHold(before: IdleSleepDrainStatus, after: IdleSleepDrainStatus
/** Inventory persisted work after the task-drain admission hold lands.
* Only a caller holding the exact idle owner may receive an empty report.
*/
export interface IdleSleepSafetyOptions {
collectDiagnostics?: true;
}
export async function readIdleSleepSafety(
db: Db,
getDrainStatus: () => IdleSleepDrainStatus,
now: () => number = Date.now,
ownerId?: string,
inspectLocalWork: () => Promise<IdleLocalWork> = readIdleLocalWork,
inspectLocalWork: (diagnostic?: IdleSleepDiagnosticSink) => Promise<IdleLocalWork> =
diagnostic => readIdleLocalWork({ diagnostic }),
inspectPlugins?: InspectIdlePlugins,
prepareBackup?: () => Promise<boolean>,
scheduledRoutines = false,
scheduledIssueMonitors = false,
options: IdleSleepSafetyOptions = {},
): Promise<IdleSleepSafety> {
const unknown: IdleSleepSafety = { version: 1, backgroundWork: "unknown" };
const diagnostic = options.collectDiagnostics ? createIdleSleepDiagnostics() : undefined;
const report = (backgroundWork: IdleLocalWork, scheduledWake?: ScheduledWake): IdleSleepSafety => {
const diagnostics = diagnostic?.report();
if (backgroundWork === "none" && diagnostics?.checks.some(check =>
check.state === "unknown" || check.state === "not_checked")) backgroundWork = "unknown";
return { version: backgroundWork === "none" && scheduledWake ? 2 : 1, backgroundWork,
...(backgroundWork === "none" && scheduledWake ? { scheduledWake } : {}),
...(diagnostics ? { diagnostics } : {}) };
};
const before = getDrainStatus();
const localBefore = idleWorkSnapshot();
if (!sameQuietHold(before, before, now())) return unknown;
const owned = () => Boolean(ownerId && ownerId === before.ownerId && before.expiresAt !== null);
if (!sameQuietHold(before, before, now()) || (diagnostic && !owned())) {
diagnostic?.record("inspection.hold", "unknown");
diagnostic?.record("durable.snapshot", "not_checked");
return report("unknown");
}
diagnostic?.record("inspection.hold", "none");
let failureReason = "local.inspection";
try {
// Reject known local blockers before querying (and potentially waking)
// the tenant database. Keep the final inspection below as a second fence.
const localFirst = await inspectLocalWork();
if (localFirst !== "none") return { version: 1, backgroundWork: localFirst };
const localFirst = await inspectLocalWork(diagnostic?.record);
diagnostic?.record("local.inspection", localFirst);
if (localFirst !== "none" && !diagnostic) return report(localFirst);
let quietPlugins: string[] = [];
let pluginState: IdleLocalWork = "none";
if (inspectPlugins) {
if (!ownerId || ownerId !== before.ownerId || before.expiresAt === null || localBefore.active !== 0) return unknown;
const pluginWork = await inspectPlugins({ ownerId, expiresAt: before.expiresAt.getTime() });
if (pluginWork.backgroundWork !== "none") return { version: 1, backgroundWork: pluginWork.backgroundWork };
quietPlugins = pluginWork.pluginIds;
const currentLocal = diagnostic ? idleWorkSnapshot() : localBefore;
if (!owned() || currentLocal.active !== 0 || (diagnostic &&
(!sameQuietHold(before, getDrainStatus(), now()) || currentLocal.generation !== localBefore.generation))) {
if (!diagnostic) return report("unknown");
diagnostic.record("plugins.runtime_drain", "not_checked");
pluginState = "unknown";
} else {
failureReason = "plugins.runtime_drain";
const pluginWork = await inspectPlugins({ ownerId: ownerId!, expiresAt: before.expiresAt!.getTime() });
pluginState = pluginWork.backgroundWork;
diagnostic?.record("plugins.runtime_drain", pluginState);
if (pluginState !== "none" && !diagnostic) return report(pluginState);
if (pluginState === "none") quietPlugins = pluginWork.pluginIds;
}
}
if (diagnostic && !sameQuietHold(before, getDrainStatus(), now())) {
diagnostic.record("inspection.hold", "unknown");
diagnostic.record("durable.snapshot", "not_checked");
return report("unknown");
}
failureReason = "durable.snapshot";
let scheduledWake: ScheduledWake | undefined;
const blocked = await db.transaction(async (tx) => {
await tx.execute(sql`SET LOCAL statement_timeout = '5s'`);
const checks = WORK_CHECKS.filter(query =>
(!scheduledRoutines || query !== `SELECT 1 FROM routines WHERE status NOT IN ('paused', 'archived')`)
&& (!scheduledIssueMonitors || query !== `SELECT 1 FROM issues WHERE status NOT IN ('done', 'cancelled')`))
.map((query) => sql`EXISTS (${sql.raw(query)})`);
checks.push(sql`EXISTS (SELECT 1 FROM agents JOIN companies ON companies.id = agents.company_id
WHERE ${companyHoldMismatch()})`);
checks.push(sql`EXISTS (SELECT 1 FROM heartbeat_runs WHERE ${pendingBudgetRunCancellationPredicate()})`);
checks.push(sql`EXISTS (${nativeResponsePresentationPendingQuery()})`);
const checks = WORK_CHECKS.filter(check =>
(!scheduledRoutines || check.reason !== "routines")
&& (!scheduledIssueMonitors || check.reason !== "issues"));
checks.push(workCheck("agents.company_hold", sql`SELECT 1 FROM agents JOIN companies ON companies.id = agents.company_id
WHERE ${companyHoldMismatch()}`));
checks.push(workCheck("heartbeat_runs.budget_cancellation", sql`SELECT 1 FROM heartbeat_runs WHERE ${pendingBudgetRunCancellationPredicate()}`));
checks.push(workCheck("native_response_presentation", nativeResponsePresentationPendingQuery()));
// Only the current worker's owned runtime drain can exempt a plugin.
// Missing workers, old SDKs, and newly installed plugins still block.
checks.push(quietPlugins.length === 0
? sql`EXISTS (SELECT 1 FROM plugins WHERE status <> 'disabled')`
: sql`EXISTS (SELECT 1 FROM plugins WHERE status <> 'disabled' AND id NOT IN (${sql.join(quietPlugins.map((id) => sql`${id}::uuid`), sql`, `)}))`);
const rows = await tx.execute<{ blocked: boolean }>(sql`SELECT ${sql.join(checks, sql` OR `)} AS blocked`);
if (rows.length !== 1 || typeof rows[0]?.blocked !== "boolean") return undefined;
if (rows[0].blocked) return true;
checks.push(workCheck("plugins", quietPlugins.length === 0
? sql`SELECT 1 FROM plugins WHERE status <> 'disabled'`
: sql`SELECT 1 FROM plugins WHERE status <> 'disabled' AND id NOT IN (${sql.join(quietPlugins.map((id) => sql`${id}::uuid`), sql`, `)})`));
let found: boolean;
if (diagnostic) {
diagnostic.expect(checks.map(check => check.reason));
const rows = await tx.execute<Record<string, boolean>>(sql`SELECT ${sql.join(checks.map(check =>
sql`EXISTS (${check.query}) AS ${sql.identifier(check.reason)}`), sql`, `)}`);
if (rows.length !== 1 || Object.keys(rows[0] ?? {}).length !== checks.length
|| checks.some(check => !Object.hasOwn(rows[0]!, check.reason) || typeof rows[0]![check.reason] !== "boolean")) return undefined;
for (const check of checks) diagnostic.record(check.reason, rows[0]![check.reason] ? "present" : "none");
diagnostic.record("durable.snapshot", "none");
found = checks.some(check => rows[0]![check.reason]);
} else {
const rows = await tx.execute<{ blocked: boolean }>(sql`SELECT ${sql.join(checks.map(check => sql`EXISTS (${check.query})`), sql` OR `)} AS blocked`);
if (rows.length !== 1 || typeof rows[0]?.blocked !== "boolean") return undefined;
found = rows[0].blocked;
}
if (found && !diagnostic) return true;
if (scheduledRoutines) {
if (!ownerId || ownerId !== before.ownerId) return undefined;
const schedule = await readIdleRoutineSchedule(tx, ownerId, now());
if (schedule.blocked) return true;
if (!owned()) return undefined;
const schedule = await readIdleRoutineSchedule(tx, ownerId!, now());
diagnostic?.record("routines.schedule", schedule.blocked ? "present" : "none");
if (schedule.blocked && !diagnostic) return true;
found ||= schedule.blocked;
scheduledWake = schedule.scheduledWake;
}
if (scheduledIssueMonitors) {
if (!scheduledRoutines || !ownerId || ownerId !== before.ownerId) return undefined;
const schedule = await readIdleIssueSchedule(tx, ownerId, now());
if (schedule.blocked) return true;
if (!scheduledRoutines || !owned()) return undefined;
const schedule = await readIdleIssueSchedule(tx, ownerId!, now());
diagnostic?.record("issues.schedule", schedule.blocked ? "present" : "none");
if (schedule.blocked && !diagnostic) return true;
found ||= schedule.blocked;
scheduledWake = combineIdleSchedules(scheduledWake, schedule.scheduledWake, schedule.generation);
}
return false;
return found;
}, { 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;
if (blocked === undefined) {
diagnostic?.record("durable.snapshot", "unknown");
return report("unknown");
}
if (!sameQuietHold(before, getDrainStatus(), now())) {
diagnostic?.record("inspection.hold", "unknown");
return report("unknown");
}
if (blocked && !diagnostic) return report("present");
if (!owned() || (!diagnostic && localBefore.active !== 0)) return report("unknown");
let expectedGeneration = localBefore.generation;
if (prepareBackup) {
if (!sameQuietHold(before, getDrainStatus(), now()) ||
idleWorkSnapshot().generation !== expectedGeneration || idleWorkSnapshot().active !== 0) return unknown;
// The runner owns exactly one tracked-work receipt through dump,
// archive verification and fsync. Its start/finish are the only work
// allowed during this scan. Any concurrent work invalidates the dump.
if (!(await prepareBackup())) return unknown;
expectedGeneration += 2;
if (diagnostic) {
diagnostic.record("backup.checkpoint", "not_checked");
} else {
if (!sameQuietHold(before, getDrainStatus(), now()) ||
idleWorkSnapshot().generation !== expectedGeneration || idleWorkSnapshot().active !== 0) return report("unknown");
// The runner owns exactly one tracked-work receipt through dump,
// archive verification and fsync. Its start/finish are the only work
// allowed during this scan. Any concurrent work invalidates the dump.
if (!(await prepareBackup())) return report("unknown");
expectedGeneration += 2;
}
}
const local = await inspectLocalWork();
failureReason = "local.inspection";
const local = await inspectLocalWork(diagnostic?.record);
diagnostic?.record("local.inspection", local);
const after = getDrainStatus();
const localAfter = idleWorkSnapshot();
if (!sameQuietHold(before, after, now()) || localAfter.active !== 0 ||
expectedGeneration !== localAfter.generation) return unknown;
if (local === "none" && scheduledWake) {
if (Date.parse(scheduledWake.nextWakeAt) <= now()) return { version: 1, backgroundWork: "present" };
return { version: 2, backgroundWork: "none", scheduledWake };
expectedGeneration !== localAfter.generation) {
diagnostic?.record("inspection.generation", "unknown");
return report("unknown");
}
return { version: 1, backgroundWork: local };
diagnostic?.record("inspection.generation", "none");
if (diagnostic && (localFirst === "unknown" || local === "unknown" || pluginState === "unknown")) return report("unknown");
if (diagnostic && (blocked || localFirst === "present" || pluginState === "present")) return report("present");
if (local === "none" && scheduledWake && Date.parse(scheduledWake.nextWakeAt) <= now()) {
diagnostic?.record("inspection.hold", "unknown");
return report("present");
}
return report(local, scheduledWake);
} 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;
diagnostic?.record(failureReason, "unknown");
return report("unknown");
}
}