From 390627b46eb333309d357004384b220ecf8a65af Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 7 Jul 2026 09:27:46 -0500 Subject: [PATCH] [codex] Suppress worktree heartbeat scheduling (#9163) Suppress heartbeat scheduling in worktree and restore runtimes while keeping routine ticks and setup cleanup active. Co-Authored-By: Paperclip --- .../heartbeat-scheduling-suppression.test.ts | 29 ++ .../heartbeat-worktree-suppression.test.ts | 255 ++++++++++++++++ .../server-startup-feedback-export.test.ts | 117 ++++++-- server/src/index.ts | 275 +++++++++--------- server/src/services/heartbeat.ts | 44 +++ server/src/services/index.ts | 2 +- 6 files changed, 573 insertions(+), 149 deletions(-) create mode 100644 server/src/__tests__/heartbeat-scheduling-suppression.test.ts create mode 100644 server/src/__tests__/heartbeat-worktree-suppression.test.ts diff --git a/server/src/__tests__/heartbeat-scheduling-suppression.test.ts b/server/src/__tests__/heartbeat-scheduling-suppression.test.ts new file mode 100644 index 0000000000..4fb35931fb --- /dev/null +++ b/server/src/__tests__/heartbeat-scheduling-suppression.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from "vitest"; +import { resolveHeartbeatSchedulingSuppression } from "../services/heartbeat.ts"; + +describe("heartbeat scheduling suppression", () => { + it("suppresses heartbeat scheduling for worktree runtimes", () => { + expect(resolveHeartbeatSchedulingSuppression({ + PAPERCLIP_IN_WORKTREE: "true", + })).toEqual({ + suppressed: true, + reason: "worktree_instance", + }); + }); + + it("suppresses heartbeat scheduling while database restore is in progress", () => { + expect(resolveHeartbeatSchedulingSuppression({ + PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS: "1", + })).toEqual({ + suppressed: true, + reason: "database_restore_in_progress", + }); + }); + + it("leaves normal live-plane runtimes unsuppressed", () => { + expect(resolveHeartbeatSchedulingSuppression({})).toEqual({ + suppressed: false, + reason: null, + }); + }); +}); diff --git a/server/src/__tests__/heartbeat-worktree-suppression.test.ts b/server/src/__tests__/heartbeat-worktree-suppression.test.ts new file mode 100644 index 0000000000..a85abdf2f8 --- /dev/null +++ b/server/src/__tests__/heartbeat-worktree-suppression.test.ts @@ -0,0 +1,255 @@ +import { randomUUID } from "node:crypto"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { eq, sql } from "drizzle-orm"; +import { + activityLog, + agents, + agentWakeupRequests, + agentRuntimeState, + companySkills, + companies, + createDb, + documentRevisions, + documents, + heartbeatRunEvents, + heartbeatRuns, + issueComments, + issueDocuments, + issues, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { heartbeatService, resolveHeartbeatSchedulingSuppression } from "../services/heartbeat.ts"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres heartbeat worktree suppression tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describeEmbeddedPostgres("heartbeat worktree suppression", () => { + let db!: ReturnType; + let tempDb: Awaited> | null = null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("heartbeat-worktree-suppression-"); + db = createDb(tempDb.connectionString); + }, 20_000); + + afterEach(async () => { + await db.delete(heartbeatRunEvents); + await db.delete(activityLog); + await db.delete(issueComments); + await db.delete(issueDocuments); + await db.delete(documentRevisions); + await db.delete(documents); + await db.delete(heartbeatRuns); + await db.delete(agentWakeupRequests); + await db.delete(issues); + await db.delete(agentRuntimeState); + await db.delete(companySkills); + await db.delete(agents); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function insertAgentAndIssue() { + const companyId = randomUUID(); + const agentId = randomUUID(); + const issueId = randomUUID(); + + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + status: "active", + issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`, + requireBoardApprovalForNewAgents: false, + defaultResponsibleUserId: "responsible-user", + }); + + await db.insert(agents).values({ + id: agentId, + companyId, + name: "Worktree Agent", + role: "engineer", + status: "idle", + adapterType: "process", + adapterConfig: { + command: process.execPath, + args: ["-e", "process.exit(0)"], + }, + runtimeConfig: { + heartbeat: { + enabled: true, + intervalSec: 60, + wakeOnDemand: true, + }, + }, + permissions: {}, + }); + + await db.insert(issues).values({ + id: issueId, + companyId, + title: "Assigned work", + status: "todo", + priority: "high", + assigneeAgentId: agentId, + responsibleUserId: "responsible-user", + }); + + return { companyId, agentId, issueId }; + } + + async function waitForTerminalRun(runId: string) { + for (let attempt = 0; attempt < 20; attempt += 1) { + const run = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + if (run && run.status !== "queued" && run.status !== "running") return run.status; + await new Promise((resolve) => setTimeout(resolve, 25)); + } + return null; + } + + async function waitForRuntimeStateLastRun(agentId: string, runId: string) { + for (let attempt = 0; attempt < 100; attempt += 1) { + const state = await db + .select({ lastRunId: agentRuntimeState.lastRunId }) + .from(agentRuntimeState) + .where(eq(agentRuntimeState.agentId, agentId)) + .then((rows) => rows[0] ?? null); + if (state?.lastRunId === runId) return; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + } + + it("suppresses new assignment wakes in worktree instances without creating heartbeat runs", async () => { + const { agentId, issueId } = await insertAgentAndIssue(); + const heartbeat = heartbeatService(db, { + runtimeEnv: { PAPERCLIP_IN_WORKTREE: "true" }, + }); + + const run = await heartbeat.wakeup(agentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { issueId }, + contextSnapshot: { issueId, wakeReason: "issue_assigned" }, + requestedByActorType: "system", + requestedByActorId: "issue_assignment", + }); + + expect(run).toBeNull(); + + const runs = await db.select().from(heartbeatRuns); + expect(runs).toHaveLength(0); + + const wakeup = await db + .select({ + status: agentWakeupRequests.status, + reason: agentWakeupRequests.reason, + payload: agentWakeupRequests.payload, + }) + .from(agentWakeupRequests) + .then((rows) => rows[0] ?? null); + expect(wakeup).toMatchObject({ + status: "skipped", + reason: "heartbeat.scheduling_suppressed", + }); + expect(wakeup?.payload).toMatchObject({ + issueId, + heartbeatSkip: { reason: "worktree_instance" }, + }); + }); + + it("does not replay copied queued runs or timer wakes while worktree scheduling is suppressed", async () => { + const { companyId, agentId, issueId } = await insertAgentAndIssue(); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + invocationSource: "assignment", + triggerDetail: "system", + status: "queued", + responsibleUserId: "responsible-user", + contextSnapshot: { issueId, wakeReason: "issue_assigned" }, + }); + + const heartbeat = heartbeatService(db, { + runtimeEnv: { PAPERCLIP_IN_WORKTREE: "true" }, + }); + + await heartbeat.resumeQueuedRuns(); + const tick = await heartbeat.tickTimers(new Date("2026-07-07T00:10:00Z")); + + expect(tick).toEqual({ checked: 0, enqueued: 0, skipped: 0 }); + + const [copiedRun] = await db + .select({ status: heartbeatRuns.status, startedAt: heartbeatRuns.startedAt }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)); + expect(copiedRun).toMatchObject({ + status: "queued", + startedAt: null, + }); + + const runningCount = await db + .select({ count: sql`count(*)::int` }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.status, "running")) + .then((rows) => rows[0]?.count ?? 0); + expect(runningCount).toBe(0); + }); + + it("still creates live-plane assignment runs when suppression is not active", async () => { + const { agentId, issueId } = await insertAgentAndIssue(); + const heartbeat = heartbeatService(db, { runtimeEnv: {} }); + + const run = await heartbeat.wakeup(agentId, { + source: "assignment", + triggerDetail: "system", + reason: "issue_assigned", + payload: { issueId }, + contextSnapshot: { issueId, wakeReason: "issue_assigned" }, + requestedByActorType: "system", + requestedByActorId: "issue_assignment", + }); + + expect(run).not.toBeNull(); + const terminalStatus = await waitForTerminalRun(run!.id); + expect(["succeeded", null]).toContain(terminalStatus); + + const runCount = await db + .select({ count: sql`count(*)::int` }) + .from(heartbeatRuns) + .then((rows) => rows[0]?.count ?? 0); + expect(runCount).toBe(1); + + await db + .update(issues) + .set({ status: "done", updatedAt: new Date() }) + .where(eq(issues.id, issueId)); + await waitForRuntimeStateLastRun(agentId, run!.id); + }); + + it("recognizes explicit restore-in-progress suppression", () => { + expect(resolveHeartbeatSchedulingSuppression({ + PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS: "true", + })).toEqual({ + suppressed: true, + reason: "database_restore_in_progress", + }); + }); +}); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 37e59eb9b4..96f59ecdf2 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -12,16 +12,60 @@ const { createDbMock, detectPortMock, deriveAuthTrustedOriginsMock, + environmentCustomImagesServiceMock, + environmentCustomImagesServiceFactoryMock, feedbackExportServiceMock, feedbackServiceFactoryMock, fakeServer, + heartbeatServiceFactoryMock, + heartbeatServiceMock, loadConfigMock, + resolveHeartbeatSchedulingSuppressionMock, + routineServiceFactoryMock, + routineServiceMock, } = vi.hoisted(() => { const createAppMock = vi.fn(async () => ((_: unknown, __: unknown) => {}) as never); const createBetterAuthInstanceMock = vi.fn(() => ({})); const createDbMock = vi.fn(() => ({}) as never); const detectPortMock = vi.fn(async (port: number) => port); const deriveAuthTrustedOriginsMock = vi.fn(() => []); + const heartbeatServiceMock = { + reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })), + promoteDueScheduledRetries: vi.fn(async () => ({ promoted: 0, runIds: [] })), + resumeQueuedRuns: vi.fn(async () => undefined), + reconcileStrandedAssignedIssues: vi.fn(async () => ({ + assignmentDispatched: 0, + dispatchRequeued: 0, + continuationRequeued: 0, + successfulRunHandoffEscalated: 0, + escalated: 0, + skipped: 0, + issueIds: [], + })), + reconcileIssueGraphLiveness: vi.fn(async () => ({ + escalationsCreated: 0, + dependencyWakesHealed: 0, + })), + reconcileTaskWatchdogs: vi.fn(async () => ({ triggered: 0 })), + scanSilentActiveRuns: vi.fn(async () => ({ created: 0, escalated: 0 })), + sweepStaleIssueLocks: vi.fn(async () => ({ cleared: 0 })), + reconcileProductivityReviews: vi.fn(async () => ({ created: 0, updated: 0, failed: 0 })), + sweepExpiredRuntimeStatuses: vi.fn(() => 0), + tickTimers: vi.fn(async () => ({ checked: 0, enqueued: 0, skipped: 0 })), + }; + const heartbeatServiceFactoryMock = vi.fn(() => heartbeatServiceMock); + const environmentCustomImagesServiceMock = { + cleanupExpiredSetupSessions: vi.fn(async () => ({ scanned: 0, timedOut: 0, failed: 0 })), + }; + const environmentCustomImagesServiceFactoryMock = vi.fn(() => environmentCustomImagesServiceMock); + const routineServiceMock = { + tickScheduledTriggers: vi.fn(async () => ({ triggered: 0 })), + }; + const routineServiceFactoryMock = vi.fn(() => routineServiceMock); + const resolveHeartbeatSchedulingSuppressionMock = vi.fn(() => ({ + suppressed: false, + reason: null, + })); const feedbackExportServiceMock = { flushPendingFeedbackTraces: vi.fn(async () => ({ attempted: 0, sent: 0, failed: 0 })), }; @@ -43,10 +87,17 @@ const { createDbMock, detectPortMock, deriveAuthTrustedOriginsMock, + environmentCustomImagesServiceMock, + environmentCustomImagesServiceFactoryMock, feedbackExportServiceMock, feedbackServiceFactoryMock, fakeServer, + heartbeatServiceFactoryMock, + heartbeatServiceMock, loadConfigMock, + resolveHeartbeatSchedulingSuppressionMock, + routineServiceFactoryMock, + routineServiceMock, }; }); @@ -144,20 +195,8 @@ vi.mock("../services/index.js", () => ({ })), feedbackService: feedbackServiceFactoryMock, bootstrapExecutionPolicyFromEnv: vi.fn(async () => null), - heartbeatService: vi.fn(() => ({ - reapOrphanedRuns: vi.fn(async () => undefined), - promoteDueScheduledRetries: vi.fn(async () => ({ promoted: 0, runIds: [] })), - resumeQueuedRuns: vi.fn(async () => undefined), - reconcileStrandedAssignedIssues: vi.fn(async () => ({ - dispatchRequeued: 0, - continuationRequeued: 0, - successfulRunHandoffEscalated: 0, - escalated: 0, - skipped: 0, - issueIds: [], - })), - tickTimers: vi.fn(async () => ({ enqueued: 0 })), - })), + environmentCustomImageService: environmentCustomImagesServiceFactoryMock, + heartbeatService: heartbeatServiceFactoryMock, instanceSettingsService: vi.fn(() => ({ getGeneral: vi.fn(async () => ({ backupRetention: { @@ -179,9 +218,8 @@ vi.mock("../services/index.js", () => ({ seededAgentIds: [], })), reconcilePersistedRuntimeServicesOnStartup: vi.fn(async () => ({ reconciled: 0 })), - routineService: vi.fn(() => ({ - tickScheduledTriggers: vi.fn(async () => ({ triggered: 0 })), - })), + resolveHeartbeatSchedulingSuppression: resolveHeartbeatSchedulingSuppressionMock, + routineService: routineServiceFactoryMock, })); vi.mock("../storage/index.js", () => ({ @@ -192,6 +230,10 @@ vi.mock("../services/feedback-share-client.js", () => ({ createFeedbackTraceShareClientFromConfig: vi.fn(() => ({ id: "feedback-share-client" })), })); +vi.mock("../services/plugin-worker-manager.js", () => ({ + createPluginWorkerManager: vi.fn(() => ({ id: "plugin-worker-manager" })), +})); + vi.mock("../startup-banner.js", () => ({ printStartupBanner: vi.fn(), })); @@ -215,6 +257,10 @@ describe("startServer feedback export wiring", () => { beforeEach(() => { vi.clearAllMocks(); loadConfigMock.mockReturnValue(buildTestConfig()); + resolveHeartbeatSchedulingSuppressionMock.mockReturnValue({ + suppressed: false, + reason: null, + }); createBetterAuthInstanceMock.mockReturnValue({}); deriveAuthTrustedOriginsMock.mockReturnValue([]); process.env.BETTER_AUTH_SECRET = "test-secret"; @@ -233,6 +279,43 @@ describe("startServer feedback export wiring", () => { }); }); + it("keeps routine ticks and setup cleanup active when heartbeat scheduling is suppressed", async () => { + loadConfigMock.mockReturnValue(buildTestConfig({ + heartbeatSchedulerEnabled: true, + heartbeatSchedulerIntervalMs: 30000, + })); + resolveHeartbeatSchedulingSuppressionMock.mockReturnValue({ + suppressed: true, + reason: "worktree_instance", + }); + let intervalCallback: (() => void) | null = null; + const setIntervalSpy = vi + .spyOn(globalThis, "setInterval") + .mockImplementation(((callback: () => void) => { + intervalCallback = callback; + return 1 as unknown as ReturnType; + }) as typeof setInterval); + + try { + await startServer(); + + expect(heartbeatServiceMock.reapOrphanedRuns).not.toHaveBeenCalled(); + expect(heartbeatServiceMock.tickTimers).not.toHaveBeenCalled(); + expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(1); + + expect(intervalCallback).not.toBeNull(); + intervalCallback?.(); + await Promise.resolve(); + await Promise.resolve(); + + expect(heartbeatServiceMock.tickTimers).not.toHaveBeenCalled(); + expect(routineServiceMock.tickScheduledTriggers).toHaveBeenCalledTimes(1); + expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(2); + } finally { + setIntervalSpy.mockRestore(); + } + }); + it("refuses authenticated public startup without an external database URL", async () => { loadConfigMock.mockReturnValue(buildTestConfig({ deploymentExposure: "public", diff --git a/server/src/index.ts b/server/src/index.ts index 094952699d..f4dab7a60e 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -45,6 +45,7 @@ import { reconcileCloudUpstreamRunsOnStartup, reconcileCodexLocalManagedHomesOnStartup, reconcilePersistedRuntimeServicesOnStartup, + resolveHeartbeatSchedulingSuppression, routineService, } from "./services/index.js"; import { @@ -804,85 +805,93 @@ export async function startServer(): Promise { const heartbeat = heartbeatService(db as any, { pluginWorkerManager }); const environmentCustomImages = environmentCustomImageService(db as any, { pluginWorkerManager }); const routines = routineService(db as any, { pluginWorkerManager }); + const heartbeatSchedulingSuppression = resolveHeartbeatSchedulingSuppression(); // Reap orphaned runs before timer ticks start so wakeups cannot coalesce // into a dead "running" row during startup recovery. - await (async () => { - for (let attempt = 1; attempt <= 2; attempt++) { - try { - const result = await heartbeat.reapOrphanedRuns(); - logger.info( - { reaped: result.reaped, runIds: result.runIds }, - "startup reap of orphaned heartbeat runs complete", - ); - break; - } catch (err) { - if (attempt < 2) { - logger.warn({ err, attempt }, "startup reap failed, retrying"); - } else { - logger.error( - { err }, - "startup reap of orphaned heartbeat runs failed after retry — periodic reaper will serve as degraded backstop", + if (heartbeatSchedulingSuppression.suppressed) { + logger.warn( + { reason: heartbeatSchedulingSuppression.reason }, + "heartbeat scheduling suppressed for this runtime instance", + ); + } else { + await (async () => { + for (let attempt = 1; attempt <= 2; attempt++) { + try { + const result = await heartbeat.reapOrphanedRuns(); + logger.info( + { reaped: result.reaped, runIds: result.runIds }, + "startup reap of orphaned heartbeat runs complete", ); + break; + } catch (err) { + if (attempt < 2) { + logger.warn({ err, attempt }, "startup reap failed, retrying"); + } else { + logger.error( + { err }, + "startup reap of orphaned heartbeat runs failed after retry — periodic reaper will serve as degraded backstop", + ); + } } } - } - const promotion = await heartbeat.promoteDueScheduledRetries(); - await heartbeat.resumeQueuedRuns(); - const reconciled = await heartbeat.reconcileStrandedAssignedIssues(); - if ( - promotion.promoted > 0 || - reconciled.assignmentDispatched > 0 || - reconciled.dispatchRequeued > 0 || - reconciled.continuationRequeued > 0 || - reconciled.successfulRunHandoffEscalated > 0 || - reconciled.escalated > 0 - ) { - logger.warn( - { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled }, - "startup heartbeat recovery changed assigned issue state", - ); - } + const promotion = await heartbeat.promoteDueScheduledRetries(); + await heartbeat.resumeQueuedRuns(); + const reconciled = await heartbeat.reconcileStrandedAssignedIssues(); + if ( + promotion.promoted > 0 || + reconciled.assignmentDispatched > 0 || + reconciled.dispatchRequeued > 0 || + reconciled.continuationRequeued > 0 || + reconciled.successfulRunHandoffEscalated > 0 || + reconciled.escalated > 0 + ) { + logger.warn( + { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled }, + "startup heartbeat recovery changed assigned issue state", + ); + } - const issueGraphReconciled = await heartbeat.reconcileIssueGraphLiveness(); - if (issueGraphReconciled.escalationsCreated > 0 || issueGraphReconciled.dependencyWakesHealed > 0) { - logger.warn( - { ...issueGraphReconciled }, - "startup issue-graph liveness reconciliation changed issue graph state", - ); - } + const issueGraphReconciled = await heartbeat.reconcileIssueGraphLiveness(); + if (issueGraphReconciled.escalationsCreated > 0 || issueGraphReconciled.dependencyWakesHealed > 0) { + logger.warn( + { ...issueGraphReconciled }, + "startup issue-graph liveness reconciliation changed issue graph state", + ); + } - const taskWatchdogsReconciled = await heartbeat.reconcileTaskWatchdogs(); - if (taskWatchdogsReconciled.triggered > 0) { - logger.warn( - { ...taskWatchdogsReconciled }, - "startup task-watchdog reconciliation triggered watchdog work", - ); - } + const taskWatchdogsReconciled = await heartbeat.reconcileTaskWatchdogs(); + if (taskWatchdogsReconciled.triggered > 0) { + logger.warn( + { ...taskWatchdogsReconciled }, + "startup task-watchdog reconciliation triggered watchdog work", + ); + } - const scanned = await heartbeat.scanSilentActiveRuns(); - if (scanned.created > 0 || scanned.escalated > 0) { - logger.warn({ ...scanned }, "startup active-run output watchdog created review work"); - } + const scanned = await heartbeat.scanSilentActiveRuns(); + if (scanned.created > 0 || scanned.escalated > 0) { + logger.warn({ ...scanned }, "startup active-run output watchdog created review work"); + } - const swept = await heartbeat.sweepStaleIssueLocks(); - if (swept.cleared > 0) { - logger.warn({ ...swept }, "startup stale-lock sweeper cleared issue locks"); - } + const swept = await heartbeat.sweepStaleIssueLocks(); + if (swept.cleared > 0) { + logger.warn({ ...swept }, "startup stale-lock sweeper cleared issue locks"); + } - const reviewed = await heartbeat.reconcileProductivityReviews(); - if (reviewed.created > 0 || reviewed.updated > 0 || reviewed.failed > 0) { - logger.warn({ ...reviewed }, "startup productivity reconciliation created or updated review work"); - } + const reviewed = await heartbeat.reconcileProductivityReviews(); + if (reviewed.created > 0 || reviewed.updated > 0 || reviewed.failed > 0) { + logger.warn({ ...reviewed }, "startup productivity reconciliation created or updated review work"); + } + })().catch((err) => { + logger.error({ err }, "startup heartbeat recovery failed"); + }); + } - const setupCleanup = await environmentCustomImages.cleanupExpiredSetupSessions(); - if (setupCleanup.timedOut > 0 || setupCleanup.failed > 0) { - logger.warn({ ...setupCleanup }, "startup environment customImage setup cleanup changed sessions"); - } - })().catch((err) => { - logger.error({ err }, "startup heartbeat recovery failed"); - }); + const setupCleanup = await environmentCustomImages.cleanupExpiredSetupSessions(); + if (setupCleanup.timedOut > 0 || setupCleanup.failed > 0) { + logger.warn({ ...setupCleanup }, "startup environment customImage setup cleanup changed sessions"); + } setInterval(() => { const sweptRuntimeStatuses = heartbeat.sweepExpiredRuntimeStatuses(); @@ -893,16 +902,18 @@ export async function startServer(): Promise { ); } - void heartbeat - .tickTimers(new Date()) - .then((result) => { - if (result.enqueued > 0) { - logger.info({ ...result }, "heartbeat timer tick enqueued runs"); - } - }) - .catch((err) => { - logger.error({ err }, "heartbeat timer tick failed"); - }); + if (!resolveHeartbeatSchedulingSuppression().suppressed) { + void heartbeat + .tickTimers(new Date()) + .then((result) => { + if (result.enqueued > 0) { + logger.info({ ...result }, "heartbeat timer tick enqueued runs"); + } + }) + .catch((err) => { + logger.error({ err }, "heartbeat timer tick failed"); + }); + } void routines .tickScheduledTriggers(new Date()) @@ -926,61 +937,63 @@ export async function startServer(): Promise { logger.error({ err }, "environment customImage setup cleanup failed"); }); - // Periodically reap orphaned runs (5-min staleness threshold) and make sure - // persisted queued work is still being driven forward. - void heartbeat - .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000 }) - .then(() => heartbeat.promoteDueScheduledRetries()) - .then(async (promotion) => { - await heartbeat.resumeQueuedRuns(); - const reconciled = await heartbeat.reconcileStrandedAssignedIssues(); - if ( - promotion.promoted > 0 || - reconciled.assignmentDispatched > 0 || - reconciled.dispatchRequeued > 0 || - reconciled.continuationRequeued > 0 || - reconciled.successfulRunHandoffEscalated > 0 || - reconciled.escalated > 0 - ) { - logger.warn( - { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled }, - "periodic heartbeat recovery changed assigned issue state", - ); - } - }) - .then(async () => { - const reconciled = await heartbeat.reconcileIssueGraphLiveness(); - if (reconciled.escalationsCreated > 0 || reconciled.dependencyWakesHealed > 0) { - logger.warn({ ...reconciled }, "periodic issue-graph liveness reconciliation changed issue graph state"); - } - }) - .then(async () => { - const reconciled = await heartbeat.reconcileTaskWatchdogs(); - if (reconciled.triggered > 0) { - logger.warn({ ...reconciled }, "periodic task-watchdog reconciliation triggered watchdog work"); - } - }) - .then(async () => { - const scanned = await heartbeat.scanSilentActiveRuns(); - if (scanned.created > 0 || scanned.escalated > 0) { - logger.warn({ ...scanned }, "periodic active-run output watchdog created review work"); - } - }) - .then(async () => { - const swept = await heartbeat.sweepStaleIssueLocks(); - if (swept.cleared > 0) { - logger.warn({ ...swept }, "periodic stale-lock sweeper cleared issue locks"); - } - }) - .then(async () => { - const reviewed = await heartbeat.reconcileProductivityReviews(); - if (reviewed.created > 0 || reviewed.updated > 0 || reviewed.failed > 0) { - logger.warn({ ...reviewed }, "periodic productivity reconciliation created or updated review work"); - } - }) - .catch((err) => { - logger.error({ err }, "periodic heartbeat recovery failed"); - }); + if (!resolveHeartbeatSchedulingSuppression().suppressed) { + // Periodically reap orphaned runs (5-min staleness threshold) and make sure + // persisted queued work is still being driven forward. + void heartbeat + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000 }) + .then(() => heartbeat.promoteDueScheduledRetries()) + .then(async (promotion) => { + await heartbeat.resumeQueuedRuns(); + const reconciled = await heartbeat.reconcileStrandedAssignedIssues(); + if ( + promotion.promoted > 0 || + reconciled.assignmentDispatched > 0 || + reconciled.dispatchRequeued > 0 || + reconciled.continuationRequeued > 0 || + reconciled.successfulRunHandoffEscalated > 0 || + reconciled.escalated > 0 + ) { + logger.warn( + { promotedScheduledRetries: promotion.promoted, promotedScheduledRetryRunIds: promotion.runIds, ...reconciled }, + "periodic heartbeat recovery changed assigned issue state", + ); + } + }) + .then(async () => { + const reconciled = await heartbeat.reconcileIssueGraphLiveness(); + if (reconciled.escalationsCreated > 0 || reconciled.dependencyWakesHealed > 0) { + logger.warn({ ...reconciled }, "periodic issue-graph liveness reconciliation changed issue graph state"); + } + }) + .then(async () => { + const reconciled = await heartbeat.reconcileTaskWatchdogs(); + if (reconciled.triggered > 0) { + logger.warn({ ...reconciled }, "periodic task-watchdog reconciliation triggered watchdog work"); + } + }) + .then(async () => { + const scanned = await heartbeat.scanSilentActiveRuns(); + if (scanned.created > 0 || scanned.escalated > 0) { + logger.warn({ ...scanned }, "periodic active-run output watchdog created review work"); + } + }) + .then(async () => { + const swept = await heartbeat.sweepStaleIssueLocks(); + if (swept.cleared > 0) { + logger.warn({ ...swept }, "periodic stale-lock sweeper cleared issue locks"); + } + }) + .then(async () => { + const reviewed = await heartbeat.reconcileProductivityReviews(); + if (reviewed.created > 0 || reviewed.updated > 0 || reviewed.failed > 0) { + logger.warn({ ...reviewed }, "periodic productivity reconciliation created or updated review work"); + } + }) + .catch((err) => { + logger.error({ err }, "periodic heartbeat recovery failed"); + }); + } }, config.heartbeatSchedulerIntervalMs); } diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 248857d47d..6f420da72d 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -4883,6 +4883,26 @@ export type HeartbeatEnvironmentRuntime = ReturnType; +} + +function isTruthyRuntimeEnvValue(value: string | undefined) { + return value === "true" || value === "1" || value === "yes" || value === "on"; +} + +export function resolveHeartbeatSchedulingSuppression( + env: Record = process.env, +): { suppressed: boolean; reason: "worktree_instance" | "database_restore_in_progress" | null } { + if (isTruthyRuntimeEnvValue(env.PAPERCLIP_IN_WORKTREE)) { + return { suppressed: true, reason: "worktree_instance" }; + } + if ( + isTruthyRuntimeEnvValue(env.PAPERCLIP_DATABASE_RESTORE_IN_PROGRESS) || + isTruthyRuntimeEnvValue(env.PAPERCLIP_RESTORE_IN_PROGRESS) + ) { + return { suppressed: true, reason: "database_restore_in_progress" }; + } + return { suppressed: false, reason: null }; } export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) { @@ -4890,6 +4910,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const getCurrentUserRedactionOptions = async () => ({ enabled: (await instanceSettings.getGeneral()).censorUsernameInLogs, }); + const runtimeEnv = options.runtimeEnv ?? process.env; + const getSchedulingSuppression = () => resolveHeartbeatSchedulingSuppression(runtimeEnv); const runLogStore = getRunLogStore(); const secretsSvc = secretService(db); @@ -9799,6 +9821,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } async function resumeQueuedRuns() { + if (getSchedulingSuppression().suppressed) return; + const queuedRuns = await db .select({ agentId: heartbeatRuns.agentId }) .from(heartbeatRuns) @@ -9925,6 +9949,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } async function startNextQueuedRunForAgent(agentId: string) { + if (getSchedulingSuppression().suppressed) return []; + return withAgentStartLock(agentId, async () => { const agent = await getAgent(agentId); if (!agent) return []; @@ -10003,6 +10029,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } async function executeRun(runId: string) { + if (getSchedulingSuppression().suppressed) return; + let run = await getRun(runId); if (!run) return; if (run.status !== "queued" && run.status !== "running") return; @@ -13165,6 +13193,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) }); }; + const schedulingSuppression = getSchedulingSuppression(); + if (schedulingSuppression.suppressed) { + await writeSkippedHeartbeatRequest("heartbeat.scheduling_suppressed", { + reason: schedulingSuppression.reason, + }); + return null; + } + const company = await db .select({ status: companies.status }) .from(companies) @@ -14645,6 +14681,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) buildRunOutputSilence, tickTimers: async (now = new Date()) => { + if (getSchedulingSuppression().suppressed) { + return { + checked: 0, + enqueued: 0, + skipped: 0, + }; + } + const allAgents = await db .select({ ...getTableColumns(agents) }) .from(agents) diff --git a/server/src/services/index.ts b/server/src/services/index.ts index 13cdaf711d..4a351f3e4a 100644 --- a/server/src/services/index.ts +++ b/server/src/services/index.ts @@ -59,7 +59,7 @@ export { secretService } from "./secrets.js"; export { routineService } from "./routines.js"; export { costService } from "./costs.js"; export { financeService } from "./finance.js"; -export { heartbeatService } from "./heartbeat.js"; +export { heartbeatService, resolveHeartbeatSchedulingSuppression } from "./heartbeat.js"; export { productivityReviewService, PRODUCTIVITY_REVIEW_ORIGIN_KIND,