From f005cee885c535cad45bf8471cca7be316a522fd Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sun, 26 Jul 2026 07:49:05 -0700 Subject: [PATCH] fix(engine): surface silent stalls and add stalled-card watchdog Make planning-guard and remediation no-ops emit warnings, and detect idle non-terminal cards with no session or continuation so FN-8596-class strands show up in logs and run-audit. --- ...self-healing-stalled-card-watchdog.test.ts | 153 ++++++++++++++++++ packages/engine/src/executor.ts | 35 +++- packages/engine/src/run-audit.ts | 3 + packages/engine/src/self-healing.ts | 107 +++++++++++- packages/engine/src/triage.ts | 42 ++++- 5 files changed, 331 insertions(+), 9 deletions(-) create mode 100644 packages/engine/src/__tests__/self-healing-stalled-card-watchdog.test.ts diff --git a/packages/engine/src/__tests__/self-healing-stalled-card-watchdog.test.ts b/packages/engine/src/__tests__/self-healing-stalled-card-watchdog.test.ts new file mode 100644 index 0000000000..3680ef6b74 --- /dev/null +++ b/packages/engine/src/__tests__/self-healing-stalled-card-watchdog.test.ts @@ -0,0 +1,153 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { EventEmitter } from "node:events"; +import type { Settings, Task, TaskStore } from "@fusion/core"; + +const { recordRunAuditEventMock } = vi.hoisted(() => ({ + recordRunAuditEventMock: vi.fn(async () => undefined), +})); +vi.mock("../run-audit.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + createRunAuditor: vi.fn(() => ({ database: recordRunAuditEventMock, git: vi.fn(), filesystem: vi.fn(), sandbox: vi.fn() })), + }; +}); + +import { SelfHealingManager } from "../self-healing.js"; +import { executingTaskLock } from "../active-session-registry.js"; + +/* +FNXC:StalledCardWatchdog 2026-07-26-19:50 (FN-8596 class): +The backstop for "a card must never sit waiting". Every other sweep recovers a KNOWN strand shape; +this one exists for shapes nobody has enumerated yet — FN-8596 sat in `triage` with a finished spec +and nothing anywhere named it, so it was only caught because a human looked at the board. + +It is DETECT-ONLY on purpose, and these tests pin that: a generic mutator racing the specialized +sweeps is the exact bug class this area keeps re-fixing. So the assertions are (a) it names a real +stall, and (b) every "this card is legitimately waiting" shape stays silent — a watchdog that cries +wolf gets ignored, which is the same as not having one. +*/ + +const HOUR = 60 * 60 * 1000; +const NOW = Date.parse("2026-07-26T20:00:00.000Z"); + +function task(id: string, overrides: Partial = {}): Task { + return { + id, + title: id, + description: id, + column: "triage", + dependencies: [], + steps: [], + currentStep: 0, + log: [], + createdAt: new Date(NOW - 4 * HOUR).toISOString(), + // Idle for an hour by default — well past the 30m floor. + updatedAt: new Date(NOW - HOUR).toISOString(), + ...overrides, + } as Task; +} + +function storeFor(tasks: Task[], workItems: Record> = {}): TaskStore & EventEmitter { + return Object.assign(new EventEmitter(), { + getSettings: vi.fn(async () => ({ globalPause: false, enginePaused: false } as Settings)), + listTasks: vi.fn(async () => tasks), + getTask: vi.fn(async (id: string) => tasks.find((t) => t.id === id)), + updateTask: vi.fn(async () => undefined), + moveTask: vi.fn(async () => undefined), + logEntry: vi.fn(async () => undefined), + listWorkflowWorkItemsForTask: vi.fn(async (id: string) => workItems[id] ?? []), + }) as unknown as TaskStore & EventEmitter; +} + +function manager(store: TaskStore, opts: Record = {}) { + return new SelfHealingManager(store, { ...opts } as never); +} + +describe("stalled-card watchdog", () => { + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + vi.setSystemTime(NOW); + }); + afterEach(() => { + vi.useRealTimers(); + executingTaskLock._clearForTest(); + }); + + it("names a card idle past the floor with no session and no continuation", async () => { + const store = storeFor([task("FN-STALL", { column: "triage", status: "planning" })]); + + expect(await manager(store).detectStalledCards()).toBe(1); + expect(recordRunAuditEventMock).toHaveBeenCalledWith(expect.objectContaining({ + type: "task:stall-watchdog-detected", + target: "FN-STALL", + metadata: expect.objectContaining({ taskId: "FN-STALL", column: "triage", status: "planning" }), + })); + }); + + it("stays silent when a continuation is queued to resume the card", async () => { + const store = storeFor( + [task("FN-QUEUED")], + { "FN-QUEUED": [{ state: "held" }] }, + ); + expect(await manager(store).detectStalledCards()).toBe(0); + expect(recordRunAuditEventMock).not.toHaveBeenCalled(); + }); + + it("stays silent for a card that is actively executing", async () => { + const store = storeFor([task("FN-BUSY", { column: "in-progress" })]); + const mgr = manager(store, { getExecutingTaskIds: () => new Set(["FN-BUSY"]) }); + expect(await mgr.detectStalledCards()).toBe(0); + }); + + it("stays silent for an operator park — a deliberate wait is not a stall", async () => { + const store = storeFor([ + task("FN-PAUSED", { paused: true }), + task("FN-USER-PAUSED", { userPaused: true } as Partial), + ]); + expect(await manager(store).detectStalledCards()).toBe(0); + }); + + it("stays silent for terminal columns and for recently-touched cards", async () => { + const store = storeFor([ + task("FN-DONE", { column: "done" }), + task("FN-ARCHIVED", { column: "archived" }), + task("FN-FRESH", { updatedAt: new Date(NOW - 60_000).toISOString() }), + ]); + expect(await manager(store).detectStalledCards()).toBe(0); + }); + + it("does not re-emit for an unchanged card, but re-alerts once its shape changes", async () => { + const stalled = task("FN-DEDUP", { column: "triage", status: "planning" }); + const store = storeFor([stalled]); + const mgr = manager(store); + + expect(await mgr.detectStalledCards()).toBe(1); + expect(await mgr.detectStalledCards()).toBe(0); // same shape → quiet + + stalled.column = "todo"; // shape changed → worth saying again + expect(await mgr.detectStalledCards()).toBe(1); + }); + + it("assumes a continuation exists when the work-item lookup fails, so it never cries wolf", async () => { + const store = storeFor([task("FN-UNKNOWN")]); + (store as unknown as { listWorkflowWorkItemsForTask: unknown }).listWorkflowWorkItemsForTask = + vi.fn(async () => { throw new Error("db down"); }); + expect(await manager(store).detectStalledCards()).toBe(0); + }); + + it("takes no lifecycle action — detection must never move, pause, or fail a card", async () => { + const store = storeFor([task("FN-NO-MUTATE", { column: "triage", status: "planning" })]); + await manager(store).detectStalledCards(); + expect(store.updateTask).not.toHaveBeenCalled(); + expect(store.moveTask).not.toHaveBeenCalled(); + }); + + it("is inert while the engine is paused", async () => { + const store = storeFor([task("FN-PAUSED-ENGINE")]); + (store.getSettings as unknown as { mockResolvedValue: (v: unknown) => void }) + .mockResolvedValue({ globalPause: false, enginePaused: true } as Settings); + expect(await manager(store).detectStalledCards()).toBe(0); + }); +}); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index a3066a018e..9ad7ac9ebd 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -5008,7 +5008,15 @@ export class TaskExecutor { fallbackMaxRevisions: settings.maxPostReviewFixes ?? 3, }); const budget = resolveOptionalStepRevisionBudget(maxRevisions, settings.maxPostReviewFixes ?? 3); - if (!budget.unbounded && (!Number.isFinite(budget.max) || budget.max <= 0)) return false; + if (!budget.unbounded && (!Number.isFinite(budget.max) || budget.max <= 0)) { + // FNXC:RemediationVisibility 2026-07-26-19:20 (FN-8596 follow-up): returning false here + // makes the graph's plan-replan node fail with `remediation-not-scheduled` and leaves the + // card parked in place with nothing scheduled to fix it. Never let that be silent. + executorLog.warn( + `${taskId}: plan-review remediation NOT scheduled — revision budget is zero/invalid (max=${String(budget.max)}). Card left parked.`, + ); + return false; + } const revisionKey = optionalStepRevisionKey(info.nodeId ?? "plan-review", info.stepName); const currentCount = countOptionalStepRevisionAttempts(liveTask, revisionKey, info.stepName); if (!budget.unbounded && currentCount >= budget.max) { @@ -5076,7 +5084,14 @@ export class TaskExecutor { return true; } - if (info.verdict !== "REVISE") return false; + if (info.verdict !== "REVISE") { + // FNXC:RemediationVisibility 2026-07-26-19:20: a hard-failed gate with no parsed REVISE + // verdict schedules nothing, so the remediation node fails and the card parks. Say so. + executorLog.warn( + `${taskId}: pre-merge remediation NOT scheduled for step "${info.stepName}" — status=${info.status}, verdict=${info.verdict ?? "none"}. Card left parked.`, + ); + return false; + } const settings = await mergeEffectiveSettings(this.store, liveTask, await this.store.getSettings()); const maxRevisions = resolveOptionalReviewRevisionBudget({ optionalGroupId: info.nodeId ?? "", @@ -5085,11 +5100,23 @@ export class TaskExecutor { fallbackMaxRevisions: settings.maxPostReviewFixes ?? 3, }); const budget = resolveOptionalStepRevisionBudget(maxRevisions, settings.maxPostReviewFixes ?? 3); - if (!budget.unbounded && (!Number.isFinite(budget.max) || budget.max <= 0)) return false; + if (!budget.unbounded && (!Number.isFinite(budget.max) || budget.max <= 0)) { + executorLog.warn( + `${taskId}: pre-merge remediation NOT scheduled for step "${info.stepName}" — revision budget is zero/invalid (max=${String(budget.max)}). Card left parked.`, + ); + return false; + } const revisionKey = optionalStepRevisionKey(info.nodeId, info.stepName); const currentCount = countOptionalStepRevisionAttempts(liveTask, revisionKey, info.stepName); - if (!budget.unbounded && currentCount >= budget.max) return false; + if (!budget.unbounded && currentCount >= budget.max) { + // Budget exhaustion is a legitimate terminal outcome, but it must be visible: the card stays + // in place with a failed pre-merge step and only an operator bypass clears it. + executorLog.warn( + `${taskId}: pre-merge remediation budget EXHAUSTED for step "${info.stepName}" (${currentCount}/${String(budget.max)}). Card left parked for operator action.`, + ); + return false; + } const nextCount = currentCount + 1; const totalFixCount = (liveTask.postReviewFixCount ?? 0) + 1; diff --git a/packages/engine/src/run-audit.ts b/packages/engine/src/run-audit.ts index 6eb22858bf..32670aa3d9 100644 --- a/packages/engine/src/run-audit.ts +++ b/packages/engine/src/run-audit.ts @@ -609,6 +609,9 @@ export type DatabaseMutationType = { taskId, column, orphanedCount, resultCount }. */ | "task:reconcile-orphaned-pending-step-results" + /* FNXC:StalledCardWatchdog 2026-07-26-19:40: detect-only backstop — a non-terminal card with no + live session and no queued continuation that has not moved past the stall floor. */ + | "task:stall-watchdog-detected" /** * FNXC:MergeQueue 2026-07-15-10:05: * Wedged single-flight merge reclaim. Metadata ids/outcomes-only: diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 0a0b33c23f..c91b6ff82f 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -31,7 +31,7 @@ import { existsSync, mkdirSync, readdirSync, readFileSync, realpathSync, rmSync, import { readFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { isAbsolute, join, relative, resolve } from "node:path"; -import { resolveColumnFlags, IN_REVIEW_STALL_DEADLOCK_LOG_PREFIX, IN_REVIEW_STALL_LOG_PREFIX, IN_REVIEW_STALL_TERMINAL_LOG_PREFIX, allowsAutoMergeProcessing, resolveEffectiveAutoMerge, countRecentIdenticalStallEntries, detectDependencyCycle, detectSelfDefeatingDependency, evaluateNoCommitsNoOpFinalize, evaluateCompletedPromotionFailureProvenance, evaluateSkipBypassTaint, getInReviewStalledSignal, getInReviewStallReason, getPrimaryPrInfo, getStalePausedReviewSignal, getStalePausedTodoSignal, getTaskHardMergeBlocker, getTaskMergeBlocker, isEphemeralAgent, isMergeRequestContractShadowEnabled, isWorkflowColumnsEnabled, isWorkspaceTask, isSharedBranchGroupMemberIntegration, isNearDuplicateCanonicalInactive, parseExplicitDuplicateMarker, flagTriageDuplicate, isTriageDuplicateKeepAcknowledged, resolveMaxAutoMergeRetries, resolveOptionalStepRevisionBudget, resolveOptionalReviewRevisionBudget, resolveWorkflowIrForTask, resolveReboundTarget, planLegacyAdoption, resolveOrphanedPendingStepResults, classifyReviewLease, PLAN_REVIEW_LEASE_STALENESS_MS, AWAITING_APPROVAL_PAUSE_REASON, type Agent, type AgentStore, type ChatStore, type MessageStore, type TaskStore, type Settings, type Task, type MergeDetails, type TaskPriority, type MergeResult, type WorkflowStepResult } from "@fusion/core"; +import { resolveColumnFlags, IN_REVIEW_STALL_DEADLOCK_LOG_PREFIX, IN_REVIEW_STALL_LOG_PREFIX, IN_REVIEW_STALL_TERMINAL_LOG_PREFIX, allowsAutoMergeProcessing, resolveEffectiveAutoMerge, countRecentIdenticalStallEntries, detectDependencyCycle, detectSelfDefeatingDependency, evaluateNoCommitsNoOpFinalize, evaluateCompletedPromotionFailureProvenance, evaluateSkipBypassTaint, getInReviewStalledSignal, getInReviewStallReason, getPrimaryPrInfo, getStalePausedReviewSignal, getStalePausedTodoSignal, getTaskHardMergeBlocker, getTaskMergeBlocker, isEphemeralAgent, isMergeRequestContractShadowEnabled, isWorkflowColumnsEnabled, isWorkspaceTask, isSharedBranchGroupMemberIntegration, isNearDuplicateCanonicalInactive, parseExplicitDuplicateMarker, flagTriageDuplicate, isTriageDuplicateKeepAcknowledged, resolveMaxAutoMergeRetries, resolveOptionalStepRevisionBudget, resolveOptionalReviewRevisionBudget, resolveWorkflowIrForTask, resolveReboundTarget, planLegacyAdoption, resolveOrphanedPendingStepResults, classifyReviewLease, PLAN_REVIEW_LEASE_STALENESS_MS, ACTIVE_WORKFLOW_WORK_ITEM_STATES, AWAITING_APPROVAL_PAUSE_REASON, type Agent, type AgentStore, type ChatStore, type MessageStore, type TaskStore, type Settings, type Task, type MergeDetails, type TaskPriority, type MergeResult, type WorkflowStepResult } from "@fusion/core"; import { finalizePlanningSegment } from "@fusion/core"; import type { MeshLeaseManager } from "./mesh-lease-manager.js"; import { createLogger, schedulerLog } from "./logger.js"; @@ -437,6 +437,7 @@ const ORPHANED_EXECUTION_RECOVERY_GRACE_MS = 60_000; * reapable column (todo/triage) shorter than this is left alone, so the reaper * never races a task mid-transition out of in-progress. */ +const STALLED_CARD_WATCHDOG_MS = 30 * 60_000; const LEAKED_WORKTREE_SLOT_GRACE_MS = 60_000; /* FNXC:MergeQueue 2026-07-15-09:50: @@ -2872,6 +2873,7 @@ export class SelfHealingManager { await this.options.reconcileAllMissionFeatures(); }, }, + { name: "detect-stalled-cards", fn: () => this.detectStalledCards() }, { name: "recover-completed-tasks", fn: () => this.recoverCompletedTasks() }, { name: "recover-stranded-completed-todo", fn: () => this.recoverStrandedCompletedTodoTasks() }, { name: "recover-advanced-triage", fn: () => this.recoverAdvancedTriageTasks() }, @@ -6989,6 +6991,109 @@ export class SelfHealingManager { write so the whole-array update cannot clobber a lease written after the page snapshot. User pauses are never disturbed. */ + /* + FNXC:StalledCardWatchdog 2026-07-26-19:40 (FN-8596 class): + The BACKSTOP for "a card must never sit waiting". Every other sweep in this file recovers a KNOWN + strand shape; this one exists for the shapes nobody has enumerated yet. FN-8596 was exactly that: + a card sat in `triage` doing nothing for ten minutes with a finished spec, and no sweep, log, or + audit event named it — the strand was only found because a human noticed the board. + + DETECT-ONLY, deliberately. It emits a run-audit event and a warning; it does NOT move, requeue, + pause, or fail anything. A generic mutator racing the specialized sweeps is precisely the class of + bug this file keeps fixing, so recovery stays with the sweep that owns each shape and this one + guarantees visibility. Emitting the shape (column/status/whether a continuation or session exists) + is what makes the next unknown strand diagnosable in one query instead of an archaeology session. + + A card counts as stalled when ALL hold: + - it is in a non-terminal column (done/archived are finished, not waiting), + - nothing is executing it (executing set + executingTaskLock + live session registry), + - it has no ACTIVE workflow continuation queued to resume it, + - it is not paused (an operator park is a deliberate wait, not a stall), + - and it has not been touched for longer than the stall floor. + Deduped per (taskId, shape) so a genuinely parked card does not re-emit every sweep. + */ + private readonly stalledCardSignatures = new Map(); + + async detectStalledCards(): Promise { + try { + const settings = await this.store.getSettings().catch(() => undefined); + if (settings?.globalPause === true || settings?.enginePaused === true) return 0; + + const executingIds = this.options.getExecutingTaskIds?.() ?? new Set(); + const now = Date.now(); + const tasks = await this.store.listTasks({ slim: true, includeArchived: false }); + let detected = 0; + const seen = new Set(); + + for (const task of tasks) { + if (task.column === "done" || task.column === "archived") continue; + if (task.paused === true || task.userPaused === true) continue; + if (executingIds.has(task.id) || executingTaskLock.has(task.id)) continue; + if (this.options.isTaskActive?.(task.id) === true) continue; + const livePaths = activeSessionRegistry.pathsForTask(task.id); + if (livePaths.some((path) => activeSessionRegistry.isPathActive(path))) continue; + + const touchedAt = Date.parse(task.updatedAt ?? task.columnMovedAt ?? ""); + if (!Number.isFinite(touchedAt) || now - touchedAt < STALLED_CARD_WATCHDOG_MS) continue; + + // An active continuation means something IS queued to resume this card — not a stall. + let hasContinuation = false; + try { + const items = await this.store.listWorkflowWorkItemsForTask?.(task.id, { kinds: ["task"] }) ?? []; + hasContinuation = items.some((item) => ACTIVE_WORKFLOW_WORK_ITEM_STATES.includes(item.state)); + } catch { + // Unknown continuation state — assume one exists so the watchdog never cries wolf. + hasContinuation = true; + } + if (hasContinuation) continue; + + seen.add(task.id); + const signature = `${task.column}|${task.status ?? "null"}`; + if (this.stalledCardSignatures.get(task.id) === signature) continue; + this.stalledCardSignatures.set(task.id, signature); + detected += 1; + + const idleMinutes = Math.floor((now - touchedAt) / 60_000); + log.warn( + `Stalled card ${task.id}: idle ${idleMinutes}m in '${task.column}' (status=${task.status ?? "null"}) ` + + "with no live session and no queued continuation — nothing is scheduled to advance it", + ); + try { + await createRunAuditor(this.store, { + runId: generateSyntheticRunId("stalled-card-watchdog", task.id), + agentId: "self-healing", + taskId: task.id, + taskLineageId: task.lineageId, + phase: "detect-stalled-cards", + }).database({ + type: "task:stall-watchdog-detected", + target: task.id, + // ids/counts/outcomes only — never spec text, error prose, or reviewer output. + metadata: { + taskId: task.id, + column: task.column, + status: task.status ?? null, + idleMinutes, + hasWorktree: Boolean(task.worktree), + stepCount: task.steps?.length ?? 0, + }, + }); + } catch (error) { + log.warn(`detectStalledCards: audit emit failed for ${task.id}: ${error instanceof Error ? error.message : String(error)}`); + } + } + + // Forget cards that recovered, so a future stall on the same card re-alerts. + for (const id of [...this.stalledCardSignatures.keys()]) { + if (!seen.has(id)) this.stalledCardSignatures.delete(id); + } + return detected; + } catch (error) { + log.error(`detectStalledCards failed: ${error instanceof Error ? error.message : String(error)}`); + return 0; + } + } + async reconcileOrphanedPendingStepResults(): Promise { try { const pageSize = 500; diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index 16d22a24db..a8be4b4ef9 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -2802,10 +2802,32 @@ export class TriageProcessor { * mutation in one task-lock acquisition; row-only atomic patches cannot protect PROMPT.md. */ private async runIfStillPlanningUnderTaskLock(task: Task, operation: () => Promise): Promise { + /* + FNXC:TriageFinalizeVisibility 2026-07-26-19:05 (FN-8596 follow-up): + Every caller of this helper treats `false` as "skip silently and return". That is how the + FN-8596 strand hid: the planning-stage predicate went false (stale execution stamps), each + guarded write no-opped, and NOTHING anywhere said so. Skipping is a legitimate outcome when the + scheduler genuinely advanced the card, but it must be OBSERVABLE, so log the reason with the + live state that decided it. Logged here rather than at the four call sites so a future caller + inherits the visibility instead of re-introducing a silent branch. + */ const store = this.store as TaskStore; - if (typeof store.withTaskLock !== "function" || typeof store.readTaskForMove !== "function") return false; + if (typeof store.withTaskLock !== "function" || typeof store.readTaskForMove !== "function") { + planLog.warn( + `${task.id}: planning-guarded write skipped — store lacks withTaskLock/readTaskForMove; no recovery write performed`, + ); + return false; + } return store.withTaskLock(task.id, async () => { - if (!isTaskStillInPlanningStage(await store.readTaskForMove(task.id))) return false; + const live = await store.readTaskForMove(task.id); + if (!isTaskStillInPlanningStage(live)) { + planLog.warn( + `${task.id}: planning-guarded write skipped — no longer in the planning stage ` + + `(column=${live?.column ?? "unknown"}, status=${live?.status ?? "null"}, ` + + `executionStartedAt=${live?.executionStartedAt ?? "null"})`, + ); + return false; + } await operation(); return true; }); @@ -3535,9 +3557,21 @@ export class TriageProcessor { if (task.column !== "todo") { const moveTaskIf = (this.store as unknown as { moveTaskIf?: TaskStore["moveTaskIf"] }).moveTaskIf; - if (typeof moveTaskIf !== "function") return; + if (typeof moveTaskIf !== "function") { + // FNXC:TriageFinalizeVisibility 2026-07-26-19:05: the release move is the handoff. If it + // cannot even be attempted the card stays in the planner column with a finished spec, so + // never let that be silent. + planLog.warn(`${task.id}: planning handoff skipped — store does not expose moveTaskIf; card left in ${task.column}`); + return; + } const release = await moveTaskIf.call(this.store, task.id, "todo", isTaskStillInPlanningStage); - if (!release.moved) return; + if (!release.moved) { + planLog.warn( + `${task.id}: planning handoff to todo REFUSED by the planning-stage guard ` + + `(column=${release.task?.column ?? "unknown"}, status=${release.task?.status ?? "null"}). Card left in ${task.column}.`, + ); + return; + } } /*