From 6ae9299576d035e7d16befd54e264bdc15ff8177 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Tue, 11 Aug 2026 12:34:47 -0700 Subject: [PATCH] FN-8953: defer terminal wedge alerts during recovery Hold terminal wedge alerts until their recovery window has elapsed. - Persist and settle pending wedge notifications across restarts. - Clear pending alerts on task progress and reconcile expired holds during self-healing. - Expose the settle window in notification settings with coverage for store and notification flows. Files changed: .changeset/fn-8953-wedge-settle-window.md | 7 + AGENTS.md | 1 + docs/architecture.md | 3 +- docs/settings-reference.md | 1 + .../core/src/__tests__/store-wedge-pending.test.ts | 56 +++++ packages/core/src/config/settings-schema.ts | 1 + packages/core/src/store.ts | 46 ++++ packages/core/src/types/settings/settings-scope.ts | 2 + packages/core/src/types/task/task-core.ts | 15 ++ .../app/components/settings/save-split.ts | 1 + .../sections/NotificationsSection.search.ts | 9 + .../settings/sections/NotificationsSection.tsx | 15 ++ .../settings-default-descriptions.test.tsx | 1 + ...self-healing-pending-wedge-notification.test.ts | 148 ++++++++++++ .../__tests__/notification-service.test.ts | 7 +- .../__tests__/task-wedge-notification.test.ts | 258 ++++++++++++++++++++- .../src/notification/notification-service.ts | 260 +++++++++++++++++++++ packages/engine/src/self-healing.ts | 38 +++ packages/i18n/locales/en/app.json | 2 + 19 files changed, 850 insertions(+), 21 deletions(-) Fusion-Task-Id: FN-8953 Fusion-Task-Lineage: fd5b5827-c69d-409f-86d5-01ff23405ee3 Co-authored-by: Fusion (runfusion.ai) --- .changeset/fn-8953-wedge-settle-window.md | 7 + AGENTS.md | 1 + docs/architecture.md | 3 +- docs/settings-reference.md | 1 + .../src/__tests__/store-wedge-pending.test.ts | 56 ++++ packages/core/src/config/settings-schema.ts | 1 + packages/core/src/store.ts | 46 ++++ .../core/src/types/settings/settings-scope.ts | 2 + packages/core/src/types/task/task-core.ts | 15 + .../app/components/settings/save-split.ts | 1 + .../sections/NotificationsSection.search.ts | 9 + .../sections/NotificationsSection.tsx | 15 + .../settings-default-descriptions.test.tsx | 1 + ...healing-pending-wedge-notification.test.ts | 148 ++++++++++ .../__tests__/notification-service.test.ts | 7 +- .../__tests__/task-wedge-notification.test.ts | 258 +++++++++++++++++- .../src/notification/notification-service.ts | 258 +++++++++++++++++- packages/engine/src/self-healing.ts | 38 +++ packages/i18n/locales/en/app.json | 2 + 19 files changed, 849 insertions(+), 20 deletions(-) create mode 100644 .changeset/fn-8953-wedge-settle-window.md create mode 100644 packages/core/src/__tests__/store-wedge-pending.test.ts create mode 100644 packages/engine/src/__tests__/self-healing-pending-wedge-notification.test.ts diff --git a/.changeset/fn-8953-wedge-settle-window.md b/.changeset/fn-8953-wedge-settle-window.md new file mode 100644 index 0000000000..245920cc5e --- /dev/null +++ b/.changeset/fn-8953-wedge-settle-window.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": minor +--- + +summary: Hold terminal-failure alerts until the task fails to recover on its own. +category: fix +dev: Adds wedgeNotificationSettleMs, durable pending wedge evidence with stale-hold re-stamping, and a self-healing backstop. diff --git a/AGENTS.md b/AGENTS.md index 11ac444081..c7dfbfe2aa 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -321,6 +321,7 @@ Scoped exception (FN-5819/FN-8823): while project auto-merge is On, shared-branc - FN-8492: the self-healing sweep `reconcile-orphaned-pending-step-results` (startup, right after legacy adoption, plus periodic maintenance) emits `task:reconcile-orphaned-pending-step-results` when it REWRITES `pending` workflow-step results with no live session behind them to `failed` (canonical liveness triple: `activeSessionRegistry` path, `executingTaskLock`, `isTaskActive`). It must never DELETE an orphaned entry — the merge gate blocks on pending/failed results, not on an enabled step with no result, so deletion silently satisfies the gate and the task merges with its review skipped; the `failed` rewrite keeps the gate closed and hands re-run/bypass to the failed-pre-merge-steps recovery and FN-7720 operator-bypass paths. `in-progress` rows are always skipped (executor-owned; resume is deferred at startup), the row is re-read immediately before the write, and user pauses are never disturbed. Metadata is ids/counts-only (`taskId`, `column`, `orphanedCount`, `resultCount`). - FN-8923: self-healing emits `task:reconcile-principal-held-planning` when it re-queues planning for a card whose only active continuation is a `held` triage-role row blocked on principal routing (`workflow-principal-*`), and deduped `task:reconcile-principal-held-planning-no-action` for a live session or a lost race. A planning hold has no other retry owner: triage re-admits a hold-column card only on `status: "needs-replan"`, so the hold survives exactly as long as that status does. Candidacy requires the planning lane (hold/intake traits), effective auto-merge on, an owned (null) status, and a grace window; the re-read and write run under the shared planning lifecycle lock. Metadata stays ids/counts/outcomes-only (`taskId`, `column`, `nodeId`, `blockedReason`, `stalenessMs`, optional `reason`). This makes `self-healing.ts` a second writer of `needs-replan` alongside `executor.ts`; the U10b note below describes the graph's replan loop, which this sweep feeds rather than replaces. - FN-8356: self-healing emits `task:reconcile-stale-duplicate-decision` when it clears a triage-marker duplicate-decision pause against a missing, deleted, done, or archived canonical. Metadata is ids/outcomes-only (`taskId`, `canonicalId`, `canonicalColumn`, `canonicalDeleted`, `priorPausedReason`); active canonical decisions and user pauses remain untouched. +- FN-8953: self-healing emits `task:reconcile-pending-wedge-notification` for a durable deferred wedge hold. Metadata is ids/counts/outcomes-only (`taskId`, `reasonKey`, `pendingAgeMs`, `outcome`), where outcome is `delivered`, `suppressed`, `cleared`, `rearmed`, `held`, `absent`, `unreadable`, `deferred`, or `failed`; descriptor prose is never recorded. - U9b (R10/KTD-8): the self-healing STARTUP recovery step `adopt-legacy-task-rows` emits `task:reconcile-legacy-adoption` when it adopts a pre-cutover row through the KTD-8 adoption table (clearing a legacy `task.status` whose writer the cutover deleted so the graph re-enters at its owning node, and/or landing the one-time `reviewLevel` -> `enabledWorkflowSteps` preset backfill), and `task:reconcile-legacy-adoption-unmappable` when an UNKNOWN status parks the row `paused` for a human with its status deliberately left in place. Metadata is ids/counts/outcomes-only (`taskId`, `action`, `priorStatus`, `column`, `backfilledStepCount`, `reason`), where `reason` is a fixed adoption-table note and never row prose. Adoption runs FIRST in startup recovery (every later step reasons about `task.status`), stamps `task.legacyAdoptedAt` only on rows it actually mutates (so upgrade does not mass-write every `done` row), and never touches a user pause or a `preserve` gate. `planLegacyAdoption` in `packages/core/src/legacy-adoption.ts` is the single shared decision used by both this sweep and the store-open reconcile so the two cannot drift. - U10 (R9): the pre-graph cutover machinery is DELETED and stays deleted, ratcheted by `packages/engine/src/__tests__/legacy-tombstones.test.ts`. Gone: `workflow-cutover.ts`, `workflow-authoritative-driver.ts`, `workflow-parity-observer.ts`, the `graphCompletionInterceptors` re-entry map, triage's out-of-graph `runPlanReviewBeforeExecution` gate, and the in-session `fn_review_step` tool with its RETHINK git-reset/session-rewind, per-step conversation checkpoints, deferred reviewer provider-error channel, and review-level prompt scaffolding. Plan/code/browser review are owned EXCLUSIVELY by workflow-graph nodes — do not re-introduce a second review authority inside the implementation session; that duplicate-Plan-Review race is what the cutover removed. The tombstone test strips comments before searching, so the FNXC notes that explain each deletion are expected to remain in source while the code must not. - U10b (R9): `maybeExecuteWorkflowGraph`'s legacy fallback is DELETED. A TaskStore without `getTaskWorkflowSelection`/`getTaskWorkflowSelectionAsync` no longer falls back to a legacy execute path — graph ownership is unconditional, `graphCompletion` is mandatory rather than optional, and the three completion boundaries that used to branch on its absence are plain returns. `transferPreHeldToLegacy` and the pre-held-slot hand-off to the legacy path are gone with it. Consequence for tests: nothing can reach the pre-graph shape by deleting the selection readers from a mock store; a test that needs "this store cannot resolve a workflow" must assert the fail-closed park, not a fallback. diff --git a/docs/architecture.md b/docs/architecture.md index e179f393c5..aad9b3f8bb 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -16,7 +16,7 @@ Actionable terminal task updates are classified into bounded reasons such as a n 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. +Each task also stores `lastNotifiedAtByReason`, an independent timestamp map keyed by bounded reason. Deferred wedge delivery uses a durable pending marker and a bounded self-healing backstop: it revalidates the live task before dispatch, clears recovery evidence silently, and re-stamps holds older than its maintenance-derived horizon rather than alerting on stale process-era evidence. `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. Generic `terminal-failed` parks are engine-owned before they become operator work. A durable `wedgeNotification.autoRecovery` budget supplies bounded attempts and backoff; its claim state and rotating apply fence ensure one observer is authorized to clear and requeue a park. The grace window detects an abandoned apply only—it is not a lock. An exhausted budget receives one reason-scoped escalation, confirmed durably at the shared dispatch seam; a write-once exhaustion marker prevents an earlier drain alert or cooldown suppression from satisfying that escalation. The budget resets on terminal success, archive, explicit operator Retry, soft-delete, or a sufficiently old foreign write. Disabling auto-recovery drains an owed alert without consuming the remaining retries; re-enabling resumes them. @@ -1155,6 +1155,7 @@ The run-audit system records every mutation performed by the engine across four - **Database / `task:reattach-orphaned-execution`** — emitted by `reattachOrphanedAssignedExecutions` (FN-6336) when self-healing re-dispatches an idle assigned `in-progress` task forward via `executor.resumeTaskForAgent(agentId)` after proving the assigned agent has no active heartbeat run or active execution. - **Database / `task:reconcile-stale-agent-assignment`** — emitted when self-healing or heartbeat reconciliation clears stale durable `Agent.taskId`/`state` for a task parked in `todo`/`triage` without live execution proof. Metadata includes `{ agentId, taskId, taskColumn, agentState, status, blockedBy, overlapBlockedBy, hadFreshRun, hadActiveExecution, reason }`; task queue/lease fields are preserved. - **Database / `task:reconcile-stale-duplicate-decision`** — emitted when self-healing clears a triage-marker duplicate-decision pause whose canonical is missing, deleted, done, or archived. Metadata is ids/outcomes-only: `{ taskId, canonicalId, canonicalColumn, canonicalDeleted, priorPausedReason }`; active canonicals and user pauses are excluded. +- **Database / `task:reconcile-pending-wedge-notification`** — a bounded startup and maintenance sweep completes a restart-durable pending wedge hold through `NotificationService`, which alone revalidates the live subject and owns delivery. Holds older than the derived stale horizon are re-stamped rather than delivered on old evidence; metadata is ids/counts/outcomes-only: `{ taskId, reasonKey, pendingAgeMs, outcome }` where outcome is `delivered`, `suppressed`, `cleared`, `rearmed`, `held`, `absent`, `unreadable`, `deferred`, or `failed`. - **Database / `task:soft-delete-column-reconciled`** — emitted by `reconcileSoftDeletedColumnDrift` (FN-5566, re-land FN-5446) when a soft-deleted row (`deletedAt IS NOT NULL`) is found with legacy `column != 'archived'`; rewrites only `column` (no resurrection), with metadata `{ previousColumn }`. - **Database / `session:runtime-resolved`** — emitted once per `createResolvedAgentSession` call with metadata `{ sessionPurpose, runtimeId, wasConfigured, provider, modelId, mockProviderActive, testModeActive, runtimeHint?, credentialInstanceId?, credentialInstanceMissing?, requestedCredentialInstanceId?, resolvedCredentialInstanceId? }` for per-lane runtime/provider attribution. Credential fields are ids/outcomes-only; a missing selected instance is visibly recorded when the canonical provider default is used. - **Database / `task:reconcile-dependency-blocking-lease`** — emitted by `reconcileDependencyBlockingLeases()` (FN-6292) when self-healing rebounds an `in-progress` holder to `todo` because an unmet dependency is blocked by the holder's stale file-scope lease. Metadata includes the dependency ID, blocked-by marker, and unmet dependency list. diff --git a/docs/settings-reference.md b/docs/settings-reference.md index 64864083f2..8e9dcc599c 100644 --- a/docs/settings-reference.md +++ b/docs/settings-reference.md @@ -95,6 +95,7 @@ Fallback thinking-level values are applied at runtime when Fusion swaps from the | `agentClarificationEnabled` | `boolean` | `false` | Legacy default for programmatic Planning Mode session notification eligibility. Dashboard Planning Mode always starts its infinite, user-validated interview with follow-up questions enabled; this setting no longer suppresses questions or creates a final summary. | | `failureNotificationMode` | `"sticky-only" \| "terminal-only" \| "all"` | `"sticky-only"` | Failure notification behavior. `sticky-only` defers failed-task notifications by `failureNotificationDelayMs` and suppresses transient self-recoveries. `terminal-only` suppresses while auto-retry is still active and only dispatches when `paused === true` or `column === "in-review"` with `status === "failed"`. `all` restores legacy immediate failure notifications. | | `failureNotificationDelayMs` | `number` | `30000` | Delay window (ms) before evaluating/sending a `failed` notification in `sticky-only` and `terminal-only` modes. Set `0` for immediate dispatch in legacy `all` mode. | +| `wedgeNotificationSettleMs` | `number` | `300000` | Time a terminal wedge must persist before its operator alert. `0` restores immediate delivery and clears outstanding holds. Production stores persist holds across restart; lightweight stores retain them only in memory. A stale hold is re-stamped using a horizon derived from the 15-minute maintenance interval, so an outage can delay a still-wedged alert rather than dispatching on stale evidence. | | `ntfyTopic` | `string` | `undefined` | ntfy topic name. | | `ntfyBaseUrl` | `string` | `undefined` | Optional custom ntfy server base URL (must use `http://` or `https://`). If blank/unset, Fusion uses `https://ntfy.sh` for both runtime and test notifications. | | `ntfyAccessToken` | `string` | `undefined` | Optional ntfy access token. When set, Fusion sends `Authorization: Bearer ` with ntfy publish requests, including Settings → Notifications test sends. Leave blank/unset to publish without authentication. | diff --git a/packages/core/src/__tests__/store-wedge-pending.test.ts b/packages/core/src/__tests__/store-wedge-pending.test.ts new file mode 100644 index 0000000000..bef5ad9fee --- /dev/null +++ b/packages/core/src/__tests__/store-wedge-pending.test.ts @@ -0,0 +1,56 @@ +import { afterAll, afterEach, beforeAll, beforeEach, expect, it } from "vitest"; +import { createSharedPgTaskStoreTestHarness, pgDescribe, type SharedPgTaskStoreHarness } from "../__test-utils__/pg-test-harness.js"; + +/* +FNXC:TaskWedgeNotifications 2026-08-11-18:57: +The durable marker is deliberately separate from episode state and cooldown history. These store +assertions pin the CAS boundary so engine timers can defer an alert without a restart dropping it +or a claim/resolve accidentally being treated as a pending-marker clear. +*/ +pgDescribe("TaskStore deferred wedge marker", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ prefix: "fusion_wedge_pending" }); + + beforeAll(h.beforeAll); + beforeEach(h.beforeEach); + afterEach(h.afterEach); + afterAll(h.afterAll); + + async function seed(id: string) { + const store = h.store(); + await store.createTaskWithReservedId({ description: "deferred wedge", column: "in-review" }, { taskId: id, applyDefaultWorkflowSteps: false }); + return store; + } + + const descriptor = { reasonKey: "terminal-failed", source: "auto" as const, reason: "terminal", action: "repair" }; + + it("arms once, restamps stale evidence, and clears without changing episode state", async () => { + const store = await seed("FN-WEDGE-001"); + const first = await store.markTaskWedgeNotificationPending("FN-WEDGE-001", descriptor); + const second = await store.markTaskWedgeNotificationPending("FN-WEDGE-001", descriptor); + expect(second).toMatchObject({ since: first.since, armed: false, restamped: false }); + + const stale = await store.markTaskWedgeNotificationPending("FN-WEDGE-001", descriptor, { staleAfterMs: -1 }); + expect(stale).toMatchObject({ armed: true, restamped: true }); + expect(await store.clearTaskWedgeNotificationPending("FN-WEDGE-001", "other")).toBe(false); + expect(await store.clearTaskWedgeNotificationPending("FN-WEDGE-001")).toBe(true); + const task = await store.getTask("FN-WEDGE-001"); + expect(task.wedgeNotification).toMatchObject({ reasonKey: "terminal-failed", status: "resolved", episodeId: "" }); + expect(task.wedgeNotification?.pending).toBeUndefined(); + }); + + it("drops pending evidence on delivered and cooldown-suppressed claims but not resolve", async () => { + const store = await seed("FN-WEDGE-002"); + await store.markTaskWedgeNotificationPending("FN-WEDGE-002", descriptor); + expect(await store.claimTaskWedgeNotificationEpisode("FN-WEDGE-002", "terminal-failed")).toMatchObject({ claimed: true }); + expect((await store.getTask("FN-WEDGE-002")).wedgeNotification?.pending).toBeUndefined(); + + await store.claimTaskWedgeNotificationEpisode("FN-WEDGE-002", null); + await store.markTaskWedgeNotificationPending("FN-WEDGE-002", descriptor); + expect(await store.claimTaskWedgeNotificationEpisode("FN-WEDGE-002", "terminal-failed")).toMatchObject({ claimed: false }); + expect((await store.getTask("FN-WEDGE-002")).wedgeNotification?.pending).toBeUndefined(); + + await store.markTaskWedgeNotificationPending("FN-WEDGE-002", { ...descriptor, reasonKey: "tool-failure" }); + await store.claimTaskWedgeNotificationEpisode("FN-WEDGE-002", null); + expect((await store.getTask("FN-WEDGE-002")).wedgeNotification?.pending).toBeDefined(); + }); +}); diff --git a/packages/core/src/config/settings-schema.ts b/packages/core/src/config/settings-schema.ts index ece70f35ab..0f64833a62 100644 --- a/packages/core/src/config/settings-schema.ts +++ b/packages/core/src/config/settings-schema.ts @@ -193,6 +193,7 @@ export const DEFAULT_GLOBAL_SETTINGS = { ntfyDashboardHost: undefined, taskTokenBudget: undefined, failureNotificationDelayMs: 30000, + wedgeNotificationSettleMs: 300000, failureNotificationMode: "sticky-only", webhookEnabled: false, webhookUrl: undefined, diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index f919968b58..39f8877579 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -1763,6 +1763,52 @@ export class TaskStore extends EventEmitter { }); return result; } + /** Atomically records the durable evidence needed to complete a deferred wedge notification. */ + async markTaskWedgeNotificationPending( + taskId: string, + descriptor: { reasonKey: string; source: "auto" | "supplied"; reason: string; action: string; gate?: string }, + options?: { staleAfterMs?: number }, + ): Promise<{ since: string; armed: boolean; restamped: boolean }> { + const now = new Date().toISOString(); + let result = { since: now, armed: false, restamped: false }; + await this.updateTaskAtomic(taskId, (current) => { + const prior = current.wedgeNotification; + const pending = prior?.pending; + if (prior?.status === "active" && prior.reasonKey === descriptor.reasonKey) { + return null; + } + const pendingSince = pending ? Date.parse(pending.since) : Number.NaN; + const stale = pending?.reasonKey === descriptor.reasonKey + && typeof options?.staleAfterMs === "number" + && Number.isFinite(options.staleAfterMs) + && Number.isFinite(pendingSince) + && Date.now() - pendingSince > options.staleAfterMs; + if (pending?.reasonKey === descriptor.reasonKey && !stale) { + result = { since: pending.since, armed: false, restamped: false }; + return null; + } + const nextPending = { since: now, ...descriptor }; + result = { since: now, armed: true, restamped: pending != null }; + return { + wedgeNotification: prior + ? { ...prior, pending: nextPending } + : { reasonKey: descriptor.reasonKey, episodeId: "", status: "resolved", transitionedAt: now, pending: nextPending }, + }; + }); + return result; + } + /** Removes deferred wedge evidence without changing the active episode or cooldown history. */ + async clearTaskWedgeNotificationPending(taskId: string, reasonKey?: string): Promise { + let cleared = false; + await this.updateTaskAtomic(taskId, (current) => { + const prior = current.wedgeNotification; + if (!prior?.pending || (reasonKey !== undefined && prior.pending.reasonKey !== reasonKey)) return null; + const { pending: _pending, ...withoutPending } = prior; + cleared = true; + return { wedgeNotification: withoutPending }; + }); + return cleared; + } /* FNXC:TaskWedgeNotifications 2026-08-10-18:54: Generic terminal parks use a dedicated durable budget because transient recovery diff --git a/packages/core/src/types/settings/settings-scope.ts b/packages/core/src/types/settings/settings-scope.ts index 22afbab0d0..da39f26a2f 100644 --- a/packages/core/src/types/settings/settings-scope.ts +++ b/packages/core/src/types/settings/settings-scope.ts @@ -474,6 +474,8 @@ export interface GlobalSettings { /** How long a task must remain in `status='failed'` before a push notification fires. * Set to 0 to dispatch immediately (legacy behavior). Default: 30000 ms. */ failureNotificationDelayMs?: number; + /** How long a wedge must persist before an operator alert. 0 restores immediate legacy delivery; default 300000 ms. */ + wedgeNotificationSettleMs?: number; /** `sticky-only` (default) defers failure notifications by `failureNotificationDelayMs` * and suppresses them if the task self-recovers. `all` restores the legacy * immediate-dispatch behavior. `terminal-only` suppresses failure notifications diff --git a/packages/core/src/types/task/task-core.ts b/packages/core/src/types/task/task-core.ts index 1846a0891d..aa14f8443a 100644 --- a/packages/core/src/types/task/task-core.ts +++ b/packages/core/src/types/task/task-core.ts @@ -596,6 +596,21 @@ export interface TaskWedgeNotificationState { lastNotifiedAtByReason?: Record; /** Monotonic durable-state version used to reject stale whole-object writes. */ budgetRevision?: number; + /* + FNXC:TaskWedgeNotifications 2026-08-11-18:28: + A parked row may emit no later task update, so a settle hold must survive an engine restart + rather than relying solely on an in-memory timer. Supplied self-healing descriptors include + stage-derived prose unavailable on the task row, so preserve that operator-facing content here. + `since` moves only when the reason changes or stale evidence is deliberately re-stamped. + */ + pending?: { + since: string; + reasonKey: string; + source: "auto" | "supplied"; + reason: string; + action: string; + gate?: string; + }; /** Sweep-owned generic terminal-failure recovery state. */ autoRecovery?: { attempts: number; diff --git a/packages/dashboard/app/components/settings/save-split.ts b/packages/dashboard/app/components/settings/save-split.ts index 9c80b7ac01..84da30b4bb 100644 --- a/packages/dashboard/app/components/settings/save-split.ts +++ b/packages/dashboard/app/components/settings/save-split.ts @@ -120,6 +120,7 @@ export const GLOBAL_SECTION_KEYS: Record> = { "ntfyEvents", "ntfyDashboardHost", "failureNotificationDelayMs", + "wedgeNotificationSettleMs", "failureNotificationMode", "webhookEnabled", "webhookUrl", diff --git a/packages/dashboard/app/components/settings/sections/NotificationsSection.search.ts b/packages/dashboard/app/components/settings/sections/NotificationsSection.search.ts index d7e8670a6a..d24e3f48ea 100644 --- a/packages/dashboard/app/components/settings/sections/NotificationsSection.search.ts +++ b/packages/dashboard/app/components/settings/sections/NotificationsSection.search.ts @@ -32,6 +32,15 @@ export const notificationsSearchEntries: SettingsSearchEntry[] = [ " How long a failure must persist before a push notification is sent. 0 = notify immediately. Default: 30000 (30 seconds). ", keywords: ["debounce", "wait", "throttle"], }, + { + sectionId: "notifications", + key: "wedgeNotificationSettleMs", + labelKey: "settings.notifications.wedgeNotificationSettleMs", + labelFallback: "Terminal-wedge settle window (ms)", + helpKey: "settings.notifications.wedgeNotificationSettleMsHelp", + helpFallback: "How long a terminal failure must persist before an operator alert. 0 = notify immediately. Default: 300000 (5 minutes).", + keywords: ["noise", "debounce", "wedge", "recovered"], + }, { sectionId: "notifications", key: "ntfyDashboardHost", diff --git a/packages/dashboard/app/components/settings/sections/NotificationsSection.tsx b/packages/dashboard/app/components/settings/sections/NotificationsSection.tsx index a1b5c4ea24..76dbab843e 100644 --- a/packages/dashboard/app/components/settings/sections/NotificationsSection.tsx +++ b/packages/dashboard/app/components/settings/sections/NotificationsSection.tsx @@ -114,6 +114,21 @@ export function NotificationsSection({ form, setForm, testNotificationLoading, t failureNotificationDelayMs: v !== null && Number.isFinite(v) && v >= 0 ? v : 0, }))} /> + setForm((f) => ({ + ...f, + wedgeNotificationSettleMs: v !== null && Number.isFinite(v) && v >= 0 ? v : 0, + }))} + /> diff --git a/packages/dashboard/app/components/settings/sections/__tests__/settings-default-descriptions.test.tsx b/packages/dashboard/app/components/settings/sections/__tests__/settings-default-descriptions.test.tsx index 6f8ce98932..bfe21d94fb 100644 --- a/packages/dashboard/app/components/settings/sections/__tests__/settings-default-descriptions.test.tsx +++ b/packages/dashboard/app/components/settings/sections/__tests__/settings-default-descriptions.test.tsx @@ -141,6 +141,7 @@ const SETTING_DESCRIPTION_KEYS: Record = { */ agentClarificationEnabled: "notifications.agentClarificationHint", failureNotificationDelayMs: "notifications.howLongAFailureMustPersistBeforeA", + wedgeNotificationSettleMs: "notifications.wedgeNotificationSettleMsHelp", failureNotificationMode: "notifications.stickyFailuresOnlyDefault", ntfyEnabled: "notifications.ntfyEnabledHint", ntfyTopic: "notifications.yourNtfyShTopicName164Alphanumeric", diff --git a/packages/engine/src/__tests__/self-healing-pending-wedge-notification.test.ts b/packages/engine/src/__tests__/self-healing-pending-wedge-notification.test.ts new file mode 100644 index 0000000000..558f71a399 --- /dev/null +++ b/packages/engine/src/__tests__/self-healing-pending-wedge-notification.test.ts @@ -0,0 +1,148 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { EventEmitter } from "node:events"; +import type { Settings, Task, TaskStore } from "@fusion/core"; + +const { completePendingWedgeNotificationMock, getActiveNotificationServiceMock, recordRunAuditEventMock } = vi.hoisted(() => ({ + completePendingWedgeNotificationMock: vi.fn(), + getActiveNotificationServiceMock: vi.fn(), + recordRunAuditEventMock: vi.fn(async () => undefined), +})); +vi.mock("../util/notifier.js", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, getActiveNotificationService: getActiveNotificationServiceMock }; +}); +vi.mock("../util/run-audit.js", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, createRunAuditor: vi.fn(() => ({ database: recordRunAuditEventMock })) }; +}); + +import { SelfHealingManager } from "../self-healing.js"; +import { NotificationService } from "../notification/notification-service.js"; + +function pendingTask(id: string, since: string): Task { + return { + id, title: id, description: "", column: "in-review", status: "failed", dependencies: [], steps: [], currentStep: 0, log: [], createdAt: since, updatedAt: since, + wedgeNotification: { reasonKey: "terminal-failed", episodeId: "", status: "resolved", transitionedAt: since, pending: { since, reasonKey: "terminal-failed", source: "auto", reason: "terminal", action: "repair" } }, + } as Task; +} + +/* +FNXC:TaskWedgeNotifications 2026-08-11-18:57: +The restart sweep is only a bounded durable-marker driver. It records the NotificationService's +verbatim outcome rather than reconstructing a descriptor or guessing recovery from a slim row. +*/ +describe("reconcile pending wedge notifications", () => { + beforeEach(() => { + vi.clearAllMocks(); + getActiveNotificationServiceMock.mockReturnValue({ getWedgeNotificationSettleMs: () => 1_000, completePendingWedgeNotification: completePendingWedgeNotificationMock }); + }); + + it("selects elapsed markers and audits the completion outcome verbatim", async () => { + const old = new Date(Date.now() - 1_001).toISOString(); + const young = new Date(Date.now() - 999).toISOString(); + const store = Object.assign(new EventEmitter(), { + getSettings: vi.fn(async () => ({ globalPause: false, enginePaused: false } as Settings)), + listTasks: vi.fn(async () => [pendingTask("FN-WEDGE-OLD", old), pendingTask("FN-WEDGE-YOUNG", young)]), + }) as unknown as TaskStore; + completePendingWedgeNotificationMock.mockResolvedValue({ outcome: "rearmed" }); + const manager = new SelfHealingManager(store, { rootDir: "/repo" }); + + await expect(manager.reconcilePendingWedgeNotifications()).resolves.toBe(1); + expect(completePendingWedgeNotificationMock).toHaveBeenCalledWith("FN-WEDGE-OLD"); + expect(recordRunAuditEventMock).toHaveBeenCalledWith(expect.objectContaining({ + type: "task:reconcile-pending-wedge-notification", + metadata: expect.objectContaining({ taskId: "FN-WEDGE-OLD", reasonKey: "terminal-failed", outcome: "rearmed" }), + })); + }); + + it("delivers a durable marker through a fresh service after a simulated restart", async () => { + const since = new Date(Date.now() - 1_001).toISOString(); + let task = { ...pendingTask("FN-WEDGE-RESTART", since), error: "boom" }; + const dispatch = vi.fn(async () => ({ success: true, providerId: "test" })); + const sendMessageOnce = vi.fn(async () => ({ message: {} as any, inserted: true })); + const store = Object.assign(new EventEmitter(), { + getSettings: vi.fn(async () => ({ globalPause: false, enginePaused: false, ntfyEnabled: true, ntfyTopic: "test", wedgeNotificationSettleMs: 1_000 } as Settings)), + getTask: vi.fn(async () => task), + listTasks: vi.fn(async () => [task]), + claimTaskWedgeNotificationEpisode: vi.fn(async (taskId: string, reasonKey: string | null) => { + if (reasonKey === null) return { claimed: false }; + task = { ...task, wedgeNotification: { reasonKey, episodeId: `${taskId}:${reasonKey}`, status: "active", transitionedAt: new Date().toISOString() } }; + return { claimed: true, episodeId: `${taskId}:${reasonKey}` }; + }), + clearTaskWedgeNotificationPending: vi.fn(async () => { + const { pending: _pending, ...wedge } = task.wedgeNotification!; + task = { ...task, wedgeNotification: wedge }; + return true; + }), + on: vi.fn(), off: vi.fn(), + }) as unknown as TaskStore; + const service = new NotificationService(store as any, { messageStore: { on: () => undefined, sendMessageOnce } as any, wedgeNotificationSettleMs: 1_000 }); + service.registerProvider({ getProviderId: () => "test", isEventSupported: () => true, sendNotification: dispatch }); + await service.start(); + getActiveNotificationServiceMock.mockReturnValue(service); + const manager = new SelfHealingManager(store, { rootDir: "/repo" }); + + await expect(manager.reconcilePendingWedgeNotifications()).resolves.toBe(1); + expect(dispatch).toHaveBeenCalledTimes(1); + expect(sendMessageOnce).toHaveBeenCalledTimes(1); + expect(task.wedgeNotification?.pending).toBeUndefined(); + expect(recordRunAuditEventMock).toHaveBeenCalledWith(expect.objectContaining({ metadata: expect.objectContaining({ outcome: "delivered" }) })); + await service.stop(); + }); + + it("re-stamps an ancient durable hold then delivers it on the following maintenance tick", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-11T12:00:00.000Z")); + const since = new Date(Date.now() - 5_000).toISOString(); + let task = { ...pendingTask("FN-WEDGE-STALE", since), error: "boom" }; + const dispatch = vi.fn(async () => ({ success: true, providerId: "test" })); + const store = Object.assign(new EventEmitter(), { + getSettings: vi.fn(async () => ({ globalPause: false, enginePaused: false, ntfyEnabled: true, ntfyTopic: "test", wedgeNotificationSettleMs: 1_000, maintenanceIntervalMs: 1_000 } as Settings)), + getTask: vi.fn(async () => task), listTasks: vi.fn(async () => [task]), on: vi.fn(), off: vi.fn(), + markTaskWedgeNotificationPending: vi.fn(async (_id: string, descriptor: any) => { + const restamped = new Date().toISOString(); + task = { ...task, wedgeNotification: { ...task.wedgeNotification!, pending: { since: restamped, ...descriptor } } }; + return { since: restamped, armed: true, restamped: true }; + }), + clearTaskWedgeNotificationPending: vi.fn(async () => { + const { pending: _pending, ...wedge } = task.wedgeNotification!; + task = { ...task, wedgeNotification: wedge }; + return true; + }), + claimTaskWedgeNotificationEpisode: vi.fn(async (taskId: string, reasonKey: string | null) => { + if (reasonKey === null) return { claimed: false }; + task = { ...task, wedgeNotification: { reasonKey, episodeId: `${taskId}:${reasonKey}`, status: "active", transitionedAt: new Date().toISOString() } }; + return { claimed: true, episodeId: `${taskId}:${reasonKey}` }; + }), + }) as unknown as TaskStore; + const service = new NotificationService(store as any, { wedgeNotificationSettleMs: 1_000 }); + service.registerProvider({ getProviderId: () => "test", isEventSupported: () => true, sendNotification: dispatch }); + await service.start(); + getActiveNotificationServiceMock.mockReturnValue(service); + const manager = new SelfHealingManager(store, { rootDir: "/repo" }); + + await manager.reconcilePendingWedgeNotifications(); + expect(dispatch).not.toHaveBeenCalled(); + expect(recordRunAuditEventMock).toHaveBeenCalledWith(expect.objectContaining({ metadata: expect.objectContaining({ outcome: "rearmed" }) })); + await vi.advanceTimersByTimeAsync(1_001); + await manager.reconcilePendingWedgeNotifications(); + expect(dispatch).toHaveBeenCalledTimes(1); + expect(recordRunAuditEventMock).toHaveBeenCalledWith(expect.objectContaining({ metadata: expect.objectContaining({ outcome: "delivered" }) })); + await service.stop(); + vi.useRealTimers(); + }); + + it("leaves a marker untouched when no notification service is active", async () => { + getActiveNotificationServiceMock.mockReturnValue(undefined); + const since = new Date(Date.now() - 300_001).toISOString(); + const store = Object.assign(new EventEmitter(), { + getSettings: vi.fn(async () => ({ globalPause: false, enginePaused: false } as Settings)), + listTasks: vi.fn(async () => [pendingTask("FN-WEDGE-DEFERRED", since)]), + }) as unknown as TaskStore; + const manager = new SelfHealingManager(store, { rootDir: "/repo" }); + + await expect(manager.reconcilePendingWedgeNotifications()).resolves.toBe(1); + expect(completePendingWedgeNotificationMock).not.toHaveBeenCalled(); + expect(recordRunAuditEventMock).toHaveBeenCalledWith(expect.objectContaining({ metadata: expect.objectContaining({ outcome: "deferred" }) })); + }); +}); diff --git a/packages/engine/src/notification/__tests__/notification-service.test.ts b/packages/engine/src/notification/__tests__/notification-service.test.ts index 42a96d2827..70de1498d7 100644 --- a/packages/engine/src/notification/__tests__/notification-service.test.ts +++ b/packages/engine/src/notification/__tests__/notification-service.test.ts @@ -114,7 +114,7 @@ describe("NotificationService deferred failure notifications", () => { markTerminalFailureAutoRecoveryBudgetExhausted: marker, markTerminalFailureAutoRecoveryEscalationDelivered: stamp, }); - const service = new NotificationService(store as any); + const service = new NotificationService(store as any, { wedgeNotificationSettleMs: 0 }); await service.start(); const exhausted = task({ id: "FN-suppressed-exhaustion", @@ -166,7 +166,7 @@ describe("NotificationService deferred failure notifications", () => { isEventSupported: () => true, sendNotification, }; - const service = new NotificationService(store as any, { failedNotificationGraceMs: 100 }); + const service = new NotificationService(store as any, { failedNotificationGraceMs: 100, wedgeNotificationSettleMs: 0 }); service.registerProvider(provider); await service.start(); return { store, service, sendNotification }; @@ -196,6 +196,7 @@ describe("NotificationService deferred failure notifications", () => { const service = new NotificationService(store as any, { messageStore: { on: () => undefined, sendMessageOnce } as any, failedNotificationGraceMs: 100, + wedgeNotificationSettleMs: 0, }); service.registerProvider({ getProviderId: () => "mock", isEventSupported: () => true, sendNotification }); await service.start(); @@ -238,7 +239,7 @@ describe("NotificationService deferred failure notifications", () => { expect(sendNotification).toHaveBeenCalledTimes(1); await service.stop(); - const restarted = new NotificationService(store as any, { messageStore: { on: () => undefined, sendMessageOnce } as any }); + const restarted = new NotificationService(store as any, { messageStore: { on: () => undefined, sendMessageOnce } as any, wedgeNotificationSettleMs: 0 }); restarted.registerProvider({ getProviderId: () => "restarted", isEventSupported: () => true, sendNotification }); await restarted.start(); store.emit("task:updated", exhausted); 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 df13bad459..b42d74ff9d 100644 --- a/packages/engine/src/notification/__tests__/task-wedge-notification.test.ts +++ b/packages/engine/src/notification/__tests__/task-wedge-notification.test.ts @@ -22,8 +22,9 @@ const RENAMED_IR = { ], }; -function fixture(workflowIr?: unknown) { +function fixture(workflowIr?: unknown, settleMs = 0, maintenanceIntervalMs?: number) { const listeners = new Set(); + const movedListeners = new Set<(data: { task: Task; from: string; to: string }) => void>(); let wedge: Task["wedgeNotification"]; let liveTask: Task | undefined; let liveReadError: Error | undefined; @@ -37,22 +38,43 @@ function fixture(workflowIr?: unknown) { return { claimed: true, episodeId: wedge.episodeId }; }); const store = { - getSettings: async () => ({ ntfyEnabled: true, ntfyTopic: "test" }) as Settings, + getSettings: async () => ({ ntfyEnabled: true, ntfyTopic: "test", ...(maintenanceIntervalMs === undefined ? {} : { maintenanceIntervalMs }) }) as Settings, getTask: async () => { if (liveReadError) throw liveReadError; return liveTask; }, - on: (event: string, listener: Listener) => { if (event === "task:updated") listeners.add(listener); }, + on: (event: string, listener: Listener | ((data: { task: Task; from: string; to: string }) => void)) => { + if (event === "task:updated") listeners.add(listener as Listener); + if (event === "task:moved") movedListeners.add(listener as (data: { task: Task; from: string; to: string }) => void); + }, off: () => undefined, emit: (task: Task) => listeners.forEach((listener) => listener(task)), + emitMoved: (task: Task, from: string, to: string) => movedListeners.forEach((listener) => listener({ task, from, to })), setLiveTask: (next: Task | undefined) => { liveTask = next; }, setLiveReadError: (next: Error | undefined) => { liveReadError = next; }, claimTaskWedgeNotificationEpisode, + markTaskWedgeNotificationPending: vi.fn(async (_taskId: string, descriptor: any, options?: { staleAfterMs?: number }) => { + const now = new Date().toISOString(); + const pending = wedge?.pending; + const stale = pending !== undefined && pending.reasonKey === descriptor.reasonKey && typeof options?.staleAfterMs === "number" && Date.now() - Date.parse(pending.since) > options.staleAfterMs; + if (wedge?.status === "active" && wedge.reasonKey === descriptor.reasonKey) return { since: pending?.since ?? now, armed: false, restamped: false }; + if (pending !== undefined && pending.reasonKey === descriptor.reasonKey && !stale) return { since: pending.since, armed: false, restamped: false }; + wedge = wedge ? { ...wedge, pending: { since: now, ...descriptor } } : { reasonKey: descriptor.reasonKey, episodeId: "", status: "resolved", transitionedAt: now, pending: { since: now, ...descriptor } }; + if (liveTask) liveTask = { ...liveTask, wedgeNotification: wedge }; + return { since: now, armed: true, restamped: pending != null }; + }), + clearTaskWedgeNotificationPending: vi.fn(async () => { + if (!wedge?.pending) return false; + const { pending: _pending, ...rest } = wedge; + wedge = rest; + if (liveTask) liveTask = { ...liveTask, wedgeNotification: wedge }; + return true; + }), /* Absent → the helper keeps the legacy ids, which is every pre-existing case in this file. */ ...(workflowIr ? { listWorkflowDefinitions: async () => [{ ir: workflowIr }] } : {}), }; const sendMessageOnce = vi.fn(async (_input: unknown, _key: string) => ({ message: {} as any, inserted: true })); - const service = new NotificationService(store as any, { messageStore: { on: () => undefined, sendMessageOnce } as any, failedNotificationGraceMs: 60_000 }); + const service = new NotificationService(store as any, { messageStore: { on: () => undefined, sendMessageOnce } as any, failedNotificationGraceMs: 60_000, wedgeNotificationSettleMs: settleMs }); const sendNotification = vi.fn(async () => ({ success: true, providerId: "test" })); const provider: NotificationProvider = { getProviderId: () => "test", isEventSupported: () => true, sendNotification }; service.registerProvider(provider); @@ -61,7 +83,7 @@ function fixture(workflowIr?: unknown) { } /* Creates a restart-safe claim fake so NotificationService tests exercise delivery policy, not storage implementation. */ -function cooldownFixture({ durable = true }: { durable?: boolean } = {}) { +function cooldownFixture({ durable = true, settleMs = 0 }: { durable?: boolean; settleMs?: number } = {}) { const listeners = new Set(); let wedge: Task["wedgeNotification"]; const stamps = new Map(); @@ -90,7 +112,7 @@ function cooldownFixture({ durable = true }: { durable?: boolean } = {}) { } : {}), }; const sendMessageOnce = vi.fn(async (_input: unknown, _key: string) => ({ message: {} as any, inserted: true })); - const service = new NotificationService(store as any, { messageStore: { on: () => undefined, sendMessageOnce } as any }); + const service = new NotificationService(store as any, { messageStore: { on: () => undefined, sendMessageOnce } as any, wedgeNotificationSettleMs: settleMs }); const sendNotification = vi.fn(async () => ({ success: true, providerId: "test" })); service.registerProvider({ getProviderId: () => "test", isEventSupported: () => true, sendNotification }); const task = (overrides: Partial = {}): Task => ({ id: "FN-8691", title: "Blocked task", description: "", column: "in-review", status: "failed", error: "BLOCKED: dependency unavailable", dependencies: [], steps: [], currentStep: 0, log: [], createdAt: "2026-08-01T12:00:00.000Z", updatedAt: new Date().toISOString(), ...overrides } as Task); @@ -103,6 +125,228 @@ async function flushWedgeHandling() { } describe("task wedge notifications", () => { + it("withholds terminal alerts when the task recovers inside the settle window", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-11T12:00:00.000Z")); + const { store, service, sendMessageOnce, sendNotification, task, getWedge } = fixture(undefined, 1_000); + await service.start(); + store.setLiveTask(task({ error: "BLOCKED: upstream", column: "in-review" })); + store.emit(task({ error: "BLOCKED: upstream", column: "in-review" })); + await flushWedgeHandling(); + expect(getWedge()?.pending).toBeDefined(); + const recovered = task({ status: "in-progress", error: undefined, column: "in-progress", log: [{ action: "Auto-recovered: retry succeeded" }] as any }); + store.setLiveTask(recovered); + store.emit(recovered); + await flushWedgeHandling(); + await vi.advanceTimersByTimeAsync(1_001); + expect(sendNotification).not.toHaveBeenCalled(); + expect(sendMessageOnce).not.toHaveBeenCalled(); + expect(getWedge()?.pending).toBeUndefined(); + await service.stop(); + vi.useRealTimers(); + }); + + it("re-arms an early timer completion on durable and compatibility stores", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-11T12:00:00.000Z")); + + const durable = fixture(undefined, 1_000); + await durable.service.start(); + const durableFailed = durable.task({ error: "BLOCKED: upstream" }); + durable.store.setLiveTask(durableFailed); + durable.store.emit(durableFailed); + await flushWedgeHandling(); + await vi.advanceTimersByTimeAsync(500); + (durable.service as any).wedgeNotificationSettleMs = 2_000; + await vi.advanceTimersByTimeAsync(500); + expect(durable.sendNotification).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_000); + expect(durable.sendNotification).toHaveBeenCalledTimes(1); + await durable.service.stop(); + + const compatibility = cooldownFixture({ durable: false, settleMs: 1_000 }); + await compatibility.service.start(); + const compatibilityFailed = compatibility.task({ error: "BLOCKED: upstream" }); + compatibility.store.emit(compatibilityFailed); + await flushWedgeHandling(); + await vi.advanceTimersByTimeAsync(500); + (compatibility.service as any).wedgeNotificationSettleMs = 2_000; + await vi.advanceTimersByTimeAsync(500); + expect(compatibility.sendNotification).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_000); + expect(compatibility.sendNotification).toHaveBeenCalledTimes(1); + await compatibility.service.stop(); + vi.useRealTimers(); + }); + + it("re-arms stale timer completions before delivering on durable and compatibility stores", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-11T12:00:00.000Z")); + + const durable = fixture(undefined, 1_000); + await durable.service.start(); + const durableFailed = durable.task({ error: "BLOCKED: upstream" }); + durable.store.setLiveTask(durableFailed); + durable.store.emit(durableFailed); + await flushWedgeHandling(); + vi.setSystemTime(new Date("2026-08-11T14:00:00.000Z")); + await vi.advanceTimersByTimeAsync(1_000); + expect(durable.sendNotification).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_000); + expect(durable.sendNotification).toHaveBeenCalledTimes(1); + await durable.service.stop(); + + const compatibility = cooldownFixture({ durable: false, settleMs: 1_000 }); + await compatibility.service.start(); + const compatibilityFailed = compatibility.task({ error: "BLOCKED: upstream" }); + compatibility.store.emit(compatibilityFailed); + await flushWedgeHandling(); + vi.setSystemTime(new Date("2026-08-11T16:00:00.000Z")); + await vi.advanceTimersByTimeAsync(1_000); + expect(compatibility.sendNotification).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_000); + expect(compatibility.sendNotification).toHaveBeenCalledTimes(1); + await compatibility.service.stop(); + vi.useRealTimers(); + }); + + it("completes against a compatibility snapshot and fails quiet without one", async () => { + vi.useFakeTimers(); + const compatibility = cooldownFixture({ durable: false, settleMs: 1_000 }); + await compatibility.service.start(); + const failed = compatibility.task({ error: "BLOCKED: upstream" }); + compatibility.store.emit(failed); + await flushWedgeHandling(); + await vi.advanceTimersByTimeAsync(1_001); + expect(compatibility.sendNotification).toHaveBeenCalledTimes(1); + await compatibility.service.stop(); + + const unreadable = cooldownFixture({ durable: false, settleMs: 1_000 }); + await unreadable.service.start(); + await expect(unreadable.service.completePendingWedgeNotification("FN-missing")).resolves.toEqual({ outcome: "unreadable" }); + expect(unreadable.sendNotification).not.toHaveBeenCalled(); + await unreadable.service.stop(); + vi.useRealTimers(); + }); + + it("uses a configured maintenance interval to keep stale holds deliverable after a restart", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-11T12:00:00.000Z")); + const { store, service, sendNotification, task, setWedge } = fixture(undefined, 1_000, 1_800_000); + await service.start(); + const since = new Date(Date.now() - 1_000_000).toISOString(); + const pending = { + reasonKey: "terminal-failed", episodeId: "", status: "resolved" as const, transitionedAt: since, + pending: { since, reasonKey: "terminal-failed", source: "auto" as const, reason: "The task entered a terminal failed state and needs operator intervention.", action: "Inspect the task failure and retry or replan it." }, + }; + setWedge(pending); + store.setLiveTask(task({ error: "boom", wedgeNotification: pending })); + + // The age exceeds the old hard-coded 15-minute horizon but not maintenance + window. + expect((service as any).resolveStaleHoldHorizonMs()).toBe(1_801_000); + await expect(service.completePendingWedgeNotification("FN-8501")).resolves.toMatchObject({ outcome: "delivered" }); + expect(sendNotification).toHaveBeenCalledTimes(1); + await service.stop(); + vi.useRealTimers(); + }); + + it("clears an outstanding hold when the settle window switches to zero without a duplicate dispatch", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-11T12:00:00.000Z")); + const { store, service, sendMessageOnce, sendNotification, task, getWedge } = fixture(undefined, 1_000); + await service.start(); + const failed = task({ error: "BLOCKED: upstream", column: "in-review" }); + store.setLiveTask(failed); + store.emit(failed); + await flushWedgeHandling(); + expect(getWedge()?.pending).toBeDefined(); + + // This models a settings refresh after an operator restores legacy immediate delivery. + (service as any).wedgeNotificationSettleMs = 0; + await expect(service.completePendingWedgeNotification(failed.id)).resolves.toEqual({ outcome: "cleared" }); + expect(getWedge()?.pending).toBeUndefined(); + expect(sendNotification).not.toHaveBeenCalled(); + + store.emit(failed); + await flushWedgeHandling(); + expect(sendNotification).toHaveBeenCalledTimes(1); + expect(sendMessageOnce).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(2_000); + expect(sendNotification).toHaveBeenCalledTimes(1); + await service.stop(); + vi.useRealTimers(); + }); + + it("clears a pending hold from the subscribed task:moved recovery path", async () => { + vi.useFakeTimers(); + const { store, service, sendMessageOnce, sendNotification, task, getWedge } = fixture(RENAMED_IR, 1_000); + await service.start(); + const failed = task({ error: "BLOCKED: upstream", column: "checking" }); + store.setLiveTask(failed); + store.emit(failed); + await flushWedgeHandling(); + expect(getWedge()?.pending).toBeDefined(); + + const recovered = task({ column: "shipped", status: "done", error: undefined }); + store.setLiveTask(recovered); + store.emitMoved(recovered, "checking", "shipped"); + await flushWedgeHandling(); + await vi.advanceTimersByTimeAsync(1_001); + expect(getWedge()?.pending).toBeUndefined(); + expect(sendNotification).not.toHaveBeenCalled(); + expect(sendMessageOnce).not.toHaveBeenCalled(); + await service.stop(); + vi.useRealTimers(); + }); + + it("revalidates an auto hold and restarts it when the live reason changes", async () => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-08-11T12:00:00.000Z")); + const { store, service, sendMessageOnce, task } = fixture(undefined, 1_000); + await service.start(); + store.setLiveTask(task({ error: "BLOCKED: upstream", column: "in-review" })); + store.emit(task({ error: "BLOCKED: upstream", column: "in-review" })); + await flushWedgeHandling(); + store.setLiveTask(task({ error: "Tool failure retries exhausted", column: "in-review" })); + await vi.advanceTimersByTimeAsync(1_001); + expect(sendMessageOnce).not.toHaveBeenCalled(); + await vi.advanceTimersByTimeAsync(1_001); + expect(sendMessageOnce).toHaveBeenCalledTimes(1); + expect((sendMessageOnce.mock.calls[0]?.[0] as any).content).toContain("tool-failure retries"); + await service.stop(); + vi.useRealTimers(); + }); + + it("clears a supplied hold when the live task becomes user-paused", async () => { + vi.useFakeTimers(); + const { service, store, sendMessageOnce, task } = fixture(undefined, 1_000); + 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(); + store.setLiveTask(stalled); + await service.notifyTaskWedge(stalled, descriptor); + store.setLiveTask({ ...stalled, userPaused: true }); + await vi.advanceTimersByTimeAsync(1_001); + expect(sendMessageOnce).not.toHaveBeenCalled(); + await service.stop(); + vi.useRealTimers(); + }); + + it("keeps the void queue live after an in-chain pending completion", async () => { + vi.useFakeTimers(); + const { store, service, sendMessageOnce, task } = fixture(undefined, 1_000); + await service.start(); + store.setLiveTask(task({ error: "BLOCKED: upstream", column: "in-review" })); + store.emit(task({ error: "BLOCKED: upstream", column: "in-review" })); + await flushWedgeHandling(); + store.emit(task({ error: "BLOCKED: upstream", column: "in-review", updatedAt: "2026-08-11T12:00:01.000Z" })); + await flushWedgeHandling(); + await vi.advanceTimersByTimeAsync(1_001); + expect(sendMessageOnce).toHaveBeenCalledTimes(1); + await service.stop(); + vi.useRealTimers(); + }); + /* FNXC:TaskWedgeNotifications 2026-08-01-15:35: A BLOCKED task can resolve and re-wedge as scheduler/self-healing touch it. A @@ -364,7 +608,7 @@ describe("task wedge notifications", () => { await vi.waitFor(() => expect(sendMessageOnce).toHaveBeenCalledTimes(1)); await service.stop(); - const restarted = new NotificationService(store as any, { messageStore: { on: () => undefined, sendMessageOnce } as any }); + const restarted = new NotificationService(store as any, { messageStore: { on: () => undefined, sendMessageOnce } as any, wedgeNotificationSettleMs: 0 }); restarted.registerProvider({ getProviderId: () => "restarted", isEventSupported: () => true, sendNotification }); await restarted.start(); store.emit(task({ updatedAt: "2026-07-22T12:01:00.000Z" })); diff --git a/packages/engine/src/notification/notification-service.ts b/packages/engine/src/notification/notification-service.ts index a19347e583..5153f4d095 100644 --- a/packages/engine/src/notification/notification-service.ts +++ b/packages/engine/src/notification/notification-service.ts @@ -18,6 +18,9 @@ import { NtfyNotificationProvider } from "./ntfy-provider.js"; import { WebhookNotificationProvider } from "./webhook-provider.js"; import { classifyTerminalFailureAutoRecoveryForTask, describeTaskRecoveryOwner, describeTaskWedge, isTaskProgressing, shouldWithholdWedgeAlertForAutoRecovery, type TaskWedgeDescriptor } from "./task-wedge-notification.js"; +type PendingWedgeCompletionOutcome = "delivered" | "suppressed" | "cleared" | "rearmed" | "held" | "absent" | "unreadable"; +type PendingWedgeCompletionResult = { outcome: PendingWedgeCompletionOutcome; reasonKey?: string; remainingMs?: number }; + export interface NotificationServiceOptions { /** Project identifier for notification deep links */ projectId?: string; @@ -31,6 +34,8 @@ export interface NotificationServiceOptions { agentNameResolver?: (agentId: string) => Promise | string | null; /** Test hook to override failed-notification grace period (default 60_000ms). */ failedNotificationGraceMs?: number; + /** Test hook for the durable terminal-wedge settle window. */ + wedgeNotificationSettleMs?: number; } interface NotificationServiceStoreEvents { @@ -46,6 +51,8 @@ interface NotificationServiceStore { getTask?(id: string): Promise | Task | undefined; /** Durable compare-and-set for restart-safe wedge delivery episodes. */ claimTaskWedgeNotificationEpisode?(taskId: string, reasonKey: string | null): Promise<{ episodeId?: string; claimed: boolean }>; + markTaskWedgeNotificationPending?(taskId: string, descriptor: { reasonKey: string; source: "auto" | "supplied"; reason: string; action: string; gate?: string }, options?: { staleAfterMs?: number }): Promise<{ since: string; armed: boolean; restamped: boolean }>; + clearTaskWedgeNotificationPending?(taskId: string, reasonKey?: string): Promise; /** Optional so lightweight notification fakes retain their structural surface. */ markTerminalFailureAutoRecoveryBudgetExhausted?(taskId: string, options: { maxAttempts: number }): Promise<"stamped" | "already-stamped" | "not-exhausted" | "no-budget">; /** Optional; self-healing's concrete store is the durable backstop when absent. */ @@ -127,6 +134,10 @@ export class NotificationService { private failureNotificationSuppressedCount = 0; private failureNotificationDelayMs = 60_000; private failureNotificationMode: "sticky-only" | "all" | "terminal-only" = "sticky-only"; + private wedgeNotificationSettleMs = 300_000; + private maintenanceIntervalMs = 900_000; + private wedgeNotificationSuppressedCount = 0; + private readonly pendingWedgeNotifications = new Map(); /** Compatibility fallback for lightweight test stores without the durable TaskStore CAS. */ private readonly activeWedgeReasons = new Map(); /* @@ -158,14 +169,28 @@ export class NotificationService { links never reject: `maybeNotifyTaskWedge` already owns its own error handling, and a rejected link would poison every later notification for that task. */ - private readonly wedgeHandlingChains = new Map>(); + private readonly wedgeHandlingChains = new Map>(); - /** Queues wedge handling for one task behind any handling already in flight for it. */ - private enqueueWedgeHandling(taskId: string, run: () => Promise): Promise { + /** + * Queues wedge handling for one task behind any handling already in flight for it. + * + * FNXC:TaskWedgeNotifications 2026-08-11-19:20: + * This queue deliberately resolves only to void and never rejects. Timer and sweep callers use + * the public completion wrapper's closure-captured outcome; an in-chain observation calls the + * queue-free core directly. Re-entering this queue from a link would wait on its own successor + * and permanently deadlock all later wedge handling for that task. + */ + private enqueueWedgeHandling(taskId: string, run: () => Promise): Promise { const previous = this.wedgeHandlingChains.get(taskId) ?? Promise.resolve(); - const next = previous.then(run, run); + const execute = async (): Promise => { + try { + await run(); + } catch (error) { + schedulerLog.debug(`[notify] ${taskId} wedge handling failed: ${error instanceof Error ? error.message : String(error)}`); + } + }; + const next = previous.then(execute, execute); this.wedgeHandlingChains.set(taskId, next); - /* Drop the entry only if no later link was appended while this one ran. */ void next.finally(() => { if (this.wedgeHandlingChains.get(taskId) === next) this.wedgeHandlingChains.delete(taskId); }); @@ -179,6 +204,7 @@ export class NotificationService { this.chatStore = options.chatStore; this.failedNotificationGraceMs = options.failedNotificationGraceMs ?? 60_000; this.failureNotificationDelayMs = this.failedNotificationGraceMs; + this.wedgeNotificationSettleMs = options.wedgeNotificationSettleMs ?? 300_000; } attachChatStore(chatStore: NotificationChatStore): void { @@ -242,6 +268,8 @@ export class NotificationService { } this.pendingFailureNotifications.clear(); this.pendingFailureStartTimes.clear(); + for (const pending of this.pendingWedgeNotifications.values()) if (pending.timer) clearTimeout(pending.timer); + this.pendingWedgeNotifications.clear(); await this.dispatcher.shutdownAll(); this.started = false; @@ -341,6 +369,20 @@ export class NotificationService { private async handleTaskMovedAsync(data: { task: Task; from: Column; to: Column }): Promise { await this.maybeSuppressTransientFailedNotification(data.task, `moved to ${data.to}`); + /* + FNXC:TaskWedgeNotifications 2026-08-11-18:57: + task:moved is a production recovery path that does not enter maybeNotifyTaskWedge. A pending + durable hold must be cleared when its subject visibly progresses, including workflow-renamed + hold/WIP/terminal lanes, so its timer or restart sweep cannot alert after that recovery. + */ + const movedProgressedLanes = await resolveProjectColumnsForRoles(this.store, ["hold", "countsTowardWip", "complete", "archived"]); + if ( + movedProgressedLanes.has(data.to) + || (typeof data.task.status === "string" && data.task.status !== "failed") + ) { + await this.clearPendingWedgeNotification(data.task.id); + } + const movedLifecycle = await this.resolveLifecycleColumnsForTask(data.task.id); if ((await this.resolveReviewColumnsForTask(data.task.id)).has(data.to)) { @@ -392,7 +434,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)); + void this.enqueueWedgeHandling(task.id, async () => { await this.maybeNotifyTaskWedge(task); }); void this.maybeSuppressTransientFailedNotification(task, `status=${task.status ?? "undefined"}`); /* @@ -542,7 +584,9 @@ export class NotificationService { descriptor: TaskWedgeDescriptor, options?: { source?: "auto-recovery-escalation" }, ): Promise<"delivered" | "suppressed" | "unavailable"> { - return this.enqueueWedgeHandling(task.id, () => this.maybeNotifyTaskWedge(task, descriptor, options)); + let result: "delivered" | "suppressed" | "unavailable" = "unavailable"; + await this.enqueueWedgeHandling(task.id, async () => { result = await this.maybeNotifyTaskWedge(task, descriptor, options); }); + return result; } private async maybeNotifyTaskWedge( @@ -586,6 +630,7 @@ export class NotificationService { if (recoveryOwner && !ownEscalation) { // Recovery ownership is not a wedge episode. Resolve only an episode we // can prove active, avoiding a write/claim for a never-notified snapshot. + await this.clearPendingWedgeNotification(task.id); if (this.activeWedgeReasons.has(task.id) || liveTask.wedgeNotification?.status === "active") { this.activeWedgeReasons.delete(task.id); await this.store.claimTaskWedgeNotificationEpisode?.(task.id, null); @@ -651,11 +696,30 @@ export class NotificationService { || (!isActiveSelfHealingNoAction && typeof task.status === "string" && task.status !== "failed") || (isActiveSelfHealingNoAction && ["queued", "planning", "in-progress", "merging", "merging-pr", "merged", "done"].includes(task.status ?? "")); if (hasProgressed) { + await this.clearPendingWedgeNotification(task.id); this.activeWedgeReasons.delete(task.id); await this.store.claimTaskWedgeNotificationEpisode?.(task.id, null); } return "unavailable"; } + const source: "auto" | "supplied" = suppliedDescriptor === undefined ? "auto" : "supplied"; + if (this.wedgeNotificationSettleMs > 0) { + const marked = typeof this.store.markTaskWedgeNotificationPending === "function" + ? await this.store.markTaskWedgeNotificationPending(task.id, { ...descriptor, source }, { staleAfterMs: this.resolveStaleHoldHorizonMs() }) + : this.markFallbackWedgePending(task, descriptor, { staleAfterMs: this.resolveStaleHoldHorizonMs() }); + const existing = this.pendingWedgeNotifications.get(task.id); + this.pendingWedgeNotifications.set(task.id, { ...existing, descriptor, source, since: marked.since, task }); + if (marked.armed) { + this.armPendingWedgeTimer(task.id, this.wedgeNotificationSettleMs); + return "unavailable"; + } + const completion = await this.runPendingWedgeCompletion(task.id); + if ((completion.outcome === "held" || completion.outcome === "rearmed") && !this.pendingWedgeNotifications.get(task.id)?.timer) { + this.armPendingWedgeTimer(task.id, completion.remainingMs ?? this.wedgeNotificationSettleMs); + } + return completion.outcome === "delivered" ? "delivered" : completion.outcome === "suppressed" ? "suppressed" : "unavailable"; + } + await this.clearPendingWedgeNotification(task.id); const isAutoRecoveryEscalationDispatch = ownEscalation && descriptor.reasonKey === "terminal-failed"; if (isAutoRecoveryEscalationDispatch && classification.action === "notify" && classification.reason === "budget-exhausted") { try { @@ -727,6 +791,174 @@ export class NotificationService { return "delivered"; } + /* + FNXC:TaskWedgeNotifications 2026-08-11-18:28: + Deferred wedge delivery is serialized per task. The timer enters through the public wrapper; + in-chain observations call the queue-free core to avoid awaiting a successor link of themselves. + Lightweight stores retain this hold only in memory: restart durability requires the production + pending-marker API, and stores without getTask can only observe recovery on later task updates. + */ + getWedgeNotificationSettleMs(): number { return this.wedgeNotificationSettleMs; } + + /* + FNXC:TaskWedgeNotifications 2026-08-11-19:08: + A restart-lost timer leaves periodic maintenance as the only deferred-delivery driver. + Keep this horizon beyond the resolved maintenance interval so its next tick observes a + re-stamped hold as fresh and can deliver it, rather than re-stamping forever. + */ + private resolveStaleHoldHorizonMs(): number { + return Math.max( + this.wedgeNotificationSettleMs * 4, + this.maintenanceIntervalMs + this.wedgeNotificationSettleMs, + ); + } + + private markFallbackWedgePending( + task: Task, + descriptor: TaskWedgeDescriptor, + options: { forceRestamp?: boolean; staleAfterMs?: number } = {}, + ): { since: string; armed: boolean; restamped: boolean } { + const prior = this.pendingWedgeNotifications.get(task.id); + const sameReason = prior?.descriptor.reasonKey === descriptor.reasonKey; + const age = prior ? Date.now() - Date.parse(prior.since) : Number.NaN; + const stale = sameReason + && typeof options.staleAfterMs === "number" + && Number.isFinite(options.staleAfterMs) + && Number.isFinite(age) + && age > options.staleAfterMs; + if (sameReason && !options.forceRestamp && !stale) return { since: prior.since, armed: false, restamped: false }; + return { since: new Date().toISOString(), armed: true, restamped: prior != null }; + } + + private armPendingWedgeTimer(taskId: string, delayMs: number): void { + const pending = this.pendingWedgeNotifications.get(taskId); + if (!pending) return; + if (pending.timer) clearTimeout(pending.timer); + const timer = setTimeout(() => { + const entry = this.pendingWedgeNotifications.get(taskId); + if (entry) delete entry.timer; + void this.completePendingWedgeNotification(taskId).then((result) => { + if (result.outcome === "held" || result.outcome === "rearmed") this.armPendingWedgeTimer(taskId, result.remainingMs ?? this.wedgeNotificationSettleMs); + }).catch(() => undefined); + }, Math.max(1, Math.min(delayMs, this.wedgeNotificationSettleMs))); + timer.unref?.(); + pending.timer = timer; + } + + private async clearPendingWedgeNotification(taskId: string): Promise { + const pending = this.pendingWedgeNotifications.get(taskId); + if (pending?.timer) clearTimeout(pending.timer); + this.pendingWedgeNotifications.delete(taskId); + try { + if (typeof this.store.clearTaskWedgeNotificationPending === "function") await this.store.clearTaskWedgeNotificationPending(taskId); + } catch (error) { + schedulerLog.debug(`[notify] ${taskId} could not clear pending wedge: ${error instanceof Error ? error.message : String(error)}`); + } + } + + async completePendingWedgeNotification(taskId: string): Promise { + let result: PendingWedgeCompletionResult = { outcome: "absent" }; + await this.enqueueWedgeHandling(taskId, async () => { result = await this.runPendingWedgeCompletion(taskId); }); + return result; + } + + private async runPendingWedgeCompletion(taskId: string): Promise { + try { + /* + FNXC:TaskWedgeNotifications 2026-08-11-18:57: + A zero window is an exit from deferred delivery, not permission for a stale timer or sweep + candidate to dispatch. New observations still use maybeNotifyTaskWedge's legacy immediate + branch; existing durable evidence is cleared here to prevent an unwindowed duplicate. + */ + if (this.wedgeNotificationSettleMs === 0) { + await this.clearPendingWedgeNotification(taskId); + return { outcome: "cleared" }; + } + const cached = this.pendingWedgeNotifications.get(taskId); + let task: Task | undefined; + try { + task = typeof this.store.getTask === "function" ? await this.store.getTask(taskId) : cached?.task; + } catch { + await this.clearPendingWedgeNotification(taskId); + return { outcome: "unreadable" }; + } + if (!task) { + await this.clearPendingWedgeNotification(taskId); + /* + FNXC:TaskWedgeNotifications 2026-08-11-19:15: + A completion with neither a live read nor the hold snapshot has no subject + to revalidate or safely format for delivery. Report it as unreadable rather + than absent so compatibility-store callers cannot mistake lost evidence for + a completed deferred episode. + */ + return { outcome: "unreadable" }; + } + const durablePending = task.wedgeNotification?.pending; + if (!durablePending && !cached) return { outcome: "absent" }; + const source = durablePending?.source ?? cached!.source; + let descriptor: TaskWedgeDescriptor = durablePending + ? { reasonKey: durablePending.reasonKey, reason: durablePending.reason, action: durablePending.action, ...(durablePending.gate ? { gate: durablePending.gate } : {}) } + : cached!.descriptor; + + const terminalLanes = await resolveProjectColumnsForRoles(this.store, ["complete", "archived"]); + const progressedLanes = await resolveProjectColumnsForRoles(this.store, ["hold", "countsTowardWip", "complete", "archived"]); + const activeSelfHealing = task.wedgeNotification?.status === "active" && task.wedgeNotification.reasonKey.startsWith("self-healing-no-action:"); + const hasProgressed = progressedLanes.has(task.column) + || (!activeSelfHealing && typeof task.status === "string" && task.status !== "failed") + || (activeSelfHealing && ["queued", "planning", "in-progress", "merging", "merging-pr", "merged", "done"].includes(task.status ?? "")); + const suppliedHeldOrProgressing = source === "supplied" && (task.paused === true || task.userPaused === true || task.autoMerge === false || isTaskProgressing(task)); + if (task.deletedAt != null || terminalLanes.has(task.column) || describeTaskRecoveryOwner(task) != null || hasProgressed || suppliedHeldOrProgressing) { + await this.clearPendingWedgeNotification(taskId); + return { outcome: "cleared" }; + } + + if (source === "auto") { + const fresh = describeTaskWedge(task); + if (!fresh) { + await this.clearPendingWedgeNotification(taskId); + return { outcome: "cleared" }; + } + if (fresh.reasonKey !== descriptor.reasonKey) { + const marked = typeof this.store.markTaskWedgeNotificationPending === "function" + ? await this.store.markTaskWedgeNotificationPending(taskId, { ...fresh, source }, { staleAfterMs: this.resolveStaleHoldHorizonMs() }) + : this.markFallbackWedgePending(task, fresh); + this.pendingWedgeNotifications.set(taskId, { ...cached, descriptor: fresh, source, since: marked.since, task }); + return { outcome: "rearmed", reasonKey: fresh.reasonKey, remainingMs: this.wedgeNotificationSettleMs }; + } + descriptor = fresh; + } + + const since = Date.parse(durablePending?.since ?? cached!.since); + const age = Number.isFinite(since) ? Date.now() - since : Number.POSITIVE_INFINITY; + if (age > this.resolveStaleHoldHorizonMs()) { + const marked = typeof this.store.markTaskWedgeNotificationPending === "function" + ? await this.store.markTaskWedgeNotificationPending(taskId, { ...descriptor, source }, { staleAfterMs: 0 }) + : this.markFallbackWedgePending(task, descriptor, { forceRestamp: true }); + this.pendingWedgeNotifications.set(taskId, { ...cached, descriptor, source, since: marked.since, task }); + return { outcome: "rearmed", reasonKey: descriptor.reasonKey, remainingMs: this.wedgeNotificationSettleMs }; + } + const remainingMs = this.wedgeNotificationSettleMs - Math.max(0, age); + if (remainingMs > 0) return { outcome: "held", reasonKey: descriptor.reasonKey, remainingMs }; + + const claim = this.store.claimTaskWedgeNotificationEpisode + ? await this.store.claimTaskWedgeNotificationEpisode(taskId, descriptor.reasonKey) + : (() => { const episodeId = this.claimFallbackWedgeNotificationEpisode(taskId, descriptor.reasonKey, task.updatedAt); return { episodeId, claimed: episodeId !== undefined }; })(); + await this.clearPendingWedgeNotification(taskId); + if (!claim.claimed || !claim.episodeId) { + this.wedgeNotificationSuppressedCount += 1; + return { outcome: "suppressed", reasonKey: descriptor.reasonKey }; + } + const payload = this.createTaskPayload(task, "task-wedged", { wedgeReason: descriptor.reasonKey, reason: descriptor.reason, action: descriptor.action, ...(descriptor.gate ? { gate: descriptor.gate } : {}), notificationDedupeKey: `task-wedge:${claim.episodeId}` }); + void this.dispatch("task-wedged", payload); + const content = `**${formatTaskIdentifier(task)} needs operator action**\n\n${descriptor.reason}\nRecommended action: ${descriptor.action}`; + try { await this.options.messageStore?.sendMessageOnce?.({ fromId: "system", fromType: "system", toId: DASHBOARD_USER_ID, toType: "user", type: "system", content, metadata: { taskId, kind: "task-wedge", wedgeReason: descriptor.reasonKey } }, `task-wedge:${claim.episodeId}`); } catch { /* mailbox is independently best-effort */ } + return { outcome: "delivered", reasonKey: descriptor.reasonKey }; + } catch (error) { + schedulerLog.debug(`[notify] ${taskId} pending wedge completion failed: ${error instanceof Error ? error.message : String(error)}`); + return { outcome: "unreadable" }; + } + } + private claimFallbackWedgeNotificationEpisode(taskId: string, reasonKey: string, updatedAt: string): string | undefined { if (this.activeWedgeReasons.get(taskId) === reasonKey) return undefined; @@ -1124,6 +1356,14 @@ export class NotificationService { ? settings.failureNotificationDelayMs : this.failedNotificationGraceMs; this.failureNotificationMode = settings.failureNotificationMode ?? "sticky-only"; + this.wedgeNotificationSettleMs = typeof settings.wedgeNotificationSettleMs === "number" && Number.isFinite(settings.wedgeNotificationSettleMs) && settings.wedgeNotificationSettleMs >= 0 + ? settings.wedgeNotificationSettleMs + : this.options.wedgeNotificationSettleMs ?? 300_000; + this.maintenanceIntervalMs = typeof settings.maintenanceIntervalMs === "number" + && Number.isFinite(settings.maintenanceIntervalMs) + && settings.maintenanceIntervalMs > 0 + ? settings.maintenanceIntervalMs + : 900_000; } private async maybeNotifyImmediateFailure(task: Task): Promise { @@ -1272,8 +1512,8 @@ export class NotificationService { this.maybeNotify(task.id, eventType, eventType === "failed" ? pending.payload : this.createTaskPayload(task, eventType)); } - getMetrics(): { failureNotificationSuppressedCount: number } { - return { failureNotificationSuppressedCount: this.failureNotificationSuppressedCount }; + getMetrics(): { failureNotificationSuppressedCount: number; wedgeNotificationSuppressedCount: number } { + return { failureNotificationSuppressedCount: this.failureNotificationSuppressedCount, wedgeNotificationSuppressedCount: this.wedgeNotificationSuppressedCount }; } getPendingFailureCount(): number { diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index a17482234f..a031df6c57 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -1769,6 +1769,7 @@ export class SelfHealingManager extends SelfHealingGitEvidence { // not stalled by a leaked `status: "merging"` on an already-done task. { name: "reconcile-stale-merger-status", fn: () => this.reconcileStaleMergerStatus().then(() => undefined) }, { name: "reconcile-stale-duplicate-decision", fn: () => this.reconcileStaleDuplicateDecisionPause().then(() => undefined) }, + { name: "reconcile-pending-wedge-notification", fn: () => this.reconcilePendingWedgeNotifications().then(() => undefined) }, { name: "recover-already-merged-review", fn: () => this.recoverAlreadyMergedReviewTasks().then(() => undefined) }, { name: "recover-post-done-noncontinuable-wedge", fn: () => this.recoverPostDoneNonContinuableWedge().then(() => undefined) }, { name: "recover-completion-handoff-limbo", fn: () => this.recoverCompletionHandoffLimbo().then(() => undefined) }, @@ -2857,6 +2858,7 @@ export class SelfHealingManager extends SelfHealingGitEvidence { { name: "reconcile-done-task-integrity", fn: () => this.reconcileDoneTaskIntegrity() }, { name: "reconcile-stale-merger-status", fn: () => this.reconcileStaleMergerStatus() }, { name: "reconcile-stale-duplicate-decision", fn: () => this.reconcileStaleDuplicateDecisionPause() }, + { name: "reconcile-pending-wedge-notification", fn: () => this.reconcilePendingWedgeNotifications() }, // FNXC:OrphanedPendingSteps 2026-07-22-16:35 (FN-8492 review follow-up): also // steady-state — a step session can die without an engine restart, and startup-only // cadence left that case riding the 3×30-min stall escalator to a deadlock park. @@ -7220,6 +7222,42 @@ export class SelfHealingManager extends SelfHealingGitEvidence { * actionable. Its audit payload is ids/outcomes-only so stale-decision recovery never stores * prompt or decision prose; active canonicals and user-owned pauses remain untouched. */ + /* + FNXC:TaskWedgeNotifications 2026-08-11-18:28: + Durable pending wedge evidence survives a process restart while its timer does not. This bounded + sweep only selects candidates and records the completion result; NotificationService remains the + sole authority for live-row validation, stale-hold re-stamping, and dispatch. + */ + async reconcilePendingWedgeNotifications(): Promise { + try { + const service = getActiveNotificationService(); + const windowMs = service?.getWedgeNotificationSettleMs() ?? 300_000; + const now = Date.now(); + const tasks = await this.store.listTasks({ slim: true, includeArchived: false, limit: 500 }); + let processed = 0; + for (const task of tasks.filter((candidate) => { + const since = candidate.wedgeNotification?.pending?.since; + return typeof since === "string" && now - Date.parse(since) >= windowMs; + }).slice(0, 50)) { + const pending = task.wedgeNotification!.pending!; + let outcome: string = "deferred"; + try { + outcome = service ? (await service.completePendingWedgeNotification(task.id)).outcome : "deferred"; + } catch { + outcome = "failed"; + } + await createRunAuditor(this.store, { + runId: generateSyntheticRunId("reconcile-pending-wedge-notification", task.id), agentId: "self-healing", taskId: task.id, taskLineageId: task.lineageId, phase: "reconcile-pending-wedge-notification", + }).database({ type: "task:reconcile-pending-wedge-notification" as DatabaseMutationType, target: task.id, metadata: { taskId: task.id, reasonKey: pending.reasonKey, pendingAgeMs: Math.max(0, now - Date.parse(pending.since)), outcome } }); + processed += 1; + } + return processed; + } catch (error) { + log.warn(`reconcilePendingWedgeNotifications failed: ${error instanceof Error ? error.message : String(error)}`); + return 0; + } + } + async reconcileStaleDuplicateDecisionPause(): Promise { try { const tasks = await this.store.listTasks({ slim: true, includeArchived: false, limit: 500 }); diff --git a/packages/i18n/locales/en/app.json b/packages/i18n/locales/en/app.json index e4828a1d9a..86438596fe 100644 --- a/packages/i18n/locales/en/app.json +++ b/packages/i18n/locales/en/app.json @@ -6491,6 +6491,8 @@ "enable": " Enable ", "failureNotificationDelayMs": "Failure notification delay (ms)", "failureNotificationMode": "Failure notification mode", + "wedgeNotificationSettleMs": "Terminal-wedge settle window (ms)", + "wedgeNotificationSettleMsHelp": "How long a terminal failure must persist before an operator alert. 0 = notify immediately. Default: 300000 (5 minutes).", "format": "Format", "general": " General: ", "generic": "Generic",