From 0e8d7430641e1b382ba2b92cbac3596c52bf049b Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sun, 9 Aug 2026 00:01:44 -0700 Subject: [PATCH] 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) --- .../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 ++-- .../notification/task-wedge-notification.ts | 22 +++- 7 files changed, 137 insertions(+), 29 deletions(-) create mode 100644 .changeset/fn-8867-wedge-false-terminal-alerts.md diff --git a/.changeset/fn-8867-wedge-false-terminal-alerts.md b/.changeset/fn-8867-wedge-false-terminal-alerts.md new file mode 100644 index 0000000000..3fad119618 --- /dev/null +++ b/.changeset/fn-8867-wedge-false-terminal-alerts.md @@ -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. diff --git a/docs/agents.md b/docs/agents.md index a6628cb8ea..32b743cbf1 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -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 diff --git a/docs/architecture.md b/docs/architecture.md index 583fe64549..581d608989 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -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. diff --git a/packages/engine/src/notification/__tests__/notification-service.test.ts b/packages/engine/src/notification/__tests__/notification-service.test.ts index c75fc2194a..59b1ab5cec 100644 --- a/packages/engine/src/notification/__tests__/notification-service.test.ts +++ b/packages/engine/src/notification/__tests__/notification-service.test.ts @@ -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" }), diff --git a/packages/engine/src/notification/__tests__/task-wedge-notification.test.ts b/packages/engine/src/notification/__tests__/task-wedge-notification.test.ts index b411785cea..143f8eceb2 100644 --- a/packages/engine/src/notification/__tests__/task-wedge-notification.test.ts +++ b/packages/engine/src/notification/__tests__/task-wedge-notification.test.ts @@ -25,20 +25,24 @@ const RENAMED_IR = { function fixture(workflowIr?: unknown) { const listeners = new Set(); 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 => ({ 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"], diff --git a/packages/engine/src/notification/notification-service.ts b/packages/engine/src/notification/notification-service.ts index 59e61cb768..b974cc4bcf 100644 --- a/packages/engine/src/notification/notification-service.ts +++ b/packages/engine/src/notification/notification-service.ts @@ -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) { diff --git a/packages/engine/src/notification/task-wedge-notification.ts b/packages/engine/src/notification/task-wedge-notification.ts index c4ec0684d0..4ffb8389f2 100644 --- a/packages/engine/src/notification/task-wedge-notification.ts +++ b/packages/engine/src/notification/task-wedge-notification.ts @@ -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." };