root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(FN-7233): persist workflow transition notifications

Browse Source 
Store workflow transition notification markers as durable task state so recovery alerts do not depend on human-readable log text.
This commit is contained in:
gsxdsm
2026-06-29 12:40:00 -07:00
parent e09d45037f
commit 50f8807037
10 changed files with 296 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/workflow-transition-marker-notifications.md Normal file
View File

@@ -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.

27
packages/core/src/__tests__/store-update.test.ts
View File

@@ -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<Task["workflowTransitionNotification"]> = {
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();

7
packages/core/src/db-migrate.ts
View File

@@ -180,11 +180,11 @@ async function migrateTasks(fusionDir: string, db: Database): Promise<void> {
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<void> {
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.

12
packages/core/src/db.ts
View File

@@ -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");
});
}
}
/**

23
packages/core/src/store.ts
View File

@@ -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<TaskStoreEvents> {
*/
enabledWorkflowSteps: (() => { const e = fromJson<string[]>(row.enabledWorkflowSteps); return Array.isArray(e) ? e : undefined; })(),
modifiedFiles: (() => { const m = fromJson<string[]>(row.modifiedFiles); return m && m.length > 0 ? m : undefined; })(),
workflowTransitionNotification: fromJson<Task["workflowTransitionNotification"]>(row.workflowTransitionNotification) ?? undefined,
missionId: row.missionId || undefined,
sliceId: row.sliceId || undefined,
assignedAgentId: row.assignedAgentId || undefined,
@@ -2645,7 +2651,7 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
"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<TaskStoreEvents> {
"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<string, unknown>; 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<string, unknown> | 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<string, unknown>; 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<string, unknown> | 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<Task> {
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) {

23
packages/core/src/types.ts
View File

@@ -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. */

32
packages/engine/src/__tests__/self-healing-paused-abort-recovery.test.ts
View File

@@ -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<typeof vi.fn>).mock.invocationCallOrder[0])
.toBeLessThan((store.moveTask as ReturnType<typeof vi.fn>).mock.invocationCallOrder[0]);
expect((store.moveTask as ReturnType<typeof vi.fn>).mock.invocationCallOrder[0])
.toBeLessThan((store.updateTask as ReturnType<typeof vi.fn>).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(

128
packages/engine/src/notification/__tests__/notification-service.test.ts
View File

@@ -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();
});
});

34
packages/engine/src/notification/notification-service.ts
View File

@@ -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 {

31
packages/engine/src/self-healing.ts
View File

@@ -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.