diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index f0906a49372..40db0e77501 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1377,6 +1377,32 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { }); }); + it("does not retry a lost monitor dispatch while another monitor wake remains scheduled", async () => { + const { companyId, runId, issueId } = await seedRunFixture({ + adapterType: "openclaw_gateway", + agentStatus: "idle", + processPid: null, + processGroupId: null, + contextSnapshot: { + wakeReason: "issue_monitor_due", + }, + }); + await db + .update(issues) + .set({ monitorNextCheckAt: new Date("2026-03-19T00:00:00.000Z") }) + .where(and(eq(issues.id, issueId), eq(issues.companyId, companyId))); + + const heartbeat = heartbeatService(db); + const result = await heartbeat.reapOrphanedRuns(); + + expect(result).toEqual({ reaped: 1, runIds: [runId] }); + const retries = await db + .select() + .from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.retryOfRunId, runId))); + expect(retries).toHaveLength(0); + }); + it("interrupts running runs on graceful shutdown and queues restart recovery without recording a failure", async () => { const { agentId, runId, issueId, wakeupRequestId } = await seedRunFixture({ agentStatus: "running", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index e500ef6a84a..92da28bdd6c 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -10923,6 +10923,29 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) .innerJoin(agents, eq(heartbeatRuns.agentId, agents.id)) .where(eq(heartbeatRuns.status, "running")); + const monitorIssueIds = [...new Set(activeRuns.flatMap(({ run }) => { + const runContext = parseObject(run.contextSnapshot); + if (readNonEmptyString(runContext.wakeReason) !== "issue_monitor_due") return []; + const issueId = readNonEmptyString(runContext.issueId); + return issueId ? [issueId] : []; + }))]; + const monitorIssues = monitorIssueIds.length > 0 + ? await db + .select({ + id: issues.id, + companyId: issues.companyId, + monitorNextCheckAt: issues.monitorNextCheckAt, + }) + .from(issues) + .where(inArray(issues.id, monitorIssueIds)) + : []; + const monitorNextCheckAtByIssue = new Map( + monitorIssues.map((issue) => [ + `${issue.companyId}:${issue.id}`, + issue.monitorNextCheckAt, + ]), + ); + const reaped: string[] = []; for (const { run, adapterType, adapterConfig } of activeRuns) { @@ -10973,17 +10996,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const monitorDispatchLostWithoutFutureWake = readNonEmptyString(runContext.wakeReason) === "issue_monitor_due" && Boolean(monitorIssueId) && - await db - .select({ monitorNextCheckAt: issues.monitorNextCheckAt }) - .from(issues) - .where(and(eq(issues.id, monitorIssueId!), eq(issues.companyId, run.companyId))) - .limit(1) - .then((rows) => { - const issue = rows[0]; - if (!issue) return false; - const nextCheckAt = issue.monitorNextCheckAt; - return !nextCheckAt || nextCheckAt.getTime() <= now.getTime(); - }); + monitorNextCheckAtByIssue.get(`${run.companyId}:${monitorIssueId}`) === null; const shouldRetry = (run.processLossRetryCount ?? 0) < 1 && ( (tracksLocalChild && (!!run.processPid || !!run.processGroupId)) || monitorDispatchLostWithoutFutureWake