FN-8918: suppress false parked-task alerts
Prevent stale self-healing observations from notifying operators about active or intentionally held tasks. - Require ownerless self-healing proof before classifying a task as wedged. - Revalidate live tasks and suppress notifications for progressing, held, deleted, archived, or complete-lane rows. - Cover reviewing, typed not-found reads, and archived episode resolution; document the behavior. - Add a patch changeset for the notification fix. Files changed: .changeset/fn-8918-false-parked-task-alerts.md | 7 ++ docs/agents.md | 2 +- docs/architecture.md | 2 +- packages/engine/src/notification/__tests__/task-wedge-notification.test.ts | 121 +++++++++++++++++++-- packages/engine/src/notification/notification-service.ts | 34 ++++-- packages/engine/src/notification/task-wedge-notification.ts | 21 +++- 6 files changed, 166 insertions(+), 21 deletions(-) Fusion-Task-Id: FN-8918 Fusion-Task-Lineage: 4da385aa-c625-40c5-a781-36becdf94fe5 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-8918-false-parked-task-alerts.md
Normal file
7
.changeset/fn-8918-false-parked-task-alerts.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Stop sending needs-operator-action alerts for tasks still running or intentionally held.
|
||||
category: fix
|
||||
dev: Tightens task-wedge classifier proof handling and live-row delivery validation.
|
||||
@@ -266,7 +266,7 @@ Separation of concerns:
|
||||
|
||||
### Task wedge operator notifications
|
||||
|
||||
When a task is terminally blocked (for example, by a merge gate, exhausted execution retries, or a completion blocker), Fusion posts a system message to the dashboard mailbox and sends a `task-wedged` notification through configured providers. Pause-derived alerts require actual pause state, and an actively progressing task never alerts even if a resume path retained a pause marker. The message identifies the task, bounded reason/gate when known, and a recovery action. The active/resolved episode is persisted with the task, so it is sent once per active reason across service restarts; retrying or otherwise restoring progress clears the episode, so a later recurrence is visible again.
|
||||
When a task is terminally blocked (for example, by a merge gate, exhausted execution retries, or a completion blocker), Fusion posts a system message to the dashboard mailbox and sends a `task-wedged` notification through configured providers. Self-healing declines alert only when their proof shows no live session, no recent activity, and no intentional pause or auto-merge-off hold. Before delivery, Fusion revalidates the live row: progressing (including `reviewing`), paused, auto-merge-off, deleted, archived, and complete-lane rows do not alert. The message identifies the task, bounded reason/gate when known, and a recovery action. The active/resolved episode is persisted with the task, so it is sent once per active reason across service restarts; retrying or otherwise restoring progress clears the episode, so a later recurrence is visible again.
|
||||
|
||||
### CLI agent permission prompts and notifications
|
||||
|
||||
|
||||
@@ -14,7 +14,7 @@ Action gates emit a best-effort, idempotent system-mail item for each pending ap
|
||||
|
||||
Actionable terminal task updates are classified into bounded reasons such as a named merge gate, retry exhaustion, or a completion blocker. The PostgreSQL-backed task row persists an active/resolved episode with an opaque identity, so `NotificationService` sends one `task-wedged` provider event and one dashboard system-mailbox message per active reason even across restarts. Repeated observations remain quiet until an authoritative non-wedge task update resolves the episode; changed and resolved-then-reentered reasons notify again.
|
||||
|
||||
A failed snapshot is not actionable while persisted automatic-recovery ownership remains: a scheduled recovery has both its retry counter and deadline, while transient merge recovery has an in-budget persisted retry counter. Pause-derived wedge reasons additionally require real pause state (`paused: true` or `status: "paused"`); an actively progressing task is never wedged. `NotificationService` re-reads and reclassifies the live task immediately before a wedge claim and again when a generic failure grace timer fires, so recovery that begins after a failed event, including a resume that deliberately leaves a stale pause reason behind, cannot create a mailbox row or `task-wedged` provider event. Explicit operator-action parks and cleared/exhausted recovery markers remain terminal and claim exactly one episode.
|
||||
A failed snapshot is not actionable while persisted automatic-recovery ownership remains: a scheduled recovery has both its retry counter and deadline, while transient merge recovery has an in-budget persisted retry counter. A self-healing decline is actionable only when its proof shows no live session, no recent activity, and no intentional pause or auto-merge-off hold. Pause-derived wedge reasons additionally require real pause state (`paused: true` or `status: "paused"`); an actively progressing task, including `reviewing`, is never wedged. `NotificationService` re-reads and reclassifies the live task immediately before a wedge claim and again when a generic failure grace timer fires: it suppresses progressing, paused, auto-merge-off, deleted, and archived/complete-lane rows, and a not-found live read is quiet rather than rejecting the per-task wedge chain. Recovery that begins after a failed event, including a resume that deliberately leaves a stale pause reason behind, therefore cannot create a mailbox row or `task-wedged` provider event. Explicit operator-action parks and cleared/exhausted recovery markers remain terminal and claim exactly one episode.
|
||||
|
||||
Each task also stores `lastNotifiedAtByReason`, an independent timestamp map keyed by bounded reason. `WEDGE_RENOTIFY_COOLDOWN_MS` defaults to six hours: resolving an episode does not clear its reason's live stamp, so a scheduler/self-healing resolve→re-wedge flap sends neither a provider push nor a mailbox message until the window expires. A different reason notifies immediately, including X→Y→X while X remains within its own cooldown; expired or invalid entries are pruned during the atomic claim, and legacy rows without the map notify normally before initializing it. The no-durable-store fallback applies the same per-reason window in memory. Provider and mailbox delivery are independently best-effort after sharing this single claim decision, while run-audit metadata remains ids/counts/outcomes-only.
|
||||
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { WEDGE_RENOTIFY_COOLDOWN_MS, type NotificationProvider, type Settings, type Task } from "@fusion/core";
|
||||
import { TaskNotFoundError, WEDGE_RENOTIFY_COOLDOWN_MS, type NotificationProvider, type Settings, type Task } from "@fusion/core";
|
||||
import { NotificationService } from "../notification-service.js";
|
||||
import { MAX_AUTO_MERGE_TRANSIENT_RETRIES } from "../../errors/transient-merge-error-classifier.js";
|
||||
import { describeSelfHealingNoActionWedge, describeTaskRecoveryOwner, describeTaskWedge } from "../task-wedge-notification.js";
|
||||
import { describeSelfHealingNoActionWedge, describeTaskRecoveryOwner, describeTaskWedge, isTaskProgressing } from "../task-wedge-notification.js";
|
||||
|
||||
type Listener = (task: Task) => void;
|
||||
|
||||
@@ -26,6 +26,7 @@ function fixture(workflowIr?: unknown) {
|
||||
const listeners = new Set<Listener>();
|
||||
let wedge: Task["wedgeNotification"];
|
||||
let liveTask: Task | undefined;
|
||||
let liveReadError: Error | undefined;
|
||||
const claimTaskWedgeNotificationEpisode = vi.fn(async (taskId: string, reasonKey: string | null) => {
|
||||
if (reasonKey === null) {
|
||||
if (wedge?.status === "active") wedge = { ...wedge, status: "resolved" };
|
||||
@@ -37,11 +38,15 @@ function fixture(workflowIr?: unknown) {
|
||||
});
|
||||
const store = {
|
||||
getSettings: async () => ({ ntfyEnabled: true, ntfyTopic: "test" }) as Settings,
|
||||
getTask: async () => liveTask,
|
||||
getTask: async () => {
|
||||
if (liveReadError) throw liveReadError;
|
||||
return liveTask;
|
||||
},
|
||||
on: (event: string, listener: Listener) => { if (event === "task:updated") listeners.add(listener); },
|
||||
off: () => undefined,
|
||||
emit: (task: Task) => listeners.forEach((listener) => listener(task)),
|
||||
setLiveTask: (next: Task | undefined) => { liveTask = next; },
|
||||
setLiveReadError: (next: Error | undefined) => { liveReadError = next; },
|
||||
claimTaskWedgeNotificationEpisode,
|
||||
/* Absent → the helper keeps the legacy ids, which is every pre-existing case in this file. */
|
||||
...(workflowIr ? { listWorkflowDefinitions: async () => [{ ir: workflowIr }] } : {}),
|
||||
@@ -93,9 +98,8 @@ function cooldownFixture({ durable = true }: { durable?: boolean } = {}) {
|
||||
}
|
||||
|
||||
async function flushWedgeHandling() {
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
// FNXC:TaskWedgeNotifications 2026-08-10-04:35: Wedge delivery resolves lifecycle roles before its claim, so drain both the per-task chain and role-resolution promise turns without real timers.
|
||||
for (let index = 0; index < 8; index += 1) await Promise.resolve();
|
||||
}
|
||||
|
||||
describe("task wedge notifications", () => {
|
||||
@@ -237,6 +241,56 @@ describe("task wedge notifications", () => {
|
||||
await service.stop();
|
||||
});
|
||||
|
||||
it("does not deliver self-healing descriptors when the live row is held or reviewing", async () => {
|
||||
const { store, service, sendMessageOnce, sendNotification, task } = fixture();
|
||||
const stalled = task({ status: "in-review", error: undefined, paused: false, userPaused: false });
|
||||
const descriptor = describeSelfHealingNoActionWedge(stalled, "reconcile-in-review-unmet-dependencies", { taskActive: false })!;
|
||||
await service.start();
|
||||
|
||||
for (const live of [
|
||||
task({ status: "in-review", paused: true, error: undefined }),
|
||||
task({ status: "in-review", userPaused: true, error: undefined }),
|
||||
task({ status: "in-review", autoMerge: false, error: undefined }),
|
||||
task({ status: "reviewing", error: undefined }),
|
||||
]) {
|
||||
store.setLiveTask(live);
|
||||
await service.notifyTaskWedge(stalled, descriptor);
|
||||
}
|
||||
|
||||
expect(sendMessageOnce).not.toHaveBeenCalled();
|
||||
expect(sendNotification).not.toHaveBeenCalled();
|
||||
await service.stop();
|
||||
});
|
||||
|
||||
it("quietly handles a typed not-found live read without rejecting the wedge chain", async () => {
|
||||
const { store, service, sendMessageOnce, sendNotification, task } = fixture();
|
||||
store.setLiveReadError(new TaskNotFoundError("FN-8501"));
|
||||
await service.start();
|
||||
|
||||
await expect(service.notifyTaskWedge(task(), describeTaskWedge(task())!)).resolves.toBeUndefined();
|
||||
store.emit(task());
|
||||
await flushWedgeHandling();
|
||||
|
||||
expect(sendMessageOnce).not.toHaveBeenCalled();
|
||||
expect(sendNotification).not.toHaveBeenCalled();
|
||||
await service.stop();
|
||||
});
|
||||
|
||||
it("resolves an active episode for an archived failed live row", async () => {
|
||||
const { store, service, sendMessageOnce, sendNotification, task, setWedge, claimTaskWedgeNotificationEpisode } = fixture();
|
||||
const active = { reasonKey: "merge-blocked:changeset-format", episodeId: "archived-episode", status: "active" as const, transitionedAt: "2026-07-22T12:00:00.000Z" };
|
||||
setWedge(active);
|
||||
store.setLiveTask(task({ column: "archived", status: "failed", wedgeNotification: active }));
|
||||
await service.start();
|
||||
|
||||
await service.notifyTaskWedge(task({ wedgeNotification: active }), describeTaskWedge(task())!);
|
||||
|
||||
expect(sendMessageOnce).not.toHaveBeenCalled();
|
||||
expect(sendNotification).not.toHaveBeenCalled();
|
||||
expect(claimTaskWedgeNotificationEpisode).toHaveBeenCalledWith("FN-8501", null);
|
||||
await service.stop();
|
||||
});
|
||||
|
||||
/*
|
||||
FNXC:TaskWedgeNotifications 2026-08-01-07:44:
|
||||
A recovery and re-wedge can be emitted back-to-back by synchronous task lifecycle writers. The
|
||||
@@ -320,6 +374,59 @@ describe("task wedge notifications", () => {
|
||||
await restarted.stop();
|
||||
});
|
||||
|
||||
it.each([
|
||||
"reclaim-pr-conflict",
|
||||
"reclaim-self-owned-branch-conflict",
|
||||
"reconcile-in-review-unmet-dependencies",
|
||||
"reconcile-dependency-blocking-lease",
|
||||
"auto-rebound-paused-scope-decay",
|
||||
"stuck-merge-deadlock",
|
||||
"missing-worktree-merge-active",
|
||||
"missing-worktree-review",
|
||||
"finalize-no-op-review",
|
||||
"stale-incomplete-review",
|
||||
"ghost-review",
|
||||
"no-progress-no-task-done",
|
||||
"partial-progress-no-task-done",
|
||||
])("preserves ownerless self-healing alerts for %s", (stage) => {
|
||||
const { task } = fixture();
|
||||
const ownerless = task({ status: "in-review", error: undefined, paused: false, userPaused: false });
|
||||
expect(describeSelfHealingNoActionWedge(ownerless, stage, {
|
||||
sessionDead: true,
|
||||
noRecentActivity: true,
|
||||
worktreeUnusable: false,
|
||||
taskActive: false,
|
||||
hasExecutingTaskLock: false,
|
||||
livePaths: [],
|
||||
})).toMatchObject({ reasonKey: `self-healing-no-action:${stage}` });
|
||||
});
|
||||
|
||||
it("suppresses self-healing declines that prove liveness or intentional holds", () => {
|
||||
const { task } = fixture();
|
||||
const ownerless = task({ status: "in-review", error: undefined, paused: false, userPaused: false });
|
||||
const stage = "reconcile-in-review-unmet-dependencies";
|
||||
for (const proof of [
|
||||
{ sessionDead: false },
|
||||
{ noRecentActivity: false },
|
||||
{ taskActive: true },
|
||||
{ hasExecutingTaskLock: true },
|
||||
{ mergePending: true },
|
||||
{ livePaths: ["/live/session"] },
|
||||
{ reason: "paused-guard" },
|
||||
{ reason: "auto-merge-processing-disabled" },
|
||||
]) {
|
||||
expect(describeSelfHealingNoActionWedge(ownerless, stage, proof)).toBeNull();
|
||||
}
|
||||
expect(describeSelfHealingNoActionWedge(ownerless, stage, undefined)).toMatchObject({ reasonKey: `self-healing-no-action:${stage}` });
|
||||
});
|
||||
|
||||
it("treats reviewing as progressing while preserving pause guards", () => {
|
||||
const { task } = fixture();
|
||||
expect(isTaskProgressing(task({ status: "reviewing", paused: false }))).toBe(true);
|
||||
expect(isTaskProgressing(task({ status: "reviewing", paused: true }))).toBe(false);
|
||||
expect(isTaskProgressing(task({ status: "paused", paused: false }))).toBe(false);
|
||||
});
|
||||
|
||||
it("delivers one durable episode for an ownerless self-healing no-action escalation", async () => {
|
||||
const { service, sendMessageOnce, sendNotification, task } = fixture();
|
||||
await service.start();
|
||||
@@ -359,7 +466,7 @@ describe("task wedge notifications", () => {
|
||||
await service.stop();
|
||||
});
|
||||
|
||||
it.each(["queued", "planning", "in-progress", "merging", "merging-pr", "merging-fix", "merged", "done"])("does not classify a stale pause reason while %s is progressing", (status) => {
|
||||
it.each(["queued", "planning", "in-progress", "reviewing", "merging", "merging-pr", "merging-fix", "merged", "done"])("does not classify a stale pause reason while %s is progressing", (status) => {
|
||||
const { task } = fixture();
|
||||
expect(describeTaskWedge(task({ status: status as Task["status"], paused: false, pausedReason: "completed-blocked", error: undefined }))).toBeNull();
|
||||
});
|
||||
|
||||
@@ -11,7 +11,7 @@ import type {
|
||||
Task,
|
||||
} from "@fusion/core";
|
||||
import type { LifecycleColumns, TaskMoveLanes, WorkflowIrResolverStore } from "@fusion/core";
|
||||
import { DASHBOARD_USER_ID, NotificationDispatcher, resolveProjectColumnsForRoles, resolveReviewColumns, resolveTaskLifecycleColumns, resolveWorkflowIrForTask, WEDGE_RENOTIFY_COOLDOWN_MS } from "@fusion/core";
|
||||
import { DASHBOARD_USER_ID, isTaskNotFoundError, NotificationDispatcher, resolveProjectColumnsForRoles, resolveReviewColumns, resolveTaskLifecycleColumns, resolveWorkflowIrForTask, WEDGE_RENOTIFY_COOLDOWN_MS } from "@fusion/core";
|
||||
import { DEFAULT_NTFY_EVENTS, buildNtfyClickUrl, formatTaskIdentifier } from "../util/notifier.js";
|
||||
import { schedulerLog } from "../logger.js";
|
||||
import { NtfyNotificationProvider } from "./ntfy-provider.js";
|
||||
@@ -535,9 +535,20 @@ export class NotificationService {
|
||||
}
|
||||
|
||||
private async maybeNotifyTaskWedge(task: Task, suppliedDescriptor?: TaskWedgeDescriptor | null): Promise<void> {
|
||||
// Task events carry snapshots. Re-read before a durable claim so an immediate
|
||||
// recovery update cannot turn a stale failed event into an operator alert.
|
||||
const liveTask = this.store.getTask ? (await this.store.getTask(task.id)) ?? task : task;
|
||||
/*
|
||||
FNXC:TaskWedgeNotifications 2026-08-10-04:35:
|
||||
`getTask` throws TaskNotFoundError for soft-deleted rows instead of returning
|
||||
undefined. A missing or unreadable live row cannot prove an actionable park,
|
||||
so fail quiet and preserve enqueueWedgeHandling's never-reject contract.
|
||||
*/
|
||||
let liveTask: Task;
|
||||
try {
|
||||
liveTask = this.store.getTask ? (await this.store.getTask(task.id)) ?? task : task;
|
||||
} catch (error) {
|
||||
const detail = isTaskNotFoundError(error) ? "not found" : error instanceof Error ? error.message : String(error);
|
||||
schedulerLog.warn(`[notify] ${task.id} wedge live-read failed (${detail}) — suppressed notification`);
|
||||
return;
|
||||
}
|
||||
const recoveryOwner = describeTaskRecoveryOwner(liveTask);
|
||||
if (recoveryOwner) {
|
||||
// Recovery ownership is not a wedge episode. Resolve only an episode we
|
||||
@@ -549,12 +560,17 @@ export class NotificationService {
|
||||
return;
|
||||
}
|
||||
/*
|
||||
FNXC:TaskWedgeNotifications 2026-08-09-06:30:
|
||||
Self-healing descriptors encode a no-action proof that the generic classifier
|
||||
cannot recompute. They still cannot claim after the live task resumes, so let
|
||||
the existing no-descriptor resolution path close any active stale episode.
|
||||
FNXC:TaskWedgeNotifications 2026-08-10-04:35:
|
||||
Archived/complete lanes and deleted rows may retain failed snapshots, but they
|
||||
are terminal history rather than operator-actionable parks. Revalidate a
|
||||
self-healing descriptor's live pause and auto-merge hold too; only role
|
||||
membership is stable when workflows rename lifecycle columns.
|
||||
*/
|
||||
const descriptor = suppliedDescriptor && isTaskProgressing(liveTask)
|
||||
const terminalLanes = await resolveProjectColumnsForRoles(this.store, ["complete", "archived"]);
|
||||
const liveRowCannotBeWedge = liveTask.deletedAt != null || terminalLanes.has(liveTask.column);
|
||||
const suppliedDescriptorIsHeldOrProgressing = suppliedDescriptor != null
|
||||
&& (liveTask.paused === true || liveTask.userPaused === true || liveTask.autoMerge === false || isTaskProgressing(liveTask));
|
||||
const descriptor = liveRowCannotBeWedge || suppliedDescriptorIsHeldOrProgressing
|
||||
? null
|
||||
: suppliedDescriptor ?? describeTaskWedge(liveTask);
|
||||
task = liveTask;
|
||||
|
||||
@@ -102,8 +102,23 @@ export function describeSelfHealingNoActionWedge(task: Task, stage: string, meta
|
||||
// Test and legacy proof producers may omit metadata; absent ownership evidence
|
||||
// remains ownerless rather than turning best-effort notification into a park failure.
|
||||
const proof = metadata ?? {};
|
||||
// These proof signals mean a live executor, checkout, or queued merge owns the task.
|
||||
if (proof.taskActive === true || proof.hasExecutingTaskLock === true || proof.mergePending === true) return null;
|
||||
/*
|
||||
FNXC:TaskWedgeNotifications 2026-08-10-04:35:
|
||||
A declined self-healing move is actionable only when its proof says the card is
|
||||
ownerless. A live session, recent activity, or intentional pause/auto-merge-off
|
||||
hold means the card is working or deliberately waiting, not parked. Keep a
|
||||
usable worktree alerting: with a dead session and stale activity it is evidence
|
||||
of a genuinely stuck card, not progress.
|
||||
*/
|
||||
if (
|
||||
proof.sessionDead === false
|
||||
|| proof.noRecentActivity === false
|
||||
|| proof.taskActive === true
|
||||
|| proof.hasExecutingTaskLock === true
|
||||
|| proof.mergePending === true
|
||||
|| proof.reason === "paused-guard"
|
||||
|| proof.reason === "auto-merge-processing-disabled"
|
||||
) return null;
|
||||
if (Array.isArray(proof.livePaths) && proof.livePaths.length > 0) return null;
|
||||
return { reasonKey: `self-healing-no-action:${stage}`, ...description };
|
||||
}
|
||||
@@ -117,7 +132,7 @@ lifecycle state, is not an operator-actionable terminal wedge.
|
||||
export function isTaskProgressing(task: Task): boolean {
|
||||
return task.paused !== true
|
||||
&& task.status !== "paused"
|
||||
&& ["queued", "planning", "in-progress", "merging", "merging-pr", "merging-fix", "merged", "done"].includes(task.status ?? "");
|
||||
&& ["queued", "planning", "in-progress", "reviewing", "merging", "merging-pr", "merging-fix", "merged", "done"].includes(task.status ?? "");
|
||||
}
|
||||
|
||||
/*
|
||||
|
||||
Reference in New Issue
Block a user