FN-7011: Exclude engine downtime from task timing
Exclude proven engine-process downtime from active task duration stats and badges. - Persist a throttled engineLastActiveAt heartbeat while the scheduler is running and unpaused. - Reconcile in-progress execution segment anchors on startup when downtime exceeds the poll threshold. - Emit run-audit events for shifted and no-action downtime reconciliation outcomes. - Cover core timing, dashboard badge behavior, scheduler heartbeat throttling, and self-healing audit metadata. - Add a patch changeset for the published CLI package. Files changed: .changeset/fn-7011-engine-downtime-timing.md | 7 +++ AGENTS.md | 1 + .../src/__tests__/store-execution-timing.test.ts | 51 ++++++++++++++++++++++ packages/core/src/settings-schema.ts | 1 + packages/core/src/store.ts | 28 ++++++++++++ packages/core/src/types.ts | 5 +++ .../app/utils/__tests__/taskTiming.test.ts | 16 +++++++ .../__tests__/scheduler-workflow-cutover.test.ts | 29 +++++++++++- packages/engine/src/__tests__/self-healing.test.ts | 23 ++++++++++ packages/engine/src/run-audit.ts | 4 ++ packages/engine/src/scheduler.ts | 11 +++++ packages/engine/src/self-healing.ts | 21 +++++++++ 12 files changed, 196 insertions(+), 1 deletion(-) Fusion-Task-Id: FN-7011 Fusion-Task-Lineage: 44225b0a-0fa1-40be-812d-33fd5e446944
This commit is contained in:
gsxdsm
7
.changeset/fn-7011-engine-downtime-timing.md
Normal file
7
.changeset/fn-7011-engine-downtime-timing.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Exclude engine-down time from task duration badge and stats.
|
||||
category: fix
|
||||
dev: Adds engineLastActiveAt heartbeat and startup reconcile-engine-downtime-active-timing recovery.
|
||||
@@ -219,6 +219,7 @@ Scoped exception (FN-5819): shared-branch-group members (`branchContext.assignme
|
||||
|
||||
### Run Audit
|
||||
|
||||
- FN-7011: self-healing emits `task:reconcile-engine-downtime-active-timing` when startup recovery shifts active task segment anchors to exclude proven engine-process downtime, and `task:reconcile-engine-downtime-active-timing-no-action` when no active task qualifies.
|
||||
- FN-5419: git run-audit now includes `pull:fast-forward` and `stash:pop-conflict`; dashboard git surfaces now include the extended `POST /api/git/pull` integration-worktree path plus companion `POST /api/git/stash-resolve`, `POST /api/git/stash-drop`, and `POST /api/git/stash-apply` routes.
|
||||
- FN-6292: self-healing emits `task:reconcile-dependency-blocking-lease` when it rebounds an in-progress holder whose stale file-scope lease blocks an unmet dependency, and `task:reconcile-dependency-blocking-lease-no-action` when triple-proof blocks that backward move.
|
||||
- FN-6736: self-healing emits `task:reclaim-phantom-executor-binding` when it proves an in-memory executor-active binding is stale, clears the binding, and requeues the in-progress task with worktree/progress preserved.
|
||||
|
||||
@@ -140,4 +140,55 @@ describe("TaskStore execution timing semantics", () => {
|
||||
const reloaded = await store.getTask(task.id);
|
||||
expect(reloaded?.columnDwellMs).toEqual(final.columnDwellMs);
|
||||
});
|
||||
|
||||
it("reconciles engine-down time without changing firstExecutionAt or accrued active time", async () => {
|
||||
/*
|
||||
FNXC:TaskTiming 2026-06-25-00:00:
|
||||
Surface Enumeration: proves the core downtime helper, completion accrual, multi-task shifts, missing/future/below-threshold no-ops, after-heartbeat task exclusion, repeated restart idempotence, and legacy missing executionStartedAt tolerance.
|
||||
*/
|
||||
vi.useFakeTimers();
|
||||
const t0 = new Date("2026-06-25T00:00:00.000Z");
|
||||
vi.setSystemTime(t0);
|
||||
|
||||
const task = await store.createTask({ description: "engine downtime symptom" });
|
||||
await store.moveTask(task.id, "todo");
|
||||
const running = await store.moveTask(task.id, "in-progress");
|
||||
const second = await store.createTask({ description: "second active" });
|
||||
await store.moveTask(second.id, "todo");
|
||||
await store.moveTask(second.id, "in-progress");
|
||||
const legacy = await store.createTask({ description: "legacy active" });
|
||||
await store.moveTask(legacy.id, "todo");
|
||||
await store.moveTask(legacy.id, "in-progress");
|
||||
await store.updateTask(legacy.id, { executionStartedAt: null });
|
||||
|
||||
await store.updateSettings({ engineLastActiveAt: new Date(t0.getTime() + 5 * 60_000).toISOString(), pollIntervalMs: 15_000 });
|
||||
vi.setSystemTime(new Date(t0.getTime() + 65 * 60_000));
|
||||
const result = await store.reconcileActiveTimingForEngineDowntime();
|
||||
|
||||
expect(result.downtimeMs).toBe(60 * 60_000);
|
||||
expect(result.shiftedTaskIds.sort()).toEqual([task.id, second.id].sort());
|
||||
const shifted = await store.getTask(task.id);
|
||||
expect(shifted?.executionStartedAt).toBe(new Date(t0.getTime() + 60 * 60_000).toISOString());
|
||||
expect(shifted?.firstExecutionAt).toBe(running.firstExecutionAt);
|
||||
expect(shifted?.cumulativeActiveMs).toBe(0);
|
||||
|
||||
vi.setSystemTime(new Date(t0.getTime() + 67 * 60_000));
|
||||
const done = await store.moveTask(task.id, "done");
|
||||
expect(done.cumulativeActiveMs).toBe(7 * 60_000);
|
||||
|
||||
await store.updateSettings({ engineLastActiveAt: undefined });
|
||||
expect((await store.reconcileActiveTimingForEngineDowntime()).shiftedTaskIds).toEqual([]);
|
||||
await store.updateSettings({ engineLastActiveAt: new Date(t0.getTime() + 90 * 60_000).toISOString() });
|
||||
expect((await store.reconcileActiveTimingForEngineDowntime()).shiftedTaskIds).toEqual([]);
|
||||
await store.updateSettings({ engineLastActiveAt: new Date(t0.getTime() + 66 * 60_000).toISOString() });
|
||||
expect((await store.reconcileActiveTimingForEngineDowntime()).shiftedTaskIds).toEqual([]);
|
||||
|
||||
await store.moveTask(second.id, "done");
|
||||
await store.updateSettings({ engineLastActiveAt: new Date(t0.getTime() + 65 * 60_000).toISOString() });
|
||||
const afterHeartbeat = await store.createTask({ description: "started after heartbeat" });
|
||||
await store.moveTask(afterHeartbeat.id, "todo");
|
||||
await store.moveTask(afterHeartbeat.id, "in-progress");
|
||||
vi.setSystemTime(new Date(t0.getTime() + 70 * 60_000));
|
||||
expect((await store.reconcileActiveTimingForEngineDowntime()).shiftedTaskIds).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -267,6 +267,7 @@ export const DEFAULT_PROJECT_SETTINGS = {
|
||||
approvedWorkflowCliCommands: undefined,
|
||||
approvedCliAutonomyAdapters: undefined,
|
||||
enginePaused: false,
|
||||
engineLastActiveAt: undefined,
|
||||
maxConcurrent: 2,
|
||||
maxTriageConcurrent: 2,
|
||||
globalMaxConcurrent: 4,
|
||||
|
||||
@@ -5501,6 +5501,34 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
|
||||
);
|
||||
}
|
||||
|
||||
async reconcileActiveTimingForEngineDowntime(now: Date = new Date()): Promise<{ shiftedTaskIds: string[]; downtimeMs: number }> {
|
||||
const settings = await this.getSettings();
|
||||
const heartbeatMs = Date.parse(settings.engineLastActiveAt ?? "");
|
||||
const nowMs = now.getTime();
|
||||
const thresholdMs = Math.max((settings.pollIntervalMs ?? 15_000) * 2, 60_000);
|
||||
const downtimeMs = Number.isFinite(heartbeatMs) && Number.isFinite(nowMs) ? nowMs - heartbeatMs : 0;
|
||||
if (!settings.engineLastActiveAt || downtimeMs <= thresholdMs) {
|
||||
return { shiftedTaskIds: [], downtimeMs: Math.max(0, downtimeMs) };
|
||||
}
|
||||
|
||||
const shiftedTaskIds: string[] = [];
|
||||
const tasks = await this.listTasks({ column: "in-progress", includeArchived: false, slim: true });
|
||||
for (const task of tasks) {
|
||||
const startedMs = Date.parse(task.executionStartedAt ?? "");
|
||||
if (!Number.isFinite(startedMs) || startedMs > heartbeatMs) continue;
|
||||
const shiftedStartedMs = Math.min(nowMs, startedMs + downtimeMs);
|
||||
if (shiftedStartedMs <= startedMs) continue;
|
||||
/*
|
||||
FNXC:TaskTiming 2026-06-25-00:00:
|
||||
Engine-process downtime is proven only by a stale engineLastActiveAt heartbeat. Advance the current active segment anchor, but preserve firstExecutionAt and cumulativeActiveMs so wall-clock history and already-accrued active work remain intact.
|
||||
*/
|
||||
await this.updateTask(task.id, { executionStartedAt: new Date(shiftedStartedMs).toISOString() });
|
||||
shiftedTaskIds.push(task.id);
|
||||
}
|
||||
|
||||
return { shiftedTaskIds, downtimeMs };
|
||||
}
|
||||
|
||||
// --- Unified PR entity (PR-lifecycle-as-workflow-nodes, U1) ---
|
||||
|
||||
private rowToPrEntity(row: PrEntityRow): PrEntity {
|
||||
|
||||
@@ -3728,6 +3728,11 @@ export interface ProjectSettings {
|
||||
* effect when {@link globalPause} is also true (hard stop already
|
||||
* covers everything). */
|
||||
enginePaused?: boolean;
|
||||
/**
|
||||
* FNXC:TaskTiming 2026-06-25-00:00:
|
||||
* Records the last time the engine process proved it was alive so startup recovery can exclude process-down wall-clock time from active task duration without changing firstExecutionAt.
|
||||
*/
|
||||
engineLastActiveAt?: string;
|
||||
/** Maximum number of concurrent AI agents across all activity types
|
||||
* (triage specification, task execution, and merge operations). */
|
||||
maxConcurrent: number;
|
||||
|
||||
@@ -31,6 +31,22 @@ describe("taskTiming helpers", () => {
|
||||
expect(runtime).toBeNull();
|
||||
});
|
||||
|
||||
it("uses shifted executionStartedAt so the active badge excludes engine-down time", () => {
|
||||
const t0 = Date.parse("2026-06-25T00:00:00.000Z");
|
||||
const runtime = getActiveRuntimeMs(
|
||||
{
|
||||
column: "in-progress",
|
||||
cumulativeActiveMs: undefined,
|
||||
executionStartedAt: new Date(t0 + 60 * 60_000).toISOString(),
|
||||
columnMovedAt: new Date(t0).toISOString(),
|
||||
},
|
||||
t0 + 65 * 60_000,
|
||||
);
|
||||
|
||||
expect(runtime).toBe(5 * 60_000);
|
||||
expect(getActiveRuntimeMs({ column: "in-progress", cumulativeActiveMs: undefined, executionStartedAt: undefined, columnMovedAt: undefined }, t0)).toBeNull();
|
||||
});
|
||||
|
||||
it("returns wall-clock runtime since first execution", () => {
|
||||
const wallClock = getWallClockSinceFirstExecutionMs(
|
||||
"2026-05-15T08:42:00.000Z",
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { makeTransitionRejection, TransitionRejectionError, type Task, type TaskStore } from "@fusion/core";
|
||||
import { existsSync } from "node:fs";
|
||||
import { readFile } from "node:fs/promises";
|
||||
@@ -42,6 +42,7 @@ function storeWith(tasks: Task[], settings: Record<string, unknown> = {}): TaskS
|
||||
experimentalFeatures: { workflowColumns: false },
|
||||
...settings,
|
||||
})),
|
||||
updateSettings: vi.fn(async (patch: Record<string, unknown>) => ({ ...settings, ...patch })),
|
||||
updateTask: vi.fn(async (id: string, patch: Partial<Task>) => {
|
||||
const current = byId.get(id);
|
||||
if (current) Object.assign(current, patch);
|
||||
@@ -73,6 +74,32 @@ describe("Scheduler workflow cutover", () => {
|
||||
vi.mocked(readFile).mockResolvedValue("# Task\nBody");
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("writes the engine active heartbeat at most once per poll interval and skips while paused", async () => {
|
||||
vi.useFakeTimers();
|
||||
vi.setSystemTime(new Date("2026-06-25T00:00:00.000Z"));
|
||||
const store = storeWith([], { pollIntervalMs: 15_000 });
|
||||
const scheduler = new Scheduler(store, { onSchedule: vi.fn() });
|
||||
(scheduler as unknown as { running: boolean }).running = true;
|
||||
|
||||
await scheduler.schedule();
|
||||
await scheduler.schedule();
|
||||
expect(store.updateSettings).toHaveBeenCalledTimes(1);
|
||||
expect(store.updateSettings).toHaveBeenCalledWith({ engineLastActiveAt: "2026-06-25T00:00:00.000Z" });
|
||||
|
||||
vi.setSystemTime(new Date("2026-06-25T00:00:15.000Z"));
|
||||
await scheduler.schedule();
|
||||
expect(store.updateSettings).toHaveBeenCalledTimes(2);
|
||||
|
||||
vi.mocked(store.getSettings).mockResolvedValue({ maxConcurrent: 2, maxWorktrees: 4, pollIntervalMs: 15_000, enginePaused: true } as any);
|
||||
vi.setSystemTime(new Date("2026-06-25T00:00:30.000Z"));
|
||||
await scheduler.schedule();
|
||||
expect(store.updateSettings).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("uses the workflow sweep for todo pickup even when stale workflowColumns=false is persisted", async () => {
|
||||
const ready = task({ id: "FN-100" });
|
||||
const store = storeWith([ready]);
|
||||
|
||||
@@ -180,6 +180,7 @@ function createMockStore(overrides: Record<string, unknown> = {}): TaskStore & E
|
||||
archiveTaskAndCleanup: vi.fn().mockResolvedValue({} as Task),
|
||||
walCheckpoint: vi.fn().mockReturnValue({ busy: 0, log: 5, checkpointed: 5 }),
|
||||
listTasks: vi.fn().mockResolvedValue([]),
|
||||
reconcileActiveTimingForEngineDowntime: vi.fn().mockResolvedValue({ shiftedTaskIds: [], downtimeMs: 0 }),
|
||||
parseFileScopeFromPrompt: vi.fn().mockResolvedValue([]),
|
||||
getCompletionHandoffAcceptedMarker: vi.fn().mockReturnValue(null),
|
||||
createTask: vi.fn().mockResolvedValue({ id: "FN-RESCUE", lineageId: "lin-rescue" }),
|
||||
@@ -820,6 +821,7 @@ describe("SelfHealingManager", () => {
|
||||
const surfaceInReviewStalled = vi.spyOn(manager, "surfaceInReviewStalled").mockResolvedValue(1);
|
||||
const surfaceStalePausedReviews = vi.spyOn(manager, "surfaceStalePausedReviews").mockResolvedValue(1);
|
||||
const surfaceStalePausedTodos = vi.spyOn(manager, "surfaceStalePausedTodos").mockResolvedValue(1);
|
||||
const reconcileEngineDowntimeActiveTiming = vi.spyOn(manager, "reconcileEngineDowntimeActiveTiming").mockResolvedValue({ shiftedTaskIds: [], downtimeMs: 0 });
|
||||
|
||||
await manager.runStartupRecovery();
|
||||
|
||||
@@ -833,6 +835,7 @@ describe("SelfHealingManager", () => {
|
||||
expect(recoverOrphanedAgents).toHaveBeenCalledTimes(1);
|
||||
expect(recoverAgentsRunningOnInactiveTasks).toHaveBeenCalledTimes(1);
|
||||
expect(clearStaleBlockedBy).toHaveBeenCalledTimes(1);
|
||||
expect(reconcileEngineDowntimeActiveTiming).toHaveBeenCalledTimes(1);
|
||||
expect(surfaceInReviewStalls).toHaveBeenCalledTimes(1);
|
||||
expect(surfaceInReviewStalled).toHaveBeenCalledTimes(1);
|
||||
expect(surfaceStalePausedReviews).toHaveBeenCalledTimes(1);
|
||||
@@ -855,6 +858,26 @@ describe("SelfHealingManager", () => {
|
||||
expect(store.logEntry).toHaveBeenCalledWith("A", expect.stringContaining("Auto-recovered (FN-5488): cleared stale blockedBy"));
|
||||
});
|
||||
|
||||
it("runStartupRecovery emits engine downtime timing audit metadata", async () => {
|
||||
const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
|
||||
store = createMockStore({
|
||||
getSettings: vi.fn().mockResolvedValue({ globalPause: false, enginePaused: false }),
|
||||
reconcileActiveTimingForEngineDowntime: vi.fn().mockResolvedValue({ shiftedTaskIds: ["FN-7011"], downtimeMs: 3_600_000 }),
|
||||
recordRunAuditEvent,
|
||||
});
|
||||
manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project" });
|
||||
|
||||
await manager.reconcileEngineDowntimeActiveTiming();
|
||||
|
||||
expect(store.reconcileActiveTimingForEngineDowntime).toHaveBeenCalledTimes(1);
|
||||
expect(recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
|
||||
domain: "database",
|
||||
mutationType: "task:reconcile-engine-downtime-active-timing",
|
||||
target: "global",
|
||||
metadata: expect.objectContaining({ shiftedTaskIds: ["FN-7011"], downtimeMs: 3_600_000 }),
|
||||
}));
|
||||
});
|
||||
|
||||
it("runStartupRecovery skips while enginePaused is active", async () => {
|
||||
vi.mocked(store.getSettings).mockResolvedValue({
|
||||
globalPause: false,
|
||||
|
||||
@@ -520,6 +520,10 @@ export type DatabaseMutationType =
|
||||
| "task:resume-limbo-escalated"
|
||||
/** Metadata: { taskId, executionAgeMs, graceMs, staleBindingAgeFloorMs, checkedOutBy, agentPresent, lastActivityMs, hasRecentRunAudit, worktree, branch, worktreeExists, signalReason } */
|
||||
| "task:reclaim-phantom-executor-binding"
|
||||
/** Metadata: { shiftedTaskIds: string[], downtimeMs, reason } */
|
||||
| "task:reconcile-engine-downtime-active-timing"
|
||||
/** Metadata: { shiftedTaskIds: [], downtimeMs, reason } */
|
||||
| "task:reconcile-engine-downtime-active-timing-no-action"
|
||||
/* FNXC:Workspace 2026-06-22-09:30 (Phase D U1) — workspace-mode self-healing run-audit events. */
|
||||
/** Metadata: { taskId, landedRepos: string[], unlandedRepos: string[], failedRepos: string[], action: "re-enqueue" | "park-failed", reason } */
|
||||
| "task:reconcile-workspace-partial-land"
|
||||
|
||||
@@ -574,6 +574,7 @@ export class Scheduler {
|
||||
private lastStaleTaskReportAt = 0;
|
||||
private lastBacklogPressureReportAt = 0;
|
||||
private lastUnlinkedMissionsAdvisoryReportAt = 0;
|
||||
private lastHeartbeatWriteMs = 0;
|
||||
private idleSemaphoreLeakCandidateSince: number | null = null;
|
||||
private readonly lastHighOverlapFanoutWarningKey = new Map<string, string>();
|
||||
|
||||
@@ -1285,6 +1286,16 @@ export class Scheduler {
|
||||
}
|
||||
this.wasEnginePaused = false;
|
||||
|
||||
const heartbeatIntervalMs = Math.max(1, settings.pollIntervalMs ?? 15_000);
|
||||
if (Date.now() - this.lastHeartbeatWriteMs >= heartbeatIntervalMs) {
|
||||
/*
|
||||
FNXC:TaskTiming 2026-06-25-00:00:
|
||||
Persist a throttled engineLastActiveAt heartbeat only while the engine is unpaused so process-down wall-clock can be excluded from in-progress task active time after restart without per-tick DB churn.
|
||||
*/
|
||||
await this.store.updateSettings({ engineLastActiveAt: new Date().toISOString() });
|
||||
this.lastHeartbeatWriteMs = Date.now();
|
||||
}
|
||||
|
||||
// ── U6: hold/release sweep ─────────────────────────────────────────────
|
||||
/*
|
||||
FNXC:WorkflowScheduling 2026-06-23-10:32:
|
||||
|
||||
@@ -1237,6 +1237,7 @@ export class SelfHealingManager {
|
||||
{ name: "reconcile-self-defeating-deps", fn: () => this.reconcileSelfDefeatingDependencies().then(() => undefined) },
|
||||
{ name: "reconcile-dependency-blocking-leases", fn: () => this.reconcileDependencyBlockingLeases().then(() => undefined) },
|
||||
{ name: "reconcile-in-review-unmet-dependencies", fn: () => this.reconcileInReviewUnmetDependencies().then(() => undefined) },
|
||||
{ name: "reconcile-engine-downtime-active-timing", fn: () => this.reconcileEngineDowntimeActiveTiming().then(() => undefined) },
|
||||
{ name: "reconcile-dependency-cycles", fn: () => this.reconcileDependencyCycles().then(() => undefined) },
|
||||
{ name: "reclaim-pr-conflicts", fn: () => this.reclaimPrConflicts().then(() => undefined) },
|
||||
{ name: "reclaim-self-owned-branch-conflicts", fn: () => this.reclaimSelfOwnedBranchConflicts().then(() => undefined) },
|
||||
@@ -1264,6 +1265,26 @@ export class SelfHealingManager {
|
||||
}
|
||||
}
|
||||
|
||||
async reconcileEngineDowntimeActiveTiming(): Promise<{ shiftedTaskIds: string[]; downtimeMs: number }> {
|
||||
const result = await this.store.reconcileActiveTimingForEngineDowntime();
|
||||
const shifted = result.shiftedTaskIds.length > 0;
|
||||
const auditor = createRunAuditor(this.store, {
|
||||
runId: generateSyntheticRunId("reconcile-engine-downtime-active-timing", "global"),
|
||||
agentId: "system:self-healing",
|
||||
phase: "reconcile-engine-downtime-active-timing",
|
||||
});
|
||||
await auditor.database({
|
||||
type: (shifted ? "task:reconcile-engine-downtime-active-timing" : "task:reconcile-engine-downtime-active-timing-no-action") as DatabaseMutationType,
|
||||
target: "global",
|
||||
metadata: {
|
||||
shiftedTaskIds: result.shiftedTaskIds,
|
||||
downtimeMs: result.downtimeMs,
|
||||
reason: shifted ? "shifted-active-segments" : "no-qualifying-active-segments",
|
||||
},
|
||||
});
|
||||
return result;
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
// Remove settings listener
|
||||
if (this.settingsListener) {
|
||||
|
||||
Reference in New Issue
Block a user