From f187de99aa7f4b0af1df46a2e6b2d79aae618d4f Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 3 Nov 2026 15:40:56 -0600 Subject: [PATCH] Preserve domain deadlines across owned idle sleep (#18656) Reviewed and verified at 9b4acc456daa44f063c61879df8c6f62a25711ec. Co-Authored-By: Paperclip --- doc/idle-domain-deadlines.md | 79 ++++ maps/agent-access-map/instance-operations.md | 12 +- maps/agent-access-map/routines.md | 3 + .../__tests__/agent-timer-deadlines.test.ts | 2 +- .../fixtures/plugin-worker-deadline.cjs | 13 + .../idle-domain-deadlines-postgres.test.ts | 408 ++++++++++++++++++ .../instance-settings-routes.test.ts | 15 +- .../src/__tests__/plugin-idle-manager.test.ts | 28 +- .../provider-trace-retention-work.test.ts | 12 +- .../src/__tests__/tool-access-service.test.ts | 5 +- server/src/routes/instance-settings.ts | 12 +- server/src/routes/openapi.ts | 3 +- .../connection-credential-state.test.ts | 67 +++ server/src/services/connection-credentials.ts | 116 ++++- .../services/heartbeat/idle-timer-schedule.ts | 68 +++ server/src/services/heartbeat/queue.test.ts | 2 +- server/src/services/heartbeat/queue.ts | 252 ++--------- server/src/services/heartbeat/timer-policy.ts | 252 +++++++++++ .../src/services/idle-deadline-proof.test.ts | 218 ++++++++++ server/src/services/idle-deadline-proof.ts | 98 +++++ server/src/services/idle-deadline-types.ts | 19 + server/src/services/idle-local-work.ts | 10 +- server/src/services/idle-sleep-safety.ts | 65 ++- server/src/services/plugin-idle-catalog.ts | 9 + .../src/services/plugin-job-idle-schedule.ts | 57 +++ server/src/services/plugin-worker-manager.ts | 21 +- .../services/provider-trace-idle-schedule.ts | 38 ++ .../services/provider-trace-retention-work.ts | 9 +- .../tool-access/github-grant-continuity.ts | 65 ++- 29 files changed, 1692 insertions(+), 266 deletions(-) create mode 100644 doc/idle-domain-deadlines.md create mode 100644 server/src/__tests__/fixtures/plugin-worker-deadline.cjs create mode 100644 server/src/__tests__/idle-domain-deadlines-postgres.test.ts create mode 100644 server/src/services/connection-credential-state.test.ts create mode 100644 server/src/services/heartbeat/idle-timer-schedule.ts create mode 100644 server/src/services/heartbeat/timer-policy.ts create mode 100644 server/src/services/idle-deadline-proof.test.ts create mode 100644 server/src/services/idle-deadline-proof.ts create mode 100644 server/src/services/idle-deadline-types.ts create mode 100644 server/src/services/plugin-idle-catalog.ts create mode 100644 server/src/services/plugin-job-idle-schedule.ts create mode 100644 server/src/services/provider-trace-idle-schedule.ts diff --git a/doc/idle-domain-deadlines.md b/doc/idle-domain-deadlines.md new file mode 100644 index 00000000000..1ec36c14301 --- /dev/null +++ b/doc/idle-domain-deadlines.md @@ -0,0 +1,79 @@ +# Negotiated domain deadlines + +The `domain_deadlines_v1` catalog extends the existing owned idle-sleep handoff. +Cloud requests `scheduledWake=domain-deadlines-v1` and signs the fixed catalog in +`idleCapabilities.deadlineCatalog` on its stack-bound `task-drain:read` assertion. +A query parameter or an ordinary credential cannot grant this capability. +The response uses idle-safety version 3 with a scheduled proof of version 2, +kind `domain_deadlines`, catalog `domain_deadlines_v1`, opaque generation and +canonical UTC `nextWakeAt`. Every requested catalog also returns an outer +`deadlineProof: { catalog: "domain_deadlines_v1", generation }`, including when +there is no future alarm. Any scheduled proof must use that same generation. +Cloud rejects an omitted receipt, an older ingress-only version 3 response, and +a changed no-alarm generation on the final read. Legacy requests and versions +keep their prior rules. + +The proof combines existing routine and issue-monitor obligations with: + +- Eligible agent interval timers, using the same actionable-work, organization, + daily-run and daily-cost policy as ordinary heartbeat dispatch. Inspection + does not claim a timer or advance `lastHeartbeatAt`. Restore and worktree + scheduling overrides remain conservative blockers. +- Plugin jobs with a valid future `nextRunAt`, matching current declaration and + an exact worker-owned quiet catalog. An explicitly manual-only declaration + adds no timer. Missing worker/deadline, unfamiliar status, unfinished run or + settlement, and catalog changes refuse sleep. +- Retained finalized provider traces at their original 24-hour expiry. Active + capture/import, expired data and unfinished object deletion remain work. + Capture expiry and deletion retain their ordinary owner; sleep never extends + retention or changes reveal authorization. +- Personal GitHub grant continuity, using the owner's token-expiry-minus-one-hour + and 30-day access-refresh rules. Broker enrollment alone is insufficient: + exact ingress binding and actual grant deadline must both be proven. Missing + credentials, inactive/missing owner-scoped vault definitions or current token + versions, refresh leases, binding synchronization and unresolved retry records + remain blockers. Inspection reads credential ownership/version metadata and + never resolves plaintext token material. + +Pending watchdog first-run grace (15 seconds) and automatic merge-confirmation +checks (five minutes) remain awake. They provide no useful safe sleep window +under the unchanged Cloud minimum of five minutes. Their cadence is preserved; +there is no merge observation migration or invented future timestamp. Grok +bindings and unfinished Grok routine recovery remain outside this catalog. +`grok-bot/routine.ts` owns `recoveryRequiredAt`, stale `sending` dispatches +(after 30 seconds), and accepted dispatch `expiresAt` (created with a ten-minute +lifetime). These are authorization/recovery obligations, not ordinary recurring +routine timestamps, and their bindings remain blocked. No alarm exemption is +inferred from the shared table name or a revoked UI connection. + +Each inspection reads one repeatable-read snapshot under the owned admission +hold. Its generation binds the process, owner and expiry, passive state, +ordinary routine/monitor proof, exact domain rows and plugin worker identities. +Any changed state, cancellation, moved deadline, owner replacement or restart +invalidates the old proof. Unknown or malformed states fail closed. + +The timer owner batches UTC-day run counts, exact decimal cost sums and actionable-work inputs. Credential ownership and selected current/previous version metadata use bounded chunks; they never read credential material or value hashes. Managed GitHub OAuth user and agent grants retain their owner deadlines. Other grant shapes grant no continuity exemption. + +The additional domain projection is bounded to 64 data reads, a two-second monotonic budget for +starting further reads and 10,000 rows per family, while reserving 60 seconds of hold headroom. Overflow refuses sleep; +it never truncates the proof or changes a schedule. These limits supplement the +transaction's statement timeout. A query already in progress may still take its five-second statement timeout; +SQL execution can therefore finish about seven seconds after the projection +starts, plus bounded local projection/driver overhead. No subsequent data query +starts after budget exhaustion. This is not a strict two-second wall-clock +promise. The final hold and generation checks remain mandatory. + +Cloud persists one existing registry alarm under its current sleep lease before +final proof validation and provider stop. It compares every proof field on the +final read and checks the time window again. The worker wakes 120 seconds before +the actual obligation, retries failed/deferred wakes after 30 seconds and keeps +its existing expiring claims and replacement fences. No tenant timer definitions +or new per-timer tables are stored in Cloud. + +After wake, ordinary domain services retain authorization, budgets, concurrency, +atomic claims, cancellation and missed-work policy. Cloud wakes the stack; it +does not dispatch a business effect. Disable new registrations for rollback +while continuing to serve existing alarms, or wake their stacks before removing +the alarm service. Host support must precede enabling app exemptions. Local +unit/PostgreSQL evidence establishes the handoff logic, not deployed host or +Neon suspension, measured startup margins or fleet savings. diff --git a/maps/agent-access-map/instance-operations.md b/maps/agent-access-map/instance-operations.md index cebe15a8df6..2c547c620b1 100644 --- a/maps/agent-access-map/instance-operations.md +++ b/maps/agent-access-map/instance-operations.md @@ -173,9 +173,17 @@ API: `present` — representative operation IDs: query parameters and tenant credentials cannot grant it. Its version-3 proof permits only saved API keys and exactly bound brokered GitHub bot configuration. Agent and board API requests authenticate under current tenant authority after - bounded host wake. Personal GitHub continuity remains blocked until the stacked - deadline proof is present. See [protocol scope](../../doc/idle-incoming-wake.md) + bounded host wake. Personal GitHub also requires the signed domain deadline + receipt and its actual stored-credential continuity proof. See [protocol scope](../../doc/idle-incoming-wake.md) and [local auth/PostgreSQL evidence](../../server/src/__tests__/idle-incoming-wake.test.ts). +- `GET /api/instance/task-drain?idleSleepSafety=1&scheduledWake=domain-deadlines-v1` + additionally requires the signed fixed deadline catalog. The version-3 + `deadlineProof` receipt is mandatory even without a future alarm. It binds + actual timer, plugin, trace expiry and credential continuity state to the + owned hold. Unknown/due/recovery work and pending watchdog/merge polling + remain blocked. It grants no business execution authority. See + [domain protocol](../../doc/idle-domain-deadlines.md); local source and + fixture checks do not establish deployed sleep or external wake. - `GET /api/ready` and `HEAD /api/ready` — public deployment readiness only; 200 after recovery, 503 while starting, recovering, or draining. No SQL, session resolution, company identity, or control authority. diff --git a/maps/agent-access-map/routines.md b/maps/agent-access-map/routines.md index 2f2ff5fbca0..83a34dd306d 100644 --- a/maps/agent-access-map/routines.md +++ b/maps/agent-access-map/routines.md @@ -81,6 +81,9 @@ idle hold. It requires signed Cloud control authority; company agents and tenant admin tokens cannot use it. It does not grant routine management or execution authority. See the [hosting protocol](../../doc/idle-sleep-safety.md#scheduled-routine-wake-protocol) and [route authorization tests](../../server/src/__tests__/instance-settings-routes.test.ts). +The signed `scheduledWake=domain-deadlines-v1` catalog can combine this routine +obligation with other supported domain deadlines. Its generation receipt is +mandatory even when no future alarm remains. See [domain scope](../../doc/idle-domain-deadlines.md). These are local contract checks, not a live provider sleep/wake qualification. ## `run-now` diff --git a/server/src/__tests__/agent-timer-deadlines.test.ts b/server/src/__tests__/agent-timer-deadlines.test.ts index f925fb8dfd7..d6cc9649ea6 100644 --- a/server/src/__tests__/agent-timer-deadlines.test.ts +++ b/server/src/__tests__/agent-timer-deadlines.test.ts @@ -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 () => { diff --git a/server/src/__tests__/fixtures/plugin-worker-deadline.cjs b/server/src/__tests__/fixtures/plugin-worker-deadline.cjs new file mode 100644 index 00000000000..a60a6f735a8 --- /dev/null +++ b/server/src/__tests__/fixtures/plugin-worker-deadline.cjs @@ -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"; }, + }) }); +}); diff --git a/server/src/__tests__/idle-domain-deadlines-postgres.test.ts b/server/src/__tests__/idle-domain-deadlines-postgres.test.ts new file mode 100644 index 00000000000..35141db2ea4 --- /dev/null +++ b/server/src/__tests__/idle-domain-deadlines-postgres.test.ts @@ -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> | 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`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`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>> | 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; + 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(); + const connector = { + refresh: async (input: Parameters[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[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); + +}); diff --git a/server/src/__tests__/instance-settings-routes.test.ts b/server/src/__tests__/instance-settings-routes.test.ts index eb9fb0cb38e..e34dd5cc08b 100644 --- a/server/src/__tests__/instance-settings-routes.test.ts +++ b/server/src/__tests__/instance-settings-routes.test.ts @@ -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" })], diff --git a/server/src/__tests__/plugin-idle-manager.test.ts b/server/src/__tests__/plugin-idle-manager.test.ts index 1ea93bc172b..ef0bb175d24 100644 --- a/server/src/__tests__/plugin-idle-manager.test.ts +++ b/server/src/__tests__/plugin-idle-manager.test.ts @@ -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(); diff --git a/server/src/__tests__/provider-trace-retention-work.test.ts b/server/src/__tests__/provider-trace-retention-work.test.ts index 2c2d2fcc3ac..08caf28d8b1 100644 --- a/server/src/__tests__/provider-trace-retention-work.test.ts +++ b/server/src/__tests__/provider-trace-retention-work.test.ts @@ -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>(), - 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); diff --git a/server/src/__tests__/tool-access-service.test.ts b/server/src/__tests__/tool-access-service.test.ts index e728d320924..2c624213a23 100644 --- a/server/src/__tests__/tool-access-service.test.ts +++ b/server/src/__tests__/tool-access-service.test.ts @@ -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(); }); diff --git a/server/src/routes/instance-settings.ts b/server/src/routes/instance-settings.ts index e5d9c9f648e..75d918bb358 100644 --- a/server/src/routes/instance-settings.ts +++ b/server/src/routes/instance-settings.ts @@ -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; diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index af22909d9b3..86d55ce1d63 100644 --- a/server/src/routes/openapi.ts +++ b/server/src/routes/openapi.ts @@ -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 }, }); diff --git a/server/src/services/connection-credential-state.test.ts b/server/src/services/connection-credential-state.test.ts new file mode 100644 index 00000000000..d0bbcf7dc3a --- /dev/null +++ b/server/src/services/connection-credential-state.test.ts @@ -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) => { + 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; +} +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"); + }); +}); diff --git a/server/src/services/connection-credentials.ts b/server/src/services/connection-credentials.ts index ed87874a85a..4af4f633883 100644 --- a/server/src/services/connection-credentials.ts +++ b/server/src/services/connection-credentials.ts @@ -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, - grant: Pick, - 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; +type CredentialGrantOwner = Pick; +type CredentialSecretOwner = Pick; +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, + 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; +export type ConnectionGrantCredentialInput = { connection: CredentialConnection; grant: CredentialGrant }; +export async function readConnectionGrantCredentialStates(db: Pick, + inputs: ConnectionGrantCredentialInput[], checkBudget: () => void = () => {}): Promise | undefined> { + if (inputs.length > 10_000) return; + const states = new Map(); + 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(); + 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(); + 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, + connection: CredentialConnection, grant: CredentialGrant): Promise { + return (await readConnectionGrantCredentialStates(db, [{ connection, grant }]))?.get(grant.id); +} diff --git a/server/src/services/heartbeat/idle-timer-schedule.ts b/server/src/services/heartbeat/idle-timer-schedule.ts new file mode 100644 index 00000000000..e72f9b3c3c5 --- /dev/null +++ b/server/src/services/heartbeat/idle-timer-schedule.ts @@ -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, now: number, + env: Record = process.env, checkBudget: () => void = () => {}): Promise { + 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>(); + for (const { agent } of rows) { + const group = byCompany.get(agent.companyId) ?? new Map(); + 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(); + 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 }; +} diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index 65ff52365cb..ef92bf94c65 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -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(); diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index d11eee87ec9..b4a942b3506 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -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 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 | 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, - options: { - checkRunCap?: boolean; - checkCostCap?: boolean; - excludeRunId?: string | null; - now?: Date; - } = {}, - client: Pick = 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`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`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 { - 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) { diff --git a/server/src/services/heartbeat/timer-policy.ts b/server/src/services/heartbeat/timer-policy.ts new file mode 100644 index 00000000000..51b785011ce --- /dev/null +++ b/server/src/services/heartbeat/timer-policy.ts @@ -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 => 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; +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) { + async function getHeartbeatDailyCapBlock(agent: TimerAgent, policy: TimerPolicy, + options: { checkRunCap?: boolean; checkCostCap?: boolean; excludeRunId?: string | null; now?: Date } = {}, + client: Pick = db): Promise { + const now = options.now ?? new Date(); + if (options.checkRunCap !== false && policy.maxDailyRuns !== null) { + const [row] = await client.select({ total: sql`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`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(), costs = new Map(), actionable = new Set(); + const state: unknown[] = []; + if (runIds.length) { + checkBudget(); + const rows = await db.select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + total: sql`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`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>(); + 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 { + return (await readTimerHeartbeatSchedule(agent, now)).deadline; + } + return { parseHeartbeatPolicy, getHeartbeatDailyCapBlock, hasActionableTimerWork, nextTimerHeartbeatAt, + readTimerHeartbeatSchedule, readTimerHeartbeatSchedules }; +} diff --git a/server/src/services/idle-deadline-proof.test.ts b/server/src/services/idle-deadline-proof.test.ts new file mode 100644 index 00000000000..4f3a7bef96c --- /dev/null +++ b/server/src/services/idle-deadline-proof.test.ts @@ -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[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, onRead: () => void = () => {}) { + return { select: () => { + onRead(); + let table = ""; + const query = { from: (source: Parameters[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; +} +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 = { 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); +}); diff --git a/server/src/services/idle-deadline-proof.ts b/server/src/services/idle-deadline-proof.ts new file mode 100644 index 00000000000..089502cca7f --- /dev/null +++ b/server/src/services/idle-deadline-proof.ts @@ -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, hold: { ownerId: string; expiresAt: number }, + options: IdleDeadlineProofOptions): Promise { + 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; + 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; +} diff --git a/server/src/services/idle-deadline-types.ts b/server/src/services/idle-deadline-types.ts new file mode 100644 index 00000000000..4749d2fe33e --- /dev/null +++ b/server/src/services/idle-deadline-types.ts @@ -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; diff --git a/server/src/services/idle-local-work.ts b/server/src/services/idle-local-work.ts index 70b8cd6a921..d24b7dc61cf 100644 --- a/server/src/services/idle-local-work.ts +++ b/server/src/services/idle-local-work.ts @@ -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(); -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 { +export async function readIdleLocalWork(options: IdleLocalWorkOptions = {}): Promise { 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"; } diff --git a/server/src/services/idle-sleep-safety.ts b/server/src/services/idle-sleep-safety.ts index 265fb61a6fd..d52ca0dbb20 100644 --- a/server/src/services/idle-sleep-safety.ts +++ b/server/src/services/idle-sleep-safety.ts @@ -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 = - diagnostic => readIdleLocalWork({ diagnostic }), + diagnostic => readIdleLocalWork({ diagnostic, deadlineCapability: options.deadlineCapability }), inspectPlugins?: InspectIdlePlugins, prepareBackup?: () => Promise, scheduledRoutines = false, @@ -298,15 +304,21 @@ export async function readIdleSleepSafety( options: IdleSleepSafetyOptions = {}, ): Promise { 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 diff --git a/server/src/services/plugin-idle-catalog.ts b/server/src/services/plugin-idle-catalog.ts new file mode 100644 index 00000000000..0d7949d1178 --- /dev/null +++ b/server/src/services/plugin-idle-catalog.ts @@ -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"); +} diff --git a/server/src/services/plugin-job-idle-schedule.ts b/server/src/services/plugin-job-idle-schedule.ts new file mode 100644 index 00000000000..e840119bca5 --- /dev/null +++ b/server/src/services/plugin-job-idle-schedule.ts @@ -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(); + const declarationsByPlugin = new Map>(); + 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, quietPluginIds: readonly string[], now: number, catalog: readonly IdlePluginCatalogEntry[] = []) { + const rows = await tx.select({ job: getTableColumns(pluginJobs), plugin: getTableColumns(plugins), + unsettled: sql`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); +} diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index 3a25a8dd28d..121f731460e 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -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(); + const idleWorkerManifests = new WeakMap(); + 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); diff --git a/server/src/services/provider-trace-idle-schedule.ts b/server/src/services/provider-trace-idle-schedule.ts new file mode 100644 index 00000000000..c409b02b515 --- /dev/null +++ b/server/src/services/provider-trace-idle-schedule.ts @@ -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, now: number) { + const rows = await tx.select({ trace: getTableColumns(providerTraceRecords), + finishedRun: sql`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); +} diff --git a/server/src/services/provider-trace-retention-work.ts b/server/src/services/provider-trace-retention-work.ts index 7aa20d65f5f..c67017b8d11 100644 --- a/server/src/services/provider-trace-retention-work.ts +++ b/server/src/services/provider-trace-retention-work.ts @@ -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, diff --git a/server/src/services/tool-access/github-grant-continuity.ts b/server/src/services/tool-access/github-grant-continuity.ts index ec5842e6aa9..4c74db0472c 100644 --- a/server/src/services/tool-access/github-grant-continuity.ts +++ b/server/src/services/tool-access/github-grant-continuity.ts @@ -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, 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(); + 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() }; +}