From 50f8807037a1a60cf3ba72cdbb1ec40e206b5c36 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Mon, 29 Jun 2026 12:40:00 -0700 Subject: [PATCH] fix(FN-7233): persist workflow transition notifications Store workflow transition notification markers as durable task state so recovery alerts do not depend on human-readable log text. --- ...orkflow-transition-marker-notifications.md | 7 + .../core/src/__tests__/store-update.test.ts | 27 ++++ packages/core/src/db-migrate.ts | 7 +- packages/core/src/db.ts | 12 +- packages/core/src/store.ts | 23 +++- packages/core/src/types.ts | 23 ++++ ...self-healing-paused-abort-recovery.test.ts | 32 ++++- .../__tests__/notification-service.test.ts | 128 +++++++++++++++++- .../src/notification/notification-service.ts | 34 ++--- packages/engine/src/self-healing.ts | 31 ++++- 10 files changed, 296 insertions(+), 28 deletions(-) create mode 100644 .changeset/workflow-transition-marker-notifications.md diff --git a/.changeset/workflow-transition-marker-notifications.md b/.changeset/workflow-transition-marker-notifications.md new file mode 100644 index 0000000000..8dcfd9cf71 --- /dev/null +++ b/.changeset/workflow-transition-marker-notifications.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Prevent stale workflow recovery log entries from sending incorrect notifications. +category: fix +dev: Adds workflowTransitionNotification task markers for pause-abort recovery requeues and avoids log-text notification heuristics. diff --git a/packages/core/src/__tests__/store-update.test.ts b/packages/core/src/__tests__/store-update.test.ts index f7749ea711..d2529e0be6 100644 --- a/packages/core/src/__tests__/store-update.test.ts +++ b/packages/core/src/__tests__/store-update.test.ts @@ -37,6 +37,33 @@ describe("TaskStore", () => { const createSourceIssueFixture = () => harness.createSourceIssueFixture(); const insertLogEntryWithTimestamp = (...args: any[]) => (harness as any).insertLogEntryWithTimestamp(...args); + describe("updateTask — workflow transition notifications", () => { + it("persists and clears typed workflow transition notification markers", async () => { + const task = await createTestTask(); + const marker: NonNullable = { + kind: "recovery-requeue", + column: "todo", + transitionId: `recovery-requeue:${task.id}:pause-abort-active-work`, + nodeId: "pause-abort-recovery-router", + reason: "pause-abort-active-work", + createdAt: "2026-06-29T20:05:00.000Z", + }; + + const updated = await store.updateTask(task.id, { workflowTransitionNotification: marker }); + expect(updated.workflowTransitionNotification).toEqual(marker); + + const fetched = await store.getTask(task.id); + expect(fetched?.workflowTransitionNotification).toEqual(marker); + + const limitedDetail = await store.getTask(task.id, { activityLogLimit: 1 }); + expect(limitedDetail?.workflowTransitionNotification).toEqual(marker); + + const cleared = await store.updateTask(task.id, { workflowTransitionNotification: null }); + expect(cleared.workflowTransitionNotification).toBeUndefined(); + expect((await store.getTask(task.id))?.workflowTransitionNotification).toBeUndefined(); + }); + }); + describe("updateTask — dependencies", () => { it("adds dependencies to a task with none", async () => { const task = await createTestTask(); diff --git a/packages/core/src/db-migrate.ts b/packages/core/src/db-migrate.ts index c04411938f..e2c53a562b 100644 --- a/packages/core/src/db-migrate.ts +++ b/packages/core/src/db-migrate.ts @@ -180,11 +180,11 @@ async function migrateTasks(fusionDir: string, db: Database): Promise { columnMovedAt, dependencies, steps, log, attachments, steeringComments, comments, workflowStepResults, prInfo, issueInfo, sourceIssueProvider, sourceIssueRepository, sourceIssueExternalIssueId, sourceIssueNumber, sourceIssueUrl, sourceIssueClosedAt, - mergeDetails, breakIntoSubtasks, noCommitsExpected, enabledWorkflowSteps, modifiedFiles, sliceId, + mergeDetails, breakIntoSubtasks, noCommitsExpected, enabledWorkflowSteps, modifiedFiles, workflowTransitionNotification, sliceId, workspaceWorktrees ) VALUES ( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, - ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? + ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? ) `); @@ -254,6 +254,9 @@ async function migrateTasks(fusionDir: string, db: Database): Promise { task.noCommitsExpected ? 1 : 0, toJson(task.enabledWorkflowSteps || []), toJson(task.modifiedFiles || []), + // FNXC:WorkflowNotifications 2026-06-29-13:10: preserve typed workflow + // transition markers during task.json -> SQLite rebuilds. + toJsonNullable(task.workflowTransitionNotification), task.sliceId ?? null, // FNXC:Workspace 2026-06-24-15:30: carry the per-sub-repo worktree map through the legacy // task.json→SQLite rebuild so a workspace task migrated from disk keeps its acquired worktrees. diff --git a/packages/core/src/db.ts b/packages/core/src/db.ts index 2ead1cab8e..cb653052b0 100644 --- a/packages/core/src/db.ts +++ b/packages/core/src/db.ts @@ -183,7 +183,7 @@ export function isFts5CorruptionError(error: unknown): boolean { // ── Schema Definition ──────────────────────────────────────────────── -const SCHEMA_VERSION = 132; +const SCHEMA_VERSION = 133; const TASKS_FTS_AUTOMERGE = 8; const TASKS_FTS_CRISISMERGE = 16; @@ -350,6 +350,8 @@ CREATE TABLE IF NOT EXISTS tasks ( noCommitsExpected INTEGER DEFAULT 0, enabledWorkflowSteps TEXT DEFAULT '[]', modifiedFiles TEXT DEFAULT '[]', + -- FNXC:WorkflowNotifications 2026-06-29-13:10: typed transition markers are JSON text. + workflowTransitionNotification TEXT, missionId TEXT, sliceId TEXT, scopeOverride INTEGER, @@ -5477,6 +5479,14 @@ export class Database { }); } + if (version < 133) { + // FNXC:WorkflowNotifications 2026-06-29-13:10: add the JSON marker column so + // recovery/manual-hold notification state survives the SQLite persistence path. + this.applyMigration(133, () => { + this.addColumnIfMissing("tasks", "workflowTransitionNotification", "TEXT"); + }); + } + } /** diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index d4b6f61096..cc05a95a0c 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -325,6 +325,7 @@ interface TaskRow { noCommitsExpected: number | null; enabledWorkflowSteps: string | null; modifiedFiles: string | null; + workflowTransitionNotification: string | null; missionId: string | null; sliceId: string | null; scopeOverride: number | null; @@ -483,6 +484,10 @@ const TASK_COLUMN_DESCRIPTORS: TaskColumnDescriptor[] = [ defineTaskColumn("noCommitsExpected", (task) => task.noCommitsExpected ? 1 : 0), defineTaskColumn("enabledWorkflowSteps", (task) => toJson(task.enabledWorkflowSteps || [])), defineTaskColumn("modifiedFiles", (task) => toJson(task.modifiedFiles || [])), + // FNXC:WorkflowNotifications 2026-06-29-13:10: persist typed workflow transition + // notification markers as JSON text so self-healing recovery alerts survive + // SQLite row round-trips and task:updated emits from the durable task shape. + defineTaskColumn("workflowTransitionNotification", (task) => toJsonNullable(task.workflowTransitionNotification)), defineTaskColumn("missionId", (task) => task.missionId ?? null), defineTaskColumn("sliceId", (task) => task.sliceId ?? null), defineTaskColumn("scopeOverride", (task) => task.scopeOverride ? 1 : null), @@ -2207,6 +2212,7 @@ export class TaskStore extends EventEmitter { */ enabledWorkflowSteps: (() => { const e = fromJson(row.enabledWorkflowSteps); return Array.isArray(e) ? e : undefined; })(), modifiedFiles: (() => { const m = fromJson(row.modifiedFiles); return m && m.length > 0 ? m : undefined; })(), + workflowTransitionNotification: fromJson(row.workflowTransitionNotification) ?? undefined, missionId: row.missionId || undefined, sliceId: row.sliceId || undefined, assignedAgentId: row.assignedAgentId || undefined, @@ -2645,7 +2651,7 @@ export class TaskStore extends EventEmitter { "createdAt", "updatedAt", "columnMovedAt", "firstExecutionAt", "cumulativeActiveMs", "columnDwellMs", "executionStartedAt", "executionCompletedAt", "dependencies", "steps", "customFields", "comments", "review", "reviewState", "workflowStepResults", "steeringComments", "attachments", "prInfo", "prInfos", "issueInfo", "githubTracking", "sourceIssueProvider", "sourceIssueRepository", "sourceIssueExternalIssueId", "sourceIssueNumber", "sourceIssueUrl", "sourceIssueClosedAt", "mergeDetails", "workspaceWorktrees", - "breakIntoSubtasks", "noCommitsExpected", "enabledWorkflowSteps", "modifiedFiles", + "breakIntoSubtasks", "noCommitsExpected", "enabledWorkflowSteps", "modifiedFiles", "workflowTransitionNotification", "missionId", "sliceId", "scopeOverride", "scopeOverrideReason", "scopeAutoWiden", "assignedAgentId", "pausedByAgentId", "assigneeUserId", "nodeId", "effectiveNodeId", "effectiveNodeSource", "sourceType", "sourceAgentId", "sourceRunId", "sourceSessionId", "sourceMessageId", "sourceParentTaskId", "sourceMetadata", "checkedOutBy", "checkedOutAt", "checkoutNodeId", "checkoutRunId", "checkoutLeaseRenewedAt", "checkoutLeaseEpoch", "deletedAt", "allowResurrection", @@ -2694,7 +2700,7 @@ export class TaskStore extends EventEmitter { "createdAt", "updatedAt", "columnMovedAt", "firstExecutionAt", "cumulativeActiveMs", "columnDwellMs", "executionStartedAt", "executionCompletedAt", "dependencies", "steps", "customFields", "attachments", "steeringComments", "comments", "review", "reviewState", "workflowStepResults", "prInfo", "prInfos", "issueInfo", "githubTracking", "sourceIssueProvider", "sourceIssueRepository", "sourceIssueExternalIssueId", "sourceIssueNumber", "sourceIssueUrl", "sourceIssueClosedAt", "mergeDetails", "workspaceWorktrees", - "breakIntoSubtasks", "noCommitsExpected", "enabledWorkflowSteps", "modifiedFiles", + "breakIntoSubtasks", "noCommitsExpected", "enabledWorkflowSteps", "modifiedFiles", "workflowTransitionNotification", "missionId", "sliceId", "scopeOverride", "scopeOverrideReason", "scopeAutoWiden", "assignedAgentId", "pausedByAgentId", "assigneeUserId", "nodeId", "effectiveNodeId", "effectiveNodeSource", "sourceType", "sourceAgentId", "sourceRunId", "sourceSessionId", "sourceMessageId", "sourceParentTaskId", "sourceMetadata", "checkedOutBy", "checkedOutAt", "checkoutNodeId", "checkoutRunId", "checkoutLeaseRenewedAt", "checkoutLeaseEpoch", "deletedAt", "allowResurrection", @@ -8146,7 +8152,7 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS} async updateTask( id: string, - updates: { title?: string; description?: string; priority?: TaskPriority | null; prompt?: string; worktree?: string | null; workspaceWorktrees?: import("./types.js").Task["workspaceWorktrees"]; status?: string | null; dependencies?: string[]; steps?: import("./types.js").TaskStep[]; customFields?: Record; currentStep?: number; blockedBy?: string | null; overlapBlockedBy?: string | null; assignedAgentId?: string | null; pausedByAgentId?: string | null; pausedReason?: string | null; tokenBudgetSoftAlertedAt?: string | null; worktrunkFallbackAlertedAt?: string | null; worktrunkFailure?: import("./types.js").Task["worktrunkFailure"] | null; tokenBudgetHardAlertedAt?: string | null; tokenBudgetOverride?: import("./types.js").TaskTokenBudgetOverride | null; dispatchStormCount?: number | null; lastDispatchAt?: string | null; assigneeUserId?: string | null; scopeOverride?: boolean | null; scopeOverrideReason?: string | null; scopeAutoWiden?: string[] | null; nodeId?: string | null; effectiveNodeId?: string | null; effectiveNodeSource?: string | null; checkedOutBy?: string | null; checkedOutAt?: string | null; checkoutNodeId?: string | null; checkoutRunId?: string | null; checkoutLeaseRenewedAt?: string | null; checkoutLeaseEpoch?: number | null; paused?: boolean; baseBranch?: string | null; autoMerge?: boolean | null; branch?: string | null; executionStartBranch?: string | null; baseCommitSha?: string | null; size?: "S" | "M" | "L"; reviewLevel?: number; executionMode?: import("./types.js").ExecutionMode | null; mergeRetries?: number; workflowStepRetries?: number; stuckKillCount?: number | null; resumeLimboCount?: number | null; graphResumeRetryCount?: number | null; resumeLimboTipSha?: string | null; resumeLimboStepSignature?: string | null; postReviewFixCount?: number | null; recoveryRetryCount?: number | null; taskDoneRetryCount?: number | null; worktreeSessionRetryCount?: number | null; completionHandoffLimboRecoveryCount?: number | null; verificationFailureCount?: number | null; mergeConflictBounceCount?: number | null; mergeAuditBounceCount?: number | null; mergeTransientRetryCount?: number | null; branchConflictRecoveryCount?: number | null; reviewerContextRetryCount?: number | null; reviewerFallbackRetryCount?: number | null; nextRecoveryAt?: string | null; enabledWorkflowSteps?: string[]; noCommitsExpected?: boolean | null; modelProvider?: string | null; modelId?: string | null; validatorModelProvider?: string | null; validatorModelId?: string | null; planningModelProvider?: string | null; planningModelId?: string | null; thinkingLevel?: string | null; error?: string | null; summary?: string | null; sessionFile?: string | null; firstExecutionAt?: string | null; cumulativeActiveMs?: number | null; executionStartedAt?: string | null; executionCompletedAt?: string | null; review?: import("./types.js").TaskReview | null; reviewState?: import("./types.js").TaskReviewState | null; workflowStepResults?: import("./types.js").WorkflowStepResult[] | null; mergeDetails?: import("./types.js").MergeDetails | null; sourceIssue?: import("./types.js").TaskSourceIssue | null; sourceMetadataPatch?: Record | null; githubTracking?: import("./types.js").TaskGithubTracking | null; tokenUsage?: import("./types.js").TaskTokenUsage | null; modifiedFiles?: string[] | null; missionId?: string | null; sliceId?: string | null }, + updates: { title?: string; description?: string; priority?: TaskPriority | null; prompt?: string; worktree?: string | null; workspaceWorktrees?: import("./types.js").Task["workspaceWorktrees"]; status?: string | null; dependencies?: string[]; steps?: import("./types.js").TaskStep[]; customFields?: Record; currentStep?: number; blockedBy?: string | null; overlapBlockedBy?: string | null; assignedAgentId?: string | null; pausedByAgentId?: string | null; pausedReason?: string | null; tokenBudgetSoftAlertedAt?: string | null; worktrunkFallbackAlertedAt?: string | null; worktrunkFailure?: import("./types.js").Task["worktrunkFailure"] | null; tokenBudgetHardAlertedAt?: string | null; tokenBudgetOverride?: import("./types.js").TaskTokenBudgetOverride | null; dispatchStormCount?: number | null; lastDispatchAt?: string | null; assigneeUserId?: string | null; scopeOverride?: boolean | null; scopeOverrideReason?: string | null; scopeAutoWiden?: string[] | null; nodeId?: string | null; effectiveNodeId?: string | null; effectiveNodeSource?: string | null; checkedOutBy?: string | null; checkedOutAt?: string | null; checkoutNodeId?: string | null; checkoutRunId?: string | null; checkoutLeaseRenewedAt?: string | null; checkoutLeaseEpoch?: number | null; paused?: boolean; baseBranch?: string | null; autoMerge?: boolean | null; branch?: string | null; executionStartBranch?: string | null; baseCommitSha?: string | null; size?: "S" | "M" | "L"; reviewLevel?: number; executionMode?: import("./types.js").ExecutionMode | null; mergeRetries?: number; workflowStepRetries?: number; stuckKillCount?: number | null; resumeLimboCount?: number | null; graphResumeRetryCount?: number | null; resumeLimboTipSha?: string | null; resumeLimboStepSignature?: string | null; postReviewFixCount?: number | null; recoveryRetryCount?: number | null; taskDoneRetryCount?: number | null; worktreeSessionRetryCount?: number | null; completionHandoffLimboRecoveryCount?: number | null; verificationFailureCount?: number | null; mergeConflictBounceCount?: number | null; mergeAuditBounceCount?: number | null; mergeTransientRetryCount?: number | null; branchConflictRecoveryCount?: number | null; reviewerContextRetryCount?: number | null; reviewerFallbackRetryCount?: number | null; nextRecoveryAt?: string | null; enabledWorkflowSteps?: string[]; noCommitsExpected?: boolean | null; modelProvider?: string | null; modelId?: string | null; validatorModelProvider?: string | null; validatorModelId?: string | null; planningModelProvider?: string | null; planningModelId?: string | null; thinkingLevel?: string | null; error?: string | null; summary?: string | null; sessionFile?: string | null; firstExecutionAt?: string | null; cumulativeActiveMs?: number | null; executionStartedAt?: string | null; executionCompletedAt?: string | null; review?: import("./types.js").TaskReview | null; reviewState?: import("./types.js").TaskReviewState | null; workflowStepResults?: import("./types.js").WorkflowStepResult[] | null; mergeDetails?: import("./types.js").MergeDetails | null; sourceIssue?: import("./types.js").TaskSourceIssue | null; sourceMetadataPatch?: Record | null; githubTracking?: import("./types.js").TaskGithubTracking | null; tokenUsage?: import("./types.js").TaskTokenUsage | null; modifiedFiles?: string[] | null; workflowTransitionNotification?: import("./types.js").Task["workflowTransitionNotification"] | null; missionId?: string | null; sliceId?: string | null }, runContext?: RunMutationContext, ): Promise { return this.withTaskLock(id, () => this.updateTaskUnlocked(id, updates, runContext)); @@ -9089,6 +9095,17 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS} } else if (updates.modifiedFiles !== undefined) { task.modifiedFiles = updates.modifiedFiles; } + if (updates.workflowTransitionNotification === null) { + task.workflowTransitionNotification = undefined; + } else if (updates.workflowTransitionNotification !== undefined) { + /* + FNXC:WorkflowNotifications 2026-06-29-13:05: + Typed workflow transition notification markers must persist through the + ordinary task update authority. Self-healing and workflow nodes rely on + the emitted task:updated row, not log text, to trigger ntfy alerts. + */ + task.workflowTransitionNotification = updates.workflowTransitionNotification; + } if (updates.missionId === null) { task.missionId = undefined; } else if (updates.missionId !== undefined) { diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index e8c35f90e8..7921b04716 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -1080,6 +1080,19 @@ export interface TaskLogEntry { runContext?: RunMutationContext; } +export type WorkflowTransitionNotificationKind = + | "manual-merge-hold" + | "recovery-requeue"; + +export interface WorkflowTransitionNotificationMarker { + kind: WorkflowTransitionNotificationKind; + column: ColumnId; + transitionId: string; + nodeId?: string; + reason?: string; + createdAt: string; +} + export type ActivityEventType = | "task:created" | "task:moved" @@ -2221,6 +2234,16 @@ export interface Task { /** Server-computed stale paused todo diagnostic signal. Undefined when no rule matches. * Diagnostic-only: must not trigger automatic state mutation. */ stalePausedTodo?: StalePausedTodoSignal; + /* + * FNXC:WorkflowNotifications 2026-06-29-12:44: + * Workflow transition notifications should use typed task state instead of + * parsing human-readable task log text. Producers set this marker when a + * workflow transition needs operator notification; NotificationService only + * consumes it while the task remains in the recorded target column. The marker + * column prevents stale task movement from triggering a later notification, + * and transitionId provides stable dedupe across repeated task:updated events. + */ + workflowTransitionNotification?: WorkflowTransitionNotificationMarker; /** Heuristic stalled-review diagnostic signal (legacy compatibility contract). */ stalledReview?: StalledReviewSignal; /** Durable aggregate token usage totals for the task. Undefined when no usage has been recorded yet. */ diff --git a/packages/engine/src/__tests__/self-healing-paused-abort-recovery.test.ts b/packages/engine/src/__tests__/self-healing-paused-abort-recovery.test.ts index 093b2e0f8d..68ee2a68f0 100644 --- a/packages/engine/src/__tests__/self-healing-paused-abort-recovery.test.ts +++ b/packages/engine/src/__tests__/self-healing-paused-abort-recovery.test.ts @@ -89,7 +89,18 @@ describe("recoverPausedAbortFailures", () => { const recovered = await manager.recoverPausedAbortFailures(); expect(recovered).toBe(1); - expect(store.updateTask).toHaveBeenCalledWith("FN-7000", { status: null, error: null }); + expect(store.updateTask).toHaveBeenCalledWith("FN-7000", { + status: null, + error: null, + workflowTransitionNotification: { + kind: "recovery-requeue", + column: "todo", + transitionId: "recovery-requeue:FN-7000:pause-abort-active-work", + nodeId: "pause-abort-recovery-router", + reason: "pause-abort-active-work", + createdAt: "2026-06-20T02:30:00.000Z", + }, + }); // Already in todo — must NOT be moved. expect(store.moveTask).not.toHaveBeenCalled(); // FNXC:WorkflowLifecycle A1 releases via the wired clearPhantomExecutorBinding, @@ -117,11 +128,26 @@ describe("recoverPausedAbortFailures", () => { const recovered = await manager.recoverPausedAbortFailures(); expect(recovered).toBe(1); + expect(store.updateTask).toHaveBeenNthCalledWith(1, "FN-7001", { status: null, error: null }); expect(store.moveTask).toHaveBeenCalledWith( "FN-7001", "todo", { preserveProgress: true, moveSource: "engine", recoveryRehome: true }, ); + expect(store.updateTask).toHaveBeenNthCalledWith(2, "FN-7001", { + workflowTransitionNotification: { + kind: "recovery-requeue", + column: "todo", + transitionId: "recovery-requeue:FN-7001:pause-abort-active-work", + nodeId: "pause-abort-recovery-router", + reason: "pause-abort-active-work", + createdAt: "2026-06-20T02:30:00.000Z", + }, + }); + expect((store.updateTask as ReturnType).mock.invocationCallOrder[0]) + .toBeLessThan((store.moveTask as ReturnType).mock.invocationCallOrder[0]); + expect((store.moveTask as ReturnType).mock.invocationCallOrder[0]) + .toBeLessThan((store.updateTask as ReturnType).mock.invocationCallOrder[1]); }); it("clears a completed in-review pause-abort park without moving it backward", async () => { @@ -143,6 +169,10 @@ describe("recoverPausedAbortFailures", () => { expect(recovered).toBe(1); expect(store.updateTask).toHaveBeenCalledWith("FN-7002", { status: null, error: null }); + expect(store.updateTask).not.toHaveBeenCalledWith( + "FN-7002", + expect.objectContaining({ workflowTransitionNotification: expect.anything() }), + ); expect(store.moveTask).not.toHaveBeenCalled(); expect(clearBinding).toHaveBeenCalledWith("FN-7002"); expect(store.logEntry).toHaveBeenCalledWith( diff --git a/packages/engine/src/notification/__tests__/notification-service.test.ts b/packages/engine/src/notification/__tests__/notification-service.test.ts index ef3e4a50c8..391b3cf2fe 100644 --- a/packages/engine/src/notification/__tests__/notification-service.test.ts +++ b/packages/engine/src/notification/__tests__/notification-service.test.ts @@ -476,7 +476,14 @@ describe("NotificationService workflow transition notifications", () => { paused: false, pausedReason: undefined, status: undefined, - log: [{ timestamp: new Date().toISOString(), action: "Workflow graph failed at node 'review' with incomplete steps - moved back to todo for execution resume" }], + workflowTransitionNotification: { + kind: "recovery-requeue", + column: "todo", + transitionId: "recovery-requeue:FN-7203:pause-abort-active-work", + nodeId: "recovery-router", + reason: "pause-abort-active-work", + createdAt: new Date().toISOString(), + }, })); await vi.waitFor(() => { @@ -499,14 +506,106 @@ describe("NotificationService workflow transition notifications", () => { expect.objectContaining({ taskId: "FN-7203", metadata: expect.objectContaining({ - notificationDedupeKey: "workflow-transition:FN-7203:recovery-requeue", + notificationDedupeKey: "workflow-transition:FN-7203:recovery-requeue:FN-7203:pause-abort-active-work", notificationKind: "workflow_recovery_requeue", + nodeId: "recovery-router", + reason: "pause-abort-active-work", }), }), ); await service.stop(); }); + it("emits manual merge hold notifications from current typed markers", async () => { + const { store, service, sendNotification } = await setup(); + + store.emit("task:updated", task({ + id: "FN-7209", + column: "in-review", + paused: false, + pausedReason: undefined, + workflowTransitionNotification: { + kind: "manual-merge-hold", + column: "in-review", + transitionId: "manual-hold:merge-request:FN-7209", + nodeId: "merge-manual-hold", + reason: "merge-request-manual-required", + createdAt: new Date().toISOString(), + }, + })); + + await vi.waitFor(() => { + expect(sendNotification).toHaveBeenCalledTimes(1); + }); + expect(sendNotification).toHaveBeenCalledWith( + "workflow-notify", + expect.objectContaining({ + taskId: "FN-7209", + metadata: expect.objectContaining({ + notificationDedupeKey: "workflow-transition:FN-7209:manual-hold:merge-request:FN-7209", + notificationKind: "manual_merge_hold", + nodeId: "merge-manual-hold", + reason: "merge-request-manual-required", + }), + }), + ); + await service.stop(); + }); + + it("does not infer workflow recovery notifications from log text or stale typed markers", async () => { + const { store, service, sendNotification } = await setup(); + + store.emit("task:updated", task({ + id: "FN-7207", + column: "todo", + status: undefined, + log: [{ timestamp: new Date().toISOString(), action: "Workflow graph moved back to todo for execution resume" }], + })); + store.emit("task:updated", task({ + id: "FN-7208", + column: "in-progress", + status: undefined, + workflowTransitionNotification: { + kind: "recovery-requeue", + column: "todo", + transitionId: "stale-recovery", + createdAt: new Date().toISOString(), + }, + })); + + await Promise.resolve(); + + expect(sendNotification).not.toHaveBeenCalled(); + await service.stop(); + }); + + it("does not infer manual merge hold notifications from log text or stale typed markers", async () => { + const { store, service, sendNotification } = await setup(); + + store.emit("task:updated", task({ + id: "FN-7210", + column: "in-review", + paused: false, + pausedReason: undefined, + log: [{ timestamp: new Date().toISOString(), action: "Workflow merge-manual-hold reached manual-required" }], + })); + store.emit("task:updated", task({ + id: "FN-7211", + column: "todo", + workflowTransitionNotification: { + kind: "manual-merge-hold", + column: "in-review", + transitionId: "stale-manual-hold", + createdAt: new Date().toISOString(), + }, + })); + + await Promise.resolve(); + + expect(sendNotification).not.toHaveBeenCalled(); + await service.stop(); + }); + it("does not add a manual-hold workflow notification when the failed status already represents the task update", async () => { const { store, service, sendNotification } = await setup({ failureNotificationMode: "all" }); @@ -526,4 +625,29 @@ describe("NotificationService workflow transition notifications", () => { expect(sendNotification).not.toHaveBeenCalledWith("workflow-notify", expect.anything()); await service.stop(); }); + + it("does not add a recovery-requeue workflow notification when the failed status already represents the task update", async () => { + const { store, service, sendNotification } = await setup({ failureNotificationMode: "all" }); + + store.emit("task:updated", task({ + id: "FN-7212", + column: "todo", + status: "failed", + workflowTransitionNotification: { + kind: "recovery-requeue", + column: "todo", + transitionId: "recovery-requeue:FN-7212:pause-abort-active-work", + nodeId: "pause-abort-recovery-router", + reason: "pause-abort-active-work", + createdAt: new Date().toISOString(), + }, + })); + + await vi.waitFor(() => { + expect(sendNotification).toHaveBeenCalledTimes(1); + }); + expect(sendNotification).toHaveBeenCalledWith("failed", expect.objectContaining({ taskId: "FN-7212" })); + expect(sendNotification).not.toHaveBeenCalledWith("workflow-notify", expect.anything()); + await service.stop(); + }); }); diff --git a/packages/engine/src/notification/notification-service.ts b/packages/engine/src/notification/notification-service.ts index a84d72be9d..f3028c7aff 100644 --- a/packages/engine/src/notification/notification-service.ts +++ b/packages/engine/src/notification/notification-service.ts @@ -735,27 +735,34 @@ export class NotificationService { }; } - if (task.status !== "failed" && this.isManualMergeHold(task)) { + const typedWorkflowTransition = this.workflowTransitionNotificationMarker(task); + if (task.status !== "failed" && (this.isManualMergeHold(task) || typedWorkflowTransition?.kind === "manual-merge-hold")) { return { event: "workflow-notify", metadata: { - notificationDedupeKey: `workflow-transition:${task.id}:manual-merge-hold`, + notificationDedupeKey: typedWorkflowTransition?.transitionId + ? `workflow-transition:${task.id}:${typedWorkflowTransition.transitionId}` + : `workflow-transition:${task.id}:manual-merge-hold`, notificationKind: "manual_merge_hold", title: `Manual merge needed for ${task.id}`, message: "Workflow is holding for manual merge action.", pausedReason: task.pausedReason, + ...(typedWorkflowTransition?.nodeId ? { nodeId: typedWorkflowTransition.nodeId } : {}), + ...(typedWorkflowTransition?.reason ? { reason: typedWorkflowTransition.reason } : {}), }, }; } - if (this.isWorkflowRecoveryRequeue(task)) { + if (task.status !== "failed" && typedWorkflowTransition?.kind === "recovery-requeue") { return { event: "workflow-notify", metadata: { - notificationDedupeKey: `workflow-transition:${task.id}:recovery-requeue`, + notificationDedupeKey: `workflow-transition:${task.id}:${typedWorkflowTransition.transitionId}`, notificationKind: "workflow_recovery_requeue", title: `Workflow requeued ${task.id}`, message: "Workflow recovery moved the task back to todo for another execution pass.", + ...(typedWorkflowTransition.nodeId ? { nodeId: typedWorkflowTransition.nodeId } : {}), + ...(typedWorkflowTransition.reason ? { reason: typedWorkflowTransition.reason } : {}), }, }; } @@ -778,22 +785,15 @@ export class NotificationService { if (task.column !== "in-review") { return false; } - if (task.pausedReason === "manual-hold") { - return true; - } - const latest = this.latestLogAction(task).toLowerCase(); - return latest.includes("manual-required") - || latest.includes("manual merge required") - || latest.includes("merge-manual-hold"); + return task.pausedReason === "manual-hold"; } - private isWorkflowRecoveryRequeue(task: Task): boolean { - if (task.column !== "todo" || task.status === "failed") { - return false; + private workflowTransitionNotificationMarker(task: Task): Task["workflowTransitionNotification"] | undefined { + const marker = task.workflowTransitionNotification; + if (!marker || marker.column !== task.column) { + return undefined; } - const latest = this.latestLogAction(task).toLowerCase(); - return latest.includes("workflow") - && (latest.includes("requeued") || latest.includes("moved back to todo")); + return marker; } private latestLogAction(task: Task): string { diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index d015966a21..358dd51366 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -7177,7 +7177,10 @@ export class SelfHealingManager { continue; } - await this.store.updateTask(task.id, { status: null, error: null }); + await this.store.updateTask(task.id, { + status: null, + error: null, + }); await this.store.logEntry( task.id, "Auto-recovered: stale merge status cleared; merge will be retried", @@ -9148,13 +9151,37 @@ export class SelfHealingManager { continue; } - await this.store.updateTask(task.id, { status: null, error: null }); + const workflowTransitionNotification = route.kind === "node-requeue" + ? { + /* + * FNXC:WorkflowNotifications 2026-06-29-12:47: + * Recovery-driven workflow notifications should be keyed by + * typed task state, not by human-readable recovery log text. + * Stamp the target column so stale markers cannot describe + * later task movement. + */ + kind: "recovery-requeue" as const, + column: "todo" as const, + transitionId: `recovery-requeue:${task.id}:pause-abort-active-work`, + nodeId: "pause-abort-recovery-router", + reason: route.reason, + createdAt: new Date().toISOString(), + } + : undefined; + await this.store.updateTask(task.id, { + status: null, + error: null, + ...(fresh.column === "todo" && workflowTransitionNotification + ? { workflowTransitionNotification } + : {}), + }); if (route.kind === "node-requeue" && fresh.column !== "todo") { await this.store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine", recoveryRehome: true, }); + await this.store.updateTask(task.id, { workflowTransitionNotification }); } // Release any in-memory worktree ownership the leaked park may still // pin, so the requeued task does not re-block the concurrency gate.