root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(FN-3005): preserve card timer across reruns

Browse Source
This commit is contained in:
gsxdsm
2026-04-29 23:13:38 -07:00
parent d81fc9b2c7
commit bb9b0f1d1a
20 changed files with 258 additions and 361 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
packages/engine/src/__tests__/agent-heartbeat.test.ts
View File

@@ -4619,61 +4619,6 @@ describe("HeartbeatTriggerScheduler", () => {
expect(callback).not.toHaveBeenCalled();
});
it("phase-aligns the first tick to lastHeartbeatAt when supplied", async () => {
// Simulate: last tick was 4s ago, interval is 5s.
// The next tick is due in 1s, not in a fresh full 5s window.
vi.setSystemTime(new Date("2026-04-30T05:00:00.000Z"));
const lastHeartbeatAt = new Date("2026-04-30T04:59:56.000Z").toISOString();
scheduler.registerAgent(
"agent-001",
{ heartbeatIntervalMs: 5000 },
{ lastHeartbeatAt },
);
await vi.advanceTimersByTimeAsync(999);
expect(callback).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(callback).toHaveBeenCalledOnce();
// Subsequent ticks resume the steady cadence.
await vi.advanceTimersByTimeAsync(5000);
expect(callback).toHaveBeenCalledTimes(2);
});
it("fires promptly with jitter when lastHeartbeatAt is already overdue", async () => {
// Interval is 60s but the last tick was 10 minutes ago — fire immediately
// (within the OVERDUE_FIRE_JITTER_MS window) instead of waiting another
// full 60s. This is the core fix for "agents look unresponsive after a
// dashboard restart" — the previous setInterval-only scheduler would
// have made the user wait a full interval before the catch-up tick.
vi.setSystemTime(new Date("2026-04-30T05:00:00.000Z"));
const lastHeartbeatAt = new Date("2026-04-30T04:50:00.000Z").toISOString();
scheduler.registerAgent(
"agent-001",
{ heartbeatIntervalMs: 60_000 },
{ lastHeartbeatAt },
);
// Jitter window is 5s; advance past it to guarantee the fire happens.
await vi.advanceTimersByTimeAsync(5_000);
expect(callback).toHaveBeenCalledOnce();
});
it("falls back to full-interval delay when lastHeartbeatAt is missing", async () => {
// No options provided — preserves the original "wait one full interval"
// behavior for agents that have never ticked.
scheduler.registerAgent("agent-001", { heartbeatIntervalMs: 5000 });
await vi.advanceTimersByTimeAsync(4999);
expect(callback).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(1);
expect(callback).toHaveBeenCalledOnce();
});
it("skips tick when agent has active run", async () => {
(store.getActiveHeartbeatRun as ReturnType<typeof vi.fn>).mockResolvedValue({
id: "run-active",

39
packages/engine/src/__tests__/executor.test.ts
View File

@@ -11158,6 +11158,45 @@ describe("TaskExecutor watchdogs", () => {
expect.stringContaining("Watchdog: workflow rerun handoff stalled for 15s"),
);
});
it("preserves the original executionStartedAt during a workflow rerun bounce", async () => {
const store = createMockStore();
const originalExecutionStartedAt = "2026-04-30T05:06:43.781Z";
const mutableTask = {
id: "FN-WD-4",
title: "Workflow rerun timing",
description: "desc",
column: "in-progress" as const,
paused: false,
worktree: "/tmp/fn-wd-4",
executionStartedAt: originalExecutionStartedAt,
dependencies: [],
steps: [{ name: "Step 0", status: "done" as const }],
currentStep: 0,
log: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
};
store.getTask.mockResolvedValue(mutableTask as any);
store.moveTask.mockImplementation(async (_taskId: string, column: string) => {
mutableTask.column = column as "in-progress" | "todo";
return { ...mutableTask };
});
const executor = new TaskExecutor(store, "/tmp/test");
const outcome = await (executor as any).performWorkflowRerunBounce("FN-WD-4", "/tmp/fn-wd-4");
expect(outcome).toBe("bounced");
expect(store.updateTask).toHaveBeenCalledWith("FN-WD-4", {
worktree: "/tmp/fn-wd-4",
executionStartedAt: originalExecutionStartedAt,
});
expect(store.moveTask.mock.calls).toEqual([
["FN-WD-4", "todo"],
["FN-WD-4", "in-progress"],
]);
});
});
// ── StepSessionExecutor integration tests ──────────────────────────────────

148
packages/engine/src/agent-heartbeat.ts
View File

@@ -1927,34 +1927,12 @@ export type TriggerCallback = (
context: WakeContext,
) => Promise<void>;
/** Per-agent timer state. The active handle is either the initial
* phase-aligned `setTimeout` waiting for the first overdue tick, or the
* steady-state `setInterval` installed once that first tick fires.
*/
/** Per-agent timer state */
interface AgentTimer {
intervalMs: number;
kind: "timeout" | "interval";
handle: ReturnType<typeof setInterval>;
}
/** Optional context passed to registerAgent, used to phase-align the
* initial timer fire to the agent's persisted heartbeat history.
*/
export interface RegisterAgentOptions {
/** ISO timestamp of the agent's last heartbeat. When set, the initial
* fire is scheduled at `lastHeartbeatAt + intervalMs` rather than
* `now + intervalMs`, so a process restart does not cost agents up to
* one full interval of silence.
*/
lastHeartbeatAt?: string | null;
}
/** Maximum random jitter (ms) added to the initial fire when an agent's
* next tick is already overdue. Prevents a thundering herd when the
* scheduler boots and many agents want to fire immediately.
*/
const OVERDUE_FIRE_JITTER_MS = 5_000;
/**
* True when an agent's state indicates it should be ticking right now.
* Heartbeats track liveness while the agent is meant to be doing work.
@@ -2040,13 +2018,9 @@ export class HeartbeatTriggerScheduler {
this.unwatchAssignments();
this.unwatchAgentLifecycle();
// Clear all timers (mix of phase-alignment timeouts and steady intervals).
// Clear all timers
for (const [agentId, timer] of this.timers) {
if (timer.kind === "timeout") {
clearTimeout(timer.handle as unknown as ReturnType<typeof setTimeout>);
} else {
clearInterval(timer.handle);
}
clearInterval(timer.handle);
heartbeatLog.log(`Cleared timer for ${agentId}`);
}
this.timers.clear();
@@ -2066,19 +2040,10 @@ export class HeartbeatTriggerScheduler {
/**
* Register an agent for timer-based heartbeat triggers.
*
* The first fire is phase-aligned to `options.lastHeartbeatAt + intervalMs`
* when supplied. This means a process restart resumes each agent's
* existing schedule rather than waiting up to a full interval before the
* first tick — the previous behavior caused agents on long intervals
* (e.g. 1h) to appear "overdue" in the UI for nearly a full interval after
* every dashboard restart even though nothing was actually wrong with them.
*
* @param agentId - The agent ID
* @param config - Per-agent heartbeat config
* @param options - Optional registration context (e.g., lastHeartbeatAt)
*/
registerAgent(agentId: string, config: AgentHeartbeatConfig, options?: RegisterAgentOptions): void {
registerAgent(agentId: string, config: AgentHeartbeatConfig): void {
if (config.enabled === false) {
this.unregisterAgent(agentId);
return;
@@ -2098,14 +2063,12 @@ export class HeartbeatTriggerScheduler {
const registrationEpoch = (this.registrationEpochs.get(agentId) ?? 0) + 1;
this.registrationEpochs.set(agentId, registrationEpoch);
const lastHeartbeatAt = options?.lastHeartbeatAt ?? null;
// Register immediately with multiplier=1 so agents don't wait for async settings I/O.
this.applyTimerRegistration(agentId, intervalMs, 1, usingDefaultInterval, lastHeartbeatAt);
this.applyTimerRegistration(agentId, intervalMs, 1, usingDefaultInterval);
// If project settings are available, refresh registration with the current multiplier.
if (this.taskStore && typeof (this.taskStore as { getSettings?: () => Promise<Settings> }).getSettings === "function") {
void this.applyProjectMultiplierRegistration(agentId, intervalMs, usingDefaultInterval, registrationEpoch, lastHeartbeatAt);
void this.applyProjectMultiplierRegistration(agentId, intervalMs, usingDefaultInterval, registrationEpoch);
}
}
@@ -2114,7 +2077,6 @@ export class HeartbeatTriggerScheduler {
baseIntervalMs: number,
usingDefaultInterval: boolean,
expectedEpoch: number,
lastHeartbeatAt: string | null,
): Promise<void> {
let multiplier = 1;
@@ -2133,33 +2095,7 @@ export class HeartbeatTriggerScheduler {
return;
}
this.applyTimerRegistration(agentId, baseIntervalMs, multiplier, usingDefaultInterval, lastHeartbeatAt);
}
/**
* Compute the delay until the agent's next scheduled fire, given when it
* last heartbeat. When `lastHeartbeatAt` is missing or unparseable, falls
* back to a full-interval delay (matching the original behavior for
* agents that have never ticked). When the next fire is already overdue,
* returns a small randomized jitter to spread thundering herds at boot.
*/
private static computeInitialDelayMs(
intervalMs: number,
lastHeartbeatAt: string | null,
now: number = Date.now(),
): number {
if (!lastHeartbeatAt) {
return intervalMs;
}
const lastMs = Date.parse(lastHeartbeatAt);
if (!Number.isFinite(lastMs)) {
return intervalMs;
}
const remaining = lastMs + intervalMs - now;
if (remaining <= 0) {
return Math.floor(Math.random() * OVERDUE_FIRE_JITTER_MS);
}
return Math.min(remaining, intervalMs);
this.applyTimerRegistration(agentId, baseIntervalMs, multiplier, usingDefaultInterval);
}
private applyTimerRegistration(
@@ -2167,67 +2103,28 @@ export class HeartbeatTriggerScheduler {
baseIntervalMs: number,
multiplier: number,
usingDefaultInterval: boolean,
lastHeartbeatAt: string | null,
): void {
const effectiveIntervalMs = Math.max(1000, Math.round(baseIntervalMs * multiplier));
const initialDelayMs = HeartbeatTriggerScheduler.computeInitialDelayMs(
effectiveIntervalMs,
lastHeartbeatAt,
);
this.clearAgentTimer(agentId);
const armSteadyInterval = () => {
// The setTimeout fired and was consumed; replace it with the long-lived
// setInterval that drives every subsequent tick. Use the same
// effectiveIntervalMs so the cadence remains correct.
const intervalHandle = setInterval(() => {
void this.onTimerTick(agentId, effectiveIntervalMs);
}, effectiveIntervalMs);
this.timers.set(agentId, {
intervalMs: effectiveIntervalMs,
kind: "interval",
handle: intervalHandle,
});
};
const handle = setInterval(() => {
void this.onTimerTick(agentId, effectiveIntervalMs);
}, effectiveIntervalMs);
if (initialDelayMs >= effectiveIntervalMs) {
// No phase-shift needed (agent has never ticked, or the saved
// lastHeartbeatAt is somehow in the future). Skip the timeout hop and
// arm the steady-state interval directly so the behavior matches the
// pre-phase-alignment scheduler.
armSteadyInterval();
} else {
const timeoutHandle = setTimeout(() => {
// Fire the overdue/phase-aligned tick first, then transition to the
// steady cadence. The tick fires regardless of whether the steady
// interval install succeeds, so a missed tick can never silently
// happen here.
void this.onTimerTick(agentId, effectiveIntervalMs);
armSteadyInterval();
}, initialDelayMs);
this.timers.set(agentId, {
intervalMs: effectiveIntervalMs,
kind: "timeout",
handle: timeoutHandle as unknown as ReturnType<typeof setInterval>,
});
}
const phaseSuffix = lastHeartbeatAt
? `, first fire in ${initialDelayMs}ms (phase-aligned to lastHeartbeatAt)`
: "";
this.timers.set(agentId, { intervalMs: effectiveIntervalMs, handle });
if (multiplier !== 1) {
heartbeatLog.log(
`Registered timer for ${agentId} (every ${baseIntervalMs}ms, multiplier ${multiplier} → ${effectiveIntervalMs}ms effective${phaseSuffix})`,
`Registered timer for ${agentId} (every ${baseIntervalMs}ms, multiplier ${multiplier} → ${effectiveIntervalMs}ms effective)`,
);
return;
}
heartbeatLog.log(
usingDefaultInterval
? `Registered timer for ${agentId} (every ${effectiveIntervalMs}ms, default interval${phaseSuffix})`
: `Registered timer for ${agentId} (every ${effectiveIntervalMs}ms${phaseSuffix})`,
? `Registered timer for ${agentId} (every ${effectiveIntervalMs}ms, default interval)`
: `Registered timer for ${agentId} (every ${effectiveIntervalMs}ms)`,
);
}
@@ -2236,14 +2133,7 @@ export class HeartbeatTriggerScheduler {
if (!timer) {
return;
}
// Both kinds share the same opaque handle type at runtime, but we route
// through the matching clear function for clarity and to satisfy strict
// type narrowing on platforms that distinguish the two.
if (timer.kind === "timeout") {
clearTimeout(timer.handle as unknown as ReturnType<typeof setTimeout>);
} else {
clearInterval(timer.handle);
}
clearInterval(timer.handle);
this.timers.delete(agentId);
}
@@ -2401,9 +2291,7 @@ export class HeartbeatTriggerScheduler {
return;
}
this.registerAgent(agent.id, this.getAgentTimerConfig(agent), {
lastHeartbeatAt: agent.lastHeartbeatAt,
});
this.registerAgent(agent.id, this.getAgentTimerConfig(agent));
heartbeatLog.log(`Timer armed for ${agent.id} (${reason})`);
}
@@ -2419,9 +2307,7 @@ export class HeartbeatTriggerScheduler {
return;
}
this.registerAgent(agent.id, this.getAgentTimerConfig(agent), {
lastHeartbeatAt: agent.lastHeartbeatAt,
});
this.registerAgent(agent.id, this.getAgentTimerConfig(agent));
heartbeatLog.log(`Timer refreshed for ${agent.id} (${reason})`);
}

6
packages/engine/src/executor.ts
View File

@@ -1103,8 +1103,12 @@ export class TaskExecutor {
}
if (latestTask.column === "in-progress") {
const originalExecutionStartedAt = latestTask.executionStartedAt;
await this.store.moveTask(taskId, "todo");
await this.store.updateTask(taskId, { worktree: worktreePath });
await this.store.updateTask(taskId, {
worktree: worktreePath,
executionStartedAt: originalExecutionStartedAt ?? null,
});
await this.store.moveTask(taskId, "in-progress");
return "bounced";
}

14
packages/engine/src/runtimes/in-process-runtime.ts
View File

@@ -563,15 +563,11 @@ export class InProcessRuntime
for (const agent of agents) {
if (!isTimerManagedAgent(agent)) continue;
const rc = agent.runtimeConfig;
this.triggerScheduler.registerAgent(
agent.id,
{
enabled: rc?.enabled as boolean | undefined,
heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,
maxConcurrentRuns: rc?.maxConcurrentRuns as number | undefined,
},
{ lastHeartbeatAt: agent.lastHeartbeatAt },
);
this.triggerScheduler.registerAgent(agent.id, {
enabled: rc?.enabled as boolean | undefined,
heartbeatIntervalMs: rc?.heartbeatIntervalMs as number | undefined,
maxConcurrentRuns: rc?.maxConcurrentRuns as number | undefined,
});
registeredCount++;
}
if (agents.length > 0) {