fix(scheduler): respect heartbeat eligibility for task cadence
This commit is contained in:
1 parent
d9fd231be3
commit
d3e108ec18
2 files changed
+42
-6
No files matched your search
@@ -223,11 +223,12 @@ export function createScheduler(
|
|||||||
if (hasActiveRun({ agentId: ceo.id, companyId: company.id })) return;
|
if (hasActiveRun({ agentId: ceo.id, companyId: company.id })) return;
|
||||||
|
|
||||||
const lastHeartbeat = ceo.lastHeartbeatAt ? new Date(ceo.lastHeartbeatAt).getTime() : 0;
|
const lastHeartbeat = ceo.lastHeartbeatAt ? new Date(ceo.lastHeartbeatAt).getTime() : 0;
|
||||||
const lastAttempt = bootstrapAttemptByAgent.get(ceo.id) ?? 0;
|
const attemptKey = agentAttemptKey(company.id, ceo.id);
|
||||||
|
const lastAttempt = bootstrapAttemptByAgent.get(attemptKey) ?? 0;
|
||||||
const bootstrapCooldown = 120_000;
|
const bootstrapCooldown = 120_000;
|
||||||
if (now - Math.max(lastHeartbeat, lastAttempt) <= bootstrapCooldown) return;
|
if (now - Math.max(lastHeartbeat, lastAttempt) <= bootstrapCooldown) return;
|
||||||
|
|
||||||
bootstrapAttemptByAgent.set(ceo.id, now);
|
bootstrapAttemptByAgent.set(attemptKey, now);
|
||||||
logger.info(`[scheduler] Company ${company.name} has ${goalCount} goal(s) and no advancing tasks; bootstrapping CEO ${ceo.name}`);
|
logger.info(`[scheduler] Company ${company.name} has ${goalCount} goal(s) and no advancing tasks; bootstrapping CEO ${ceo.name}`);
|
||||||
wakeAgentIfPossible(ceo.id, company.id, "bootstrap");
|
wakeAgentIfPossible(ceo.id, company.id, "bootstrap");
|
||||||
}
|
}
|
||||||
@@ -243,6 +244,8 @@ export function createScheduler(
|
|||||||
const wokenAgentIds = new Set<string>();
|
const wokenAgentIds = new Set<string>();
|
||||||
const pendingWakeupAgents = db.automation.wakeup.listPendingWakeupAgents(companyId);
|
const pendingWakeupAgents = db.automation.wakeup.listPendingWakeupAgents(companyId);
|
||||||
for (const { agentId } of pendingWakeupAgents) {
|
for (const { agentId } of pendingWakeupAgents) {
|
||||||
|
// Queued wakeups are explicit durable events; they bypass the periodic
|
||||||
|
// heartbeat_enabled scan and let wakeAgentIfPossible own re-enabling.
|
||||||
if (hasActiveRun({ agentId, companyId })) continue;
|
if (hasActiveRun({ agentId, companyId })) continue;
|
||||||
if (deps.checkpointEngine?.shouldPauseHeartbeat(companyId, agentId)) continue;
|
if (deps.checkpointEngine?.shouldPauseHeartbeat(companyId, agentId)) continue;
|
||||||
|
|
||||||
@@ -264,21 +267,22 @@ export function createScheduler(
|
|||||||
}
|
}
|
||||||
|
|
||||||
function runAssignedTaskWakeCadence(companyId: string, now: number, wokenAgentIds: Set<string>) {
|
function runAssignedTaskWakeCadence(companyId: string, now: number, wokenAgentIds: Set<string>) {
|
||||||
const activeAgents = db.core.agents.listActiveAgentsForCompany(companyId);
|
const activeAgents = db.core.agents.listHeartbeatEligibleAgents(companyId);
|
||||||
for (const activeAgent of activeAgents) {
|
for (const activeAgent of activeAgents) {
|
||||||
if (wokenAgentIds.has(activeAgent.id)) continue;
|
if (wokenAgentIds.has(activeAgent.id)) continue;
|
||||||
if (hasActiveRun({ agentId: activeAgent.id, companyId })) continue;
|
if (hasActiveRun({ agentId: activeAgent.id, companyId })) continue;
|
||||||
if (deps.checkpointEngine?.shouldPauseHeartbeat(companyId, activeAgent.id)) continue;
|
if (deps.checkpointEngine?.shouldPauseHeartbeat(companyId, activeAgent.id)) continue;
|
||||||
|
|
||||||
const lastHeartbeat = activeAgent.lastHeartbeatAt ? new Date(activeAgent.lastHeartbeatAt).getTime() : 0;
|
const lastHeartbeat = activeAgent.lastHeartbeatAt ? new Date(activeAgent.lastHeartbeatAt).getTime() : 0;
|
||||||
const lastAttempt = taskWakeAttemptByAgent.get(activeAgent.id) ?? 0;
|
const attemptKey = agentAttemptKey(companyId, activeAgent.id);
|
||||||
const intervalMs = (activeAgent.heartbeatIntervalSec || 120) * 1000;
|
const lastAttempt = taskWakeAttemptByAgent.get(attemptKey) ?? 0;
|
||||||
|
const intervalMs = (activeAgent.heartbeatIntervalSec ?? 120) * 1000;
|
||||||
if (now - Math.max(lastHeartbeat, lastAttempt) < intervalMs) continue;
|
if (now - Math.max(lastHeartbeat, lastAttempt) < intervalMs) continue;
|
||||||
|
|
||||||
const actionableTasks = db.core.tasks.listAgentTasks(activeAgent.id, companyId, ["todo"]);
|
const actionableTasks = db.core.tasks.listAgentTasks(activeAgent.id, companyId, ["todo"]);
|
||||||
if (!actionableTasks.some((task) => !task.activeRunId)) continue;
|
if (!actionableTasks.some((task) => !task.activeRunId)) continue;
|
||||||
|
|
||||||
taskWakeAttemptByAgent.set(activeAgent.id, now);
|
taskWakeAttemptByAgent.set(attemptKey, now);
|
||||||
wakeAgentIfPossible(activeAgent.id, companyId, "timer");
|
wakeAgentIfPossible(activeAgent.id, companyId, "timer");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -199,6 +199,38 @@ describe("unified scheduler heartbeat cadence", () => {
|
|||||||
expect(wakeAgentIfPossible).toHaveBeenCalledWith(agentId, companyId, "timer");
|
expect(wakeAgentIfPossible).toHaveBeenCalledWith(agentId, companyId, "timer");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("does not wake assigned work for an agent with heartbeat disabled", () => {
|
||||||
|
const companyId = createCompany(db);
|
||||||
|
createConnector(db);
|
||||||
|
const workspaceId = createWorkspace(db, companyId, tempDir);
|
||||||
|
const agentId = createAgent(db, companyId, workspaceId);
|
||||||
|
db.core.agents.setHeartbeat(agentId, false, 1);
|
||||||
|
db.core.tasks.saveTask({
|
||||||
|
companyId,
|
||||||
|
title: "Paused operating work",
|
||||||
|
description: "Manual heartbeat disable should pause periodic recovery.",
|
||||||
|
assigneeAgentId: agentId,
|
||||||
|
workspaceId,
|
||||||
|
priority: "high",
|
||||||
|
status: "todo",
|
||||||
|
});
|
||||||
|
const wakeAgentIfPossible = vi.fn();
|
||||||
|
const scheduler = createScheduler(
|
||||||
|
createDeps(db),
|
||||||
|
() => false,
|
||||||
|
wakeAgentIfPossible,
|
||||||
|
);
|
||||||
|
|
||||||
|
try {
|
||||||
|
scheduler.startAllSchedulers();
|
||||||
|
vi.advanceTimersByTime(60_000);
|
||||||
|
} finally {
|
||||||
|
scheduler.stopAllSchedulers();
|
||||||
|
}
|
||||||
|
|
||||||
|
expect(wakeAgentIfPossible).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
it("wakes an agent with a queued wakeup request on heartbeat cadence", () => {
|
it("wakes an agent with a queued wakeup request on heartbeat cadence", () => {
|
||||||
const companyId = createCompany(db);
|
const companyId = createCompany(db);
|
||||||
createConnector(db);
|
createConnector(db);
|
||||||
|
|||||||
Reference in new issue
Block a user