RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[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 <noreply@paperclip.ing>
This commit is contained in:
Dotta 2026-07-07 09:27:46 -05:00 committed by GitHub
parent 57a7da81ee
commit 390627b46e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 573 additions and 149 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
server/src/__tests__/heartbeat-scheduling-suppression.test.ts Normal file
View File

@ -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,
});
});
});

255
server/src/__tests__/heartbeat-worktree-suppression.test.ts Normal file
View File

@ -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<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<number>`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<number>`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",
});
});
});

117
server/src/__tests__/server-startup-feedback-export.test.ts
View File

@ -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<typeof setInterval>;
}) 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",

275
server/src/index.ts
View File

@ -45,6 +45,7 @@ import {
reconcileCloudUpstreamRunsOnStartup,
reconcileCodexLocalManagedHomesOnStartup,
reconcilePersistedRuntimeServicesOnStartup,
resolveHeartbeatSchedulingSuppression,
routineService,
} from "./services/index.js";
import {
@ -804,85 +805,93 @@ export async function startServer(): Promise<StartedServer> {
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<StartedServer> {
);
}
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<StartedServer> {
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);
}

44
server/src/services/heartbeat.ts
View File

@ -4883,6 +4883,26 @@ export type HeartbeatEnvironmentRuntime = ReturnType<typeof environmentRuntimeSe
export interface HeartbeatServiceOptions {
pluginWorkerManager?: PluginWorkerManager;
environmentRuntime?: HeartbeatEnvironmentRuntime;
runtimeEnv?: Record<string, string | undefined>;
}
function isTruthyRuntimeEnvValue(value: string | undefined) {
return value === "true" || value === "1" || value === "yes" || value === "on";
}
export function resolveHeartbeatSchedulingSuppression(
env: Record<string, string | undefined> = 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)

2
server/src/services/index.ts
View File

@ -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,