Preserve domain deadlines across owned idle sleep (#18656)

Reviewed and verified at 9b4acc456d.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed 2026-11-03 15:40:56 -06:00
1 parent 0a1e4c3d0f
commit f187de99aa
29 files changed
+1692 -266

No files matched your search

@@ -44,7 +44,7 @@ describe.skipIf(!support.supported)("agent timer owned deadlines", () => {
beforeAll(async () => {
if (external) await applyPendingMigrations(external);
else temporary = await startEmbeddedPostgresTestDatabase("agent-timer-deadlines-");
db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
db = createDb(external ?? temporary!.connectionString, { maxConnections: 4, idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
}, 30_000);
beforeEach(async () => {
@@ -0,0 +1,13 @@
void Promise.all([
import("../../../../packages/plugins/sdk/src/define-plugin.ts"),
import("../../../../packages/plugins/sdk/src/worker-rpc-host.ts"),
]).then(([{ definePlugin }, { startWorkerRpcHost }]) => {
startWorkerRpcHost({ plugin: definePlugin({
async setup(ctx) {
ctx.jobs.register("hourly", async job => {
await ctx.state.set({ scopeKind: "instance", stateKey: job.runId }, { runId: job.runId, scheduledAt: job.scheduledAt });
});
},
async onIdleDrain() { return "none"; },
}) });
});
@@ -0,0 +1,408 @@
import { createGitHubGrantContinuity, readIdleGitHubGrantContinuity } from "../services/tool-access/github-grant-continuity.js";
import { createOAuthCredentialStore } from "../services/tool-access/oauth/credential-store.js";
import { createOAuthRefreshLeases } from "../services/tool-access/oauth/refresh-leases.js";
import { createOAuthGrantRefresh } from "../services/tool-access/oauth/grant-refresh.js";
import { secretService } from "../services/secrets.js";
import { writeConnectionCredential } from "../services/connection-credentials.js";
import type { PaperclipCloudConnector } from "../services/paperclip-cloud-connector.js";
import { GITHUB_CONNECTOR_PROFILES } from "@paperclipai/shared";
import { promises as fs } from "node:fs";
import os from "node:os";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { createRequire } from "node:module";
import { setTimeout as delay } from "node:timers/promises";
import { createPluginWorkerManager, type PluginWorkerManager } from "../services/plugin-worker-manager.js";
import { createPluginJobScheduler } from "../services/plugin-job-scheduler.js";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { pluginStateStore } from "../services/plugin-state-store.js";
import { startTaskDrain, stopTaskDrain, beginIdleTrackedWork } from "../services/task-admission.js";
import { providerTraceStore } from "../services/provider-trace-store.js";
import { readIdleSleepSafety } from "../services/idle-sleep-safety.js";
import { randomUUID } from "node:crypto";
import { eq, sql, inArray } from "drizzle-orm";
import { agents, companies, heartbeatRuns, plugins, pluginJobs, pluginJobRuns, pluginState, providerTraceRecords, issues, issueWatchdogs, issueThreadInteractions, toolApplications, toolConnections, connectionGrants, companySecrets, companySecretVersions, routines, routineTriggers,
applyPendingMigrations, createDb, type Db } from "@paperclipai/db";
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { readIdleDeadlineProof } from "../services/idle-deadline-proof.js";
import type { VerifiedIdleCapabilities } from "../services/idle-incoming-wake-capability.js";
import { claimDueTimerHeartbeatForAgent } from "../services/heartbeat/queue.js";
import { pluginJobStore } from "../services/plugin-job-store.js";
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_IDLE_DEADLINE_TEST_DATABASE_URL");
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
(support.supported ? describe : describe.skip)("durable domain deadline snapshot with PostgreSQL", () => {
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let db: Db;
let directory: string;
beforeAll(async () => {
if (external) await applyPendingMigrations(external);
else temporary = await startEmbeddedPostgresTestDatabase("idle-domain-deadlines-");
db = createDb(external ?? temporary!.connectionString, { maxConnections: 4, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
directory = await fs.mkdtemp(path.join(os.tmpdir(), "idle-deadline-producers-"));
vi.stubEnv("PAPERCLIP_CONFIG", path.join(directory, "config.json"));
vi.stubEnv("PROVIDER_TRACE_BASE_PATH", path.join(directory, "traces"));
vi.stubEnv("PAPERCLIP_STORAGE_PROVIDER", "local_disk");
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "d".repeat(64));
}, 60_000);
afterAll(async () => { vi.useRealTimers(); vi.unstubAllEnvs(); await db?.$client.end({ timeout: 1 }); await temporary?.cleanup();
if (directory) await fs.rm(directory, { recursive: true, force: true }); });
it.each(["routines", "issues"] as const)("marks overflowed %s incomplete under the negotiated deadline catalog", async family => {
const now = Date.now(), companyId = randomUUID(), agentId = randomUUID(), ownerId = randomUUID();
const capability = { deadlineCatalog: "domain_deadlines_v1" } as VerifiedIdleCapabilities;
const read = () => readIdleSleepSafety(db, () => ({ ownerId, draining: true,
startedAt: new Date(now - 1000), expiresAt: new Date(now + 300_000), activeRuns: 0, pendingWakes: 0 }),
() => now, ownerId, async () => "none", undefined, undefined, true, true,
{ deadlineCapability: capability, collectDiagnostics: true });
const reason = `${family}.schedule`;
const ids = Array.from({ length: 10_001 }, () => randomUUID());
const insert = async (batch: string[]) => {
if (family === "routines") {
await db.insert(routines).values(batch.map(id => ({ id, companyId, title: "Future routine",
status: "active", assigneeAgentId: agentId })));
await db.insert(routineTriggers).values(batch.map(id => ({ id, routineId: id, companyId,
kind: "schedule", cronExpression: "*/15 * * * *", timezone: "UTC", nextRunAt: new Date(now + 600_000),
enabled: true, archived: false, setupPending: false })));
} else {
await db.insert(issues).values(batch.map(id => ({ id, companyId, title: "Passive backlog", status: "backlog" })));
}
};
await db.insert(companies).values({ id: companyId, name: "Deadline overflow", issuePrefix: `O${companyId.slice(0, 6)}` });
try {
await db.insert(agents).values({ id: agentId, companyId, name: "On-demand worker", role: "engineer",
status: "idle", adapterType: "process" });
for (let offset = 0; offset < 10_000; offset += 500) await insert(ids.slice(offset, Math.min(offset + 500, 10_000)));
const atLimit = await read();
expect(atLimit).toMatchObject({ version: 3, backgroundWork: "none", diagnostics: { complete: true },
deadlineProof: { catalog: "domain_deadlines_v1" } });
expect(atLimit.diagnostics?.checks).toContainEqual({ reason, state: "none" });
if (family === "routines") expect(atLimit.scheduledWake?.nextWakeAt).toBe(new Date(now + 600_000).toISOString());
else expect(atLimit.scheduledWake).toBeUndefined();
await insert([ids[10_000]!]);
const [rowCount] = await db.select({ count: sql<number>`count(*)::integer` })
.from(family === "routines" ? routines : issues).where(eq(family === "routines" ? routines.companyId : issues.companyId, companyId));
expect(rowCount?.count).toBe(10_001);
if (family === "routines") {
const [triggers] = await db.select({ count: sql<number>`count(*)::integer` }).from(routineTriggers)
.where(eq(routineTriggers.companyId, companyId));
expect(triggers?.count).toBe(10_001);
}
const overflowed = await read();
expect(overflowed.backgroundWork).not.toBe("none");
expect(overflowed.diagnostics?.complete).toBe(false);
expect(overflowed.diagnostics?.checks).toContainEqual({ reason, state: "unknown" });
expect(overflowed.deadlineProof).toBeUndefined();
expect(overflowed.scheduledWake).toBeUndefined();
if (family === "routines") {
await db.delete(routines).where(eq(routines.id, ids[10_000]!));
await db.update(routineTriggers).set({ nextRunAt: new Date(now - 1) }).where(eq(routineTriggers.id, ids[0]!));
} else {
await db.delete(issues).where(eq(issues.id, ids[10_000]!));
await db.update(issues).set({ status: "in_progress", assigneeAgentId: agentId }).where(eq(issues.id, ids[0]!));
}
const blocked = await read();
expect(blocked.backgroundWork).not.toBe("none");
expect(blocked.diagnostics?.complete).toBe(true);
expect(blocked.diagnostics?.checks).toContainEqual({ reason, state: "present" });
expect(blocked.deadlineProof).toBeUndefined();
expect(blocked.scheduledWake).toBeUndefined();
} finally {
await db.delete(routines).where(eq(routines.companyId, companyId));
await db.delete(issues).where(eq(issues.companyId, companyId));
await db.delete(agents).where(eq(agents.companyId, companyId));
await db.delete(companies).where(eq(companies.id, companyId));
}
}, 30_000);
it("projects durable timers across client reconnection and executes the original plugin and trace owners", async () => {
const now = Date.now(), companyId = randomUUID(), agentId = randomUUID(), pluginId = randomUUID(), runId = randomUUID();
let hold: { ownerId: string; expiresAt: number } = { ownerId: randomUUID(), expiresAt: now + 300_000 };
let pluginReceipt: Awaited<ReturnType<NonNullable<PluginWorkerManager["inspectIdleSleep"]>>> | undefined;
const capability = { deadlineCatalog: "domain_deadlines_v1" } as VerifiedIdleCapabilities;
let projectionReads = 0;
const read = (quietPluginIds: string[] = []) => db.transaction(async tx => {
projectionReads = 0;
await tx.execute(sql`SET LOCAL statement_timeout = '5s'`);
const counted = { select: (...args: unknown[]) => { projectionReads++; return Reflect.apply(tx.select, tx, args); } } as Pick<Db, "select">;
return readIdleDeadlineProof(counted, hold, { now: Date.now(), verifiedCapability: capability, quietPluginIds,
quietPluginGeneration: pluginReceipt?.catalogGeneration, quietPluginCatalog: pluginReceipt?.catalog });
}, { isolationLevel: "repeatable read", accessMode: "read only" });
await db.insert(companies).values({ id: companyId, name: "Deadline fixture", issuePrefix: `D${companyId.slice(0, 6)}` });
await db.insert(agents).values({ id: agentId, companyId, name: "Timer", role: "engineer", status: "idle", adapterType: "codex_local",
runtimeConfig: { heartbeat: { enabled: true, intervalSec: 600 } }, createdAt: new Date(now), lastHeartbeatAt: new Date(now) });
const central = (enabled: boolean, diagnostics = false) => readIdleSleepSafety(db, () => ({ ...hold, ownerId: hold.ownerId,
draining: true, startedAt: new Date(now - 1000), expiresAt: new Date(hold.expiresAt), activeRuns: 0, pendingWakes: 0 }),
() => now, hold.ownerId, async () => "none", undefined, undefined, enabled, enabled,
enabled ? { deadlineCapability: capability, ...(diagnostics ? { collectDiagnostics: true as const } : {}) } : {});
expect((await central(false)).backgroundWork).toBe("present");
const negotiated = await central(true);
expect(negotiated).toMatchObject({ version: 3, backgroundWork: "none", deadlineProof: { catalog: "domain_deadlines_v1" },
scheduledWake: { version: 2, kind: "domain_deadlines" } });
expect(negotiated.deadlineProof?.generation).toBe(negotiated.scheduledWake?.generation);
const initial = await read();
expect(initial.blocked).toBe(false);
expect(initial.scheduledWake?.nextWakeAt).toBe(new Date(now + 600_000).toISOString());
expect((await db.select().from(agents).where(eq(agents.id, agentId)))[0].lastHeartbeatAt?.getTime()).toBe(now);
await db.$client.end({ timeout: 1 });
db = createDb(external ?? temporary!.connectionString, { maxConnections: 4, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
expect((await read()).scheduledWake).toEqual(initial.scheduledWake);
await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: true, intervalSec: 450 } } }).where(eq(agents.id, agentId));
const earlier = await read();
expect(earlier.generation).not.toBe(initial.generation);
expect(earlier.scheduledWake?.nextWakeAt).toBe(new Date(now + 450_000).toISOString());
await db.update(agents).set({ lastHeartbeatAt: new Date(now - 600_000) }).where(eq(agents.id, agentId));
expect((await read()).blocked).toBe(true);
const [agent] = await db.select().from(agents).where(eq(agents.id, agentId));
const claimed = await Promise.all([claimDueTimerHeartbeatForAgent(db, agent, new Date(now), 450), claimDueTimerHeartbeatForAgent(db, agent, new Date(now), 450)]);
expect(claimed.filter(Boolean)).toHaveLength(1);
expect((await read()).scheduledWake?.nextWakeAt).toBe(new Date(now + 450_000).toISOString());
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "succeeded", invocationSource: "timer",
startedAt: new Date(now - 1000), finishedAt: new Date(now), costAccountingPending: false });
await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: true, intervalSec: 450, maxDailyRuns: 1 } } }).where(eq(agents.id, agentId));
const date = new Date(now);
expect((await read()).scheduledWake?.nextWakeAt).toBe(new Date(Date.UTC(date.getUTCFullYear(), date.getUTCMonth(), date.getUTCDate() + 1)).toISOString());
await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: true, intervalSec: 450, skipTimerWhenNoActionableWork: true } } }).where(eq(agents.id, agentId));
expect((await read()).scheduledWake).toBeUndefined();
await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: "true", intervalSec: 450 } } }).where(eq(agents.id, agentId));
expect((await read()).blocked).toBe(true);
await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: false } } }).where(eq(agents.id, agentId));
expect((await read()).scheduledWake).toBeUndefined();
const noAlarm = await central(true);
expect(noAlarm).toMatchObject({ version: 3, backgroundWork: "none", deadlineProof: { catalog: "domain_deadlines_v1" } });
expect(noAlarm.scheduledWake).toBeUndefined();
expect(noAlarm.deadlineProof?.generation).not.toBe(negotiated.deadlineProof?.generation);
const scaleIds = Array.from({ length: 160 }, () => randomUUID());
const scaleAgent = (id: string, cost = false) => ({ id, companyId, name: "Scale timer", role: "engineer", status: "idle", adapterType: "codex_local",
runtimeConfig: { heartbeat: { enabled: true, intervalSec: 900, maxDailyRuns: 2, ...(cost ? { maxDailyCostCents: 100 } : {}) } },
createdAt: new Date(now), lastHeartbeatAt: new Date(now) });
await db.insert(agents).values(scaleIds.slice(0, 80).map(id => scaleAgent(id)));
expect((await read()).scheduledWake?.nextWakeAt).toBe(new Date(now + 900_000).toISOString());
expect(projectionReads).toBe(5);
await db.insert(agents).values(scaleIds.slice(80).map(id => scaleAgent(id, true)));
const inactiveIds = [randomUUID(), randomUUID(), randomUUID()];
await db.insert(agents).values(inactiveIds.map((id, i) => ({ ...scaleAgent(id), status: i === 0 ? "paused" : i === 1 ? "terminated" : "idle",
...(i === 2 ? { runtimeConfig: { heartbeat: { enabled: false } } } : {}) })));
const scaled = await read();
expect(scaled.blocked).toBe(false);
expect(scaled.scheduledWake?.nextWakeAt).toBe(new Date(now + 900_000).toISOString());
expect(projectionReads).toBe(6);
await db.update(agents).set({ lastHeartbeatAt: new Date(now - 900_000) }).where(eq(agents.id, scaleIds[0]));
expect((await read()).blocked).toBe(true);
await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: false } } }).where(inArray(agents.id, [...scaleIds, ...inactiveIds]));
const cancelledScale = await read();
expect(cancelledScale.scheduledWake).toBeUndefined();
expect(cancelledScale.generation).not.toBe(scaled.generation);
await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: true, intervalSec: 0.5 } }, lastHeartbeatAt: new Date(now) }).where(eq(agents.id, agentId));
let checkpointClock = now;
const becameDue = await readIdleSleepSafety(db, () => ({ ownerId: hold.ownerId, draining: true, startedAt: new Date(now - 1000),
expiresAt: new Date(hold.expiresAt), activeRuns: 0, pendingWakes: 0 }), () => checkpointClock, hold.ownerId, async () => "none", undefined,
async () => { const finish = beginIdleTrackedWork(); try { checkpointClock += 1000; return true; } finally { finish(); } },
true, true, { deadlineCapability: capability });
expect(becameDue.backgroundWork).toBe("present");
await db.update(agents).set({ runtimeConfig: { heartbeat: { enabled: false } } }).where(eq(agents.id, agentId));
const manifest = { id: `deadline-${pluginId}`, apiVersion: 1 as const, version: "1.0.0", displayName: "Fixture", description: "Deadline fixture", author: "Test",
categories: [], capabilities: ["jobs.schedule" as const, "plugin.state.write" as const], entrypoints: { worker: "worker.cjs" },
jobs: [{ jobKey: "hourly", displayName: "Hourly", schedule: "0 * * * *" }] };
await db.insert(plugins).values({ id: pluginId, pluginKey: manifest.id, packageName: "fixture-plugin", version: manifest.version, status: "ready", manifestJson: manifest });
const store = pluginJobStore(db), state = pluginStateStore(db), manager = createPluginWorkerManager();
let allowed = true;
const errors: unknown[] = [];
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => allowed, onError: error => errors.push(error) });
const scheduler = createPluginJobScheduler({ db, jobStore: store, workerManager: manager, isBackgroundWorkEnabled: () => allowed, tickIntervalMs: 10 });
const prepare = async () => {
allowed = false;
const lease = startTaskDrain({ purpose: "idle", ttlMs: 300_000 });
hold = { ownerId: lease.ownerId!, expiresAt: lease.expiresAt!.getTime() };
pluginReceipt = await manager.inspectIdleSleep!({ ...hold });
expect(pluginReceipt.backgroundWork).toBe("none");
return read([pluginId]);
};
const resume = () => { stopTaskDrain(); manager.releaseIdleSleep!(); allowed = true; };
try {
await manager.startWorker(pluginId, { manifest, apiVersion: 1, config: {}, instanceInfo: { instanceId: "fixture", hostVersion: "1.0.0" },
entrypointPath: fileURLToPath(new URL("./fixtures/plugin-worker-deadline.cjs", import.meta.url)),
execArgv: ["--import", createRequire(import.meta.url).resolve("tsx")],
hostHandlers: { "state.set": async input => {
if (input.scopeKind !== "instance") throw new Error("Fixture job must publish instance-scoped state");
await state.set(pluginId, { ...input, scopeKind: "instance" });
} } });
await store.syncJobDeclarations(pluginId, manifest.jobs);
const [job] = await db.select().from(pluginJobs).where(eq(pluginJobs.pluginId, pluginId));
expect((await prepare()).blocked).toBe(true);
resume();
await scheduler.start(coordinator);
await scheduler.tick();
await store.updateRunTimestamps(job.id, new Date(now - 3_600_000), new Date(now + 900_000));
expect((await read()).blocked).toBe(true);
const transferred = await prepare();
expect(transferred.scheduledWake?.nextWakeAt).toBe(new Date(now + 900_000).toISOString());
resume();
vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(now + 900_001);
await Promise.all([scheduler.tick(), scheduler.tick()]);
await expect.poll(async () => (await db.select().from(pluginJobRuns).where(eq(pluginJobRuns.jobId, job.id))).map(run => run.status), { timeout: 5_000 }).toEqual(["succeeded"]);
const effects = await db.select().from(pluginState).where(eq(pluginState.pluginId, pluginId));
const runs = await db.select().from(pluginJobRuns).where(eq(pluginJobRuns.jobId, job.id));
expect(effects).toHaveLength(1); expect(effects[0].stateKey).toBe(runs[0].id);
expect(runs[0].finishedAt).not.toBeNull();
const [after] = await db.select().from(pluginJobs).where(eq(pluginJobs.id, job.id));
expect(after.nextRunAt!.getTime()).toBeGreaterThan(now + 900_001);
vi.useRealTimers();
expect((await prepare()).scheduledWake?.nextWakeAt).toBe(after.nextRunAt!.toISOString());
resume(); vi.useRealTimers();
await store.updateJobStatus(job.id, "paused");
expect(errors).toEqual([]);
} finally { resume(); vi.useRealTimers(); await scheduler.stop(); await coordinator.stop(); await manager.stopAll(); }
hold = { ownerId: randomUUID(), expiresAt: Date.now() + 300_000 };
expect((await read()).scheduledWake).toBeUndefined();
const traces = providerTraceStore(db);
const capture = await traces.prepare({ companyId, runId, provider: "opencode", requestedBy: "fixture" });
expect((await read()).blocked).toBe(true);
const bytes = JSON.stringify({ kind: "trace_status", status: "complete", debugChannel: "typescript_opencode_native", debugSequence: 1, acknowledgedDebugSequence: 0 }) + "\n";
await fs.writeFile(capture.path, bytes);
expect((await traces.finalize(runId, companyId))?.status).toBe("complete");
expect((await read()).scheduledWake?.nextWakeAt).toBe(new Date(capture.metadata.expiresAt).toISOString());
const savedCapture = (await traces.getByRun(runId, companyId))!;
await db.update(providerTraceRecords).set({ status: "deleted", deletedAt: new Date(), traceRef: `s3-segments-v1:${companyId}/${runId}/${randomUUID()}.ndjson` }).where(eq(providerTraceRecords.runId, runId));
expect((await read()).blocked).toBe(true);
await db.update(providerTraceRecords).set({ status: "complete", deletedAt: null, traceRef: savedCapture.traceRef }).where(eq(providerTraceRecords.runId, runId));
await traces.remove(runId, companyId);
await expect(fs.stat(capture.path)).rejects.toMatchObject({ code: "ENOENT" });
expect((await read()).blocked).toBe(false);
const expiringRunId = randomUUID();
await db.insert(heartbeatRuns).values({ id: expiringRunId, companyId, agentId, status: "succeeded", invocationSource: "timer", startedAt: new Date(now), finishedAt: new Date(), costAccountingPending: false });
const expiring = await traces.prepare({ companyId, runId: expiringRunId, provider: "opencode", requestedBy: "fixture" });
await fs.writeFile(expiring.path, bytes);
expect((await traces.finalize(expiringRunId, companyId))?.status).toBe("complete");
await fs.unlink(expiring.path); await fs.mkdir(expiring.path);
const expiry = new Date(new Date(expiring.metadata.expiresAt).getTime() + 1);
await expect(traces.cleanupExpired(expiry)).rejects.toThrow();
expect((await read()).blocked).toBe(true);
await fs.rmdir(expiring.path); await fs.writeFile(expiring.path, bytes);
await traces.cleanupExpired(expiry);
const cleaned = await traces.getByRun(expiringRunId, companyId);
expect(cleaned).toMatchObject({ status: "expired", traceRef: "00000000-0000-0000-0000-000000000000.ndjson" });
await expect(fs.stat(expiring.path)).rejects.toMatchObject({ code: "ENOENT" });
expect((await read()).blocked).toBe(false);
const watched = randomUUID();
await db.insert(issues).values({ id: watched, companyId, title: "Watchdog grace fixture", status: "done" });
await db.insert(issueWatchdogs).values({ companyId, issueId: watched, watchdogAgentId: agentId, status: "active" });
expect((await central(true, true)).diagnostics?.checks).toContainEqual({ reason: "issue_watchdogs", state: "present" });
const mergeIssue = randomUUID();
await db.insert(issues).values({ id: mergeIssue, companyId, title: "Merge confirmation fixture", status: "in_review", assigneeAgentId: agentId });
await db.insert(issueThreadInteractions).values({ companyId, issueId: mergeIssue, kind: "request_confirmation", status: "pending",
title: "Merge fixture/repo#39?", payload: { version: 1, prompt: "Merge fixture/repo#39?" } });
expect((await central(true, true)).diagnostics?.checks).toContainEqual({ reason: "issues.schedule", state: "present" });
}, 30_000);
it("uses real vault refresh leases, rotates credentials and settles GitHub binding debt after wake", async () => {
const now = Date.now(), companyId = randomUUID(), owner = "deadline-owner";
await db.insert(companies).values({ id: companyId, name: "GitHub deadline", issuePrefix: `G${companyId.slice(0, 6)}` });
const [app] = await db.insert(toolApplications).values({ companyId, applicationKey: randomUUID(), name: "GitHub", type: "mcp_http", status: "active" }).returning();
const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: app.id, uid: randomUUID(), name: "GitHub", status: "active", enabled: true,
transport: "mcp_remote", authKind: "oauth", credentialPolicy: "per_user", credentialSource: "paperclip_vault", transportConfig: {},
config: { sourceTemplateKey: "github", oauth: { strategy: "paperclip_cloud_connector", connectorProfile: "github.code" } } }).returning();
const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId: connection.id, kind: "user", subjectUserId: owner, status: "active",
credentialSecretRefs: [{ configPath: "oauth.access_token", secretId: randomUUID(), versionSelector: "latest" }],
providerTenant: { oauth: { strategy: "paperclip_cloud_connector", accessTokenExpiresAt: new Date(now + 8 * 3_600_000).toISOString(), refreshedAt: new Date(now).toISOString() },
github: { userId: "42", login: "fixture", installationCount: 1, repositoryCount: 1, repositorySelection: "all", installationIds: ["101"],
installationOwnerLogins: ["fixture"], accessRevision: randomUUID(), lastAccessRefreshAt: new Date(now).toISOString(), webhookHealth: "healthy" } } }).returning();
const read = () => db.transaction(async tx => {
await tx.execute(sql`SET LOCAL statement_timeout = '5s'`);
return readIdleGitHubGrantContinuity(tx, Date.now());
}, { isolationLevel: "repeatable read", accessMode: "read only" });
expect((await read()).blocked).toBe(true);
const refs = [];
for (const [configPath, value] of [["oauth.access_token", "saved-access"], ["oauth.refresh_token", "saved-refresh"]] as const) {
const saved = await writeConnectionCredential(db, { companyId, connectionName: "GitHub", configPath, label: configPath, value, ownerUserId: owner });
refs.push({ configPath, secretId: saved.secret.id, versionSelector: "latest" as const, required: true });
}
await db.update(connectionGrants).set({ credentialSecretRefs: refs }).where(eq(connectionGrants.id, grant.id));
const vault = secretService(db), credentials = createOAuthCredentialStore(db, vault);
await credentials.syncCredentialBindings(connection, refs);
const pinned = await writeConnectionCredential(db, { companyId, connectionName: "GitHub pinned", configPath: "oauth.access_token",
label: "Pinned access", value: "pinned-initial", ownerUserId: owner });
const pinnedRef = { configPath: "oauth.access_token", secretId: pinned.secret.id, versionSelector: 1 as const, required: true };
await writeConnectionCredential(db, { companyId, connectionName: "GitHub pinned", configPath: "oauth.access_token", label: "Pinned access",
value: "pinned-rotated", ownerUserId: owner, existingRef: pinnedRef });
await db.update(connectionGrants).set({ credentialSecretRefs: [pinnedRef, refs[1]] }).where(eq(connectionGrants.id, grant.id));
await credentials.syncCredentialBindings(connection, [pinnedRef, refs[1]]);
const [pinnedGrant] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id));
expect((await credentials.resolveOAuthGrantSecret(connection, pinnedGrant, pinnedRef,
{ actorType: "system", actorId: "fixture" }, undefined)).value).toBe("pinned-initial");
expect((await read()).blocked).toBe(false);
for (const status of ["disabled", "destroyed"]) {
await db.update(companySecretVersions).set({ status }).where(eq(companySecretVersions.secretId, pinned.secret.id));
expect((await read()).blocked).toBe(true);
}
await db.update(companySecretVersions).set({ status: "previous", revokedAt: new Date() }).where(eq(companySecretVersions.secretId, pinned.secret.id));
expect((await read()).blocked).toBe(true);
await db.update(companySecretVersions).set({ revokedAt: null }).where(eq(companySecretVersions.secretId, pinned.secret.id));
expect((await read()).blocked).toBe(false);
await db.update(connectionGrants).set({ credentialSecretRefs: refs }).where(eq(connectionGrants.id, grant.id));
await credentials.syncCredentialBindings(connection, refs);
const baseline = await read();
expect(baseline.blocked).toBe(false);
expect(baseline.connectionIds).toContain(connection.id);
expect(new Date(baseline.deadlines[0]).toISOString()).toBe(new Date(now + 7 * 3_600_000).toISOString());
const getConnection = async () => (await db.select().from(toolConnections).where(eq(toolConnections.id, connection.id)))[0];
const leases = createOAuthRefreshLeases(db, () => new Date(), getConnection, credentials.getOAuthGrantRow);
let failBindings = true, rotations = 0;
const successfulBindings = new Map<string, { active: boolean; grantId: string }>();
const connector = {
refresh: async (input: Parameters<PaperclipCloudConnector["refresh"]>[0]) => {
rotations++; expect(input.refreshToken).toBe("saved-refresh"); expect(input.subject).toBe(owner);
return { v: 1 as const, accessToken: "renewed-access", refreshToken: "renewed-refresh", tokenType: "Bearer", accessTokenExpiresAt: new Date(Date.now() + 8 * 3_600_000).toISOString(),
refreshTokenExpiresAt: null, scopes: [...GITHUB_CONNECTOR_PROFILES["github.code"].scopes], subject: owner, companyId, instanceId: "fixture", environment: "test" as const, provider: "github" as const, profile: "github.code" };
},
setWebhookBinding: async (input: Parameters<PaperclipCloudConnector["setWebhookBinding"]>[0]) => {
expect(input.accessToken).toBe("renewed-access"); expect(input.companyId).toBe(companyId);
if (failBindings) throw new Error("named Cloud binding provider unavailable");
successfulBindings.set(input.id, { active: input.active, grantId: input.grantId });
},
} as PaperclipCloudConnector;
const refresh = createOAuthGrantRefresh(db, { now: () => new Date(), store: credentials, leases, getConnectionRow: getConnection,
currentCloudConnector: () => connector, exchangeOAuthToken: async () => { throw new Error("Direct OAuth must not run for managed GitHub"); } });
const continuity = createGitHubGrantContinuity(db, { now: () => new Date(), currentCloudConnector: () => connector,
refreshOAuthGrantCredentials: refresh, resolveOAuthGrantSecret: credentials.resolveOAuthGrantSecret });
vi.spyOn(globalThis, "fetch").mockImplementation(async (input, init) => {
const url = new URL(String(input)); expect(url.origin).toBe("https://api.github.com");
expect(new Headers(init?.headers).get("authorization")).toBe("Bearer renewed-access");
const body = url.pathname === "/user" ? { id: 42, login: "fixture" }
: url.pathname === "/user/installations" ? { installations: [{ id: 101, repository_selection: "all", account: { login: "fixture" } }] }
: { repositories: [{ id: 999, full_name: "fixture/repo" }] };
return new Response(JSON.stringify(body), { status: 200, headers: { "content-type": "application/json" } });
});
try {
vi.useFakeTimers({ toFake: ["Date"] }); vi.setSystemTime(now + 7 * 3_600_000 + 1);
expect(await continuity.sweepGitHubConnectionContinuity(companyId)).toMatchObject({ due: 1, failed: 1 });
expect(rotations).toBe(1);
const [debt] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id));
expect(debt.providerTenant?.github?.bindingSync).toBeDefined();
expect(debt.providerTenant?.github?.continuityRetry).toBeDefined();
expect((await read()).blocked).toBe(true);
failBindings = false; vi.setSystemTime(Date.now() + 31_000);
const reconstructed = createGitHubGrantContinuity(db, { now: () => new Date(), currentCloudConnector: () => connector,
refreshOAuthGrantCredentials: refresh, resolveOAuthGrantSecret: credentials.resolveOAuthGrantSecret });
expect(await reconstructed.sweepGitHubConnectionContinuity(companyId)).toMatchObject({ due: 1, failed: 0 });
expect(rotations).toBe(1); expect(successfulBindings.size).toBe(1);
const [settled] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id));
expect(settled.providerTenant?.github?.bindingSync).toBeUndefined();
expect(settled.providerTenant?.github?.continuityRetry).toBeUndefined();
const [stored] = await db.select().from(companySecrets).where(eq(companySecrets.id, refs[0].secretId));
expect(stored.latestVersion).toBe(2);
expect((await credentials.resolveOAuthGrantSecret(connection, settled, refs[0], { actorType: "system", actorId: "fixture" }, undefined)).value).toBe("renewed-access");
const after = await read();
expect(after.blocked).toBe(false); expect(after.connectionIds).toContain(connection.id);
expect(Math.min(...after.deadlines)).toBeGreaterThan(Date.now());
} finally { vi.useRealTimers(); vi.restoreAllMocks(); }
}, 30_000);
});
@@ -1209,9 +1209,9 @@ describe("instance settings routes", () => {
expect(response.body.idleSleepSafety).toEqual({ version: 3, backgroundWork: "none" });
expect(mockReadIdleSleepSafety).toHaveBeenCalledExactlyOnceWith(mockDb, expect.any(Function), Date.now,
ownerId, expect.any(Function), undefined, undefined, false, false,
{ incomingWake: catalog, deadlineCapability: catalog });
{ incomingWake: catalog, deadlineCapability: undefined });
const options = mockReadIdleSleepSafety.mock.calls[0][9];
expect(options.incomingWake).toBe(options.deadlineCapability);
expect(options.deadlineCapability).toBeUndefined();
expect(Object.isFrozen(options.incomingWake)).toBe(true);
expect(Object.isFrozen(options.incomingWake.incomingWake)).toBe(true);
expect(Object.isFrozen(options.incomingWake.incomingWake.routes)).toBe(true);
@@ -1219,6 +1219,17 @@ describe("instance settings routes", () => {
expect(mockReadIdleSleepSafety).toHaveBeenCalledTimes(1);
});
it("passes the signed deadline catalog only for its explicit named query", async () => {
await request(createControlApp({ type: "none" })).get(`${path}&scheduledWake=domain-deadlines-v1`)
.set("x-paperclip-cloud-control", assertion()).expect(200);
expect(mockReadIdleSleepSafety).toHaveBeenCalledExactlyOnceWith(mockDb, expect.any(Function), Date.now,
ownerId, expect.any(Function), undefined, undefined, true, true,
{ incomingWake: catalog, deadlineCapability: catalog });
const options = mockReadIdleSleepSafety.mock.calls[0][9];
expect(options.incomingWake).toBe(options.deadlineCapability);
expect(Object.isFrozen(options.deadlineCapability)).toBe(true);
});
it.each([
["wrong stack", () => ({ sub: "other-stack" })],
["wrong action", () => ({ action: "task-drain:start" })],
@@ -1,8 +1,9 @@
import { pluginIdleManifestGeneration } from "../services/plugin-idle-catalog.js";
import { fileURLToPath } from "node:url";
import { createRequire } from "node:module";
import { setTimeout as delay } from "node:timers/promises";
import { afterEach, describe, expect, it } from "vitest";
import { createPluginWorkerHandle } from "../services/plugin-worker-manager.js";
import { createPluginWorkerHandle, createPluginWorkerManager } from "../services/plugin-worker-manager.js";
import { startTaskDrain, stopTaskDrain } from "../services/task-admission.js";
const manifest = { id: "test.idle", apiVersion: 1 as const, version: "1.0.0", displayName: "Idle fixture",
@@ -23,6 +24,31 @@ function hold() {
describe("plugin manager idle receipts", () => {
afterEach(() => stopTaskDrain());
it("binds a quiet catalog to the actual started manifest and replacement worker", async () => {
const manager = createPluginWorkerManager();
const options = (version: string) => ({
entrypointPath: fileURLToPath(new URL("./fixtures/plugin-worker-idle.cjs", import.meta.url)),
execArgv: ["--import", createRequire(import.meta.url).resolve("tsx")],
manifest: { ...manifest, version }, config: {}, instanceInfo: { instanceId: "fixture", hostVersion: "1.0.0" }, apiVersion: 1,
hostHandlers: { "events.subscribe": async () => ({}) },
});
try {
await manager.startWorker("test.idle", options("1.0.0"));
const lease = hold();
const first = await manager.inspectIdleSleep!(lease);
expect(first.backgroundWork).toBe("none");
expect(first.catalog).toEqual([{ pluginId: "test.idle", manifestGeneration: pluginIdleManifestGeneration(manifest) }]);
expect((await manager.inspectIdleSleep!(lease)).catalogGeneration).toBe(first.catalogGeneration);
stopTaskDrain(); manager.releaseIdleSleep!();
await manager.stopWorker("test.idle");
await manager.startWorker("test.idle", options("1.0.1"));
const replacement = await manager.inspectIdleSleep!(hold());
expect(replacement.backgroundWork).toBe("none");
expect(replacement.catalogGeneration).not.toBe(first.catalogGeneration);
expect(replacement.catalog?.[0].manifestGeneration).not.toBe(first.catalog?.[0].manifestGeneration);
} finally { stopTaskDrain(); await manager.stopAll(); }
});
it("retains timed-out worker operations and admits work after the owned hold releases", async () => {
const handle = worker();
await handle.start();
@@ -1,4 +1,6 @@
import type { Db } from "@paperclipai/db";
import type { IdleLocalWorkOptions } from "../services/idle-local-work.js";
import type { VerifiedIdleCapabilities } from "../services/idle-incoming-wake-capability.js";
import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js";
import { afterEach, expect, it, vi } from "vitest";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
@@ -7,14 +9,14 @@ import { registerProviderTraceRetentionWork } from "../services/provider-trace-r
const state = vi.hoisted(() => ({
cleanup: vi.fn(async () => []),
next: vi.fn<() => Promise<number | null>>(),
probe: undefined as undefined | (() => string),
probe: undefined as undefined | ((options?: IdleLocalWorkOptions) => string),
unregister: vi.fn(),
}));
vi.mock("../services/provider-trace-store.js", () => ({
providerTraceStore: () => ({ cleanupExpired: state.cleanup, nextRetentionDeadline: state.next }),
}));
vi.mock("../services/idle-local-work.js", () => ({
registerIdleLocalWorkProbe: (probe: () => string) => {
registerIdleLocalWorkProbe: (probe: (options?: IdleLocalWorkOptions) => string) => {
state.probe = probe;
return state.unregister;
},
@@ -51,6 +53,12 @@ it("reconstructs trace deadlines, stays quiet between obligations, and discharge
worker.wake();
await vi.advanceTimersByTimeAsync(0);
expect(state.probe!()).toBe("present");
expect(state.probe!({})).toBe("present");
const negotiated = { deadlineCapability: { deadlineCatalog: "domain_deadlines_v1" } as VerifiedIdleCapabilities };
expect(state.probe!(negotiated)).toBe("none");
vi.setSystemTime(deadline);
expect(state.probe!(negotiated)).toBe("present");
vi.setSystemTime(deadline - 60_000);
expect(coordinator.nextWakeAt()).toBe(deadline);
const scheduled = state.cleanup.mock.calls.length;
await vi.advanceTimersByTimeAsync(59_999);
@@ -931,14 +931,14 @@ describeEmbeddedPostgres("tool access service", () => {
beforeAll(async () => {
if (externalDatabaseUrl) {
await applyPendingMigrations(externalDatabaseUrl);
db = createDb(externalDatabaseUrl);
db = createDb(externalDatabaseUrl, { maxConnections: 4 });
await instanceSettingsService(db).updateExperimental({ enableMemoryConnectors: true });
return;
}
tempDb = await startEmbeddedPostgresTestDatabase(
"paperclip-tool-access-service-",
);
db = createDb(tempDb.connectionString);
db = createDb(tempDb.connectionString, { maxConnections: 4 });
await instanceSettingsService(db).updateExperimental({ enableMemoryConnectors: true });
}, 20_000);
@@ -977,6 +977,7 @@ describeEmbeddedPostgres("tool access service", () => {
});
afterAll(async () => {
await db?.$client.end({ timeout: 1 });
await tempDb?.cleanup();
});
+9 -3
View File
@@ -320,7 +320,12 @@ export function instanceSettingsRoutes(db: Db, pluginWorkers?: PluginWorkerManag
// The report covers every company in this process. Ordinary company
// members may read process counters, but not instance-wide work state.
assertCanManageInstanceSettings(req);
const scheduledIssueMonitors = req.query.scheduledWake === "2";
const domainDeadlines = req.query.scheduledWake === "domain-deadlines-v1";
const capabilities = verifiedIdleCapabilities(req);
if (domainDeadlines && capabilities?.deadlineCatalog !== "domain_deadlines_v1") {
throw forbidden("Domain deadline reports require the signed Cloud deadline catalog");
}
const scheduledIssueMonitors = req.query.scheduledWake === "2" || domainDeadlines;
const scheduledRoutines = req.query.scheduledWake === "1" || scheduledIssueMonitors;
if (scheduledRoutines && req.actor.source !== "cloud_control") {
throw forbidden("Scheduled wake reports require Cloud control authority");
@@ -335,9 +340,10 @@ 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,
diagnostic => readIdleLocalWork({ backupCheckpoint: Boolean(prepareIdleDatabaseBackup), diagnostic }),
diagnostic => readIdleLocalWork({ backupCheckpoint: Boolean(prepareIdleDatabaseBackup), diagnostic,
deadlineCapability: domainDeadlines ? capabilities : undefined }),
pluginWorkers?.inspectIdleSleep?.bind(pluginWorkers), prepareIdleDatabaseBackup, scheduledRoutines, scheduledIssueMonitors,
{ incomingWake: verifiedIdleCapabilities(req), deadlineCapability: verifiedIdleCapabilities(req),
{ incomingWake: capabilities, deadlineCapability: domainDeadlines ? capabilities : undefined,
...(req.query.idleSleepDiagnostics === "1" ? { collectDiagnostics: true } : {}) });
res.json({ ...heartbeat.getTaskDrainStatus(), idleSleepSafety });
return;
+2 -1
View File
@@ -7081,7 +7081,8 @@ registry.registerPath({
tags: ["instance"],
summary:
"Get process task-drain status, with optional bounded upgrade-readiness or idle-safety reports for instance operators",
request: { query: z.object({ upgradeReadiness: z.literal("1").optional().describe("Instance operators can request a bounded aggregate report of current remote-runner recovery eligibility. This read does not hold admission or detach runners."), 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."), scheduledWake: z.enum(["1", "2"]).optional().describe("Cloud control callers may request routine deadlines (1), or routines and issue monitors (2). Cloud must acknowledge a durable alarm and revalidate this report before stopping the instance.") }) },
description: "With idleSleepSafety=1, scheduledWake=domain-deadlines-v1 requires signed Cloud authority with idleCapabilities.deadlineCatalog=domain_deadlines_v1 and returns a mandatory deadlineProof even when no wake is scheduled. Cloud must acknowledge any returned durable alarm and revalidate the report before stopping the instance.",
request: { query: z.object({ upgradeReadiness: z.literal("1").optional().describe("Instance operators can request a bounded aggregate report of current remote-runner recovery eligibility. This read does not hold admission or detach runners."), 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."), scheduledWake: z.enum(["1", "2", "domain-deadlines-v1"]).optional().describe("Cloud control callers may request routine deadlines (1), routines and issue monitors (2), or the domain deadline catalog (domain-deadlines-v1). The domain catalog requires signed Cloud authority with idleCapabilities.deadlineCatalog=domain_deadlines_v1 and returns a deadlineProof even when no wake is scheduled. Cloud must acknowledge any returned durable alarm and revalidate this report before stopping the instance.") }) },
responses: { 200: r.ok(), 401: r.unauthorized, 403: r.forbidden },
});
@@ -0,0 +1,67 @@
import { describe, expect, it } from "vitest";
import { companySecrets, companySecretVersions, type Db } from "@paperclipai/db";
import { PgDialect } from "drizzle-orm/pg-core";
import type { SQL } from "drizzle-orm";
import { readConnectionGrantCredentialStates, type ConnectionGrantCredentialInput } from "./connection-credentials.js";
const input = (index = 0): ConnectionGrantCredentialInput => ({ connection: { id: "connection", companyId: "company" },
grant: { id: `grant-${index}`, companyId: "company", connectionId: "connection", kind: "user", subjectUserId: "owner",
credentialSecretRefs: [{ configPath: "oauth.access_token", secretId: `secret-${index}`, versionSelector: 1 }] } });
const metadata = (index = 0) => ({ secret: { id: `secret-${index}`, companyId: "company", scope: "user", ownerUserId: "owner",
userSecretDefinitionId: `definition-${index}`, status: "active", deletedAt: null, latestVersion: 2, updatedAt: new Date(0) },
definition: { id: `definition-${index}`, companyId: "company", status: "active", deletedAt: null, updatedAt: new Date(0) } });
const version = (index = 0) => ({ id: `version-${index}`, secretId: `secret-${index}`, version: 1, status: "previous", revokedAt: null as Date | null });
function metadataDb(rows = [metadata()], versions = [version()], onRead: () => void = () => {}) {
return { select: (columns: Record<string, unknown>) => {
onRead();
for (const selected of [columns, columns.secret ?? {}, columns.definition ?? {}]) {
expect(Object.keys(selected as object)).not.toEqual(expect.arrayContaining(["material"]));
for (const name of ["valueSha256", "fingerprintSha256", "providerMetadata"]) expect(selected).not.toHaveProperty(name);
}
let table: unknown, params: unknown[] = [];
const query = { from: (source: unknown) => { table = source; return query; }, leftJoin: () => query,
where: (where: SQL) => { params = new PgDialect().sqlToQuery(where).params; return query; }, orderBy: () => query,
limit: () => Promise.resolve(table === companySecrets ? rows.filter(row => params.includes(row.secret.id))
: table === companySecretVersions ? versions.filter(row => params.includes(row.secretId) && params.includes(row.version)) : []) };
return query;
} } as unknown as Pick<Db, "select">;
}
describe("connection grant credential metadata", () => {
it("accepts the exact non-revoked pinned previous version and also selects current latest", async () => {
const request = input(), row = metadata(), saved = version();
expect((await readConnectionGrantCredentialStates(metadataDb([row], [saved]), [request]))?.size).toBe(1);
request.grant.credentialSecretRefs[0].versionSelector = "latest";
expect(await readConnectionGrantCredentialStates(metadataDb([row], [saved]), [request])).toBeUndefined();
saved.version = 2; saved.status = "current";
expect((await readConnectionGrantCredentialStates(metadataDb([row], [saved]), [request]))?.size).toBe(1);
});
it("keeps revoked, disabled, destroyed, failed, absent and foreign metadata unproven", async () => {
for (const status of ["disabled", "destroyed", "failed", "unknown"]) {
const saved = { ...version(), status };
expect(await readConnectionGrantCredentialStates(metadataDb([metadata()], [saved]), [input()])).toBeUndefined();
}
const revoked = { ...version(), revokedAt: new Date() };
expect(await readConnectionGrantCredentialStates(metadataDb([metadata()], [revoked]), [input()])).toBeUndefined();
expect(await readConnectionGrantCredentialStates(metadataDb([], []), [input()])).toBeUndefined();
expect(await readConnectionGrantCredentialStates(metadataDb([metadata()], []), [input()])).toBeUndefined();
for (const field of ["companyId", "ownerUserId", "userSecretDefinitionId"] as const) {
const row = metadata(); row.secret[field] = "foreign";
expect(await readConnectionGrantCredentialStates(metadataDb([row]), [input()])).toBeUndefined();
}
const row = metadata(); row.definition.status = "disabled";
expect(await readConnectionGrantCredentialStates(metadataDb([row]), [input()])).toBeUndefined();
const duplicate = input(); duplicate.grant.credentialSecretRefs.push({ ...duplicate.grant.credentialSecretRefs[0] });
expect(await readConnectionGrantCredentialStates(metadataDb(), [duplicate])).toBeUndefined();
});
it("bounds1001 grants to four metadata reads and checks budget during collection", async () => {
const inputs = Array.from({ length: 1001 }, (_, i) => input(i));
let reads = 0;
const db = metadataDb(inputs.map((_, i) => metadata(i)), inputs.map((_, i) => version(i)), () => { reads++; });
expect((await readConnectionGrantCredentialStates(db, inputs))?.size).toBe(1001);
expect(reads).toBe(4);
let checks = 0;
await expect(readConnectionGrantCredentialStates(db, inputs, () => {
if (++checks > 3) throw new Error("budget exhausted");
})).rejects.toThrow("budget exhausted");
});
});
+103 -13
View File
@@ -1,6 +1,6 @@
import { randomUUID } from "node:crypto";
import { and, eq, inArray, isNull, ne, or } from "drizzle-orm";
import { companySecrets, companySecretProposals, managedAgentProfiles, routineTriggers, userSecretDefinitions, type connectionGrants, type toolConnections } from "@paperclipai/db";
import { companySecrets, companySecretVersions, companySecretProposals, managedAgentProfiles, routineTriggers, userSecretDefinitions, type Db, type connectionGrants, type toolConnections } from "@paperclipai/db";
import type { ToolCredentialSecretRef } from "@paperclipai/shared";
import { connectionCredentialConfigPath } from "@paperclipai/shared";
import { unprocessable } from "../errors.js";
@@ -117,26 +117,36 @@ export async function writeConnectionCredential(
return { secret, created: true, definitionId: createdDefinitionId };
}
/** Validate metadata without reading secret values, including for connection pickers. */
export async function validateConnectionGrantSecretOwnership(
db: CredentialDb,
connection: Pick<typeof toolConnections.$inferSelect, "id" | "companyId">,
grant: Pick<typeof connectionGrants.$inferSelect, "id" | "companyId" | "connectionId" | "kind" | "subjectUserId">,
ref: ToolCredentialSecretRef,
) {
const [secret] = await db.select().from(companySecrets).where(and(
eq(companySecrets.id, ref.secretId), eq(companySecrets.companyId, connection.companyId),
)).limit(1);
const credentialSecretColumns = {
id: companySecrets.id, companyId: companySecrets.companyId, scope: companySecrets.scope,
ownerUserId: companySecrets.ownerUserId, userSecretDefinitionId: companySecrets.userSecretDefinitionId,
status: companySecrets.status, deletedAt: companySecrets.deletedAt, latestVersion: companySecrets.latestVersion,
updatedAt: companySecrets.updatedAt,
};
type CredentialConnection = Pick<typeof toolConnections.$inferSelect, "id" | "companyId">;
type CredentialGrantOwner = Pick<typeof connectionGrants.$inferSelect,
"id" | "companyId" | "connectionId" | "kind" | "subjectUserId">;
type CredentialSecretOwner = Pick<typeof companySecrets.$inferSelect,
"id" | "companyId" | "scope" | "ownerUserId" | "userSecretDefinitionId">;
function assertConnectionGrantSecretOwnership(connection: CredentialConnection, grant: CredentialGrantOwner,
ref: ToolCredentialSecretRef, secret: CredentialSecretOwner | undefined) {
const personal = grant.kind === "user";
if (grant.companyId !== connection.companyId || grant.connectionId !== connection.id || !secret
|| (personal
|| secret.companyId !== connection.companyId || (personal
? !grant.subjectUserId || secret.scope !== "user" || secret.ownerUserId !== grant.subjectUserId || !secret.userSecretDefinitionId
: secret.scope !== "company")) {
throw unprocessable("This connection's credential does not belong to the selected identity. Its owner must reconnect it.", {
code: "grant_credential_invalid", connectionId: connection.id, grantId: grant.id, credential: ref.configPath,
});
}
return secret;
}
export async function validateConnectionGrantSecretOwnership(db: Pick<Db, "select">,
connection: CredentialConnection, grant: CredentialGrantOwner, ref: ToolCredentialSecretRef) {
const [secret] = await db.select(credentialSecretColumns).from(companySecrets).where(and(
eq(companySecrets.id, ref.secretId), eq(companySecrets.companyId, connection.companyId),
)).limit(1);
assertConnectionGrantSecretOwnership(connection, grant, ref, secret);
return secret!;
}
/** The same owner/declaration checks apply during setup, testing and execution. */
@@ -165,3 +175,83 @@ export async function resolveConnectionGrantSecret(
});
return { value: resolved.value, latestVersion: secret.latestVersion };
}
type CredentialGrant = CredentialGrantOwner & Pick<typeof connectionGrants.$inferSelect, "credentialSecretRefs">;
export type ConnectionGrantCredentialInput = { connection: CredentialConnection; grant: CredentialGrant };
export async function readConnectionGrantCredentialStates(db: Pick<Db, "select">,
inputs: ConnectionGrantCredentialInput[], checkBudget: () => void = () => {}): Promise<Map<string, unknown[]> | undefined> {
if (inputs.length > 10_000) return;
const states = new Map<string, unknown[]>();
const requests: { input: ConnectionGrantCredentialInput; ref: ToolCredentialSecretRef }[] = [];
for (const input of inputs) {
checkBudget();
if (states.has(input.grant.id)) return;
states.set(input.grant.id, []);
const paths = new Set<string>();
for (const ref of input.grant.credentialSecretRefs) {
checkBudget();
if (!["oauth.access_token", "oauth.refresh_token"].includes(ref.configPath)) continue;
if (paths.has(ref.configPath)) return;
paths.add(ref.configPath);
requests.push({ input, ref });
}
if (!paths.has("oauth.access_token")) return;
}
if (requests.length > 20_000) return;
for (let offset = 0; offset < requests.length; offset += 1000) {
checkBudget();
const chunk = requests.slice(offset, offset + 1000);
const ids = [...new Set(chunk.map(request => request.ref.secretId))];
const rows = await db.select({ secret: credentialSecretColumns, definition: {
id: userSecretDefinitions.id, companyId: userSecretDefinitions.companyId, status: userSecretDefinitions.status,
deletedAt: userSecretDefinitions.deletedAt, updatedAt: userSecretDefinitions.updatedAt,
} }).from(companySecrets).leftJoin(userSecretDefinitions, eq(userSecretDefinitions.id, companySecrets.userSecretDefinitionId))
.where(inArray(companySecrets.id, ids)).orderBy(companySecrets.id).limit(1001);
if (rows.length > ids.length) return;
const secrets = new Map(rows.map(row => [row.secret.id, row]));
if (secrets.size !== rows.length) return;
const selected = [];
const pairs = new Map<string, { secretId: string; version: number }>();
for (const request of chunk) {
checkBudget();
const row = secrets.get(request.ref.secretId);
try { assertConnectionGrantSecretOwnership(request.input.connection, request.input.grant, request.ref, row?.secret); }
catch (error) {
if (error && typeof error === "object" && "status" in error && error.status === 422) return;
throw error;
}
if (!row) return;
const { secret, definition } = row;
if (secret.status !== "active" || secret.deletedAt || !Number.isSafeInteger(secret.latestVersion) || secret.latestVersion <= 0) return;
if (secret.scope === "user" && (!definition || definition.id !== secret.userSecretDefinitionId
|| definition.companyId !== request.input.connection.companyId || definition.status !== "active" || definition.deletedAt)) return;
const version = request.ref.versionSelector === undefined || request.ref.versionSelector === "latest"
? secret.latestVersion : request.ref.versionSelector;
if (typeof version !== "number" || !Number.isSafeInteger(version) || version <= 0) return;
const key = JSON.stringify([secret.id, version]);
pairs.set(key, { secretId: secret.id, version });
selected.push({ request, secret, definition: secret.scope === "user" ? definition : undefined, key });
}
checkBudget();
const versions = await db.select({ id: companySecretVersions.id, secretId: companySecretVersions.secretId,
version: companySecretVersions.version, status: companySecretVersions.status, revokedAt: companySecretVersions.revokedAt })
.from(companySecretVersions).where(or(...[...pairs.values()].map(pair => and(
eq(companySecretVersions.secretId, pair.secretId), eq(companySecretVersions.version, pair.version)))))
.orderBy(companySecretVersions.secretId, companySecretVersions.version).limit(1001);
if (versions.length > pairs.size) return;
const saved = new Map(versions.map(version => [JSON.stringify([version.secretId, version.version]), version]));
if (saved.size !== versions.length || versions.some(version => !pairs.has(JSON.stringify([version.secretId, version.version])))) return;
for (const { request, secret, definition, key } of selected) {
checkBudget();
const version = saved.get(key);
if (!version || !["current", "previous"].includes(version.status) || version.revokedAt) return;
states.get(request.input.grant.id)!.push([request.ref.configPath, secret.id, secret.scope, secret.ownerUserId,
secret.latestVersion, secret.updatedAt, definition, version]);
}
}
return states;
}
export async function readConnectionGrantCredentialState(db: Pick<Db, "select">,
connection: CredentialConnection, grant: CredentialGrant): Promise<unknown[] | undefined> {
return (await readConnectionGrantCredentialStates(db, [{ connection, grant }]))?.get(grant.id);
}
@@ -0,0 +1,68 @@
import { agents, companies, type Db } from "@paperclipai/db";
import { AGENT_STATUSES, COMPANY_STATUSES } from "@paperclipai/shared";
import { eq, getTableColumns } from "drizzle-orm";
import { evaluateAgentInvokability } from "../agent-invokability.js";
import { supportsUnscopedTimerHeartbeat } from "../heartbeat-policy.js";
import { createHeartbeatTimerPolicy, heartbeatTimerPolicyIsWellFormed } from "./timer-policy.js";
import { idleDeadlineRowLimit, isFutureIdleDeadline, type IdleDeadlineSnapshot } from "../idle-deadline-types.js";
export async function readIdleAgentTimerSchedule(tx: Pick<Db, "select">, now: number,
env: Record<string, string | undefined> = process.env, checkBudget: () => void = () => {}): Promise<IdleDeadlineSnapshot> {
const flags = [env.HEARTBEAT_SCHEDULER_ENABLED, env.PAPERCLIP_IN_WORKTREE,
env.PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS, env.PAPERCLIP_RESTORE_IN_PROGRESS];
const state: unknown[] = [flags];
const blocked = () => ({ blocked: true, state, deadlines: [] });
if (flags[0] === "false" || flags.slice(1).some(v => ["true", "1", "yes", "on"].includes(v ?? ""))) return blocked();
const rows = await tx.select({ agent: getTableColumns(agents), companyStatus: companies.status,
companyUpdatedAt: companies.updatedAt }).from(agents).leftJoin(companies, eq(companies.id, agents.companyId))
.orderBy(agents.id).limit(idleDeadlineRowLimit + 1);
state.push(rows);
if (rows.length > idleDeadlineRowLimit) return blocked();
const byCompany = new Map<string, Map<string, typeof agents.$inferSelect>>();
for (const { agent } of rows) {
const group = byCompany.get(agent.companyId) ?? new Map<string, typeof agents.$inferSelect>();
if (group.has(agent.id)) return blocked();
group.set(agent.id, agent);
byCompany.set(agent.companyId, group);
}
const owner = createHeartbeatTimerPolicy(tx);
const candidates: typeof agents.$inferSelect[] = [];
const deadlines: number[] = [];
for (const { agent, companyStatus } of rows) {
checkBudget();
if (!AGENT_STATUSES.includes(agent.status as typeof AGENT_STATUSES[number]) || !companyStatus || !COMPANY_STATUSES.includes(companyStatus as typeof COMPANY_STATUSES[number]) || agent.status === "running") return blocked();
if (companyStatus !== "active") continue;
if (!heartbeatTimerPolicyIsWellFormed(agent)) return blocked();
const policy = owner.parseHeartbeatPolicy(agent);
if (!Number.isFinite(policy.intervalSec)) return blocked();
if (!policy.enabled || policy.intervalSec <= 0 || !supportsUnscopedTimerHeartbeat(agent)) continue;
const ancestors: typeof agents.$inferSelect[] = [];
const seen = new Set<string>();
let current: typeof agents.$inferSelect | undefined = agent;
while (current && !seen.has(current.id)) {
checkBudget();
seen.add(current.id);
ancestors.push(current);
current = current.reportsTo ? byCompany.get(agent.companyId)!.get(current.reportsTo) : undefined;
}
const invokability = evaluateAgentInvokability(agent, ancestors);
if (!invokability.invokable) {
if (invokability.invalidOrgChain || invokability.reason === "unknown_status") return blocked();
continue;
}
candidates.push(agent);
}
const batch = await owner.readTimerHeartbeatSchedules(candidates, new Date(now), checkBudget);
if (!batch || batch.schedules.size !== candidates.length) return blocked();
state.push(batch.state);
for (const agent of candidates) {
checkBudget();
const snapshot = batch.schedules.get(agent.id);
if (!snapshot) return blocked();
state.push([agent.id, snapshot]);
if (snapshot.deadline === null) continue;
if (!isFutureIdleDeadline(snapshot.deadline, now)) return blocked();
deadlines.push(snapshot.deadline);
}
return { blocked: false, state, deadlines };
}
+1 -1
View File
@@ -86,7 +86,7 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => {
let db: Db;
beforeAll(async () => {
if (!external) database = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-queue-module");
db = createDb(external ?? database!.connectionString);
db = createDb(external ?? database!.connectionString, { maxConnections: 4 });
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
afterAll(async () => {
await db?.$client.end();
+35 -217
View File
@@ -1,3 +1,4 @@
import { createHeartbeatTimerPolicy } from "./timer-policy.js";
import { notifyExecutionRunWork } from "../execution-control-notifications.js";
import { PersistedExecutionWaitRefusal, writeDeferredExecutionWait } from "../deferred-execution-wait.js";
import { notifyDeferredExecutionWork } from "../deferred-execution-notifications.js";
@@ -34,11 +35,7 @@ import {
CHAT_COMPLETION_WAKE_REASON,
isCompletedOnboardingHandoffWake,
} from "../chat-completion-delivery.js";
import {
compareCents,
hasWorkspaceRestoreFailure,
} from "@paperclipai/shared";
import { nonIdleSlackIssueCondition } from "../slack-conversation-state.js";
import { hasWorkspaceRestoreFailure } from "@paperclipai/shared";
import { readQueuedInteractionResponse } from "../queued-interaction-response.js";
import {
isConversation,
@@ -86,15 +83,12 @@ import {
gte,
inArray,
isNull,
lt,
lte,
ne,
notInArray,
or,
sql,
} from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import { AGENT_DEFAULT_MAX_CONCURRENT_RUNS } from "@paperclipai/shared";
import {
agents,
caseHandoffs,
@@ -104,7 +98,6 @@ import {
chatConversations,
chatEndpoints,
companies,
costEvents,
executionWorkspaces,
heartbeatRunEvents,
heartbeatRuns,
@@ -127,11 +120,7 @@ import {
withQueuedCommentIdsInWakePayload,
withQueuedCommentIdsInRunContext,
} from "../issue-queued-comment-queue.js";
import {
parseObject,
asBoolean,
asNumber,
} from "../../adapters/utils.js";
import { parseObject } from "../../adapters/utils.js";
import { isExternalChatWaitAuthorizationContention } from "../native-runtime/chat-attachment-reuse.js";
import { emitAgentTaskRun } from "../agent-task-run-telemetry.js";
import {
@@ -175,7 +164,7 @@ import {
shouldCancelRunsForNonInvokableAgent,
type AgentOrgRow,
} from "../agent-invokability.js";
import { isHeartbeatWakeOnDemandEnabled, supportsUnscopedTimerHeartbeat } from "../heartbeat-policy.js";
import { supportsUnscopedTimerHeartbeat } from "../heartbeat-policy.js";
import { isUuidLike } from "@paperclipai/shared";
import { retryChatControlAdmission } from "../chat-control-admission-retry.js";
@@ -235,11 +224,8 @@ export interface HeartbeatQueueDependencies extends Pick<HeartbeatRetryDependenc
formatIssueIdentifierLink: (identifier: string | null, fallback: string) => string;
}
const HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT = AGENT_DEFAULT_MAX_CONCURRENT_RUNS;
const HEARTBEAT_MAX_CONCURRENT_RUNS_MIN = 1;
const HEARTBEAT_MAX_CONCURRENT_RUNS_MAX = 50;
export const DEFERRED_WAKE_CONTEXT_KEY = "_paperclipWakeContext";
@@ -249,7 +235,6 @@ export const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = [
"scheduled_retry",
] as const;
const TIMER_ACTIONABLE_ISSUE_STATUSES = ["todo", "in_progress"] as const;
export class ChatControlRecoveryUnresolvedError extends Error {
constructor() {
@@ -266,16 +251,6 @@ const RUNNING_ISSUE_WAKE_REASONS_REQUIRING_FOLLOWUP = new Set([
"issue_recovery_action_restored",
]);
function normalizeMaxConcurrentRuns(value: unknown) {
const parsed = Math.floor(
asNumber(value, HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT),
);
if (!Number.isFinite(parsed)) return HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT;
return Math.max(
HEARTBEAT_MAX_CONCURRENT_RUNS_MIN,
Math.min(HEARTBEAT_MAX_CONCURRENT_RUNS_MAX, parsed),
);
}
export function shouldRequireIssueCommentForWake(
contextSnapshot: Record<string, unknown> | null | undefined,
@@ -365,6 +340,34 @@ function readNonEmptyString(value: unknown): string | null {
return typeof value === "string" && value.trim().length > 0 ? value : null;
}
export async function claimDueTimerHeartbeatForAgent(
db: Db,
agent: typeof agents.$inferSelect,
now: Date,
intervalSec: number,
) {
const dueBefore = new Date(now.getTime() - intervalSec * 1000);
const claimed = await db.transaction(async tx => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
return tx
.update(agents)
.set({ lastHeartbeatAt: now, updatedAt: now })
.where(and(
eq(agents.id, agent.id),
eq(agents.companyId, agent.companyId),
or(
lte(agents.lastHeartbeatAt, dueBefore),
and(isNull(agents.lastHeartbeatAt), lte(agents.createdAt, dueBefore)),
),
))
.returning({ id: agents.id })
.then((rows) => rows[0] ?? null);
});
if (!claimed) return null;
return { wasFirstHeartbeat: !agent.lastHeartbeatAt };
}
/** Queue admission and dispatch share the service lifecycle callbacks and execution ownership. */
export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDependencies) {
const {
@@ -416,137 +419,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
filterZombieCoalesceTarget,
} = dependencies;
function parseHeartbeatPolicy(agent: typeof agents.$inferSelect) {
const runtimeConfig = parseObject(agent.runtimeConfig);
const heartbeat = parseObject(runtimeConfig.heartbeat);
return {
enabled: asBoolean(heartbeat.enabled, false),
intervalSec: Math.max(0, asNumber(heartbeat.intervalSec, 0)),
wakeOnDemand: isHeartbeatWakeOnDemandEnabled(agent),
// A Dot binding has one external turn. Competing assignments must retain
// their queue position instead of claiming a second run that cannot bind.
maxConcurrentRuns: agent.adapterType === "paperclip_runner" &&
["openai_dot", "grok_bot", "muse"].includes(String(parseObject(agent.adapterConfig).provider))
? 1 : normalizeMaxConcurrentRuns(heartbeat.maxConcurrentRuns),
skipTimerWhenNoActionableWork: asBoolean(
heartbeat.skipTimerWhenNoActionableWork ??
heartbeat.requireActionableTimerWork ??
heartbeat.issueOnlyTimer,
false,
),
maxDailyRuns: normalizeOptionalNonNegativeInteger(
heartbeat.maxDailyRuns ??
heartbeat.dailyRunLimit ??
heartbeat.dailyRunCap ??
heartbeat.maxRunsPerDay,
),
maxDailyCostCents: normalizeOptionalNonNegativeInteger(
heartbeat.maxDailyCostCents ??
heartbeat.dailyCostCentsLimit ??
heartbeat.dailySpendCentsLimit ??
heartbeat.dailyBudgetCents,
),
};
}
function normalizeOptionalNonNegativeInteger(value: unknown) {
if (value === null || value === undefined || value === "") return null;
const normalized = Math.floor(asNumber(value, 0));
return normalized >= 0 ? normalized : null;
}
function currentUtcDayWindow(now = new Date()) {
const start = new Date(
Date.UTC(
now.getUTCFullYear(),
now.getUTCMonth(),
now.getUTCDate(),
0,
0,
0,
0,
),
);
const end = new Date(
Date.UTC(
now.getUTCFullYear(),
now.getUTCMonth(),
now.getUTCDate() + 1,
0,
0,
0,
0,
),
);
return { start, end };
}
async function getHeartbeatDailyCapBlock(
agent: typeof agents.$inferSelect,
policy: ReturnType<typeof parseHeartbeatPolicy>,
options: {
checkRunCap?: boolean;
checkCostCap?: boolean;
excludeRunId?: string | null;
now?: Date;
} = {},
client: Pick<Db, "select"> = db,
) {
const checkRunCap = options.checkRunCap ?? true;
const checkCostCap = options.checkCostCap ?? true;
const { start, end } = currentUtcDayWindow(options.now);
if (checkRunCap && policy.maxDailyRuns !== null) {
const conditions = [
eq(heartbeatRuns.companyId, agent.companyId),
eq(heartbeatRuns.agentId, agent.id),
gte(heartbeatRuns.startedAt, start),
lt(heartbeatRuns.startedAt, end),
notInArray(heartbeatRuns.status, ["queued", "scheduled_retry"]),
];
if (options.excludeRunId) {
conditions.push(sql`${heartbeatRuns.id} <> ${options.excludeRunId}`);
}
const [row] = await client
.select({ total: sql<number>`count(*)::integer` })
.from(heartbeatRuns)
.where(and(...conditions));
const observed = Number(row?.total ?? 0);
if (observed >= policy.maxDailyRuns) {
return {
reason: "heartbeat.daily_run_limit",
observed,
limit: policy.maxDailyRuns,
};
}
}
if (checkCostCap && policy.maxDailyCostCents !== null) {
const [row] = await client
.select({
total: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text`,
})
.from(costEvents)
.where(
and(
eq(costEvents.companyId, agent.companyId),
eq(costEvents.agentId, agent.id),
gte(costEvents.occurredAt, start),
lt(costEvents.occurredAt, end),
),
);
const observed = Number(row?.total ?? 0);
if (compareCents(String(row?.total ?? 0), policy.maxDailyCostCents) >= 0) {
return {
reason: "heartbeat.daily_cost_limit",
observed,
limit: policy.maxDailyCostCents,
};
}
}
return null;
}
const { parseHeartbeatPolicy, getHeartbeatDailyCapBlock, hasActionableTimerWork, nextTimerHeartbeatAt } = createHeartbeatTimerPolicy(db);
async function cancelQueuedRunForHeartbeatDailyCap(
run: typeof heartbeatRuns.$inferSelect,
@@ -598,38 +471,6 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
return cancelled;
}
async function hasActionableTimerWork(agent: typeof agents.$inferSelect) {
const row = await db
.select({ id: issues.id })
.from(issues)
.where(
and(
eq(issues.companyId, agent.companyId),
eq(issues.assigneeAgentId, agent.id),
isNull(issues.assigneeUserId),
isNull(issues.hiddenAt),
inArray(issues.status, [...TIMER_ACTIONABLE_ISSUE_STATUSES]),
isNull(issues.conversationAgentId),
nonIdleSlackIssueCondition(),
),
)
.limit(1)
.then((rows) => rows[0] ?? null);
return Boolean(row);
}
async function nextTimerHeartbeatAt(agent: typeof agents.$inferSelect, now = new Date()): Promise<number | null> {
const policy = parseHeartbeatPolicy(agent);
if (!policy.enabled || policy.intervalSec <= 0) return null;
if (policy.maxDailyRuns === 0 || policy.maxDailyCostCents === 0) return null;
if (policy.skipTimerWhenNoActionableWork && !(await hasActionableTimerWork(agent))) return null;
const dueAt = new Date(agent.lastHeartbeatAt ?? agent.createdAt).getTime() + policy.intervalSec * 1000;
const cap = await getHeartbeatDailyCapBlock(agent, policy, { now });
if (!cap) return dueAt;
return Math.max(dueAt, currentUtcDayWindow(now).end.getTime());
}
async function markTimerHeartbeatChecked(
agentId: string,
source: WakeupOptions["source"],
@@ -643,31 +484,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
});
}
async function claimDueTimerHeartbeat(
agent: typeof agents.$inferSelect,
now: Date,
intervalSec: number,
) {
const dueBefore = new Date(now.getTime() - intervalSec * 1000);
const claimed = await db.transaction(async tx => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
return tx
.update(agents)
.set({ lastHeartbeatAt: now, updatedAt: now })
.where(and(
eq(agents.id, agent.id),
eq(agents.companyId, agent.companyId),
or(
lte(agents.lastHeartbeatAt, dueBefore),
and(isNull(agents.lastHeartbeatAt), lte(agents.createdAt, dueBefore)),
),
))
.returning({ id: agents.id })
.then((rows) => rows[0] ?? null);
});
if (!claimed) return null;
return { wasFirstHeartbeat: !agent.lastHeartbeatAt };
}
const claimDueTimerHeartbeat = (agent: typeof agents.$inferSelect, now: Date, intervalSec: number) =>
claimDueTimerHeartbeatForAgent(db, agent, now, intervalSec);
function issueRunPriorityRank(priority: string | null | undefined) {
switch (priority) {
@@ -0,0 +1,252 @@
import { agents, heartbeatRuns, costEvents, issues, type Db } from "@paperclipai/db";
import { AGENT_DEFAULT_MAX_CONCURRENT_RUNS, compareCents } from "@paperclipai/shared";
import { and, eq, gte, lt, notInArray, sql, isNull, inArray, type SQL } from "drizzle-orm";
import { asBoolean, asNumber, parseObject } from "../../adapters/utils.js";
import { isHeartbeatWakeOnDemandEnabled } from "../heartbeat-policy.js";
import { nonIdleSlackIssueCondition } from "../slack-conversation-state.js";
const HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT = AGENT_DEFAULT_MAX_CONCURRENT_RUNS;
const HEARTBEAT_MAX_CONCURRENT_RUNS_MIN = 1;
const HEARTBEAT_MAX_CONCURRENT_RUNS_MAX = 50;
const TIMER_ACTIONABLE_ISSUE_STATUSES = ["todo", "in_progress"] as const;
function normalizeMaxConcurrentRuns(value: unknown) {
const parsed = Math.floor(
asNumber(value, HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT),
);
if (!Number.isFinite(parsed)) return HEARTBEAT_MAX_CONCURRENT_RUNS_DEFAULT;
return Math.max(
HEARTBEAT_MAX_CONCURRENT_RUNS_MIN,
Math.min(HEARTBEAT_MAX_CONCURRENT_RUNS_MAX, parsed),
);
}
export function heartbeatTimerPolicyIsWellFormed(agent: typeof agents.$inferSelect): boolean {
const record = (v: unknown): v is Record<string, unknown> => v !== null && typeof v === "object" && !Array.isArray(v);
if (!record(agent.runtimeConfig)) return false;
const heartbeat = agent.runtimeConfig.heartbeat;
if (heartbeat === undefined) return true;
if (!record(heartbeat)) return false;
for (const key of ["enabled", "wakeOnDemand", "wakeOnAssignment", "wakeOnOnDemand", "wakeOnAutomation",
"skipTimerWhenNoActionableWork", "requireActionableTimerWork", "issueOnlyTimer"]) {
if (heartbeat[key] !== undefined && typeof heartbeat[key] !== "boolean") return false;
}
for (const key of ["intervalSec", "maxConcurrentRuns", "maxDailyRuns", "dailyRunLimit", "dailyRunCap", "maxRunsPerDay",
"maxDailyCostCents", "dailyCostCentsLimit", "dailySpendCentsLimit", "dailyBudgetCents"]) {
const v = heartbeat[key];
if (v !== undefined && v !== null && v !== "" && (typeof v !== "number" || !Number.isFinite(v) || v < 0)) return false;
}
return true;
}
export function parseHeartbeatPolicy(agent: typeof agents.$inferSelect) {
const runtimeConfig = parseObject(agent.runtimeConfig);
const heartbeat = parseObject(runtimeConfig.heartbeat);
return {
enabled: asBoolean(heartbeat.enabled, false),
intervalSec: Math.max(0, asNumber(heartbeat.intervalSec, 0)),
wakeOnDemand: isHeartbeatWakeOnDemandEnabled(agent),
maxConcurrentRuns: agent.adapterType === "paperclip_runner" &&
["openai_dot", "grok_bot", "muse"].includes(String(parseObject(agent.adapterConfig).provider))
? 1 : normalizeMaxConcurrentRuns(heartbeat.maxConcurrentRuns),
skipTimerWhenNoActionableWork: asBoolean(
heartbeat.skipTimerWhenNoActionableWork ??
heartbeat.requireActionableTimerWork ??
heartbeat.issueOnlyTimer,
false,
),
maxDailyRuns: normalizeOptionalNonNegativeInteger(
heartbeat.maxDailyRuns ??
heartbeat.dailyRunLimit ??
heartbeat.dailyRunCap ??
heartbeat.maxRunsPerDay,
),
maxDailyCostCents: normalizeOptionalNonNegativeInteger(
heartbeat.maxDailyCostCents ??
heartbeat.dailyCostCentsLimit ??
heartbeat.dailySpendCentsLimit ??
heartbeat.dailyBudgetCents,
),
};
}
function normalizeOptionalNonNegativeInteger(value: unknown) {
if (value === null || value === undefined || value === "") return null;
const normalized = Math.floor(asNumber(value, 0));
return normalized >= 0 ? normalized : null;
}
function currentUtcDayWindow(now = new Date()) {
const start = new Date(
Date.UTC(
now.getUTCFullYear(),
now.getUTCMonth(),
now.getUTCDate(),
0,
0,
0,
0,
),
);
const end = new Date(
Date.UTC(
now.getUTCFullYear(),
now.getUTCMonth(),
now.getUTCDate() + 1,
0,
0,
0,
0,
),
);
return { start, end };
}
type TimerAgent = typeof agents.$inferSelect;
type TimerPolicy = ReturnType<typeof parseHeartbeatPolicy>;
type DailyCap = { reason: string; observed: number; limit: number } | null;
const timerKey = (companyId: string, agentId: string) => JSON.stringify([companyId, agentId]);
function dailyRunConditions(scope: SQL, now: Date) {
const { start, end } = currentUtcDayWindow(now);
return and(scope, gte(heartbeatRuns.startedAt, start), lt(heartbeatRuns.startedAt, end),
notInArray(heartbeatRuns.status, ["queued", "scheduled_retry"]));
}
function dailyCostConditions(scope: SQL, now: Date) {
const { start, end } = currentUtcDayWindow(now);
return and(scope, gte(costEvents.occurredAt, start), lt(costEvents.occurredAt, end));
}
function actionableConditions(scope: SQL) {
return and(scope, isNull(issues.assigneeUserId), isNull(issues.hiddenAt),
inArray(issues.status, [...TIMER_ACTIONABLE_ISSUE_STATUSES]), isNull(issues.conversationAgentId), nonIdleSlackIssueCondition());
}
export function heartbeatDailyCapFromTotals(policy: TimerPolicy, totals: { runs?: number; cost?: string },
checks: { checkRunCap?: boolean; checkCostCap?: boolean } = {}): DailyCap {
if (checks.checkRunCap !== false && policy.maxDailyRuns !== null && (totals.runs ?? 0) >= policy.maxDailyRuns)
return { reason: "heartbeat.daily_run_limit", observed: totals.runs ?? 0, limit: policy.maxDailyRuns };
if (checks.checkCostCap !== false && policy.maxDailyCostCents !== null
&& compareCents(totals.cost ?? "0", policy.maxDailyCostCents) >= 0)
return { reason: "heartbeat.daily_cost_limit", observed: Number(totals.cost ?? "0"), limit: policy.maxDailyCostCents };
return null;
}
export function projectTimerHeartbeatSchedule(agent: TimerAgent, now: Date,
input: { actionable: boolean | null; cap: DailyCap }) {
const policy = parseHeartbeatPolicy(agent);
if (!policy.enabled || policy.intervalSec <= 0 || policy.maxDailyRuns === 0 || policy.maxDailyCostCents === 0)
return { deadline: null, policy, actionable: null, cap: null };
if (input.actionable === false) return { deadline: null, policy, actionable: false, cap: null };
const dueAt = new Date(agent.lastHeartbeatAt ?? agent.createdAt).getTime() + policy.intervalSec * 1000;
return { deadline: input.cap ? Math.max(dueAt, currentUtcDayWindow(now).end.getTime()) : dueAt,
policy, actionable: input.actionable, cap: input.cap };
}
export function createHeartbeatTimerPolicy(db: Pick<Db, "select">) {
async function getHeartbeatDailyCapBlock(agent: TimerAgent, policy: TimerPolicy,
options: { checkRunCap?: boolean; checkCostCap?: boolean; excludeRunId?: string | null; now?: Date } = {},
client: Pick<Db, "select"> = db): Promise<DailyCap> {
const now = options.now ?? new Date();
if (options.checkRunCap !== false && policy.maxDailyRuns !== null) {
const [row] = await client.select({ total: sql<number>`count(*)::integer` }).from(heartbeatRuns)
.where(dailyRunConditions(and(eq(heartbeatRuns.companyId, agent.companyId), eq(heartbeatRuns.agentId, agent.id),
options.excludeRunId ? sql`${heartbeatRuns.id} <> ${options.excludeRunId}` : undefined)!, now));
const cap = heartbeatDailyCapFromTotals(policy, { runs: Number(row?.total ?? 0) }, { checkCostCap: false });
if (cap) return cap;
}
if (options.checkCostCap !== false && policy.maxDailyCostCents !== null) {
const [row] = await client.select({ total: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text` }).from(costEvents)
.where(dailyCostConditions(and(eq(costEvents.companyId, agent.companyId), eq(costEvents.agentId, agent.id))!, now));
return heartbeatDailyCapFromTotals(policy, { cost: String(row?.total ?? 0) }, { checkRunCap: false });
}
return null;
}
async function hasActionableTimerWork(agent: TimerAgent) {
const rows = await db.select({ id: issues.id }).from(issues)
.where(actionableConditions(and(eq(issues.companyId, agent.companyId), eq(issues.assigneeAgentId, agent.id))!)).limit(1);
return rows.length > 0;
}
async function readTimerHeartbeatSchedule(agent: TimerAgent, now = new Date()) {
const policy = parseHeartbeatPolicy(agent);
if (!policy.enabled || policy.intervalSec <= 0 || policy.maxDailyRuns === 0 || policy.maxDailyCostCents === 0)
return projectTimerHeartbeatSchedule(agent, now, { actionable: null, cap: null });
const actionable = policy.skipTimerWhenNoActionableWork ? await hasActionableTimerWork(agent) : null;
const cap = actionable === false ? null : await getHeartbeatDailyCapBlock(agent, policy, { now });
return projectTimerHeartbeatSchedule(agent, now, { actionable, cap });
}
async function readTimerHeartbeatSchedules(candidates: TimerAgent[], now: Date, checkBudget: () => void = () => {}) {
if (candidates.length > 10_000) return;
const allowed = new Set(candidates.map(agent => timerKey(agent.companyId, agent.id)));
if (allowed.size !== candidates.length) return;
const eligible = candidates.filter(agent => {
const p = parseHeartbeatPolicy(agent);
return p.enabled && p.intervalSec > 0 && p.maxDailyRuns !== 0 && p.maxDailyCostCents !== 0;
});
const runIds = eligible.filter(a => parseHeartbeatPolicy(a).maxDailyRuns !== null).map(a => a.id);
const costIds = eligible.filter(a => parseHeartbeatPolicy(a).maxDailyCostCents !== null).map(a => a.id);
const actionableIds = eligible.filter(a => parseHeartbeatPolicy(a).skipTimerWhenNoActionableWork).map(a => a.id);
const runs = new Map<string, number>(), costs = new Map<string, string>(), actionable = new Set<string>();
const state: unknown[] = [];
if (runIds.length) {
checkBudget();
const rows = await db.select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
total: sql<number>`count(*)::integer` }).from(heartbeatRuns)
.where(dailyRunConditions(inArray(heartbeatRuns.agentId, runIds), now))
.groupBy(heartbeatRuns.companyId, heartbeatRuns.agentId).orderBy(heartbeatRuns.companyId, heartbeatRuns.agentId).limit(10_001);
if (rows.length > 10_000) return;
for (const row of rows) {
checkBudget();
const key = timerKey(row.companyId, row.agentId);
if (!allowed.has(key) || runs.has(key) || typeof row.total !== "number" || !Number.isSafeInteger(row.total) || row.total < 0) return;
runs.set(key, row.total);
}
state.push(rows);
}
if (costIds.length) {
checkBudget();
const rows = await db.select({ companyId: costEvents.companyId, agentId: costEvents.agentId,
total: sql<string>`coalesce(sum(${costEvents.costCents}), 0)::text` }).from(costEvents)
.where(dailyCostConditions(inArray(costEvents.agentId, costIds), now))
.groupBy(costEvents.companyId, costEvents.agentId).orderBy(costEvents.companyId, costEvents.agentId).limit(10_001);
if (rows.length > 10_000) return;
for (const row of rows) {
checkBudget();
if (!row.agentId) return;
const key = timerKey(row.companyId, row.agentId);
if (!allowed.has(key) || costs.has(key) || typeof row.total !== "string" || !/^-?\d+(?:\.\d+)?$/.test(row.total)) return;
costs.set(key, row.total);
}
state.push(rows);
}
if (actionableIds.length) {
checkBudget();
const rows = await db.select({ companyId: issues.companyId, agentId: issues.assigneeAgentId }).from(issues)
.where(actionableConditions(inArray(issues.assigneeAgentId, actionableIds)))
.groupBy(issues.companyId, issues.assigneeAgentId).orderBy(issues.companyId, issues.assigneeAgentId).limit(10_001);
if (rows.length > 10_000) return;
for (const row of rows) {
checkBudget();
if (!row.agentId) return;
const key = timerKey(row.companyId, row.agentId);
if (!allowed.has(key) || actionable.has(key)) return;
actionable.add(key);
}
state.push(rows);
}
const schedules = new Map<string, ReturnType<typeof projectTimerHeartbeatSchedule>>();
for (const agent of candidates) {
checkBudget();
const policy = parseHeartbeatPolicy(agent), key = timerKey(agent.companyId, agent.id);
schedules.set(agent.id, projectTimerHeartbeatSchedule(agent, now, {
actionable: policy.skipTimerWhenNoActionableWork ? actionable.has(key) : null,
cap: heartbeatDailyCapFromTotals(policy, { runs: runs.get(key), cost: costs.get(key) }),
}));
}
return { schedules, state };
}
async function nextTimerHeartbeatAt(agent: TimerAgent, now = new Date()): Promise<number | null> {
return (await readTimerHeartbeatSchedule(agent, now)).deadline;
}
return { parseHeartbeatPolicy, getHeartbeatDailyCapBlock, hasActionableTimerWork, nextTimerHeartbeatAt,
readTimerHeartbeatSchedule, readTimerHeartbeatSchedules };
}
@@ -0,0 +1,218 @@
import { pluginIdleManifestGeneration } from "./plugin-idle-catalog.js";
import { agents, heartbeatRuns, costEvents, issues, type Db } from "@paperclipai/db";
import { getTableName } from "drizzle-orm";
import { readIdleAgentTimerSchedule } from "./heartbeat/idle-timer-schedule.js";
import { createHeartbeatTimerPolicy } from "./heartbeat/timer-policy.js";
import { describe, expect, it } from "vitest";
import { composeIdleDeadlineProof, createIdleDeadlineInspectionBudget, deadlineWorkReplacement, readIdleDeadlineProof,
IdleDeadlineInspectionBudgetExceeded, type IdleDeadlineProofOptions } from "./idle-deadline-proof.js";
import type { VerifiedIdleCapabilities } from "./idle-incoming-wake-capability.js";
import { projectIdlePluginJobSchedule, type IdlePluginJobRow } from "./plugin-job-idle-schedule.js";
import { projectIdleProviderTraceSchedule, type IdleProviderTraceRow } from "./provider-trace-idle-schedule.js";
import { projectIdleGitHubGrantContinuity } from "./tool-access/github-grant-continuity.js";
const now = Date.parse("2099-01-01T00:00:00Z");
const verifiedCapability = { deadlineCatalog: "domain_deadlines_v1" } as VerifiedIdleCapabilities;
const options: IdleDeadlineProofOptions = { now, verifiedCapability, quietPluginIds: [] };
const hold = { ownerId: "owned", expiresAt: now + 300_000 };
const snapshot = (state: unknown = [], deadlines: number[] = []) => ({ blocked: false, state, deadlines });
const fixtures = () => ({ timers: snapshot(["agent"], [now + 600_000]), jobs: snapshot(), traces: snapshot(),
github: { ...snapshot(), connectionIds: [] as string[] } });
describe("domain deadline handoff", () => {
it("preserves earliest legacy and owner obligations without moving an inspected deadline", () => {
const all = fixtures(); all.traces = snapshot(["trace"], [now + 500_000]);
const routine = { version: 1 as const, kind: "routines_and_issue_monitors" as const,
generation: "b".repeat(64), nextWakeAt: new Date(now + 400_000).toISOString() };
const first = composeIdleDeadlineProof(hold, { ...options, scheduledWake: routine }, all);
expect(first.scheduledWake).toMatchObject({ version: 2, kind: "domain_deadlines", catalog: "domain_deadlines_v1", nextWakeAt: routine.nextWakeAt });
expect(composeIdleDeadlineProof(hold, { ...options, now: now + 1, scheduledWake: routine }, all)).toEqual(first);
});
it("binds passive state, drain owner/expiry and exact plugin worker generation", () => {
const all = fixtures(), first = composeIdleDeadlineProof(hold, options, all);
const passive = fixtures(); passive.jobs.state = ["changed-paused-job"];
expect(composeIdleDeadlineProof(hold, options, passive).generation).not.toBe(first.generation);
expect(composeIdleDeadlineProof({ ...hold, ownerId: "replacement" }, options, all).generation).not.toBe(first.generation);
expect(composeIdleDeadlineProof({ ...hold, expiresAt: hold.expiresAt + 1 }, options, all).generation).not.toBe(first.generation);
const plugin = { ...options, quietPluginIds: ["plugin"], quietPluginGeneration: "worker-1" };
expect(composeIdleDeadlineProof(hold, plugin, all).generation).not.toBe(composeIdleDeadlineProof(hold, { ...plugin, quietPluginGeneration: "worker-2" }, all).generation);
});
it("invalidates moved-earlier/cancelled obligations and refuses due/unknown work", () => {
const all = fixtures(), first = composeIdleDeadlineProof(hold, options, all);
all.timers = snapshot(["edited"], [now + 450_000]);
expect(composeIdleDeadlineProof(hold, options, all).generation).not.toBe(first.generation);
all.timers = snapshot(["cancelled"]);
expect(composeIdleDeadlineProof(hold, options, all).scheduledWake).toBeUndefined();
all.timers = snapshot(["due"], [now]);
expect(composeIdleDeadlineProof(hold, options, all)).toEqual({ blocked: true });
expect(composeIdleDeadlineProof(hold, { ...options, verifiedCapability: undefined }, fixtures())).toEqual({ blocked: true });
expect(composeIdleDeadlineProof(hold, { ...options, quietPluginIds: ["missing-worker"] }, fixtures())).toEqual({ blocked: true });
});
it("keeps watchdogs, pending merge cards, and recovery predicates intact", () => {
expect(deadlineWorkReplacement("agents.execution_or_timer")).toBe("SELECT 1 FROM agents WHERE status = 'running'");
expect(deadlineWorkReplacement("plugin_jobs")).toContain("NOT IN ('active', 'paused')");
for (const reason of ["issue_watchdogs", "issue_thread_interactions", "heartbeat_runs", "unknown"]) expect(deadlineWorkReplacement(reason)).toBeUndefined();
});
it("bounds whole-inspection query count and monotonic elapsed time before final hold headroom", () => {
let elapsed = 0;
const budget = createIdleDeadlineInspectionBudget(now, hold.expiresAt, () => elapsed);
for (let i = 0; i < 64; i++) budget.beforeQuery();
expect(budget.queries).toBe(64);
expect(() => budget.beforeQuery()).toThrow(IdleDeadlineInspectionBudgetExceeded);
const timed = createIdleDeadlineInspectionBudget(now, hold.expiresAt, () => elapsed);
elapsed = 2_000;
expect(() => timed.beforeQuery()).toThrow(IdleDeadlineInspectionBudgetExceeded);
const short = createIdleDeadlineInspectionBudget(now, now + 60_000, () => 0);
expect(() => short.beforeQuery()).toThrow(IdleDeadlineInspectionBudgetExceeded);
expect(short.queries).toBe(0);
});
});
const pluginJob = (): IdlePluginJobRow => ({ job: { id: "job", pluginId: "plugin", jobKey: "daily", status: "active", schedule: "0 * * * *", nextRunAt: new Date(now + 600_000) } as IdlePluginJobRow["job"], plugin: { id: "plugin", pluginKey: "fixture", version: "1.0.0", apiVersion: 1, status: "ready", manifestJson: { id: "fixture", version: "1.0.0", apiVersion: 1, jobs: [{ jobKey: "daily", schedule: "0 * * * *" }] } } as IdlePluginJobRow["plugin"], unsettled: false });
it("plugin jobs require exact quiet catalog/declaration and retain unsettled/missing/due work", () => {
const row = pluginJob();
const catalog = [{ pluginId: "plugin", manifestGeneration: pluginIdleManifestGeneration(row.plugin!.manifestJson) }];
expect(projectIdlePluginJobSchedule([row], ["plugin"], now, catalog).deadlines).toEqual([now + 600_000]);
expect(projectIdlePluginJobSchedule([row], [], now).blocked).toBe(true);
for (const edit of [{ unsettled: true }, { job: { ...row.job, nextRunAt: null } }, { job: { ...row.job, nextRunAt: new Date(now) } }])
expect(projectIdlePluginJobSchedule([{ ...row, ...edit }], ["plugin"], now, catalog).blocked).toBe(true);
row.plugin!.manifestJson.jobs![0].schedule = "*/5 * * * *";
expect(projectIdlePluginJobSchedule([row], ["plugin"], now, catalog).blocked).toBe(true);
row.job.status = "paused";
expect(projectIdlePluginJobSchedule([row], [], now)).toMatchObject({ blocked: false, deadlines: [] });
});
const trace = (): IdleProviderTraceRow => ({ finishedRun: true, trace: { id: "trace", companyId: "company", runId: "run", status: "complete", traceRef: "s3-segments-v1:company/run/11111111-1111-4111-8111-111111111111.ndjson", expiresAt: new Date(now + 86_400_000), deletedAt: null } as IdleProviderTraceRow["trace"] });
it("trace expiry transfers actual retained data; capture/import/deletion/cross-company work stays blocked", () => {
const row = trace(), original = JSON.stringify(row);
expect(projectIdleProviderTraceSchedule([row], now)).toMatchObject({ blocked: false, deadlines: [now + 86_400_000] });
expect(JSON.stringify(row)).toBe(original);
for (const patch of [{ status: "capturing" }, { traceRef: row.trace.traceRef.replace("s3-segments-v1:", "s3-import-v1:") }, { companyId: "foreign" }, { expiresAt: new Date(now) }, { status: "deleted", deletedAt: new Date(now) }])
expect(projectIdleProviderTraceSchedule([{ ...trace(), trace: { ...trace().trace, ...patch } }], now).blocked).toBe(true);
expect(projectIdleProviderTraceSchedule([{ ...trace(), finishedRun: false }], now).blocked).toBe(true);
const expired = trace(); expired.trace.status = "expired"; expired.trace.deletedAt = new Date(now); expired.trace.traceRef = "00000000-0000-0000-0000-000000000000.ndjson";
expect(projectIdleProviderTraceSchedule([expired], now)).toMatchObject({ blocked: false, deadlines: [] });
});
type ContinuityRow = Parameters<typeof projectIdleGitHubGrantContinuity>[0][number];
const github = (): ContinuityRow => ({ companyStatus: "active", connection: { id: "connection", companyId: "company", status: "active", enabled: true, authKind: "oauth", credentialSource: "paperclip_vault", config: { sourceTemplateKey: "github", oauth: { strategy: "paperclip_cloud_connector", connectorProfile: "github.code" } } } as unknown as ContinuityRow["connection"], grant: { id: "grant", companyId: "company", connectionId: "connection", kind: "user", subjectUserId: "user", credentialSecretRefs: [{ configPath: "oauth.access_token", secretId: "secret" }], providerTenant: { oauth: { accessTokenExpiresAt: new Date(now + 8 * 3_600_000).toISOString(), refreshedAt: new Date(now).toISOString() }, github: { userId: "user", installationIds: ["101"], lastAccessRefreshAt: new Date(now).toISOString() } } } as ContinuityRow["grant"] });
it("GitHub enrollment alone never bypasses its credential continuity deadline or unresolved authority", () => {
expect(projectIdleGitHubGrantContinuity([github()], now)).toMatchObject({ blocked: false, deadlines: [now + 7 * 3_600_000], connectionIds: ["connection"] });
const nonexpiring = github(); delete nonexpiring.grant.providerTenant!.oauth;
expect(projectIdleGitHubGrantContinuity([nonexpiring], now).deadlines).toEqual([now + 30 * 86_400_000]);
const missing = github(); missing.grant.credentialSecretRefs = [];
expect(projectIdleGitHubGrantContinuity([missing], now).blocked).toBe(true);
const foreign = github(); foreign.grant.companyId = "foreign";
expect(projectIdleGitHubGrantContinuity([foreign], now).blocked).toBe(true);
const expired = github(); expired.grant.providerTenant!.oauth!.accessTokenExpiresAt = new Date(now).toISOString();
expect(projectIdleGitHubGrantContinuity([expired], now).blocked).toBe(true);
});
function timerAgent(index: number): typeof agents.$inferSelect {
return { id: `agent-${index}`, companyId: "company", name: "Timer", status: "idle", reportsTo: null,
adapterType: "codex_local", adapterConfig: {}, runtimeConfig: { heartbeat: { enabled: true, intervalSec: 3600, maxDailyRuns: 1 } },
createdAt: new Date(now), lastHeartbeatAt: new Date(now) } as unknown as typeof agents.$inferSelect;
}
function fixtureDb(tables: Record<string, unknown[]>, onRead: () => void = () => {}) {
return { select: () => {
onRead();
let table = "";
const query = { from: (source: Parameters<typeof getTableName>[0]) => { table = getTableName(source); return query; },
leftJoin: () => query, orderBy: () => query, where: () => query, groupBy: () => query,
limit: () => Promise.resolve(tables[table] ?? []),
then: (resolve: (rows: unknown[]) => unknown) => Promise.resolve(tables[table] ?? []).then(resolve) };
return query;
} } as unknown as Pick<Db, "select">;
}
it("batches more than128 eligible timers and the old80 run-cap shape within a fixed query budget", async () => {
const rows = Array.from({ length: 160 }, (_, i) => ({ agent: timerAgent(i), companyStatus: "active", companyUpdatedAt: new Date(now) }));
rows.push(...["paused", "terminated", "idle"].map((status, i) => ({ agent: { ...timerAgent(200 + i), status,
runtimeConfig: { heartbeat: { enabled: false } } }, companyStatus: "active", companyUpdatedAt: new Date(now) })));
let reads = 0;
const tables: Record<string, unknown[]> = { agents: rows };
const tx = fixtureDb(tables, () => { reads++; });
const future = await readIdleDeadlineProof(tx, hold, options);
expect(future.blocked).toBe(false);
expect(future.scheduledWake?.nextWakeAt).toBe(new Date(now + 3_600_000).toISOString());
expect(reads).toBe(5);
tables.agents = rows.slice(0, 80); reads = 0;
expect((await readIdleDeadlineProof(tx, hold, options)).blocked).toBe(false);
expect(reads).toBe(5);
rows[0].agent.lastHeartbeatAt = new Date(now - 3_600_000); tables.agents = rows; reads = 0;
expect((await readIdleDeadlineProof(tx, hold, options)).blocked).toBe(true);
expect(reads).toBe(2);
rows[0].agent.lastHeartbeatAt = new Date(now);
tables.agents = Array.from({ length: 10_001 }, () => rows[0]); reads = 0;
expect(await readIdleDeadlineProof(tx, hold, options)).toEqual({ blocked: true });
expect(reads).toBe(1);
tables.agents = rows;
for (const total of [null, "", true, false, "0", NaN, Infinity, -1, 1.5]) {
tables.heartbeat_runs = [{ companyId: rows[0].agent.companyId, agentId: rows[0].agent.id, total }];
expect((await readIdleDeadlineProof(tx, hold, options)).blocked).toBe(true);
}
tables.heartbeat_runs = [{ companyId: "foreign", agentId: rows[0].agent.id, total: 0 }];
expect((await readIdleDeadlineProof(tx, hold, options)).blocked).toBe(true);
});
it("single and batch owner schedules share cap, actionable, UTC and decimal semantics", async () => {
for (const heartbeat of [
{ enabled: false }, { enabled: true, intervalSec: 600, maxDailyRuns: 0 },
{ enabled: true, intervalSec: 600, maxDailyRuns: 2 },
{ enabled: true, intervalSec: 600, maxDailyCostCents: 1 },
{ enabled: true, intervalSec: 600, maxDailyRuns: 2, maxDailyCostCents: 1, skipTimerWhenNoActionableWork: true },
]) for (const actionable of [false, true]) for (const runTotal of [0, 2]) for (const cost of ["0.999999", "1.000001"]) {
const agent = { ...timerAgent(0), runtimeConfig: { heartbeat } };
const owner = createHeartbeatTimerPolicy(fixtureDb({
[getTableName(heartbeatRuns)]: [{ companyId: agent.companyId, agentId: agent.id, total: runTotal }],
[getTableName(costEvents)]: [{ companyId: agent.companyId, agentId: agent.id, total: cost }],
[getTableName(issues)]: actionable ? [{ companyId: agent.companyId, agentId: agent.id, id: "issue" }] : [],
}));
const single = await owner.readTimerHeartbeatSchedule(agent, new Date(now));
const batch = await owner.readTimerHeartbeatSchedules([agent], new Date(now));
expect(batch?.schedules.get(agent.id)).toEqual(single);
}
});
it("checks global suppression even without eligible timers and retains owner org-chain checks", async () => {
for (const rows of [[], [timerAgent(1)], [{ ...timerAgent(1), status: "paused" }]]) {
if (rows[0]?.status === "idle") rows[0].runtimeConfig = { heartbeat: { enabled: false } };
for (const env of [{ PAPERCLIP_IN_WORKTREE: "true" }, { PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS: "1" },
{ PAPERCLIP_RESTORE_IN_PROGRESS: "yes" }, { HEARTBEAT_SCHEDULER_ENABLED: "false" }]) {
let reads = 0;
expect((await readIdleAgentTimerSchedule(fixtureDb({ agents: rows }, () => { reads++; }), now, env)).blocked).toBe(true);
expect(reads).toBe(0);
}
}
const parent = timerAgent(1), child = { ...timerAgent(2), reportsTo: parent.id };
const rows = [parent, child].map(agent => ({ agent, companyStatus: "active", companyUpdatedAt: new Date(now) }));
const tx = fixtureDb({ agents: rows });
expect((await readIdleAgentTimerSchedule(tx, now, {})).blocked).toBe(false);
parent.status = "paused";
expect((await readIdleAgentTimerSchedule(tx, now, {})).blocked).toBe(false);
parent.status = "terminated";
expect((await readIdleAgentTimerSchedule(tx, now, {})).blocked).toBe(true);
parent.status = "idle"; parent.reportsTo = child.id;
expect((await readIdleAgentTimerSchedule(tx, now, {})).blocked).toBe(true);
let checks = 0;
await expect(readIdleAgentTimerSchedule(tx, now, {}, () => {
if (++checks > 1) throw new IdleDeadlineInspectionBudgetExceeded();
})).rejects.toThrow(IdleDeadlineInspectionBudgetExceeded);
});
it("does not turn unrelated GitHub grant shapes into continuity authority", () => {
const unrelated = github(); unrelated.connection!.authKind = "api_key";
expect(projectIdleGitHubGrantContinuity([unrelated], now)).toMatchObject({ blocked: false, deadlines: [], connectionIds: [] });
unrelated.connection!.authKind = "oauth"; unrelated.connection!.config = { sourceTemplateKey: "other" };
expect(projectIdleGitHubGrantContinuity([unrelated], now)).toMatchObject({ blocked: false, connectionIds: [] });
for (const patch of [{ oauth: { connectorProfile: "github.code", strategy: "unknown_strategy" } },
{ oauth: { connectorProfile: "github.code" } }, { oauth: null }, { oauth: { connectorProfile: 123 } }]) {
const malformed = github(); malformed.connection!.config = { sourceTemplateKey: "github", ...patch };
expect(projectIdleGitHubGrantContinuity([malformed], now).blocked).toBe(true);
}
for (const patch of [{ authKind: "unknown_auth" }, { credentialSource: "unknown_source" }]) {
const malformed = github(); Object.assign(malformed.connection!, patch);
expect(projectIdleGitHubGrantContinuity([malformed], now).blocked).toBe(true);
}
const persisted = github(); persisted.connection!.config.oauth = { connectorProfile: "github.code", strategy: "paperclip_id_connector" };
expect(projectIdleGitHubGrantContinuity([persisted], now).blocked).toBe(false);
const agent = github(); agent.grant.kind = "agent"; agent.grant.subjectUserId = null; agent.grant.subjectAgentId = "dedicated-agent";
expect(projectIdleGitHubGrantContinuity([agent], now).connectionIds).toEqual(["connection"]);
const orphan = github(); orphan.connection = null;
expect(projectIdleGitHubGrantContinuity([orphan], now).blocked).toBe(true);
});
@@ -0,0 +1,98 @@
import type { IdlePluginCatalogEntry } from "./plugin-idle-catalog.js";
import { createHash, randomUUID } from "node:crypto";
import type { Db } from "@paperclipai/db";
import type { VerifiedIdleCapabilities } from "./idle-incoming-wake-capability.js";
import type { ScheduledWake } from "./idle-scheduled-work.js";
import { IDLE_DEADLINE_CATALOG, type DomainScheduledWake, type IdleDeadlineSnapshot } from "./idle-deadline-types.js";
import { readIdleAgentTimerSchedule } from "./heartbeat/idle-timer-schedule.js";
import { readIdlePluginJobSchedule } from "./plugin-job-idle-schedule.js";
import { readIdleProviderTraceSchedule } from "./provider-trace-idle-schedule.js";
import { readIdleGitHubGrantContinuity } from "./tool-access/github-grant-continuity.js";
const processGeneration = randomUUID();
export class IdleDeadlineInspectionBudgetExceeded extends Error {}
export function createIdleDeadlineInspectionBudget(now: number, expiresAt: number,
monotonicNow: () => number = () => performance.now()) {
const started = monotonicNow();
const elapsedLimit = Math.min(2_000, expiresAt - now - 60_000);
let queries = 0;
const check = () => {
if (!Number.isFinite(elapsedLimit) || elapsedLimit <= 0 || monotonicNow() - started >= elapsedLimit)
throw new IdleDeadlineInspectionBudgetExceeded("Idle deadline inspection exhausted its time budget");
};
return { check, beforeQuery() {
check();
if (++queries > 64) throw new IdleDeadlineInspectionBudgetExceeded("Idle deadline inspection exhausted its query budget");
}, get queries() { return queries; } };
}
export type IdleDeadlineProof = {
blocked: boolean;
generation?: string;
scheduledWake?: DomainScheduledWake;
githubContinuity?: { connectionIds: string[]; generation: string };
};
export type IdleDeadlineProofOptions = {
now: number;
verifiedCapability?: VerifiedIdleCapabilities;
quietPluginIds: readonly string[];
quietPluginGeneration?: string;
quietPluginCatalog?: readonly IdlePluginCatalogEntry[];
scheduledWake?: ScheduledWake;
passiveGeneration?: string;
};
const generation = (state: unknown) => createHash("sha256").update(JSON.stringify(state)).digest("hex");
export function composeIdleDeadlineProof(hold: { ownerId: string; expiresAt: number }, options: IdleDeadlineProofOptions,
snapshots: { timers: IdleDeadlineSnapshot; jobs: IdleDeadlineSnapshot; traces: IdleDeadlineSnapshot;
github: IdleDeadlineSnapshot & { connectionIds: string[] } }): IdleDeadlineProof {
if (!Number.isFinite(options.now) || options.verifiedCapability?.deadlineCatalog !== IDLE_DEADLINE_CATALOG || !hold.ownerId
|| !Number.isFinite(hold.expiresAt) || hold.expiresAt <= options.now
|| Object.values(snapshots).some(snapshot => snapshot.blocked)
|| (options.quietPluginIds.length > 0 && !options.quietPluginGeneration)) return { blocked: true };
const deadlines = Object.values(snapshots).flatMap(snapshot => snapshot.deadlines);
if (options.scheduledWake) deadlines.push(Date.parse(options.scheduledWake.nextWakeAt));
if (deadlines.some(at => !Number.isFinite(at) || at <= options.now)) return { blocked: true };
const proofGeneration = generation([IDLE_DEADLINE_CATALOG, processGeneration, hold,
options.quietPluginIds, options.quietPluginGeneration, options.quietPluginCatalog, options.scheduledWake, options.passiveGeneration,
Object.entries(snapshots).map(([owner, snapshot]) => [owner, snapshot.state, snapshot.deadlines])]);
const githubContinuity = { connectionIds: snapshots.github.connectionIds,
generation: generation([proofGeneration, snapshots.github.connectionIds]) };
return { blocked: false, generation: proofGeneration, githubContinuity,
...(deadlines.length ? { scheduledWake: { version: 2, kind: "domain_deadlines", catalog: IDLE_DEADLINE_CATALOG,
generation: proofGeneration, nextWakeAt: new Date(Math.min(...deadlines)).toISOString() } } : {}) };
}
export async function readIdleDeadlineProof(tx: Pick<Db, "select">, hold: { ownerId: string; expiresAt: number },
options: IdleDeadlineProofOptions): Promise<IdleDeadlineProof> {
if (options.verifiedCapability?.deadlineCatalog !== IDLE_DEADLINE_CATALOG) return { blocked: true };
const budget = createIdleDeadlineInspectionBudget(options.now, hold.expiresAt);
const guarded = { select: (...args: unknown[]) => {
budget.beforeQuery();
return Reflect.apply(tx.select, tx, args);
} } as Pick<Db, "select">;
try {
const timers = await readIdleAgentTimerSchedule(guarded, options.now, process.env, budget.check);
budget.check();
if (timers.blocked) return { blocked: true };
const jobs = await readIdlePluginJobSchedule(guarded, options.quietPluginIds, options.now, options.quietPluginCatalog);
budget.check();
if (jobs.blocked) return { blocked: true };
const traces = await readIdleProviderTraceSchedule(guarded, options.now);
budget.check();
if (traces.blocked) return { blocked: true };
const github = await readIdleGitHubGrantContinuity(guarded, options.now, budget.check);
budget.check();
const proof = composeIdleDeadlineProof(hold, options, { timers, jobs, traces, github });
budget.check();
return proof;
} catch (error) {
if (error instanceof IdleDeadlineInspectionBudgetExceeded) return { blocked: true };
throw error;
}
}
export function deadlineWorkReplacement(reason: string): string | undefined {
if (reason === "agents.execution_or_timer") return `SELECT 1 FROM agents WHERE status = 'running'`;
if (reason === "plugin_jobs") return `SELECT 1 FROM plugin_jobs WHERE status NOT IN ('active', 'paused')`;
return undefined;
}
@@ -0,0 +1,19 @@
export const IDLE_DEADLINE_CATALOG = "domain_deadlines_v1" as const;
export interface IdleDeadlineReceipt {
catalog: typeof IDLE_DEADLINE_CATALOG;
generation: string;
}
export interface DomainScheduledWake {
version: 2;
kind: "domain_deadlines";
catalog: typeof IDLE_DEADLINE_CATALOG;
generation: string;
nextWakeAt: string;
}
export interface IdleDeadlineSnapshot {
blocked: boolean;
state: unknown;
deadlines: number[];
}
export const idleDeadlineRowLimit = 10_000;
export const isFutureIdleDeadline = (at: number, now: number) => Number.isFinite(at) && at > now;
+6 -4
View File
@@ -1,3 +1,4 @@
import type { VerifiedIdleCapabilities } from "./idle-incoming-wake-capability.js";
import { idleApiActivity } from "./idle-api-activity.js";
import { promises as fs } from "node:fs";
import path from "node:path";
@@ -5,10 +6,11 @@ import { resolvePaperclipInstanceRoot } from "../home-paths.js";
import { idleWorkSnapshot } from "./task-admission.js";
import type { IdleSleepDiagnosticSink } from "./idle-sleep-diagnostics.js";
export type IdleLocalWorkOptions = { backupCheckpoint?: boolean; diagnostic?: IdleSleepDiagnosticSink; deadlineCapability?: VerifiedIdleCapabilities };
export type IdleLocalWork = "none" | "present" | "unknown";
const spoolDirectories = new Set<string>();
const localWorkProbes = new Set<() => IdleLocalWork>();
export function registerIdleLocalWorkProbe(probe: () => IdleLocalWork): () => void {
const localWorkProbes = new Set<(options: IdleLocalWorkOptions) => IdleLocalWork>();
export function registerIdleLocalWorkProbe(probe: (options: IdleLocalWorkOptions) => IdleLocalWork): () => void {
localWorkProbes.add(probe);
return () => { localWorkProbes.delete(probe); };
}
@@ -63,7 +65,7 @@ export async function inspectIdleQuotaRecovery(directory: string): Promise<IdleL
}
}
export async function readIdleLocalWork(options: { backupCheckpoint?: boolean; diagnostic?: IdleSleepDiagnosticSink } = {}): Promise<IdleLocalWork> {
export async function readIdleLocalWork(options: IdleLocalWorkOptions = {}): Promise<IdleLocalWork> {
const record = options.diagnostic;
const states: IdleLocalWork[] = [];
record?.("local.startup", startupComplete ? "none" : "unknown");
@@ -84,7 +86,7 @@ export async function readIdleLocalWork(options: { backupCheckpoint?: boolean; d
}
let probes: IdleLocalWork;
try {
const local = [idleApiActivity.active() ? "present" : "none", ...[...localWorkProbes].map(probe => probe())];
const local = [idleApiActivity.active() ? "present" : "none", ...[...localWorkProbes].map(probe => probe(options))];
probes = local.some(state => state !== "none" && state !== "present") ? "unknown"
: local.includes("present") ? "present" : "none";
} catch { probes = "unknown"; }
+55 -10
View File
@@ -1,3 +1,6 @@
import { readIdleDeadlineProof, deadlineWorkReplacement, type IdleDeadlineProof } from "./idle-deadline-proof.js";
import type { DomainScheduledWake, IdleDeadlineReceipt } from "./idle-deadline-types.js";
import type { IdlePluginCatalogEntry } from "./plugin-idle-catalog.js";
import type { ApiWakeAdmission } from "./idle-api-wake-admission.js";
import { readIdleIncomingWakeProof, incomingWakeWorkQuery, type IdleIncomingWakeProof } from "./idle-incoming-wake.js";
import type { VerifiedIdleCapabilities, IdleIncomingWakeCapability } from "./idle-incoming-wake-capability.js";
@@ -43,15 +46,18 @@ export interface IdleSleepDrainStatus {
export interface IdleSleepSafety {
version: 1 | 2 | 3;
backgroundWork: "none" | "present" | "unknown";
scheduledWake?: ScheduledWake;
scheduledWake?: ScheduledWake | DomainScheduledWake;
diagnostics?: IdleSleepDiagnostics;
incomingWake?: IdleIncomingWakeCapability;
deadlineProof?: IdleDeadlineReceipt;
apiWakeAdmission?: ApiWakeAdmission;
}
export type InspectIdlePlugins = (hold: { ownerId: string; expiresAt: number }) => Promise<{
backgroundWork: "none" | "present" | "unknown";
pluginIds: string[];
catalogGeneration?: string;
catalog?: IdlePluginCatalogEntry[];
}>;
// A failed turn may have successfully torn down its ephemeral allocation. A
@@ -290,7 +296,7 @@ export async function readIdleSleepSafety(
now: () => number = Date.now,
ownerId?: string,
inspectLocalWork: (diagnostic?: IdleSleepDiagnosticSink) => Promise<IdleLocalWork> =
diagnostic => readIdleLocalWork({ diagnostic }),
diagnostic => readIdleLocalWork({ diagnostic, deadlineCapability: options.deadlineCapability }),
inspectPlugins?: InspectIdlePlugins,
prepareBackup?: () => Promise<boolean>,
scheduledRoutines = false,
@@ -298,15 +304,21 @@ export async function readIdleSleepSafety(
options: IdleSleepSafetyOptions = {},
): Promise<IdleSleepSafety> {
let incomingProof: IdleIncomingWakeProof | undefined;
let deadlineProof: IdleDeadlineProof | undefined;
const deadlines = options.deadlineCapability?.deadlineCatalog === "domain_deadlines_v1";
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" && incomingProof ? 3 : backgroundWork === "none" && scheduledWake ? 2 : 1, backgroundWork,
if (backgroundWork === "none" && deadlines && (!deadlineProof || deadlineProof.blocked || !deadlineProof.generation)) backgroundWork = "unknown";
const wake = deadlineProof ? deadlineProof.scheduledWake : scheduledWake;
return { version: backgroundWork === "none" && (incomingProof || deadlineProof) ? 3 : backgroundWork === "none" && wake ? 2 : 1, backgroundWork,
...(backgroundWork === "none" && incomingProof ? { incomingWake: incomingProof.capability,
...(incomingProof.apiWakeAdmission ? { apiWakeAdmission: incomingProof.apiWakeAdmission } : {}) } : {}),
...(backgroundWork === "none" && scheduledWake ? { scheduledWake } : {}),
...(backgroundWork === "none" && deadlineProof?.generation ? { deadlineProof: {
catalog: "domain_deadlines_v1" as const, generation: deadlineProof.generation } } : {}),
...(backgroundWork === "none" && wake ? { scheduledWake: wake } : {}),
...(diagnostics ? { diagnostics } : {}) };
};
const before = getDrainStatus();
@@ -324,6 +336,8 @@ export async function readIdleSleepSafety(
diagnostic?.record("local.inspection", localFirst);
if (localFirst !== "none" && !diagnostic) return report(localFirst);
let quietPlugins: string[] = [];
let quietPluginGeneration: string | undefined;
let quietPluginCatalog: IdlePluginCatalogEntry[] | undefined;
let pluginState: IdleLocalWork = "none";
if (inspectPlugins) {
const currentLocal = diagnostic ? idleWorkSnapshot() : localBefore;
@@ -338,7 +352,11 @@ export async function readIdleSleepSafety(
pluginState = pluginWork.backgroundWork;
diagnostic?.record("plugins.runtime_drain", pluginState);
if (pluginState !== "none" && !diagnostic) return report(pluginState);
if (pluginState === "none") quietPlugins = pluginWork.pluginIds;
if (pluginState === "none") {
quietPlugins = pluginWork.pluginIds;
quietPluginGeneration = pluginWork.catalogGeneration;
quietPluginCatalog = pluginWork.catalog;
}
}
}
if (diagnostic && !sameQuietHold(before, getDrainStatus(), now())) {
@@ -354,10 +372,35 @@ export async function readIdleSleepSafety(
if (!ownerId || ownerId !== before.ownerId) return undefined;
incomingProof = await readIdleIncomingWakeProof(tx, { capabilities: options.incomingWake, ownerId, receiverOwner: db });
}
if (deadlines) {
if (!owned() || !scheduledRoutines || !scheduledIssueMonitors) return undefined;
const routine = await readIdleRoutineSchedule(tx, ownerId!, now());
const issue = await readIdleIssueSchedule(tx, ownerId!, now());
diagnostic?.record("routines.schedule", routine.inspectionComplete === false ? "unknown" : routine.blocked ? "present" : "none");
diagnostic?.record("issues.schedule", issue.inspectionComplete === false ? "unknown" : issue.blocked ? "present" : "none");
diagnostic?.expect(["deadlines.catalog"]);
failureReason = "deadlines.catalog";
if (routine.blocked || issue.blocked) deadlineProof = { blocked: true };
else {
scheduledWake = combineIdleSchedules(routine.scheduledWake, issue.scheduledWake, issue.generation);
deadlineProof = await readIdleDeadlineProof(tx, { ownerId: ownerId!, expiresAt: before.expiresAt!.getTime() }, {
now: now(), verifiedCapability: options.deadlineCapability,
quietPluginIds: quietPlugins, quietPluginGeneration, quietPluginCatalog,
scheduledWake, passiveGeneration: issue.generation,
});
}
diagnostic?.record("deadlines.catalog", deadlineProof.blocked ? "present" : "none");
if (deadlineProof.blocked && !diagnostic) return true;
}
failureReason = "durable.snapshot";
const checks = WORK_CHECKS.filter(check =>
(!scheduledRoutines || check.reason !== "routines")
&& (!scheduledIssueMonitors || check.reason !== "issues"))
.map(check => ({ ...check, query: incomingWakeWorkQuery(check.reason, incomingProof) ?? check.query }));
.map(check => {
const replacement = deadlineProof?.blocked === false ? deadlineWorkReplacement(check.reason) : undefined;
return { ...check, query: incomingWakeWorkQuery(check.reason, incomingProof,
deadlineProof?.githubContinuity?.connectionIds) ?? (replacement ? sql.raw(replacement) : check.query) };
});
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()}`));
@@ -382,8 +425,9 @@ export async function readIdleSleepSafety(
if (rows.length !== 1 || typeof rows[0]?.blocked !== "boolean") return undefined;
found = rows[0].blocked;
}
found ||= deadlineProof?.blocked ?? false;
if (found && !diagnostic) return true;
if (scheduledRoutines) {
if (scheduledRoutines && !deadlines) {
if (!owned()) return undefined;
const schedule = await readIdleRoutineSchedule(tx, ownerId!, now());
diagnostic?.record("routines.schedule", schedule.inspectionComplete === false ? "unknown" : schedule.blocked ? "present" : "none");
@@ -391,7 +435,7 @@ export async function readIdleSleepSafety(
found ||= schedule.blocked && schedule.inspectionComplete !== false;
scheduledWake = schedule.scheduledWake;
}
if (scheduledIssueMonitors) {
if (scheduledIssueMonitors && !deadlines) {
if (!scheduledRoutines || !owned()) return undefined;
const schedule = await readIdleIssueSchedule(tx, ownerId!, now());
diagnostic?.record("issues.schedule", schedule.inspectionComplete === false ? "unknown" : schedule.blocked ? "present" : "none");
@@ -438,13 +482,14 @@ export async function readIdleSleepSafety(
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()) {
const finalWake = deadlineProof ? deadlineProof.scheduledWake : scheduledWake;
if (local === "none" && finalWake && Date.parse(finalWake.nextWakeAt) <= now()) {
diagnostic?.record("inspection.hold", "unknown");
return report("present");
}
// Legacy alarm kinds cannot be silently combined with a new ingress proof.
// The negotiated deadline catalog supplies the composed v3 alarm separately.
if (incomingProof && scheduledWake) return report("unknown");
if (incomingProof && scheduledWake && !deadlineProof) return report("unknown");
return report(local, scheduledWake);
} catch {
// Missing migrations, unknown state, malformed configuration, timeouts and
@@ -0,0 +1,9 @@
import { createHash } from "node:crypto";
import type { PaperclipPluginManifestV1 } from "@paperclipai/shared";
export interface IdlePluginCatalogEntry { pluginId: string; manifestGeneration: string }
export function pluginIdleManifestGeneration(manifest: PaperclipPluginManifestV1): string {
const ordered = (value: unknown): unknown => Array.isArray(value) ? value.map(ordered)
: value !== null && typeof value === "object"
? Object.fromEntries(Object.entries(value).sort(([a], [b]) => a.localeCompare(b)).map(([key, v]) => [key, ordered(v)])) : value;
return createHash("sha256").update(JSON.stringify(ordered(manifest))).digest("hex");
}
@@ -0,0 +1,57 @@
import { pluginIdleManifestGeneration, type IdlePluginCatalogEntry } from "./plugin-idle-catalog.js";
import type { PluginJobDeclaration } from "@paperclipai/shared";
import { plugins, pluginJobs, type Db } from "@paperclipai/db";
import { eq, getTableColumns, sql } from "drizzle-orm";
import { validateCron } from "./cron.js";
import { idleDeadlineRowLimit, isFutureIdleDeadline, type IdleDeadlineSnapshot } from "./idle-deadline-types.js";
type Job = typeof pluginJobs.$inferSelect;
type Plugin = typeof plugins.$inferSelect;
export type IdlePluginJobRow = { job: Job; plugin: Plugin | null; unsettled: boolean };
export function projectIdlePluginJobSchedule(rows: IdlePluginJobRow[], quietPluginIds: readonly string[], now: number, catalog: readonly IdlePluginCatalogEntry[] = []): IdleDeadlineSnapshot {
const state: unknown[] = [];
const deadlines: number[] = [];
const blocked = () => ({ blocked: true, state, deadlines: [] });
if (rows.length > idleDeadlineRowLimit) return blocked();
const quiet = new Set(quietPluginIds);
const catalogByPlugin = new Map(catalog.map(entry => [entry.pluginId, entry.manifestGeneration]));
const manifests = new Map<string, string>();
const declarationsByPlugin = new Map<string, Map<string, PluginJobDeclaration>>();
if (catalogByPlugin.size !== catalog.length) return blocked();
for (const { job, plugin, unsettled } of rows) {
if (!["active", "paused"].includes(job.status) || unsettled !== false) return blocked();
if (job.status === "paused") { state.push([job, unsettled]); continue; }
if (!plugin || plugin.id !== job.pluginId || plugin.status !== "ready" || !quiet.has(plugin.id)) return blocked();
let manifestGeneration = manifests.get(plugin.id);
if (!manifestGeneration) {
manifestGeneration = pluginIdleManifestGeneration(plugin.manifestJson);
manifests.set(plugin.id, manifestGeneration);
}
if (plugin.manifestJson?.id !== plugin.pluginKey || plugin.manifestJson?.version !== plugin.version || plugin.manifestJson?.apiVersion !== plugin.apiVersion
|| catalogByPlugin.get(plugin.id) !== manifestGeneration) return blocked();
state.push([job, [plugin.id, plugin.pluginKey, plugin.version, plugin.apiVersion, plugin.status, plugin.updatedAt, manifestGeneration], unsettled]);
let declarations = declarationsByPlugin.get(plugin.id);
if (!declarations) {
const jobs = plugin.manifestJson?.jobs;
if (!Array.isArray(jobs) || jobs.some(d => !d || typeof d.jobKey !== "string")) return blocked();
declarations = new Map(jobs.map(d => [d.jobKey, d]));
if (declarations.size !== jobs.length) return blocked();
declarationsByPlugin.set(plugin.id, declarations);
}
const declaration = declarations.get(job.jobKey);
if (!declaration) return blocked();
if ((declaration.schedule ?? "") !== job.schedule) return blocked();
if (!job.schedule && job.nextRunAt === null) continue;
const at = job.nextRunAt?.getTime() ?? NaN;
if (validateCron(job.schedule) || !isFutureIdleDeadline(at, now)) return blocked();
deadlines.push(at);
}
return { blocked: false, state, deadlines };
}
export async function readIdlePluginJobSchedule(tx: Pick<Db, "select">, quietPluginIds: readonly string[], now: number, catalog: readonly IdlePluginCatalogEntry[] = []) {
const rows = await tx.select({ job: getTableColumns(pluginJobs), plugin: getTableColumns(plugins),
unsettled: sql<boolean>`EXISTS (SELECT 1 FROM plugin_job_runs r WHERE r.job_id = ${pluginJobs.id}
AND (r.plugin_id <> ${pluginJobs.pluginId} OR r.status NOT IN ('succeeded', 'failed', 'cancelled') OR r.finished_at IS NULL))`,
}).from(pluginJobs).leftJoin(plugins, eq(plugins.id, pluginJobs.pluginId)).orderBy(pluginJobs.id).limit(idleDeadlineRowLimit + 1);
return projectIdlePluginJobSchedule(rows, quietPluginIds, now, catalog);
}
+19 -2
View File
@@ -1,3 +1,5 @@
import { pluginIdleManifestGeneration, type IdlePluginCatalogEntry } from "./plugin-idle-catalog.js";
import { createHash, randomUUID } from "node:crypto";
/**
* PluginWorkerManager — spawns and manages out-of-process plugin worker child
* processes, routes JSON-RPC 2.0 calls over stdio, and handles lifecycle
@@ -19,7 +21,6 @@
*/
import { fork, type ChildProcess } from "node:child_process";
import { randomUUID } from "node:crypto";
import { beginIdleTrackedWork, readTaskDrain, trackIdleWork } from "./task-admission.js";
import { EventEmitter } from "node:events";
import { createInterface, type Interface as ReadlineInterface } from "node:readline";
@@ -814,6 +815,8 @@ export interface PluginWorkerManager {
onConfigApplied(listener: () => void): () => void;
inspectIdleSleep?(hold: { ownerId: string; expiresAt: number }): Promise<{
backgroundWork: "none" | "present" | "unknown"; pluginIds: string[];
catalogGeneration?: string;
catalog?: IdlePluginCatalogEntry[];
}>;
releaseIdleSleep?(): void;
/**
@@ -3960,6 +3963,10 @@ export function createPluginWorkerManager(
managerOptions?.maxConcurrentDuplexRoutes,
);
const idleWorkerIdentities = new WeakMap<PluginWorkerHandle, string>();
const idleWorkerManifests = new WeakMap<PluginWorkerHandle, string>();
const idleManagerGeneration = randomUUID();
let nextIdleWorkerIdentity = 0;
return {
onConfigApplied(listener) {
configApplied.add(listener);
@@ -3974,7 +3981,16 @@ export function createPluginWorkerManager(
return { backgroundWork: "unknown", pluginIds: [] };
}
const backgroundWork = results.includes("present") ? "present" : results.includes("unknown") ? "unknown" : "none";
return { backgroundWork, pluginIds: backgroundWork === "none" ? snapshot.map(([id]) => id) : [] };
const catalogGeneration = createHash("sha256").update(JSON.stringify([idleManagerGeneration, snapshot.map(([id, worker]) => {
let identity = idleWorkerIdentities.get(worker);
if (!identity) { identity = String(++nextIdleWorkerIdentity); idleWorkerIdentities.set(worker, identity); }
return [id, identity, idleWorkerManifests.get(worker)];
})])).digest("hex");
const catalog = snapshot.flatMap(([pluginId, worker]) => {
const manifestGeneration = idleWorkerManifests.get(worker);
return manifestGeneration ? [{ pluginId, manifestGeneration }] : [];
});
return { backgroundWork, pluginIds: backgroundWork === "none" ? snapshot.map(([id]) => id) : [], catalogGeneration, catalog };
},
releaseIdleSleep() {
for (const worker of workers.values()) worker.releaseIdleSleep?.();
@@ -4003,6 +4019,7 @@ export function createPluginWorkerManager(
duplexRouteSlots,
...options,
}, () => pruneStartupWorker(handle));
idleWorkerManifests.set(handle, pluginIdleManifestGeneration(options.manifest));
if (existing) retainStartupWorker(existing);
workers.set(pluginId, handle);
@@ -0,0 +1,38 @@
import { providerTraceRecords, type Db } from "@paperclipai/db";
import { getTableColumns, sql } from "drizzle-orm";
import { importingProviderTraceRef, objectProviderTraceRef } from "./provider-trace-object-store.js";
import { finishedIdleRun } from "./idle-work-proof.js";
import { idleDeadlineRowLimit, isFutureIdleDeadline, type IdleDeadlineSnapshot } from "./idle-deadline-types.js";
const expiredTraceRef = "00000000-0000-0000-0000-000000000000.ndjson";
export type IdleProviderTraceRow = { trace: typeof providerTraceRecords.$inferSelect; finishedRun: boolean };
export function projectIdleProviderTraceSchedule(rows: IdleProviderTraceRow[], now: number): IdleDeadlineSnapshot {
const state = rows;
const deadlines: number[] = [];
const blocked = () => ({ blocked: true, state, deadlines: [] });
if (rows.length > idleDeadlineRowLimit) return blocked();
for (const { trace, finishedRun } of rows) {
if (finishedRun !== true) return blocked();
if (trace.traceRef === expiredTraceRef) {
if (!["expired", "deleted"].includes(trace.status) || !trace.deletedAt) return blocked();
continue;
}
const objectRef = objectProviderTraceRef(trace.traceRef);
const localRef = /^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}\.ndjson$/.test(trace.traceRef);
if (localRef && trace.status === "deleted" && trace.deletedAt) continue;
if (importingProviderTraceRef(trace.traceRef) || (objectRef && (objectRef[1] !== trace.companyId || objectRef[2] !== trace.runId))
|| (!objectRef && !localRef)
|| trace.deletedAt || !["complete", "incomplete", "truncated"].includes(trace.status)) return blocked();
const at = trace.expiresAt.getTime();
if (!isFutureIdleDeadline(at, now)) return blocked();
deadlines.push(at);
}
return { blocked: false, state, deadlines };
}
export async function readIdleProviderTraceSchedule(tx: Pick<Db, "select">, now: number) {
const rows = await tx.select({ trace: getTableColumns(providerTraceRecords),
finishedRun: sql<boolean>`EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id = ${providerTraceRecords.runId}
AND r.company_id = ${providerTraceRecords.companyId} AND ${sql.raw(finishedIdleRun("r"))})`,
}).from(providerTraceRecords).orderBy(providerTraceRecords.id).limit(idleDeadlineRowLimit + 1);
return projectIdleProviderTraceSchedule(rows, now);
}
@@ -13,9 +13,12 @@ export function registerProviderTraceRetentionWork(
const traces = providerTraceStore(db);
let deadline: number | null | undefined;
const unregister = registerIdleLocalWorkProbe(() =>
deadline === undefined ? "unknown" : deadline === null ? "none" : "present",
);
const unregister = registerIdleLocalWorkProbe((options = {}) => {
if (deadline === undefined) return "unknown";
if (deadline === null) return "none";
return options.deadlineCapability?.deadlineCatalog === "domain_deadlines_v1" && deadline > Date.now()
? "none" : "present";
});
const worker = registerDeadlineWork(coordinator, DELIVERY_QUEUES.providerTraceRetention, {
canRun,
retryMs: 30_000,
@@ -1,14 +1,15 @@
import { readConnectionGrantCredentialStates } from "../connection-credentials.js";
import { randomUUID } from "node:crypto";
import { notifyGitHubCredentialChange } from "../github-connection-notifications.js";
import type { ActorInfo } from "./types.js";
import { loadGitHubGrantMetadata } from "./github-access-discovery.js";
import { measureRunPhase } from "../execution-preparation-timing.js";
import { measureRunTransaction } from "../execution-transaction-timing.js";
import { and, eq, sql } from "drizzle-orm";
import { and, eq, sql, or, isNull, ne } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import { companies, connectionGrants, toolConnections } from "@paperclipai/db";
import { conflict, HttpError, notFound, unprocessable } from "../../errors.js";
import { PaperclipCloudConnectorError, type PaperclipCloudConnector } from "../paperclip-cloud-connector.js";
import { isPaperclipCloudConnectorStrategy, PaperclipCloudConnectorError, type PaperclipCloudConnector } from "../paperclip-cloud-connector.js";
import type { createOAuthGrantRefresh } from "./oauth/grant-refresh.js";
import type { OAuthCredentialStore } from "./oauth/credential-store.js";
@@ -395,3 +396,63 @@ export function githubContinuityRevision(grant: Pick<typeof connectionGrants.$in
return JSON.stringify([grant.providerTenant?.oauth?.accessTokenExpiresAt, grant.providerTenant?.oauth?.refreshedAt,
grant.providerTenant?.github?.lastAccessRefreshAt, grant.providerTenant?.github?.bindingSync?.id]);
}
export async function readIdleGitHubGrantContinuity(tx: Pick<Db, "select">, now: number,
checkBudget: () => void = () => {}) {
const rows = await tx.select({ grant: connectionGrants, connection: toolConnections, companyStatus: companies.status })
.from(connectionGrants).leftJoin(toolConnections, eq(toolConnections.id, connectionGrants.connectionId))
.leftJoin(companies, eq(companies.id, connectionGrants.companyId))
.where(and(eq(connectionGrants.status, "active"), sql`${connectionGrants.providerTenant}->'github' IS NOT NULL`,
or(isNull(toolConnections.id), ne(toolConnections.companyId, connectionGrants.companyId),
sql`${toolConnections.config}->>'sourceTemplateKey' = 'github'`)))
.orderBy(connectionGrants.id).limit(10_001);
const proof = projectIdleGitHubGrantContinuity(rows, now, checkBudget);
if (proof.blocked) return proof;
const eligible = new Set(proof.connectionIds);
const inputs = rows.flatMap(row => row.connection && eligible.has(row.connection.id)
&& managedGitHubContinuityConnection(row.connection) === "managed" ? [{ connection: row.connection, grant: row.grant }] : []);
const credentials = await readConnectionGrantCredentialStates(tx, inputs, checkBudget);
if (!credentials) return { ...proof, blocked: true, connectionIds: [], deadlines: [] };
return { ...proof, state: [rows, [...credentials]] };
}
function managedGitHubContinuityConnection(connection: typeof toolConnections.$inferSelect): "managed" | "unrelated" | "invalid" {
const config = asRecord(connection.config), oauth = asRecord(config.oauth);
if (config.sourceTemplateKey !== "github" || connection.authKind !== "oauth"
|| connection.credentialSource === "vercel_connect") return "unrelated";
if (typeof oauth.connectorProfile !== "string" || !oauth.connectorProfile) return "invalid";
if (oauth.connectorProfile !== "github.code") return "unrelated";
return isPaperclipCloudConnectorStrategy(oauth.strategy) ? "managed" : "invalid";
}
export function projectIdleGitHubGrantContinuity(rows: {
grant: typeof connectionGrants.$inferSelect;
connection: typeof toolConnections.$inferSelect | null;
companyStatus: string | null;
}[], now: number, checkBudget: () => void = () => {}) {
const deadlines: number[] = [];
const connectionIds = new Set<string>();
const blocked = () => ({ blocked: true, state: rows, deadlines: [], connectionIds: [] as string[] });
if (rows.length > 10_000) return blocked();
for (const { grant, connection, companyStatus } of rows) {
checkBudget();
if (!connection || connection.companyId !== grant.companyId || !companyStatus
|| !["active", "paused", "archived"].includes(companyStatus)
|| !["draft", "active", "disabled", "archived"].includes(connection.status)
|| !["oauth", "api_key", "none"].includes(connection.authKind)
|| !["paperclip_vault", "vercel_connect"].includes(connection.credentialSource)) return blocked();
if (companyStatus !== "active" || connection.status !== "active" || !connection.enabled) continue;
const ownership = managedGitHubContinuityConnection(connection);
if (ownership === "invalid") return blocked();
if (ownership === "unrelated") continue;
if (grant.connectionId !== connection.id || (grant.kind === "user" ? !grant.subjectUserId
: grant.kind === "agent" ? !grant.subjectAgentId : true)) return blocked();
const oauth = asRecord(grant.providerTenant?.oauth);
const github = grant.providerTenant?.github;
if (!github || github.bindingSync || github.continuityRetry || oauth.refreshLease) return blocked();
const deadline = githubGrantContinuityDeadline(connection, grant);
if (deadline === null || !Number.isFinite(deadline) || deadline <= now) return blocked();
if (!grant.credentialSecretRefs.some(ref => ref.configPath === "oauth.access_token")) return blocked();
deadlines.push(deadline);
connectionIds.add(connection.id);
}
return { blocked: false, state: rows, deadlines, connectionIds: [...connectionIds].sort() };
}