FN-8867: prevent false task wedge notifications
Prevent stale terminal snapshots and pause markers from notifying for progressing tasks. - Reclassify the live task immediately before claiming wedge episodes. - Require actual pause state for pause-derived wedge reasons. - Cover resumed-task races and stale pause markers with notification tests. - Document the wedge notification behavior and add a patch changeset. Files changed: .changeset/fn-8867-wedge-false-terminal-alerts.md | 7 ++ docs/agents.md | 2 +- docs/architecture.md | 2 +- .../__tests__/notification-service.test.ts | 9 +- .../__tests__/task-wedge-notification.test.ts | 102 +++++++++++++++++++-- .../src/notification/notification-service.ts | 22 +++-- .../src/notification/task-wedge-notification.ts | 22 ++++- 7 files changed, 137 insertions(+), 29 deletions(-) Fusion-Task-Id: FN-8867 Fusion-Task-Lineage: c92c7735-e8b9-4979-8028-10a0555810f1 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-8867-wedge-false-terminal-alerts.md
Normal file
7
.changeset/fn-8867-wedge-false-terminal-alerts.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Stop sending "needs operator action" alerts for tasks that are running normally.
|
||||
category: fix
|
||||
dev: Wedge classification now requires real pause state for pause-reason-derived reasons, and NotificationService revalidates the descriptor against the live task before claiming an episode.
|
||||
@@ -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. 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. 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.
|
||||
|
||||
### CLI agent permission prompts and notifications
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ This document describes the actual architecture of Fusion as implemented in this
|
||||
|
||||
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. `NotificationService` re-reads 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 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. 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.
|
||||
|
||||
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.
|
||||
|
||||
|
||||
@@ -230,7 +230,7 @@ describe("NotificationService deferred failure notifications", () => {
|
||||
await service.stop();
|
||||
});
|
||||
|
||||
it("delivers a new wedge episode when an opaque terminal failure gains a specific cause", async () => {
|
||||
it("delivers only the live wedge cause when a snapshot is superseded", async () => {
|
||||
const { store, service, sendNotification } = await setup();
|
||||
const genericFailure = task({ id: "FN-wedge", status: "failed", error: "unexpected failure" });
|
||||
store.setTask(genericFailure);
|
||||
@@ -246,11 +246,8 @@ describe("NotificationService deferred failure notifications", () => {
|
||||
store.emit("task:updated", wedge);
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
|
||||
expect(sendNotification).toHaveBeenCalledTimes(2);
|
||||
expect(sendNotification).toHaveBeenCalledWith("task-wedged", expect.objectContaining({
|
||||
taskId: "FN-wedge",
|
||||
metadata: expect.objectContaining({ wedgeReason: "terminal-failed" }),
|
||||
}));
|
||||
// FNXC:TaskWedgeNotifications 2026-08-09-06:30: A superseded snapshot must not claim an obsolete episode.
|
||||
expect(sendNotification).toHaveBeenCalledTimes(1);
|
||||
expect(sendNotification).toHaveBeenCalledWith("task-wedged", expect.objectContaining({
|
||||
taskId: "FN-wedge",
|
||||
metadata: expect.objectContaining({ wedgeReason: "merge-blocked:changeset-format" }),
|
||||
|
||||
@@ -25,20 +25,24 @@ const RENAMED_IR = {
|
||||
function fixture(workflowIr?: unknown) {
|
||||
const listeners = new Set<Listener>();
|
||||
let wedge: Task["wedgeNotification"];
|
||||
let liveTask: Task | undefined;
|
||||
const claimTaskWedgeNotificationEpisode = vi.fn(async (taskId: string, reasonKey: string | null) => {
|
||||
if (reasonKey === null) {
|
||||
if (wedge?.status === "active") wedge = { ...wedge, status: "resolved" };
|
||||
return { claimed: false };
|
||||
}
|
||||
if (wedge?.status === "active" && wedge.reasonKey === reasonKey) return { claimed: false };
|
||||
wedge = { reasonKey, episodeId: `${taskId}-${reasonKey}-${Date.now()}`, status: "active", transitionedAt: new Date().toISOString() };
|
||||
return { claimed: true, episodeId: wedge.episodeId };
|
||||
});
|
||||
const store = {
|
||||
getSettings: async () => ({ ntfyEnabled: true, ntfyTopic: "test" }) as Settings,
|
||||
getTask: async () => liveTask,
|
||||
on: (event: string, listener: Listener) => { if (event === "task:updated") listeners.add(listener); },
|
||||
off: () => undefined,
|
||||
emit: (task: Task) => listeners.forEach((listener) => listener(task)),
|
||||
claimTaskWedgeNotificationEpisode: async (taskId: string, reasonKey: string | null) => {
|
||||
if (reasonKey === null) {
|
||||
if (wedge?.status === "active") wedge = { ...wedge, status: "resolved" };
|
||||
return { claimed: false };
|
||||
}
|
||||
if (wedge?.status === "active" && wedge.reasonKey === reasonKey) return { claimed: false };
|
||||
wedge = { reasonKey, episodeId: `${taskId}-${reasonKey}-${Date.now()}`, status: "active", transitionedAt: new Date().toISOString() };
|
||||
return { claimed: true, episodeId: wedge.episodeId };
|
||||
},
|
||||
setLiveTask: (next: Task | undefined) => { liveTask = next; },
|
||||
claimTaskWedgeNotificationEpisode,
|
||||
/* Absent → the helper keeps the legacy ids, which is every pre-existing case in this file. */
|
||||
...(workflowIr ? { listWorkflowDefinitions: async () => [{ ir: workflowIr }] } : {}),
|
||||
};
|
||||
@@ -48,7 +52,7 @@ function fixture(workflowIr?: unknown) {
|
||||
const provider: NotificationProvider = { getProviderId: () => "test", isEventSupported: () => true, sendNotification };
|
||||
service.registerProvider(provider);
|
||||
const task = (overrides: Partial<Task> = {}): Task => ({ id: "FN-8501", title: "Fix changeset", description: "", column: "in-review", status: "failed", error: "merge verification failed: check:changeset-format", dependencies: [], steps: [], currentStep: 0, log: [], createdAt: "2026-07-22T12:00:00.000Z", updatedAt: "2026-07-22T12:00:00.000Z", ...overrides } as Task);
|
||||
return { store, service, sendMessageOnce, sendNotification, task, getWedge: () => wedge };
|
||||
return { store, service, sendMessageOnce, sendNotification, task, getWedge: () => wedge, setWedge: (next: Task["wedgeNotification"]) => { wedge = next; }, claimTaskWedgeNotificationEpisode };
|
||||
}
|
||||
|
||||
/* Creates a restart-safe claim fake so NotificationService tests exercise delivery policy, not storage implementation. */
|
||||
@@ -183,6 +187,56 @@ describe("task wedge notifications", () => {
|
||||
await service.stop();
|
||||
});
|
||||
|
||||
/*
|
||||
FNXC:TaskWedgeNotifications 2026-08-09-06:30:
|
||||
The event can race a completed resume. Delivery must ignore its stale failed
|
||||
descriptor, re-read the executing row, and resolve the previously active episode.
|
||||
*/
|
||||
it("does not alert from a failed snapshot after the live task resumes", async () => {
|
||||
const { store, service, sendMessageOnce, sendNotification, task, setWedge, claimTaskWedgeNotificationEpisode } = fixture();
|
||||
const active = { reasonKey: "merge-blocked:changeset-format", episodeId: "active-episode", status: "active" as const, transitionedAt: "2026-07-22T12:00:00.000Z" };
|
||||
setWedge(active);
|
||||
store.setLiveTask(task({ status: "in-progress", column: "in-progress", error: undefined, wedgeNotification: active }));
|
||||
await service.start();
|
||||
|
||||
store.emit(task({ status: "failed", wedgeNotification: active }));
|
||||
await flushWedgeHandling();
|
||||
|
||||
expect(sendMessageOnce).not.toHaveBeenCalled();
|
||||
expect(sendNotification).not.toHaveBeenCalled();
|
||||
expect(claimTaskWedgeNotificationEpisode).toHaveBeenCalledWith("FN-8501", null);
|
||||
await service.stop();
|
||||
});
|
||||
|
||||
it("does not alert repeatedly for a live executing task with a stale pause reason", async () => {
|
||||
const { store, service, sendMessageOnce, sendNotification, task } = fixture();
|
||||
const resumed = task({ status: "in-progress", column: "in-progress", paused: false, pausedReason: "completed-blocked", error: undefined });
|
||||
store.setLiveTask(resumed);
|
||||
await service.start();
|
||||
|
||||
store.emit(task({ paused: true, pausedReason: "completed-blocked", status: "queued" }));
|
||||
store.emit(task({ paused: true, pausedReason: "completed-blocked", status: "queued", updatedAt: "2026-07-22T12:01:00.000Z" }));
|
||||
await flushWedgeHandling();
|
||||
|
||||
expect(sendMessageOnce).not.toHaveBeenCalled();
|
||||
expect(sendNotification).not.toHaveBeenCalled();
|
||||
await service.stop();
|
||||
});
|
||||
|
||||
it("does not deliver a self-healing descriptor after the live task progresses", 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 })!;
|
||||
store.setLiveTask(task({ status: "in-progress", column: "in-progress", error: undefined }));
|
||||
await service.start();
|
||||
|
||||
await service.notifyTaskWedge(stalled, descriptor);
|
||||
|
||||
expect(sendMessageOnce).not.toHaveBeenCalled();
|
||||
expect(sendNotification).not.toHaveBeenCalled();
|
||||
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
|
||||
@@ -305,6 +359,34 @@ 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) => {
|
||||
const { task } = fixture();
|
||||
expect(describeTaskWedge(task({ status: status as Task["status"], paused: false, pausedReason: "completed-blocked", error: undefined }))).toBeNull();
|
||||
});
|
||||
|
||||
it.each([
|
||||
["completed-blocked", "completion-blocked"],
|
||||
["error-retry-exhausted", "heartbeat-retry-exhausted"],
|
||||
["error-unrecoverable", "heartbeat-error-unrecoverable"],
|
||||
["branch-cross-contamination", "branch-cross-contamination"],
|
||||
["branch-conflict-tripwire", "branch-conflict-tripwire"],
|
||||
["branch-conflict-recovery-exhausted", "branch-conflict-recovery-exhausted"],
|
||||
["branch-conflict-unrecoverable", "branch-conflict-unrecoverable"],
|
||||
["stuck-loop-exhausted-manual-intervention-required", "stuck-loop-exhausted"],
|
||||
["non-retryable-provider-error", "non-retryable-provider-error"],
|
||||
["in-review-stall-deadlock", "in-review-stall-deadlock"],
|
||||
])("requires pause proof but preserves terminal pause reason %s as %s", (pausedReason, reasonKey) => {
|
||||
const { task } = fixture();
|
||||
expect(describeTaskWedge(task({ status: "failed", paused: false, pausedReason }))).not.toMatchObject({ reasonKey });
|
||||
expect(describeTaskWedge(task({ status: "paused", paused: false, pausedReason }))).toMatchObject({ reasonKey });
|
||||
expect(describeTaskWedge(task({ status: "queued", paused: true, pausedReason }))).toMatchObject({ reasonKey });
|
||||
});
|
||||
|
||||
it("keeps the FN-7926 completed-blocked park shape actionable", () => {
|
||||
const { task } = fixture();
|
||||
expect(describeTaskWedge(task({ status: "queued", paused: true, pausedReason: "completed-blocked", error: undefined }))).toMatchObject({ reasonKey: "completion-blocked" });
|
||||
});
|
||||
|
||||
it.each([
|
||||
["branch-cross-contamination", "branch-cross-contamination"],
|
||||
["branch-conflict-tripwire", "branch-conflict-tripwire"],
|
||||
|
||||
@@ -16,7 +16,7 @@ import { DEFAULT_NTFY_EVENTS, buildNtfyClickUrl, formatTaskIdentifier } from "..
|
||||
import { schedulerLog } from "../logger.js";
|
||||
import { NtfyNotificationProvider } from "./ntfy-provider.js";
|
||||
import { WebhookNotificationProvider } from "./webhook-provider.js";
|
||||
import { describeTaskRecoveryOwner, describeTaskWedge, type TaskWedgeDescriptor } from "./task-wedge-notification.js";
|
||||
import { describeTaskRecoveryOwner, describeTaskWedge, isTaskProgressing, type TaskWedgeDescriptor } from "./task-wedge-notification.js";
|
||||
|
||||
export interface NotificationServiceOptions {
|
||||
/** Project identifier for notification deep links */
|
||||
@@ -371,10 +371,10 @@ export class NotificationService {
|
||||
|
||||
private handleTaskUpdated = (task: Task, meta?: { lanes?: TaskMoveLanes }): void => {
|
||||
/*
|
||||
FNXC:TaskWedgeNotifications 2026-08-05-04:53:
|
||||
A task update is a point-in-time snapshot. The pure classifier recognizes
|
||||
only persisted recovery ownership, while `maybeNotifyTaskWedge` re-reads
|
||||
live state before a durable claim so recovery cannot produce a false alert.
|
||||
FNXC:TaskWedgeNotifications 2026-08-09-06:30:
|
||||
An update is a snapshot and resume paths can leave a pause marker behind.
|
||||
Use snapshot classification only for generic-failure scheduling; wedge delivery
|
||||
reclassifies the live row immediately before its episode claim.
|
||||
*/
|
||||
const recoveryOwner = task.status === "failed" ? describeTaskRecoveryOwner(task) : null;
|
||||
const wedge = describeTaskWedge(task);
|
||||
@@ -385,7 +385,7 @@ export class NotificationService {
|
||||
only operator notification; dispatch-time suppression below covers races.
|
||||
*/
|
||||
if (wedge) this.cancelPendingFailureNotification(task.id, "classified-terminal-wedge");
|
||||
void this.enqueueWedgeHandling(task.id, () => this.maybeNotifyTaskWedge(task, wedge));
|
||||
void this.enqueueWedgeHandling(task.id, () => this.maybeNotifyTaskWedge(task));
|
||||
void this.maybeSuppressTransientFailedNotification(task, `status=${task.status ?? "undefined"}`);
|
||||
|
||||
/*
|
||||
@@ -548,7 +548,15 @@ export class NotificationService {
|
||||
}
|
||||
return;
|
||||
}
|
||||
const descriptor = suppliedDescriptor ?? describeTaskWedge(liveTask);
|
||||
/*
|
||||
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.
|
||||
*/
|
||||
const descriptor = suppliedDescriptor && isTaskProgressing(liveTask)
|
||||
? null
|
||||
: suppliedDescriptor ?? describeTaskWedge(liveTask);
|
||||
task = liveTask;
|
||||
let episode: string | undefined;
|
||||
if (!descriptor) {
|
||||
|
||||
@@ -108,6 +108,18 @@ export function describeSelfHealingNoActionWedge(task: Task, stage: string, meta
|
||||
return { reasonKey: `self-healing-no-action:${stage}`, ...description };
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:TaskWedgeNotifications 2026-08-09-06:30:
|
||||
Resume paths deliberately retain pause markers for await-input and CLI-approval
|
||||
protocols. A stale marker without real pause state, or any actively progressing
|
||||
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 ?? "");
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:TaskWedgeNotifications 2026-07-22-12:00:
|
||||
Terminal task updates are the shared delivery seam for merger, executor, heartbeat,
|
||||
@@ -115,14 +127,16 @@ and self-healing writers. Classify only states that have no scheduled owner; raw
|
||||
error output is never used as an idempotency key or forwarded into audit metadata.
|
||||
*/
|
||||
export function describeTaskWedge(task: Task): TaskWedgeDescriptor | null {
|
||||
if (isTaskProgressing(task)) return null;
|
||||
const error = task.error ?? "";
|
||||
if (task.pausedReason === "completed-blocked") {
|
||||
const hasPauseProof = task.paused === true || task.status === "paused";
|
||||
if (hasPauseProof && task.pausedReason === "completed-blocked") {
|
||||
return { reasonKey: "completion-blocked", reason: "Completed work is blocked from advancing to review.", action: "Clear the blocker or reset the task to todo." };
|
||||
}
|
||||
if (task.pausedReason === "error-retry-exhausted") {
|
||||
if (hasPauseProof && task.pausedReason === "error-retry-exhausted") {
|
||||
return { reasonKey: "heartbeat-retry-exhausted", reason: "The assigned agent exhausted its heartbeat recovery budget.", action: "Repair the agent configuration, then retry the task." };
|
||||
}
|
||||
if (task.pausedReason === "error-unrecoverable") {
|
||||
if (hasPauseProof && task.pausedReason === "error-unrecoverable") {
|
||||
return { reasonKey: "heartbeat-error-unrecoverable", reason: "The assigned agent needs operator repair before it can resume.", action: "Repair credentials, access, or configuration, then retry the task." };
|
||||
}
|
||||
/*
|
||||
@@ -140,7 +154,7 @@ export function describeTaskWedge(task: Task): TaskWedgeDescriptor | null {
|
||||
"non-retryable-provider-error": { reasonKey: "non-retryable-provider-error", reason: "A non-retryable provider error stopped the task.", action: "Repair provider access or configuration, then retry the task." },
|
||||
"in-review-stall-deadlock": { reasonKey: "in-review-stall-deadlock", reason: "Review stalled in a deadlock that needs operator intervention.", action: "Inspect review ownership and retry or reset to todo." },
|
||||
};
|
||||
if (task.pausedReason && pausedDescriptors[task.pausedReason]) return pausedDescriptors[task.pausedReason];
|
||||
if (hasPauseProof && task.pausedReason && pausedDescriptors[task.pausedReason]) return pausedDescriptors[task.pausedReason];
|
||||
if (task.status !== "failed") return null;
|
||||
if (error.startsWith("EXECUTION_DISPATCH_LOOP_EXHAUSTED")) {
|
||||
return { reasonKey: "execution-dispatch-loop-exhausted", reason: "Execution re-queued without progress until its retry budget was exhausted.", action: "Retry, decompose, or rescope the task." };
|
||||
|
||||
Reference in New Issue
Block a user