FN-7975: exclude engine-paused wall-clock from task active timing
Reconcile active task segment anchors on full Global/Engine unpause so stopped-engine wall-clock does not inflate execution time, reusing the FN-7011 downtime path with a transition-captured heartbeat. - Pass optional engineLastActiveAtOverride into reconcileActiveTimingForEngineDowntime so unpause callers freeze the stopped-window proof against racing scheduler heartbeats - Await downtime reconciliation in resumeAfterUnpauseAndSweepInReview before resuming agentic work or sweeping in-review tasks - Fold Global/Engine unpause into the unified pause-lifecycle listener (single reconcile when both clear together; no-op while either pause remains) - Soft-fail reconcile errors so unpause resume still proceeds - Add store and project-engine coverage for override, await-before-resume, dual-source clear, and fail-soft paths; document FN-7975 in AGENTS.md run-audit notes - Add patch changeset for the operator-facing timing fix Files changed: .changeset/fn-7975-engine-pause-active-timing.md | 7 ++ AGENTS.md | 2 +- .../core/src/__tests__/store-active-timing.test.ts | 86 +++++++++++++ packages/core/src/store.ts | 23 ++-- .../project-engine-unpause-active-timing.test.ts | 94 ++++++++++++++ .../engine/src/__tests__/project-engine.test.ts | 139 +++++++++++++++++++++ packages/engine/src/project-engine.ts | 64 +++++----- packages/engine/src/self-healing.ts | 6 +- 8 files changed, 378 insertions(+), 43 deletions(-) Fusion-Task-Id: FN-7975 Fusion-Task-Lineage: 84a46e6f-92bf-452a-ab67-c25ba85cbffb Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
7
.changeset/fn-7975-engine-pause-active-timing.md
Normal file
7
.changeset/fn-7975-engine-pause-active-timing.md
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
---
|
||||||
|
"@runfusion/fusion": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
summary: Exclude long engine pauses from in-progress task execution time.
|
||||||
|
category: fix
|
||||||
|
dev: Reuses FN-7011 active-timing reconciliation on full Global/Engine unpause. Engine-pause time is excluded even if in-flight agents continue, matching restart behavior.
|
||||||
@@ -253,7 +253,7 @@ Scoped exception (FN-5819): shared-branch-group members (`branchContext.assignme
|
|||||||
- FN-7802: self-healing emits `task:reconcile-missing-worktree-merge-active` when it proves an `in-review` merge-active task (`merging`/`merging-pr`/`merging-fix`) is stranded by an unusable-worktree session-start failure, clears stale `worktree`/`branch`/`sessionFile`, resets the worktree-session retry budget, increments `recoveryRetryCount` as the bounded stale-metadata clear counter, and requeues to `todo`; it emits `task:reconcile-missing-worktree-merge-active-no-action` when `autoMerge:false`, workspace-task ownership, or triple-proof blocks the backward move.
|
- FN-7802: self-healing emits `task:reconcile-missing-worktree-merge-active` when it proves an `in-review` merge-active task (`merging`/`merging-pr`/`merging-fix`) is stranded by an unusable-worktree session-start failure, clears stale `worktree`/`branch`/`sessionFile`, resets the worktree-session retry budget, increments `recoveryRetryCount` as the bounded stale-metadata clear counter, and requeues to `todo`; it emits `task:reconcile-missing-worktree-merge-active-no-action` when `autoMerge:false`, workspace-task ownership, or triple-proof blocks the backward move.
|
||||||
- FN-7863: executor emits `task:execution-dispatch-loop-terminalized` when an execute-node self-requeue loop reaches `MAX_EXECUTE_REQUEUE_LOOP_CYCLES` with an unchanged progress signature; metadata stays ids/counts/outcomes-only (`taskId`, `cycleCount`, `maxCycles`, `progressSignature`, `failureValue`) and the task is visibly failed with `EXECUTION_DISPATCH_LOOP_EXHAUSTED:` while preserving worktree/branch/step progress.
|
- FN-7863: executor emits `task:execution-dispatch-loop-terminalized` when an execute-node self-requeue loop reaches `MAX_EXECUTE_REQUEUE_LOOP_CYCLES` with an unchanged progress signature; metadata stays ids/counts/outcomes-only (`taskId`, `cycleCount`, `maxCycles`, `progressSignature`, `failureValue`) and the task is visibly failed with `EXECUTION_DISPATCH_LOOP_EXHAUSTED:` while preserving worktree/branch/step progress.
|
||||||
- FN-7926: executor emits `task:completed-blocked-parked` when completed implementation work is held by a live `getTaskCompletionBlocker()` reason instead of re-entering the execute self-requeue loop; self-healing emits `task:completed-blocked-advanced` when the blocker clears and the parked work advances to review. Metadata stays ids/outcomes-only (`taskId`, blocker/source/prior column/status).
|
- FN-7926: executor emits `task:completed-blocked-parked` when completed implementation work is held by a live `getTaskCompletionBlocker()` reason instead of re-entering the execute self-requeue loop; self-healing emits `task:completed-blocked-advanced` when the blocker clears and the parked work advances to review. Metadata stays ids/outcomes-only (`taskId`, blocker/source/prior column/status).
|
||||||
- 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-7011/FN-7975: self-healing emits `task:reconcile-engine-downtime-active-timing` when startup recovery or a full Global/Engine unpause shifts active task segment anchors to exclude proven stopped-engine wall-clock, 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-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-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.
|
- 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.
|
||||||
|
|||||||
86
packages/core/src/__tests__/store-active-timing.test.ts
Normal file
86
packages/core/src/__tests__/store-active-timing.test.ts
Normal file
@@ -0,0 +1,86 @@
|
|||||||
|
import { describe, expect, it, vi } from "vitest";
|
||||||
|
import { TaskStore } from "../store.js";
|
||||||
|
|
||||||
|
const now = new Date("2026-07-15T12:00:00.000Z");
|
||||||
|
const staleHeartbeat = "2026-07-15T11:58:00.000Z";
|
||||||
|
const startedBeforeHeartbeat = "2026-07-15T11:50:00.000Z";
|
||||||
|
|
||||||
|
type TimingTask = {
|
||||||
|
id: string;
|
||||||
|
executionStartedAt?: string;
|
||||||
|
};
|
||||||
|
|
||||||
|
function createStoreDouble(settings: Record<string, unknown>, tasks: TimingTask[]) {
|
||||||
|
const updateTask = vi.fn(async (id: string, patch: { executionStartedAt: string }) => {
|
||||||
|
const task = tasks.find((candidate) => candidate.id === id);
|
||||||
|
if (task) task.executionStartedAt = patch.executionStartedAt;
|
||||||
|
});
|
||||||
|
return {
|
||||||
|
getSettings: vi.fn(async () => settings),
|
||||||
|
listTasks: vi.fn(async () => tasks),
|
||||||
|
updateTask,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
async function reconcile(
|
||||||
|
store: ReturnType<typeof createStoreDouble>,
|
||||||
|
opts?: { engineLastActiveAtOverride?: string },
|
||||||
|
) {
|
||||||
|
return TaskStore.prototype.reconcileActiveTimingForEngineDowntime.call(store as never, now, opts);
|
||||||
|
}
|
||||||
|
|
||||||
|
describe("TaskStore.reconcileActiveTimingForEngineDowntime", () => {
|
||||||
|
it("uses a stale captured override despite a fresh settings heartbeat and shifts exactly once by downtime", async () => {
|
||||||
|
const tasks = [{ id: "FN-active", executionStartedAt: startedBeforeHeartbeat }];
|
||||||
|
const store = createStoreDouble({ pollIntervalMs: 15_000, engineLastActiveAt: now.toISOString() }, tasks);
|
||||||
|
|
||||||
|
const result = await reconcile(store, { engineLastActiveAtOverride: staleHeartbeat });
|
||||||
|
|
||||||
|
expect(result).toEqual({ shiftedTaskIds: ["FN-active"], downtimeMs: 120_000 });
|
||||||
|
expect(tasks[0].executionStartedAt).toBe("2026-07-15T11:52:00.000Z");
|
||||||
|
// The subsequent in-progress exit accrues only the pre-pause eight-minute segment.
|
||||||
|
expect(now.getTime() - Date.parse(tasks[0].executionStartedAt!)).toBe(8 * 60_000);
|
||||||
|
expect(store.updateTask).toHaveBeenCalledTimes(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("treats missing or invalid supplied overrides as no-action without falling back to settings", async () => {
|
||||||
|
for (const engineLastActiveAtOverride of [undefined, "not-a-date"]) {
|
||||||
|
const tasks = [{ id: "FN-active", executionStartedAt: startedBeforeHeartbeat }];
|
||||||
|
const store = createStoreDouble({ pollIntervalMs: 15_000, engineLastActiveAt: staleHeartbeat }, tasks);
|
||||||
|
|
||||||
|
await expect(reconcile(store, { engineLastActiveAtOverride })).resolves.toEqual({
|
||||||
|
shiftedTaskIds: [],
|
||||||
|
downtimeMs: 0,
|
||||||
|
});
|
||||||
|
expect(tasks[0].executionStartedAt).toBe(startedBeforeHeartbeat);
|
||||||
|
expect(store.updateTask).not.toHaveBeenCalled();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it("keeps the no-options startup recovery fallback and ignores recent or absent settings heartbeats", async () => {
|
||||||
|
const staleTasks = [{ id: "FN-stale", executionStartedAt: startedBeforeHeartbeat }];
|
||||||
|
const staleStore = createStoreDouble({ pollIntervalMs: 15_000, engineLastActiveAt: staleHeartbeat }, staleTasks);
|
||||||
|
await expect(reconcile(staleStore)).resolves.toEqual({ shiftedTaskIds: ["FN-stale"], downtimeMs: 120_000 });
|
||||||
|
|
||||||
|
for (const engineLastActiveAt of [now.toISOString(), undefined]) {
|
||||||
|
const tasks = [{ id: "FN-active", executionStartedAt: startedBeforeHeartbeat }];
|
||||||
|
const store = createStoreDouble({ pollIntervalMs: 15_000, engineLastActiveAt }, tasks);
|
||||||
|
await expect(reconcile(store)).resolves.toEqual({ shiftedTaskIds: [], downtimeMs: 0 });
|
||||||
|
expect(store.updateTask).not.toHaveBeenCalled();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it("does not shift a task started after the stale heartbeat or downtime at the threshold", async () => {
|
||||||
|
const postHeartbeatTask = [{ id: "FN-after-pause", executionStartedAt: "2026-07-15T11:59:00.000Z" }];
|
||||||
|
const staleStore = createStoreDouble({ pollIntervalMs: 15_000, engineLastActiveAt: staleHeartbeat }, postHeartbeatTask);
|
||||||
|
await expect(reconcile(staleStore)).resolves.toEqual({ shiftedTaskIds: [], downtimeMs: 120_000 });
|
||||||
|
expect(staleStore.updateTask).not.toHaveBeenCalled();
|
||||||
|
|
||||||
|
const thresholdStore = createStoreDouble(
|
||||||
|
{ pollIntervalMs: 15_000, engineLastActiveAt: "2026-07-15T11:59:00.000Z" },
|
||||||
|
[{ id: "FN-at-threshold", executionStartedAt: startedBeforeHeartbeat }],
|
||||||
|
);
|
||||||
|
await expect(reconcile(thresholdStore)).resolves.toEqual({ shiftedTaskIds: [], downtimeMs: 60_000 });
|
||||||
|
expect(thresholdStore.updateTask).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -711,19 +711,28 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
|
|||||||
return atomicWriteTaskJsonWithAuditImpl(this, dir, task, auditInput);
|
return atomicWriteTaskJsonWithAuditImpl(this, dir, task, auditInput);
|
||||||
}
|
}
|
||||||
/*
|
/*
|
||||||
FNXC:TaskTiming 2026-06-25-00:00:
|
FNXC:TaskTiming 2026-07-15-00:00:
|
||||||
Engine-process downtime is proven by a stale engineLastActiveAt heartbeat.
|
Engine-process downtime is proven by a stale engineLastActiveAt heartbeat.
|
||||||
Advance the current active segment anchor, preserving firstExecutionAt and
|
Same-process unpause callers pass the transition-captured heartbeat so a
|
||||||
cumulativeActiveMs so wall-clock history and already-accrued active work
|
racing scheduler write cannot erase the stopped-window proof. No opts keeps
|
||||||
remain intact. Ported from origin/main FN-7011 during rebase.
|
FN-7011 startup recovery's settings fallback; supplied but invalid opts are
|
||||||
|
intentionally a no-action. Preserve the existing shift arithmetic: callers
|
||||||
|
own exactly-once dispatch because this store method does not deduplicate
|
||||||
|
repeated reconciles. Advance the current active segment anchor, preserving
|
||||||
|
firstExecutionAt and cumulativeActiveMs so wall-clock history and
|
||||||
|
already-accrued active work remain intact.
|
||||||
*/
|
*/
|
||||||
async reconcileActiveTimingForEngineDowntime(now: Date = new Date()): Promise<{ shiftedTaskIds: string[]; downtimeMs: number }> {
|
async reconcileActiveTimingForEngineDowntime(
|
||||||
|
now: Date = new Date(),
|
||||||
|
opts?: { engineLastActiveAtOverride?: string },
|
||||||
|
): Promise<{ shiftedTaskIds: string[]; downtimeMs: number }> {
|
||||||
const settings = await this.getSettings();
|
const settings = await this.getSettings();
|
||||||
const heartbeatMs = Date.parse(settings.engineLastActiveAt ?? "");
|
const heartbeatValue = opts === undefined ? settings.engineLastActiveAt : opts.engineLastActiveAtOverride;
|
||||||
|
const heartbeatMs = Date.parse(heartbeatValue ?? "");
|
||||||
const nowMs = now.getTime();
|
const nowMs = now.getTime();
|
||||||
const thresholdMs = Math.max((settings.pollIntervalMs ?? 15_000) * 2, 60_000);
|
const thresholdMs = Math.max((settings.pollIntervalMs ?? 15_000) * 2, 60_000);
|
||||||
const downtimeMs = Number.isFinite(heartbeatMs) && Number.isFinite(nowMs) ? nowMs - heartbeatMs : 0;
|
const downtimeMs = Number.isFinite(heartbeatMs) && Number.isFinite(nowMs) ? nowMs - heartbeatMs : 0;
|
||||||
if (!settings.engineLastActiveAt || downtimeMs <= thresholdMs) {
|
if (!heartbeatValue || !Number.isFinite(heartbeatMs) || downtimeMs <= thresholdMs) {
|
||||||
return { shiftedTaskIds: [], downtimeMs: Math.max(0, downtimeMs) };
|
return { shiftedTaskIds: [], downtimeMs: Math.max(0, downtimeMs) };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,94 @@
|
|||||||
|
import { describe, expect, it, vi } from "vitest";
|
||||||
|
import type { Settings, TaskStore } from "@fusion/core";
|
||||||
|
import { ProjectEngine } from "../project-engine.js";
|
||||||
|
import { SelfHealingManager } from "../self-healing.js";
|
||||||
|
|
||||||
|
const activeSettings = {
|
||||||
|
globalPause: false,
|
||||||
|
enginePaused: false,
|
||||||
|
autoMerge: true,
|
||||||
|
} as Settings;
|
||||||
|
|
||||||
|
type SettingsUpdatedHandler = (event: { settings: Settings; previous: Settings }) => void;
|
||||||
|
|
||||||
|
function wireUnpauseTimingHandler(manager: SelfHealingManager) {
|
||||||
|
const handlers: SettingsUpdatedHandler[] = [];
|
||||||
|
const store = {
|
||||||
|
on: vi.fn((event: string, handler: SettingsUpdatedHandler) => {
|
||||||
|
if (event === "settings:updated") handlers.push(handler);
|
||||||
|
}),
|
||||||
|
};
|
||||||
|
const engine = {
|
||||||
|
runtime: { stuckTaskDetector: { pause: vi.fn(), resume: vi.fn() } },
|
||||||
|
settingsHandlers: [],
|
||||||
|
getSelfHealingManager: () => manager,
|
||||||
|
resumeAfterUnpauseAndSweepInReview: (
|
||||||
|
ProjectEngine.prototype as unknown as { resumeAfterUnpauseAndSweepInReview: unknown }
|
||||||
|
).resumeAfterUnpauseAndSweepInReview,
|
||||||
|
} as unknown as ProjectEngine;
|
||||||
|
|
||||||
|
(
|
||||||
|
ProjectEngine.prototype as unknown as {
|
||||||
|
wireSettingsListeners(store: TaskStore): void;
|
||||||
|
}
|
||||||
|
).wireSettingsListeners.call(engine, store as unknown as TaskStore);
|
||||||
|
|
||||||
|
return handlers[0]!;
|
||||||
|
}
|
||||||
|
|
||||||
|
function createManager(
|
||||||
|
reconcileActiveTimingForEngineDowntime: ReturnType<typeof vi.fn>,
|
||||||
|
recordRunAuditEvent: ReturnType<typeof vi.fn>,
|
||||||
|
) {
|
||||||
|
return new SelfHealingManager({
|
||||||
|
reconcileActiveTimingForEngineDowntime,
|
||||||
|
recordRunAuditEvent,
|
||||||
|
} as unknown as TaskStore, { rootDir: "/tmp/fn-7975" });
|
||||||
|
}
|
||||||
|
|
||||||
|
describe("ProjectEngine unpause active timing reconciliation", () => {
|
||||||
|
it.each([
|
||||||
|
["globalPause", "Global unpause"],
|
||||||
|
["enginePaused", "Engine unpause"],
|
||||||
|
] as const)("routes %s resume through the downtime audit with shifted tasks", async (pauseKey) => {
|
||||||
|
const reconcileActiveTimingForEngineDowntime = vi.fn().mockResolvedValue({
|
||||||
|
shiftedTaskIds: ["FN-active"],
|
||||||
|
downtimeMs: 3_600_000,
|
||||||
|
});
|
||||||
|
const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
|
||||||
|
const handler = wireUnpauseTimingHandler(createManager(reconcileActiveTimingForEngineDowntime, recordRunAuditEvent));
|
||||||
|
const capturedHeartbeat = "2026-07-15T11:00:00.000Z";
|
||||||
|
|
||||||
|
handler({
|
||||||
|
settings: { ...activeSettings },
|
||||||
|
previous: { ...activeSettings, [pauseKey]: true, engineLastActiveAt: capturedHeartbeat },
|
||||||
|
});
|
||||||
|
await vi.waitFor(() => expect(recordRunAuditEvent).toHaveBeenCalledTimes(1));
|
||||||
|
|
||||||
|
expect(reconcileActiveTimingForEngineDowntime).toHaveBeenCalledWith(
|
||||||
|
expect.any(Date),
|
||||||
|
{ engineLastActiveAtOverride: capturedHeartbeat },
|
||||||
|
);
|
||||||
|
expect(recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
|
||||||
|
mutationType: "task:reconcile-engine-downtime-active-timing",
|
||||||
|
metadata: expect.objectContaining({ shiftedTaskIds: ["FN-active"], downtimeMs: 3_600_000 }),
|
||||||
|
}));
|
||||||
|
});
|
||||||
|
|
||||||
|
it("emits the no-action audit for a below-threshold unpause", async () => {
|
||||||
|
const reconcileActiveTimingForEngineDowntime = vi.fn().mockResolvedValue({ shiftedTaskIds: [], downtimeMs: 60_000 });
|
||||||
|
const recordRunAuditEvent = vi.fn().mockResolvedValue(undefined);
|
||||||
|
const handler = wireUnpauseTimingHandler(createManager(reconcileActiveTimingForEngineDowntime, recordRunAuditEvent));
|
||||||
|
|
||||||
|
handler({
|
||||||
|
settings: { ...activeSettings },
|
||||||
|
previous: { ...activeSettings, globalPause: true, engineLastActiveAt: "2026-07-15T11:59:00.000Z" },
|
||||||
|
});
|
||||||
|
await vi.waitFor(() => expect(recordRunAuditEvent).toHaveBeenCalledTimes(1));
|
||||||
|
|
||||||
|
expect(recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
|
||||||
|
mutationType: "task:reconcile-engine-downtime-active-timing-no-action",
|
||||||
|
metadata: expect.objectContaining({ shiftedTaskIds: [], downtimeMs: 60_000 }),
|
||||||
|
}));
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -21,6 +21,7 @@ const mocks = vi.hoisted(() => ({
|
|||||||
runtimeStart: vi.fn(async () => undefined),
|
runtimeStart: vi.fn(async () => undefined),
|
||||||
runtimeStop: vi.fn(async () => undefined),
|
runtimeStop: vi.fn(async () => undefined),
|
||||||
runtimeResumeAfterUnpause: vi.fn(async () => undefined),
|
runtimeResumeAfterUnpause: vi.fn(async () => undefined),
|
||||||
|
getSelfHealingManager: vi.fn(() => undefined),
|
||||||
runAiMerge: vi.fn(),
|
runAiMerge: vi.fn(),
|
||||||
landWorkspaceTask: vi.fn(),
|
landWorkspaceTask: vi.fn(),
|
||||||
execFile: vi.fn(),
|
execFile: vi.fn(),
|
||||||
@@ -210,6 +211,7 @@ vi.mock("../runtimes/in-process-runtime.js", () => ({
|
|||||||
getRoutineRunner: vi.fn(),
|
getRoutineRunner: vi.fn(),
|
||||||
getHeartbeatMonitor: vi.fn(),
|
getHeartbeatMonitor: vi.fn(),
|
||||||
getTriggerScheduler: vi.fn(),
|
getTriggerScheduler: vi.fn(),
|
||||||
|
getSelfHealingManager: mocks.getSelfHealingManager,
|
||||||
configurePrMonitoring: mocks.runtimeConfigurePrMonitoring,
|
configurePrMonitoring: mocks.runtimeConfigurePrMonitoring,
|
||||||
};
|
};
|
||||||
}),
|
}),
|
||||||
@@ -358,6 +360,8 @@ beforeEach(() => {
|
|||||||
mocks.deliverPostgresMigrationCompleteNotice.mockReset();
|
mocks.deliverPostgresMigrationCompleteNotice.mockReset();
|
||||||
mocks.deliverPostgresMigrationCompleteNotice.mockResolvedValue("no-migration");
|
mocks.deliverPostgresMigrationCompleteNotice.mockResolvedValue("no-migration");
|
||||||
mocks.runtimeResumeAfterUnpause.mockClear();
|
mocks.runtimeResumeAfterUnpause.mockClear();
|
||||||
|
mocks.getSelfHealingManager.mockReset();
|
||||||
|
mocks.getSelfHealingManager.mockReturnValue(undefined);
|
||||||
mocks.notifierStart.mockClear();
|
mocks.notifierStart.mockClear();
|
||||||
mocks.notifierStop.mockClear();
|
mocks.notifierStop.mockClear();
|
||||||
mocks.notifierNotifyGridlock.mockClear();
|
mocks.notifierNotifyGridlock.mockClear();
|
||||||
@@ -2759,6 +2763,141 @@ describe("ProjectEngine paused in-review auto-merge behavior", () => {
|
|||||||
await engine.stop();
|
await engine.stop();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("reconciles active timing exactly once when both pause sources clear together", async () => {
|
||||||
|
const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
|
||||||
|
mocks.currentStore = mockStore.store;
|
||||||
|
const reconcileEngineDowntimeActiveTiming = vi.fn(async () => ({ shiftedTaskIds: [], downtimeMs: 120_000 }));
|
||||||
|
mocks.getSelfHealingManager.mockReturnValue({ reconcileEngineDowntimeActiveTiming });
|
||||||
|
const engine = createEngine();
|
||||||
|
|
||||||
|
await engine.start();
|
||||||
|
await mockStore.emitSettingsUpdated(
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: false, enginePaused: false },
|
||||||
|
{
|
||||||
|
...baseSettings,
|
||||||
|
autoMerge: true,
|
||||||
|
globalPause: true,
|
||||||
|
enginePaused: true,
|
||||||
|
engineLastActiveAt: "2026-07-15T11:58:00.000Z",
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(reconcileEngineDowntimeActiveTiming).toHaveBeenCalledTimes(1);
|
||||||
|
expect(reconcileEngineDowntimeActiveTiming).toHaveBeenCalledWith({
|
||||||
|
engineLastActiveAtOverride: "2026-07-15T11:58:00.000Z",
|
||||||
|
});
|
||||||
|
await engine.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("waits for active-timing reconciliation before resuming agentic work", async () => {
|
||||||
|
const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
|
||||||
|
mocks.currentStore = mockStore.store;
|
||||||
|
let resolveReconcile!: () => void;
|
||||||
|
const reconcileEngineDowntimeActiveTiming = vi.fn(() => new Promise<{ shiftedTaskIds: string[]; downtimeMs: number }>((resolve) => {
|
||||||
|
resolveReconcile = () => resolve({ shiftedTaskIds: ["FN-active"], downtimeMs: 120_000 });
|
||||||
|
}));
|
||||||
|
mocks.getSelfHealingManager.mockReturnValue({ reconcileEngineDowntimeActiveTiming });
|
||||||
|
const engine = createEngine();
|
||||||
|
await engine.start();
|
||||||
|
|
||||||
|
const unpause = mockStore.emitSettingsUpdated(
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: false },
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: true, engineLastActiveAt: "2026-07-15T11:58:00.000Z" },
|
||||||
|
);
|
||||||
|
await vi.waitFor(() => expect(reconcileEngineDowntimeActiveTiming).toHaveBeenCalledTimes(1));
|
||||||
|
expect(mocks.runtimeResumeAfterUnpause).not.toHaveBeenCalled();
|
||||||
|
|
||||||
|
resolveReconcile();
|
||||||
|
await unpause;
|
||||||
|
expect(mocks.runtimeResumeAfterUnpause).toHaveBeenCalledTimes(1);
|
||||||
|
await engine.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("reconciles once for either individual unpause, but not while another pause remains", async () => {
|
||||||
|
const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
|
||||||
|
mocks.currentStore = mockStore.store;
|
||||||
|
const reconcileEngineDowntimeActiveTiming = vi.fn(async () => ({ shiftedTaskIds: [], downtimeMs: 0 }));
|
||||||
|
mocks.getSelfHealingManager.mockReturnValue({ reconcileEngineDowntimeActiveTiming });
|
||||||
|
const engine = createEngine();
|
||||||
|
|
||||||
|
await engine.start();
|
||||||
|
await mockStore.emitSettingsUpdated(
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: false, enginePaused: true },
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: true, enginePaused: true },
|
||||||
|
);
|
||||||
|
expect(reconcileEngineDowntimeActiveTiming).not.toHaveBeenCalled();
|
||||||
|
|
||||||
|
await mockStore.emitSettingsUpdated(
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: false, enginePaused: false },
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: false, enginePaused: true, engineLastActiveAt: "engine-only" },
|
||||||
|
);
|
||||||
|
await mockStore.emitSettingsUpdated(
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: false, enginePaused: false },
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: true, enginePaused: false, engineLastActiveAt: "global-only" },
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(reconcileEngineDowntimeActiveTiming).toHaveBeenCalledTimes(2);
|
||||||
|
expect(reconcileEngineDowntimeActiveTiming).toHaveBeenNthCalledWith(1, { engineLastActiveAtOverride: "engine-only" });
|
||||||
|
expect(reconcileEngineDowntimeActiveTiming).toHaveBeenNthCalledWith(2, { engineLastActiveAtOverride: "global-only" });
|
||||||
|
await engine.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("fails soft when timing reconciliation rejects or its manager is unavailable", async () => {
|
||||||
|
const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
|
||||||
|
mocks.currentStore = mockStore.store;
|
||||||
|
const reconcileEngineDowntimeActiveTiming = vi.fn(async () => {
|
||||||
|
throw new Error("timing unavailable");
|
||||||
|
});
|
||||||
|
mocks.getSelfHealingManager.mockReturnValue({ reconcileEngineDowntimeActiveTiming });
|
||||||
|
const warn = vi.spyOn(runtimeLog, "warn").mockImplementation(() => undefined);
|
||||||
|
const engine = createEngine();
|
||||||
|
await engine.start();
|
||||||
|
const resume = vi.fn();
|
||||||
|
Object.defineProperty(engine.getRuntime(), "stuckTaskDetector", { get: () => ({ resume }), configurable: true });
|
||||||
|
|
||||||
|
await mockStore.emitSettingsUpdated(
|
||||||
|
{ ...baseSettings, autoMerge: true, enginePaused: false },
|
||||||
|
{ ...baseSettings, autoMerge: true, enginePaused: true },
|
||||||
|
);
|
||||||
|
await Promise.resolve();
|
||||||
|
expect(resume).toHaveBeenCalledTimes(1);
|
||||||
|
expect(warn).toHaveBeenCalledWith(expect.stringContaining("failed to reconcile engine downtime active timing"));
|
||||||
|
|
||||||
|
mocks.getSelfHealingManager.mockReturnValue(undefined);
|
||||||
|
await expect(mockStore.emitSettingsUpdated(
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: false },
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: true },
|
||||||
|
)).resolves.toBeUndefined();
|
||||||
|
expect(resume).toHaveBeenCalledTimes(2);
|
||||||
|
warn.mockRestore();
|
||||||
|
await engine.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("passes the frozen heartbeat once so paused task time is discounted once", async () => {
|
||||||
|
const mockStore = createMockStore({ ...baseSettings, autoMerge: true });
|
||||||
|
mocks.currentStore = mockStore.store;
|
||||||
|
const startedMs = Date.parse("2026-07-15T11:50:00.000Z");
|
||||||
|
const capturedHeartbeat = "2026-07-15T11:58:00.000Z";
|
||||||
|
let executionStartedAt = new Date(startedMs).toISOString();
|
||||||
|
const reconcileEngineDowntimeActiveTiming = vi.fn(async ({ engineLastActiveAtOverride }: { engineLastActiveAtOverride?: string }) => {
|
||||||
|
const downtimeMs = Date.parse("2026-07-15T12:00:00.000Z") - Date.parse(engineLastActiveAtOverride ?? "");
|
||||||
|
executionStartedAt = new Date(startedMs + downtimeMs).toISOString();
|
||||||
|
return { shiftedTaskIds: ["FN-active"], downtimeMs };
|
||||||
|
});
|
||||||
|
mocks.getSelfHealingManager.mockReturnValue({ reconcileEngineDowntimeActiveTiming });
|
||||||
|
const engine = createEngine();
|
||||||
|
await engine.start();
|
||||||
|
|
||||||
|
await mockStore.emitSettingsUpdated(
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: false, enginePaused: false, engineLastActiveAt: "2026-07-15T12:00:00.000Z" },
|
||||||
|
{ ...baseSettings, autoMerge: true, globalPause: true, enginePaused: true, engineLastActiveAt: capturedHeartbeat },
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(reconcileEngineDowntimeActiveTiming).toHaveBeenCalledTimes(1);
|
||||||
|
expect(executionStartedAt).toBe("2026-07-15T11:52:00.000Z");
|
||||||
|
await engine.stop();
|
||||||
|
});
|
||||||
|
|
||||||
it("resumes deferred startup recovery on engine unpause", async () => {
|
it("resumes deferred startup recovery on engine unpause", async () => {
|
||||||
const mockStore = createMockStore(baseSettings);
|
const mockStore = createMockStore(baseSettings);
|
||||||
mocks.currentStore = mockStore.store;
|
mocks.currentStore = mockStore.store;
|
||||||
|
|||||||
@@ -4819,7 +4819,26 @@ export class ProjectEngine {
|
|||||||
store: TaskStore,
|
store: TaskStore,
|
||||||
settings: Settings,
|
settings: Settings,
|
||||||
source: "Global unpause" | "Engine unpause",
|
source: "Global unpause" | "Engine unpause",
|
||||||
|
engineLastActiveAtOverride?: string,
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
|
/*
|
||||||
|
FNXC:TaskTiming 2026-07-15-00:00:
|
||||||
|
Reconcile paused wall-clock before resuming agentic work or sweeping tasks.
|
||||||
|
Settings listeners do not await one another, so a detached reconcile lets a
|
||||||
|
task leave in-progress before its anchor shifts and incorrectly accrues the
|
||||||
|
paused span. The captured heartbeat preserves the FN-7011 downtime proof
|
||||||
|
even if the scheduler writes a fresh heartbeat during this await.
|
||||||
|
*/
|
||||||
|
try {
|
||||||
|
await this.getSelfHealingManager()?.reconcileEngineDowntimeActiveTiming({
|
||||||
|
engineLastActiveAtOverride,
|
||||||
|
});
|
||||||
|
} catch (err: unknown) {
|
||||||
|
runtimeLog.warn(
|
||||||
|
`${source}: failed to reconcile engine downtime active timing: ${err instanceof Error ? err.message : String(err)}`,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const runtime = this.runtime as any;
|
const runtime = this.runtime as any;
|
||||||
runtime.resumeAfterUnpause?.().catch((err: Error) =>
|
runtime.resumeAfterUnpause?.().catch((err: Error) =>
|
||||||
@@ -4874,7 +4893,7 @@ export class ProjectEngine {
|
|||||||
|
|
||||||
// 1. Unified pause lifecycle — detector only resumes once BOTH pause sources
|
// 1. Unified pause lifecycle — detector only resumes once BOTH pause sources
|
||||||
// are clear, and pauses when either source engages.
|
// are clear, and pauses when either source engages.
|
||||||
const onPauseLifecycleTransition = ({
|
const onPauseLifecycleTransition = async ({
|
||||||
settings: s,
|
settings: s,
|
||||||
previous: prev,
|
previous: prev,
|
||||||
}: {
|
}: {
|
||||||
@@ -4891,6 +4910,13 @@ export class ProjectEngine {
|
|||||||
|
|
||||||
if (wasPaused && !isPaused) {
|
if (wasPaused && !isPaused) {
|
||||||
const source = prev.globalPause && !s.globalPause ? "Global unpause" : "Engine unpause";
|
const source = prev.globalPause && !s.globalPause ? "Global unpause" : "Engine unpause";
|
||||||
|
runtimeLog.log(`${source} — resuming agentic activity`);
|
||||||
|
await this.resumeAfterUnpauseAndSweepInReview(
|
||||||
|
store,
|
||||||
|
s,
|
||||||
|
source,
|
||||||
|
prev.engineLastActiveAt,
|
||||||
|
);
|
||||||
applyDetectorPauseLifecycle(false, source);
|
applyDetectorPauseLifecycle(false, source);
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
@@ -4934,39 +4960,11 @@ export class ProjectEngine {
|
|||||||
store.on("settings:updated", onAutoMergeDisabled);
|
store.on("settings:updated", onAutoMergeDisabled);
|
||||||
this.settingsHandlers.push(onAutoMergeDisabled);
|
this.settingsHandlers.push(onAutoMergeDisabled);
|
||||||
|
|
||||||
// 4. Global unpause — resume orphaned tasks + sweep in-review
|
// 4. The unified lifecycle listener above owns unpause. It waits for timing
|
||||||
const onGlobalUnpause = async ({
|
// reconciliation before any agentic resume, avoiding duplicate work when
|
||||||
settings: s,
|
// globalPause and enginePaused clear in one settings update.
|
||||||
previous: prev,
|
|
||||||
}: {
|
|
||||||
settings: Settings;
|
|
||||||
previous: Settings;
|
|
||||||
}) => {
|
|
||||||
if (prev.globalPause && !s.globalPause) {
|
|
||||||
runtimeLog.log("Global unpause — resuming agentic activity");
|
|
||||||
await this.resumeAfterUnpauseAndSweepInReview(store, s, "Global unpause");
|
|
||||||
}
|
|
||||||
};
|
|
||||||
store.on("settings:updated", onGlobalUnpause);
|
|
||||||
this.settingsHandlers.push(onGlobalUnpause);
|
|
||||||
|
|
||||||
// 5. Engine unpause — same as global unpause
|
// 5. Maintenance interval change — reschedule mergeActive reconciliation
|
||||||
const onEngineUnpause = async ({
|
|
||||||
settings: s,
|
|
||||||
previous: prev,
|
|
||||||
}: {
|
|
||||||
settings: Settings;
|
|
||||||
previous: Settings;
|
|
||||||
}) => {
|
|
||||||
if (prev.enginePaused && !s.enginePaused) {
|
|
||||||
runtimeLog.log("Engine unpaused — resuming agentic activity");
|
|
||||||
await this.resumeAfterUnpauseAndSweepInReview(store, s, "Engine unpause");
|
|
||||||
}
|
|
||||||
};
|
|
||||||
store.on("settings:updated", onEngineUnpause);
|
|
||||||
this.settingsHandlers.push(onEngineUnpause);
|
|
||||||
|
|
||||||
// 6. Maintenance interval change — reschedule mergeActive reconciliation
|
|
||||||
const onMaintenanceIntervalChange = ({
|
const onMaintenanceIntervalChange = ({
|
||||||
settings: s,
|
settings: s,
|
||||||
previous: prev,
|
previous: prev,
|
||||||
|
|||||||
@@ -1450,8 +1450,10 @@ export class SelfHealingManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
async reconcileEngineDowntimeActiveTiming(): Promise<{ shiftedTaskIds: string[]; downtimeMs: number }> {
|
async reconcileEngineDowntimeActiveTiming(
|
||||||
const result = await this.store.reconcileActiveTimingForEngineDowntime();
|
opts?: { engineLastActiveAtOverride?: string },
|
||||||
|
): Promise<{ shiftedTaskIds: string[]; downtimeMs: number }> {
|
||||||
|
const result = await this.store.reconcileActiveTimingForEngineDowntime(new Date(), opts);
|
||||||
const shifted = result.shiftedTaskIds.length > 0;
|
const shifted = result.shiftedTaskIds.length > 0;
|
||||||
const auditor = createRunAuditor(this.store, {
|
const auditor = createRunAuditor(this.store, {
|
||||||
runId: generateSyntheticRunId("reconcile-engine-downtime-active-timing", "global"),
|
runId: generateSyntheticRunId("reconcile-engine-downtime-active-timing", "global"),
|
||||||
|
|||||||
Reference in New Issue
Block a user