diff --git a/docs/architecture.md b/docs/architecture.md index 3109da5b67..d2a6df9fae 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1783,6 +1783,7 @@ This section preserves the detailed lifecycle/self-healing contracts that were f - **Dual-observe parity seam (FN-5742 Phase 2)**: with the same flag ON, legacy remains authoritative while shadow reads compute/emit parity telemetry only. Scheduler emits `merge:dependency-parity-diff` when `in-review|done|archived` dependency satisfaction diverges from completion-handoff marker satisfaction, and `merge:lease-parity-diff` when legacy in-review overlap leasing diverges from shadow lease decomposition. Merger emits `merge:request-dequeued-shadow` (agree/disagree metadata) by comparing legacy dequeue selection to shadow merge-request selection while explicitly skipping `manual-required` rows. Phase 3 dequeue cutover is gated on sustained parity (low disagreement rate) from these additive events; no lifecycle authority changes in Phase 2. - **Authoritative cutover seam (FN-5743 Phase 3)**: with the flag ON, merge-request records and `completion_handoff_accepted` markers become authoritative enforcement signals for dequeue/retry ownership and dependency/lease gates. Accepted handoffs stop stamping `in-review` executor overlap leases, transient merge retries stay in merge-request state (`running → retrying → queued`, terminal `exhausted|succeeded|cancelled`) without `todo` rebounds, and user hard-cancel (`in-review → todo`) deterministically cancels pending merge-request records while keeping FN-5147/FN-5704 behavior unchanged. - **No-progress churn terminalization (FN-5168)**: `StuckTaskDetector` now tracks ignored `fn_task_update` rebuffs via `recordIgnoredStepUpdate(taskId)` and, after one loop/compact-and-resume recovery has already fired in the same `execute()` lifecycle, escalates `ignoredStepUpdateCount >= 25` to the terminal reason `no-progress-churn`. `SelfHealingManager.checkStuckBudget()` maps that reason directly to `STUCK_NO_PROGRESS_CHURN`, emits `task:stuck-no-progress-churn-terminalized` with `{ taskId, ignoredStepUpdateCount, stuckKillStreak, lastReason }`, and parks the task in `in-review` without consuming the normal stuck-kill budget. Under FN-5147 `autoMerge: false`, that failed in-review task remains terminal-until-merged just like `STUCK_LOOP_EXHAUSTED`; the new class adds an earlier bounded exit, not a re-execution path. +- **Todo↔in-progress flapping convergence (FN-5941)**: live backward-recovery paths now share a `getFalsePositiveRequeueSignal(...)` guard that suppresses `in-progress → todo` recovery when any hard liveness proof exists (`getExecutingTaskIds`, recent active-heartbeat run, checked-out lease, live worktree+branch binding, or recent `executionStartedAt` inside the relevant grace window). Suppressed candidates emit observation-only `task:*no-action` audits instead of silently mutating lifecycle state. Scheduler adds a short `recentEngineTodoRequeues` settle window so engine-sourced requeues cannot be re-dispatched immediately on the same `task:moved → todo` tick. The durable convergence backstop is the dispatch-oscillation breaker: scheduler reuses `task.dispatchStormCount` + `task.lastDispatchAt` as a sliding-window counter (`dispatchOscillationThreshold`, `dispatchOscillationWindowMs`) and, when the threshold is exceeded, leaves the task parked in `todo`, sets `paused: true` with `pausedReason: "dispatch-oscillation"`, records `task:dispatch-oscillation-terminalized`, and requires an operator unpause or forward move to reset the counter. - **Landed-files attribution (FN-5103)**: Rebase-strategy `mergeDetails.landedFiles` / `filesChanged` / `insertions` / `deletions` are captured from task-attributable commits only via `filterFilesToOwnTaskCommits` (subject-prefix + trailer + bracket-prefix evidence), tagged `landedFilesAttributionRestricted: true`. Zero own commits → `landedFiles: []` and `noOpVerifiedShortCircuit: true`. FN-5304 guard: when `..HEAD` reports zero own commits, merger must also validate the source `fusion/` tip; if that source tip still has attributable own commits relative to `rebaseBaseSha`, throw `SilentNoOpAttributionMismatchError`, refuse writing `mergeConfirmed: true`, park the task in `in-review` with `status: "failed"`, and emit `merge:no-op-attribution-mismatch`. If source ref is unavailable, skip with diagnostic + `merge:no-op-attribution-mismatch-skipped` (`reason: "source-ref-unavailable"`). Attribution-helper failures fall back to the unrestricted `rebaseBaseSha..sha` walk and set `landedFilesCaptureFallback: 'attribution-failed'`. Self-healing `recoverDoneTaskMergeMetadata` skips reconcile when `landedFilesAttributionRestricted` or `noOpVerifiedShortCircuit` is set so the narrower set is not overwritten with the full range. Squash-strategy capture is unchanged. - **Soft-delete scheduler invalidation (FN-5137)**: `task:deleted` events must invalidate `AutoClaimSnapshotManager` and clear scheduler bookkeeping (`pausedTaskIds`, `failedTaskIds`, `wasNodeDispatchValidationBlocked`, `wasNodeBlocked`); `executor.execute()` / `resumeOrphaned()` / `resumeTaskForAgent()` refuse any task with `deletedAt` set. - **Soft-delete in-flight abort (FN-5142)**: `task:deleted` must immediately abort/dispose active executor work (`activeSessions`, `activeStepExecutors`, `activeWorkflowStepSessions`, reviewer subagents), interrupt active merge state (`mergeAbortController`, `activeMergeSession`, `activeMergeTaskId`, `mergeActive`, `mergeQueue`, `pausedReviewTaskIds`), and abort triage specify/subagent sessions for that id. Handlers are per-task and idempotent. diff --git a/docs/diagnostics.md b/docs/diagnostics.md index ab0ca6b047..176cea2633 100644 --- a/docs/diagnostics.md +++ b/docs/diagnostics.md @@ -71,6 +71,15 @@ Time-based stuck/stalled/stale surfaces now floor activity timestamps using `set - Audit event: `task:stuck-no-progress-churn-terminalized` with `{ taskId, ignoredStepUpdateCount, stuckKillStreak, lastReason: "no-progress-churn" }`. - Outcome: task is marked `status: "failed"`, moved to `in-review`, and not requeued; operators should decompose/rescope the task instead of waiting for more automatic stuck-kill retries. +## Dispatch oscillation breaker (`[scheduler]`, `[self-healing]`) + +FN-5941 adds a convergence backstop for repeated `todo↔in-progress` churn. + +- Suppressed false-positive backward recoveries emit `task:reclaim-self-owned-branch-conflict-no-action`, `task:auto-recover-in-progress-limbo-no-action`, or `task:stuck-loop-exhausted-no-action` with liveness metadata (`taskId`, `branch`, `worktree`, `checkedOutBy`, `executionStartedAt`, `executionAgeMs`, `graceMs`, `liveWorktreeBoundBranch`, `reason`). +- Scheduler settle-window diagnostic: `Task was engine-requeued ms ago — waiting ms settle window before redispatch`. +- Terminal audit event: `task:dispatch-oscillation-terminalized` with `{ taskId, cycleCount, windowMs, lastMoveSource }`. +- Outcome: task stays in `todo`, is auto-paused with `pausedReason: "dispatch-oscillation"`, and requires operator unpause/forward progress to reset the counter. + ## Stale self-owned active-session cleanup diagnostics (`[executor]`) FN-5346 adds a same-task stale-binding reconcile marker before worktree removal: diff --git a/docs/settings-reference.md b/docs/settings-reference.md index 94712f71a3..183b6217bc 100644 --- a/docs/settings-reference.md +++ b/docs/settings-reference.md @@ -389,6 +389,9 @@ Default notes: | `specStalenessEnabled` | `boolean` | `false` | Enforce automatic re-planning for stale plans. | | `specStalenessMaxAgeMs` | `number` | `21600000` | Spec staleness threshold in ms (6 hours). | | `taskStuckTimeoutMs` | `number` | `undefined` | Inactivity timeout for stuck-task recovery. | +| `dispatchOscillationThreshold` | `number` | `5` | Number of rapid `todo↔in-progress` cycles allowed before scheduler auto-pauses the task with `pausedReason="dispatch-oscillation"`. | +| `dispatchOscillationWindowMs` | `number` | `60000` | Sliding window in ms used to count rapid `todo↔in-progress` cycles for the dispatch-oscillation breaker. | +| `dispatchOscillationSettleMs` | `number` | `5000` | Minimum settle delay after an engine-sourced `in-progress → todo` recovery before scheduler may re-dispatch the task. Prevents immediate same-tick redispatch races. | | `runtimeStopDrainMs` | `number` | `2000` | Maximum milliseconds `InProcessRuntime.stop()` waits for in-flight tasks to drain after aborting AI sessions. Set `0` to skip drain polling entirely (useful for test/CI). | | `engineActiveSinceMs` | `number` | `undefined` | Epoch ms when the in-process runtime last became active (startup or unpause). Time-based stuck/stalled/stale surfaces floor their activity anchor at this timestamp so paused/stopped downtime is not counted as quiet age. Runtime-managed; typically not set manually. | | `engineActivationGraceMs` | `number` | `300000` | Extra grace window (ms) added after `engineActiveSinceMs` before time-based stuck/stalled/stale surfaces can fire. Set `0` to disable warmup. | diff --git a/packages/core/src/__tests__/settings-parity.test.ts b/packages/core/src/__tests__/settings-parity.test.ts index eae4ca2f72..14ce670f26 100644 --- a/packages/core/src/__tests__/settings-parity.test.ts +++ b/packages/core/src/__tests__/settings-parity.test.ts @@ -181,7 +181,13 @@ describe("settings key parity", () => { it("keeps task stuck timeout active by default without coupling to workflow step timeout", () => { expect(DEFAULT_PROJECT_SETTINGS.taskStuckTimeoutMs).toBe(600_000); + expect(DEFAULT_PROJECT_SETTINGS.dispatchOscillationThreshold).toBe(5); + expect(DEFAULT_PROJECT_SETTINGS.dispatchOscillationWindowMs).toBe(60_000); + expect(DEFAULT_PROJECT_SETTINGS.dispatchOscillationSettleMs).toBe(5_000); expect(DEFAULT_PROJECT_SETTINGS.runtimeStopDrainMs).toBe(2_000); + expect(isProjectSettingsKey("dispatchOscillationThreshold")).toBe(true); + expect(isProjectSettingsKey("dispatchOscillationWindowMs")).toBe(true); + expect(isProjectSettingsKey("dispatchOscillationSettleMs")).toBe(true); // workflowStepTimeoutMs MOVED to workflow settings (U4) — no longer a project key. expect(isProjectSettingsKey("workflowStepTimeoutMs")).toBe(false); expect(PROJECT_SETTINGS_KEYS).not.toContain("workflowStepTimeoutMs"); diff --git a/packages/core/src/settings-schema.ts b/packages/core/src/settings-schema.ts index 5e3cef5c65..90c9b0f0dd 100644 --- a/packages/core/src/settings-schema.ts +++ b/packages/core/src/settings-schema.ts @@ -342,6 +342,12 @@ export const DEFAULT_PROJECT_SETTINGS = { specStalenessEnabled: false, specStalenessMaxAgeMs: 6 * 60 * 60 * 1000, taskStuckTimeoutMs: 600_000, + /** Number of rapid todo↔in-progress cycles allowed before auto-pausing the task. */ + dispatchOscillationThreshold: 5, + /** Sliding time window used to count rapid todo↔in-progress cycles. */ + dispatchOscillationWindowMs: 60_000, + /** Delay before scheduler may re-dispatch an engine-requeued todo task. */ + dispatchOscillationSettleMs: 5_000, runtimeStopDrainMs: 2_000, engineActiveSinceMs: undefined, engineActivationGraceMs: 5 * 60_000, diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index 2bbe56a69a..1940c905e5 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -3649,6 +3649,15 @@ export interface ProjectSettings { * than this duration, the task is considered stuck and will be terminated and retried. * Default: 600000 (10 minutes). Set to 0 to disable. */ taskStuckTimeoutMs?: number; + /** Number of rapid todo↔in-progress cycles allowed before auto-pausing the task. + * Default: 5. */ + dispatchOscillationThreshold?: number; + /** Sliding time window in milliseconds used to count rapid todo↔in-progress cycles. + * Default: 60000 (1 minute). */ + dispatchOscillationWindowMs?: number; + /** Delay before scheduler may re-dispatch an engine-requeued todo task. + * Default: 5000 (5 seconds). */ + dispatchOscillationSettleMs?: number; /** Maximum milliseconds InProcessRuntime.stop() waits for in-flight tasks to drain * AFTER aborting their AI sessions. Default: 2000. Set to 0 to skip drain waits * entirely (test/CI). Set to 30000 to preserve the historical 30s grace window. */ diff --git a/packages/engine/src/__tests__/reliability-interactions/todo-inprogress-flapping.test.ts b/packages/engine/src/__tests__/reliability-interactions/todo-inprogress-flapping.test.ts new file mode 100644 index 0000000000..445d1ac490 --- /dev/null +++ b/packages/engine/src/__tests__/reliability-interactions/todo-inprogress-flapping.test.ts @@ -0,0 +1,556 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { EventEmitter } from "node:events"; +import { mkdirSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { execSync } from "node:child_process"; +import type { Settings, Task, TaskStore } from "@fusion/core"; +import { Scheduler } from "../../scheduler.js"; +import { SelfHealingManager } from "../../self-healing.js"; +import * as branchConflictModule from "../../branch-conflicts.js"; +import * as worktreePoolModule from "../../worktree-pool.js"; + +function git(cwd: string, command: string): string { + return execSync(`git ${command}`, { cwd, encoding: "utf8", stdio: ["pipe", "pipe", "pipe"] }).trim(); +} + +type MutableSettings = Settings & { + autoMerge?: boolean; + globalPause?: boolean; + enginePaused?: boolean; + dispatchOscillationSettleMs?: number; +}; + +function makeTask(rootDir: string, overrides: Partial = {}): Task { + return { + id: "FN-5941", + title: "flapping", + description: "test", + column: "in-progress", + branch: "fusion/fn-5941", + worktree: join(rootDir, ".worktrees", "fn-5941"), + paused: false, + userPaused: false, + checkedOutBy: undefined, + dependencies: [], + steps: [{ id: "s1", title: "step", status: "in-progress" } as any], + currentStep: 1, + log: [], + createdAt: new Date("2026-01-01T00:00:00.000Z").toISOString(), + updatedAt: new Date("2026-01-01T00:00:00.000Z").toISOString(), + columnMovedAt: new Date("2026-01-01T00:00:00.000Z").toISOString(), + executionStartedAt: new Date("2026-01-01T00:00:00.000Z").toISOString(), + ...overrides, + } as Task; +} + +function makeStore(rootDir: string, task: Task, settingsOverrides: Partial = {}): TaskStore & EventEmitter { + const emitter = new EventEmitter(); + const settings = { + autoMerge: true, + globalPause: false, + enginePaused: false, + baseBranch: "main", + mergeStrategy: "direct", + autoRecovery: { mode: "deterministic-only", maxRetries: 3 }, + maxStuckKills: 6, + ...settingsOverrides, + } as unknown as Settings; + + return Object.assign(emitter, { + getSettings: vi.fn(async () => settings), + getTask: vi.fn(async (id: string) => (id === task.id ? task : null)), + listTasks: vi.fn(async ({ column }: { column?: string } = {}) => (column === task.column ? [task] : [])), + updateTask: vi.fn(async (_id: string, updates: Partial) => Object.assign(task, updates)), + moveTask: vi.fn(async (_id: string, column: Task["column"], opts?: Record) => { + task.column = column; + (task as any).__lastMoveOpts = opts; + return task; + }), + logEntry: vi.fn(async () => undefined), + recordRunAuditEvent: vi.fn(async () => undefined), + appendAgentLog: vi.fn(async () => undefined), + updateSettings: vi.fn(async () => settings), + clearStaleExecutionStartBranchReferences: vi.fn(() => []), + walCheckpoint: vi.fn(() => ({ busy: 0, log: 0, checkpointed: 0 })), + archiveTaskAndCleanup: vi.fn(async () => ({})), + mergeTask: vi.fn(async () => undefined), + handoffToReview: vi.fn(async () => task), + getRootDir: vi.fn(() => rootDir), + }) as unknown as TaskStore & EventEmitter; +} + +function makeSchedulerStore(rootDir: string, task: Task, settingsOverrides: Partial = {}): TaskStore & EventEmitter { + const emitter = new EventEmitter(); + const settings = { + autoMerge: true, + globalPause: false, + enginePaused: false, + maxConcurrent: 1, + maxWorktrees: 10, + groupOverlappingFiles: false, + dispatchOscillationSettleMs: 5_000, + worktreeNaming: "task-id", + ...settingsOverrides, + } as unknown as Settings; + + return Object.assign(emitter, { + getSettings: vi.fn(async () => settings), + getTask: vi.fn(async (id: string) => (id === task.id ? task : null)), + listTasks: vi.fn(async ({ column }: { column?: string } = {}) => { + if (!column) return [task]; + return task.column === column ? [task] : []; + }), + updateTask: vi.fn(async (_id: string, updates: Partial) => Object.assign(task, updates)), + moveTask: vi.fn(async (_id: string, column: Task["column"], opts?: Record) => { + const from = task.column; + task.column = column; + task.columnMovedAt = new Date(Date.now()).toISOString(); + emitter.emit("task:moved", { task, from, to: column, source: (opts?.moveSource as "user" | "engine" | "scheduler" | undefined) ?? "engine" }); + return task; + }), + logEntry: vi.fn(async () => undefined), + recordRunAuditEvent: vi.fn(async () => undefined), + getRootDir: vi.fn(() => rootDir), + getTasksDir: vi.fn(() => join(rootDir, ".fusion", "tasks")), + parseFileScopeFromPrompt: vi.fn(async () => []), + on: emitter.on.bind(emitter), + off: emitter.off.bind(emitter), + }) as unknown as TaskStore & EventEmitter; +} + +describe("FN-5941 reliability interactions: todo/in-progress flapping", () => { + let rootDir = ""; + + beforeEach(() => { + vi.restoreAllMocks(); + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T02:00:00.000Z")); + vi.spyOn(Scheduler.prototype as any, "validateTaskFilesystem").mockResolvedValue({ valid: true }); + rootDir = join(tmpdir(), `fn-5941-${Math.random().toString(36).slice(2)}`); + mkdirSync(rootDir, { recursive: true }); + git(rootDir, "init -b main"); + git(rootDir, "config user.name 'Fusion'"); + git(rootDir, "config user.email 'hi@runfusion.ai'"); + writeFileSync(join(rootDir, "README.md"), "root\n"); + git(rootDir, "add README.md"); + git(rootDir, "commit -m 'init'"); + mkdirSync(join(rootDir, ".worktrees"), { recursive: true }); + mkdirSync(join(rootDir, ".worktrees", "fn-5941"), { recursive: true }); + vi.spyOn(worktreePoolModule, "isUsableTaskWorktree").mockResolvedValue(true); + vi.spyOn(branchConflictModule, "inspectBranchConflict").mockResolvedValue({ + kind: "reclaimable", + taskAttributedCommitCount: 1, + strandedCommits: [{ sha: "c1", authorName: "a", subject: "s", timestamp: Date.now() }], + livePath: join(rootDir, ".worktrees", "fn-5941"), + tipSha: "abc123abc123abc123abc123abc123abc123abcd", + } as any); + }); + + afterEach(() => { + vi.useRealTimers(); + if (rootDir) rmSync(rootDir, { recursive: true, force: true }); + }); + + it("does not requeue a genuinely executing in-progress task during resume-limbo reclaim", async () => { + const task = makeTask(rootDir, { + resumeLimboCount: 1, + resumeLimboTipSha: "abc123abc123abc123abc123abc123abc123abcd", + resumeLimboStepSignature: JSON.stringify({ currentStep: 1, steps: ["in-progress"] }), + executionStartedAt: new Date(Date.now() - 30_000).toISOString(), + }); + const store = makeStore(rootDir, task); + const manager = new SelfHealingManager(store as any, { + rootDir, + getExecutingTaskIds: () => new Set([task.id]), + } as any); + + const recovered = await manager.reclaimSelfOwnedBranchConflicts(); + + expect(recovered).toBe(0); + expect(store.moveTask).not.toHaveBeenCalled(); + expect(task.column).toBe("in-progress"); + expect(task.resumeLimboCount).toBe(1); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:reclaim-self-owned-branch-conflict-no-action", + target: task.id, + metadata: expect.objectContaining({ reason: "executor-active" }), + })); + + manager.stop(); + }); + + it("defers resume-limbo reclaim when execution started within the grace window", async () => { + const task = makeTask(rootDir, { + resumeLimboCount: 1, + resumeLimboTipSha: "abc123abc123abc123abc123abc123abc123abcd", + resumeLimboStepSignature: JSON.stringify({ currentStep: 1, steps: ["in-progress"] }), + executionStartedAt: new Date(Date.now() - 60_000).toISOString(), + }); + const store = makeStore(rootDir, task); + const manager = new SelfHealingManager(store as any, { rootDir } as any); + + const recovered = await manager.reclaimSelfOwnedBranchConflicts(); + + expect(recovered).toBe(0); + expect(store.moveTask).not.toHaveBeenCalled(); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:reclaim-self-owned-branch-conflict-no-action", + target: task.id, + metadata: expect.objectContaining({ reason: "recent-execution-started" }), + })); + + manager.stop(); + }); + + it("does not requeue a task tracked by an active heartbeat run during resume-limbo reclaim", async () => { + const task = makeTask(rootDir, { + id: "FN-5941-HEARTBEAT", + resumeLimboCount: 1, + resumeLimboTipSha: "abc123abc123abc123abc123abc123abc123abcd", + resumeLimboStepSignature: JSON.stringify({ currentStep: 1, steps: ["in-progress"] }), + executionStartedAt: new Date("2025-12-31T20:00:00.000Z").toISOString(), + }); + const store = makeStore(rootDir, task); + const agentStore = { + listActiveHeartbeatRuns: vi.fn().mockResolvedValue([ + { startedAt: new Date().toISOString(), contextSnapshot: { taskId: task.id } }, + ]), + }; + const manager = new SelfHealingManager(store as any, { rootDir, agentStore } as any); + + const recovered = await manager.reclaimSelfOwnedBranchConflicts(); + + expect(recovered).toBe(0); + expect(store.moveTask).not.toHaveBeenCalled(); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:reclaim-self-owned-branch-conflict-no-action", + target: task.id, + metadata: expect.objectContaining({ reason: "active-heartbeat-run" }), + })); + + manager.stop(); + }); + + it("does not requeue an executor-active task when stuck-kill budget is exhausted", async () => { + const task = makeTask(rootDir, { + id: "FN-5941-STUCK", + stuckKillCount: 6, + executionStartedAt: new Date(Date.now() - 15_000).toISOString(), + steps: [{ id: "s1", title: "step", status: "in-progress" } as any], + currentStep: 1, + }); + const store = makeStore(rootDir, task, { maxStuckKills: 6 }); + const manager = new SelfHealingManager(store as any, { + rootDir, + getExecutingTaskIds: () => new Set([task.id]), + } as any); + + const allowedRetry = await manager.checkStuckBudget(task.id, "inactivity"); + + expect(allowedRetry).toBe(false); + expect(store.moveTask).not.toHaveBeenCalled(); + expect(task.column).toBe("in-progress"); + expect(task.stuckKillCount).toBe(6); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:stuck-loop-exhausted-no-action", + target: task.id, + metadata: expect.objectContaining({ + reason: "executor-active", + attemptedStuckKillCount: 7, + maxStuckKills: 6, + }), + })); + + manager.stop(); + }); + + it("does not requeue a recent resume-limbo task when execution just started", async () => { + const task = makeTask(rootDir, { + id: "FN-5941-RESUME", + resumeLimboCount: 1, + resumeLimboTipSha: "abc123abc123abc123abc123abc123abc123abcd", + resumeLimboStepSignature: JSON.stringify({ currentStep: 1, steps: ["in-progress"] }), + executionStartedAt: new Date(Date.now() - 15_000).toISOString(), + }); + const store = makeStore(rootDir, task); + const manager = new SelfHealingManager(store as any, { rootDir } as any); + + const recovered = await manager.reclaimSelfOwnedBranchConflicts(); + + expect(recovered).toBe(0); + expect(store.moveTask).not.toHaveBeenCalled(); + expect(task.column).toBe("in-progress"); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:reclaim-self-owned-branch-conflict-no-action", + target: task.id, + metadata: expect.objectContaining({ reason: "recent-execution-started" }), + })); + + manager.stop(); + }); + + it("still requeues a genuinely dead task when stuck-kill budget is exhausted", async () => { + const task = makeTask(rootDir, { + id: "FN-5941-DEAD", + stuckKillCount: 6, + branch: undefined, + worktree: join(rootDir, ".worktrees", "fn-5941-missing-dead"), + executionStartedAt: new Date("2025-12-31T20:00:00.000Z").toISOString(), + steps: [{ id: "s1", title: "step", status: "in-progress" } as any], + currentStep: 1, + }); + const store = makeStore(rootDir, task, { maxStuckKills: 6 }); + const manager = new SelfHealingManager(store as any, { rootDir } as any); + + const allowedRetry = await manager.checkStuckBudget(task.id, "inactivity"); + + expect(allowedRetry).toBe(false); + expect(store.moveTask).toHaveBeenCalledWith(task.id, "todo", expect.objectContaining({ + preserveProgress: true, + preserveStatus: true, + moveSource: "engine", + recoveryRehome: true, + })); + expect(task.column).toBe("todo"); + expect(task.stuckKillCount).toBe(7); + expect(task.status).toBe("queued"); + + manager.stop(); + }); + + it("does not requeue an in-progress limbo candidate while a checkout lease is still live", async () => { + const task = makeTask(rootDir, { + branch: undefined, + worktree: join(rootDir, ".worktrees", "fn-5941-missing"), + checkedOutBy: "agent-123", + executionStartedAt: new Date("2025-12-31T23:00:00.000Z").toISOString(), + updatedAt: new Date("2025-12-31T23:00:00.000Z").toISOString(), + columnMovedAt: new Date("2025-12-31T23:00:00.000Z").toISOString(), + steps: [{ id: "s1", title: "step", status: "pending" } as any], + currentStep: 1, + }); + const store = makeStore(rootDir, task); + const manager = new SelfHealingManager(store as any, { rootDir } as any); + + const recovered = await manager.recoverInProgressLimbo(); + + expect(recovered).toBe(0); + expect(store.moveTask).not.toHaveBeenCalled(); + expect(task.column).toBe("in-progress"); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:auto-recover-in-progress-limbo-no-action", + target: task.id, + metadata: expect.objectContaining({ reason: "checked-out-lease-active" }), + })); + + manager.stop(); + }); + + it("waits for the settle window before re-dispatching an engine-requeued todo task", async () => { + const task = makeTask(rootDir, { + id: "FN-5941-SCHED", + column: "todo", + branch: undefined, + worktree: undefined, + steps: [], + currentStep: 0, + status: "queued", + columnMovedAt: new Date().toISOString(), + }); + const store = makeSchedulerStore(rootDir, task); + const scheduler = new Scheduler(store as any); + + store.emit("task:moved", { task, from: "in-progress", to: "todo", source: "engine" }); + (scheduler as any).running = true; + + await scheduler.schedule(); + expect(store.moveTask).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(5_001); + await scheduler.schedule(); + + expect(store.moveTask).toHaveBeenCalledWith(task.id, "in-progress", expect.objectContaining({ moveSource: "scheduler" })); + }); + + it("auto-pauses a task once dispatch oscillation exceeds the threshold", async () => { + const task = makeTask(rootDir, { + id: "FN-5941-BREAKER", + column: "todo", + branch: undefined, + worktree: undefined, + steps: [], + currentStep: 0, + status: "queued", + dispatchStormCount: 5, + lastDispatchAt: new Date(Date.now() - 1_000).toISOString(), + columnMovedAt: new Date().toISOString(), + error: undefined, + }); + const store = makeSchedulerStore(rootDir, task, { + dispatchOscillationThreshold: 5, + dispatchOscillationWindowMs: 60_000, + }); + const scheduler = new Scheduler(store as any); + (scheduler as any).running = true; + + await scheduler.schedule(); + + expect(store.moveTask).not.toHaveBeenCalled(); + expect(task.paused).toBe(true); + expect(task.pausedReason).toBe("dispatch-oscillation"); + expect(task.dispatchStormCount).toBe(6); + expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:dispatch-oscillation-terminalized", + target: task.id, + metadata: expect.objectContaining({ cycleCount: 6, windowMs: 60_000 }), + })); + }); + + it("keeps dispatching while oscillation count stays under the threshold", async () => { + const task = makeTask(rootDir, { + id: "FN-5941-UNDER", + column: "todo", + branch: undefined, + worktree: undefined, + steps: [], + currentStep: 0, + status: "queued", + dispatchStormCount: 1, + lastDispatchAt: new Date(Date.now() - 1_000).toISOString(), + columnMovedAt: new Date().toISOString(), + }); + const store = makeSchedulerStore(rootDir, task, { + dispatchOscillationThreshold: 5, + dispatchOscillationWindowMs: 60_000, + }); + const scheduler = new Scheduler(store as any); + (scheduler as any).running = true; + + await scheduler.schedule(); + + expect(store.moveTask).toHaveBeenCalledWith(task.id, "in-progress", expect.objectContaining({ moveSource: "scheduler" })); + expect(task.dispatchStormCount).toBe(2); + expect(task.paused).toBe(false); + }); + + it("resets the oscillation counter after the window ages out", async () => { + const task = makeTask(rootDir, { + id: "FN-5941-AGED", + column: "todo", + branch: undefined, + worktree: undefined, + steps: [], + currentStep: 0, + status: "queued", + dispatchStormCount: 4, + lastDispatchAt: new Date(Date.now() - 120_000).toISOString(), + columnMovedAt: new Date().toISOString(), + }); + const store = makeSchedulerStore(rootDir, task, { + dispatchOscillationThreshold: 5, + dispatchOscillationWindowMs: 60_000, + }); + const scheduler = new Scheduler(store as any); + (scheduler as any).running = true; + + await scheduler.schedule(); + + expect(store.moveTask).toHaveBeenCalledWith(task.id, "in-progress", expect.objectContaining({ moveSource: "scheduler" })); + expect(task.dispatchStormCount).toBe(1); + }); + + it("resets dispatch oscillation state on forward transition to in-review", async () => { + const task = makeTask(rootDir, { + id: "FN-5941-RESET-MOVE", + column: "todo", + branch: undefined, + worktree: undefined, + steps: [], + currentStep: 0, + dispatchStormCount: 3, + lastDispatchAt: new Date().toISOString(), + }); + const store = makeSchedulerStore(rootDir, task); + const scheduler = new Scheduler(store as any); + + store.emit("task:moved", { task, from: "todo", to: "in-review", source: "engine" }); + await Promise.resolve(); + + expect(task.dispatchStormCount).toBeNull(); + expect(task.lastDispatchAt).toBeNull(); + expect((scheduler as any).recentEngineTodoRequeues.has(task.id)).toBe(false); + }); + + it("resets dispatch oscillation state on manual unpause", async () => { + const task = makeTask(rootDir, { + id: "FN-5941-RESET-UNPAUSE", + column: "todo", + branch: undefined, + worktree: undefined, + steps: [], + currentStep: 0, + paused: true, + userPaused: false, + dispatchStormCount: 4, + lastDispatchAt: new Date().toISOString(), + }); + const store = makeSchedulerStore(rootDir, task); + new Scheduler(store as any); + + store.emit("task:updated", task); + task.paused = false; + store.emit("task:updated", task); + await Promise.resolve(); + + expect(task.dispatchStormCount).toBeNull(); + expect(task.lastDispatchAt).toBeNull(); + }); + + it("clears the settle-window guard when the task is deleted", async () => { + const task = makeTask(rootDir, { + id: "FN-5941-DELETE", + column: "todo", + branch: undefined, + worktree: undefined, + steps: [], + currentStep: 0, + status: "queued", + columnMovedAt: new Date().toISOString(), + }); + const store = makeSchedulerStore(rootDir, task); + const scheduler = new Scheduler(store as any); + + store.emit("task:moved", { task, from: "in-progress", to: "todo", source: "engine" }); + store.emit("task:deleted", task); + (scheduler as any).running = true; + + await scheduler.schedule(); + + expect(store.moveTask).toHaveBeenCalledWith(task.id, "in-progress", expect.objectContaining({ moveSource: "scheduler" })); + }); + + it("does not delay a user-moved todo task with the settle-window guard", async () => { + const task = makeTask(rootDir, { + id: "FN-5941-USER", + column: "todo", + branch: undefined, + worktree: undefined, + steps: [], + currentStep: 0, + status: "queued", + columnMovedAt: new Date().toISOString(), + }); + const store = makeSchedulerStore(rootDir, task); + const scheduler = new Scheduler(store as any); + + store.emit("task:moved", { task, from: "in-progress", to: "todo", source: "engine" }); + task.columnMovedAt = new Date().toISOString(); + store.emit("task:moved", { task, from: "in-progress", to: "todo", source: "user" }); + (scheduler as any).running = true; + + await scheduler.schedule(); + + expect(store.moveTask).toHaveBeenCalledWith(task.id, "in-progress", expect.objectContaining({ moveSource: "scheduler" })); + }); +}); diff --git a/packages/engine/src/__tests__/scheduler.test.ts b/packages/engine/src/__tests__/scheduler.test.ts index 4b4ef4fff5..b07223c40b 100644 --- a/packages/engine/src/__tests__/scheduler.test.ts +++ b/packages/engine/src/__tests__/scheduler.test.ts @@ -578,13 +578,14 @@ describe("Scheduler", () => { expect(onMoves).toContainEqual(["FN-1", "in-progress"]); }); - it("flag-OFF: the sweep never issues a scheduler-sourced move (legacy path byte-identical)", async () => { + it("flag-OFF: todo dispatch is tagged as scheduler-sourced for redispatch guards", async () => { const off = setupTodoStore(false); await off.scheduler.schedule(); const schedulerSourcedMoves = vi .mocked(off.store.moveTask) .mock.calls.filter((c) => (c[2] as { moveSource?: string } | undefined)?.moveSource === "scheduler"); - expect(schedulerSourcedMoves.length).toBe(0); + expect(schedulerSourcedMoves.length).toBe(1); + expect(schedulerSourcedMoves[0]?.slice(0, 2)).toEqual(["FN-1", "in-progress"]); }); }); @@ -2466,22 +2467,23 @@ describe("Scheduler", () => { (scheduler as any).running = true; await scheduler.schedule(); - expect(updateTask).toHaveBeenNthCalledWith(1, "FN-011", { + const dispatchPrepCalls = updateTask.mock.calls.filter(([, patch]) => Object.prototype.hasOwnProperty.call(patch, "mergeRetries")); + expect(dispatchPrepCalls[0]).toEqual(["FN-011", { status: null, blockedBy: null, executionStartBranch: undefined, effectiveNodeId: null, effectiveNodeSource: "local", mergeRetries: 0, - }); - expect(updateTask).toHaveBeenNthCalledWith(2, "FN-012", { + }]); + expect(dispatchPrepCalls[1]).toEqual(["FN-012", { status: null, blockedBy: null, executionStartBranch: undefined, effectiveNodeId: null, effectiveNodeSource: "local", mergeRetries: 0, - }); + }]); randomSpy.mockRestore(); }); diff --git a/packages/engine/src/__tests__/self-healing-in-progress-limbo.test.ts b/packages/engine/src/__tests__/self-healing-in-progress-limbo.test.ts index 5ef1222233..9f99467c59 100644 --- a/packages/engine/src/__tests__/self-healing-in-progress-limbo.test.ts +++ b/packages/engine/src/__tests__/self-healing-in-progress-limbo.test.ts @@ -98,7 +98,7 @@ describe("recoverInProgressLimbo", () => { rootDir: "/tmp/test-project", getExecutingTaskIds: () => new Set(), leaseManager: { - recoverAbandonedLease: vi.fn().mockResolvedValue(false), + recoverAbandonedLease: vi.fn().mockResolvedValue(true), reconcileLeaseRow, } as any, }); diff --git a/packages/engine/src/__tests__/self-healing.test.ts b/packages/engine/src/__tests__/self-healing.test.ts index 12813ab527..f6f18f0e63 100644 --- a/packages/engine/src/__tests__/self-healing.test.ts +++ b/packages/engine/src/__tests__/self-healing.test.ts @@ -7951,15 +7951,21 @@ describe("SelfHealingManager reclaimSelfOwnedBranchConflicts", () => { expect(store.updateTask).not.toHaveBeenCalled(); }); - it("skips checked out tasks", async () => { + it("skips checked out tasks and emits no-action telemetry", async () => { (store.listTasks as any) .mockResolvedValueOnce([{ id: "FN-501", checkedOutBy: "agent-1", branch: "fusion/fn-501", worktree: "/tmp/fn-501" }]) + .mockResolvedValueOnce([]) .mockResolvedValueOnce([]); const inspectSpy = vi.spyOn(branchConflictModule, "inspectBranchConflict"); const recovered = await manager.reclaimSelfOwnedBranchConflicts(); expect(recovered).toBe(0); expect(inspectSpy).not.toHaveBeenCalled(); + expect((store as any).recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({ + mutationType: "task:reclaim-self-owned-branch-conflict-no-action", + target: "FN-501", + metadata: expect.objectContaining({ reason: "checked-out-lease-active" }), + })); }); it("skips tasks with recent active heartbeat runs", async () => { diff --git a/packages/engine/src/run-audit.ts b/packages/engine/src/run-audit.ts index 0b560535c2..6f52b46221 100644 --- a/packages/engine/src/run-audit.ts +++ b/packages/engine/src/run-audit.ts @@ -471,10 +471,18 @@ export type DatabaseMutationType = | "task:auto-recover-post-done-noncontinuable-wedge-exhausted" | "task:auto-recover-worktree-session-exhausted" | "task:auto-recover-in-progress-limbo" + /** Metadata: { taskId, branch, worktree, checkedOutBy, executionStartedAt, executionAgeMs, graceMs, liveWorktreeBoundBranch, reason } */ + | "task:auto-recover-in-progress-limbo-no-action" | "task:resume-limbo-escalated" + /** Metadata: { taskId, branch, worktree, checkedOutBy, executionStartedAt, executionAgeMs, graceMs, liveWorktreeBoundBranch, reason } */ + | "task:reclaim-self-owned-branch-conflict-no-action" | "task:orphan-detected-no-action" + /** Metadata: { taskId, lastReason, stuckKillCount, attemptedStuckKillCount, maxStuckKills, checkedOutBy, executionStartedAt, executionAgeMs, graceMs, liveWorktreeBoundBranch } */ + | "task:stuck-loop-exhausted-no-action" /** Metadata: { taskId: string; ignoredStepUpdateCount: number; stuckKillStreak: number; lastReason: "no-progress-churn" } */ | "task:stuck-no-progress-churn-terminalized" + /** Metadata: { taskId, cycleCount, windowMs, lastMoveSource } */ + | "task:dispatch-oscillation-terminalized" | "task:auto-recover-starved-refinement" /** Metadata: { rawDiffFileCount: number; attributedFileCount: number; foreignCommitCount: number; foreignCommitShas: string[]; source: string } */ | "task:worktree-contamination-detected" diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index 5a014383eb..c099ccfcdb 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -142,6 +142,10 @@ const COORDINATION_SAFE_SCOPE_PREFIXES = [ ".changeset/", ]; +const DEFAULT_DISPATCH_OSCILLATION_SETTLE_MS = 5_000; +const DEFAULT_DISPATCH_OSCILLATION_THRESHOLD = 5; +const DEFAULT_DISPATCH_OSCILLATION_WINDOW_MS = 60_000; + function isCoordinationSafeScopeEntry(entry: string): boolean { const normalized = normalizeOverlapPath(entry).toLowerCase(); if (!normalized) return false; @@ -483,6 +487,8 @@ export class Scheduler { private dispatchQueuedConcurrencyAuditMemo = new Map(); /** Tracks per-task candidacy fingerprints for task:updated auto-claim invalidation gating. */ private lastAutoClaimFingerprint = new Map(); + /** Tracks recent engine-sourced in-progress → todo requeues to prevent immediate re-dispatch races. */ + private recentEngineTodoRequeues = new Map(); private readonly staleTaskReporter: StaleTaskReporter; private readonly backlogPressureReporter: BacklogPressureReporter; private readonly unlinkedMissionsAdvisoryReporter: UnlinkedMissionsAdvisoryReporter; @@ -559,7 +565,7 @@ export class Scheduler { * Also handles mission auto-advance: when a linked task completes, * update feature status and potentially activate next pending slice. */ - this.store.on("task:moved", async ({ task, from, to }) => { + this.store.on("task:moved", async ({ task, from, to, source }) => { this.lastAutoClaimFingerprint.set(task.id, computeAutoClaimFingerprint(task)); if (from === "todo" || to === "todo") { this.options.snapshotManager?.invalidate(`task:moved:${from}->${to}`); @@ -670,6 +676,24 @@ export class Scheduler { } } + if (from === "in-progress" && to === "todo") { + if (source === "engine") { + this.recentEngineTodoRequeues.set(task.id, task.columnMovedAt ?? new Date().toISOString()); + } else { + this.recentEngineTodoRequeues.delete(task.id); + } + } else if (to === "in-review" || to === "done" || to === "archived") { + this.recentEngineTodoRequeues.delete(task.id); + if (task.dispatchStormCount != null || task.lastDispatchAt != null) { + void this.store.updateTask(task.id, { + dispatchStormCount: null, + lastDispatchAt: null, + }).catch((error) => { + schedulerLog.warn(`Failed to reset dispatch oscillation state for ${task.id} on move to ${to}: ${error instanceof Error ? error.message : String(error)}`); + }); + } + } + // Event-driven scheduling: when a task moves to "done" (completion) or "todo" (retry/manual move), // trigger scheduling immediately so waiting tasks can start without waiting // for the next poll interval (up to 15 seconds). @@ -706,6 +730,14 @@ export class Scheduler { } else if (this.pausedTaskIds.has(task.id)) { // Task was paused, now unpaused — trigger scheduling this.pausedTaskIds.delete(task.id); + if (task.userPaused === false && (task.dispatchStormCount != null || task.lastDispatchAt != null)) { + void this.store.updateTask(task.id, { + dispatchStormCount: null, + lastDispatchAt: null, + }).catch((error) => { + schedulerLog.warn(`Failed to reset dispatch oscillation state for ${task.id} on unpause: ${error instanceof Error ? error.message : String(error)}`); + }); + } if (this.running && (task.column === "todo" || task.column === "triage")) { schedulerLog.log(`Task ${task.id} unpaused — triggering scheduling`); this.schedule(); @@ -734,6 +766,7 @@ export class Scheduler { this.options.snapshotManager?.invalidate("task:deleted"); this.pausedTaskIds.delete(task.id); this.failedTaskIds.delete(task.id); + this.recentEngineTodoRequeues.delete(task.id); this.wasNodeDispatchValidationBlocked.delete(task.id); this.wasNodeBlocked.delete(task.id); this.wasPermanentAgentUnavailable.delete(task.id); @@ -1687,6 +1720,31 @@ export class Scheduler { } const latestSettings = await this.store.getSettings(); + const oscillationSettings = latestSettings as Settings & { + dispatchOscillationSettleMs?: number; + dispatchOscillationThreshold?: number; + dispatchOscillationWindowMs?: number; + }; + const dispatchSettleMs = oscillationSettings.dispatchOscillationSettleMs + ?? DEFAULT_DISPATCH_OSCILLATION_SETTLE_MS; + const dispatchOscillationThreshold = oscillationSettings.dispatchOscillationThreshold + ?? DEFAULT_DISPATCH_OSCILLATION_THRESHOLD; + const dispatchOscillationWindowMs = oscillationSettings.dispatchOscillationWindowMs + ?? DEFAULT_DISPATCH_OSCILLATION_WINDOW_MS; + const recentEngineTodoMovedAt = this.recentEngineTodoRequeues.get(task.id); + if (recentEngineTodoMovedAt) { + if (freshTask.columnMovedAt !== recentEngineTodoMovedAt) { + this.recentEngineTodoRequeues.delete(task.id); + } else { + const movedAtMs = Date.parse(recentEngineTodoMovedAt); + const settleAgeMs = Number.isFinite(movedAtMs) ? Math.max(0, Date.now() - movedAtMs) : dispatchSettleMs; + if (settleAgeMs < dispatchSettleMs) { + schedulerLog.log(`Task ${task.id} was engine-requeued ${settleAgeMs}ms ago — waiting ${dispatchSettleMs}ms settle window before redispatch`); + continue; + } + this.recentEngineTodoRequeues.delete(task.id); + } + } if (latestSettings.globalPause) { schedulerLog.log(`Task ${task.id} dispatch aborted — globalPause became active mid-pass`); continue; @@ -1892,6 +1950,53 @@ export class Scheduler { // with mergeRetries=MAX, the merger refuses it (canMergeTask false), // and the ghost-review fallback bounces it back to todo every 10 min // before the 30-min cooldown can elapse — infinite loop. See FN-3305. + const dispatchTimestamp = new Date().toISOString(); + const lastDispatchAtMs = freshTask.lastDispatchAt ? Date.parse(freshTask.lastDispatchAt) : Number.NaN; + const priorDispatchWithinWindow = Number.isFinite(lastDispatchAtMs) + && Date.now() - lastDispatchAtMs <= dispatchOscillationWindowMs; + const nextDispatchStormCount = priorDispatchWithinWindow + ? (freshTask.dispatchStormCount ?? 0) + 1 + : 1; + + if (nextDispatchStormCount > dispatchOscillationThreshold) { + const oscillationError = freshTask.error + ?? `DISPATCH_OSCILLATION: detected ${nextDispatchStormCount} todo↔in-progress cycles within ${dispatchOscillationWindowMs}ms. Task auto-paused for operator review.`; + await this.store.updateTask(task.id, { + dispatchStormCount: nextDispatchStormCount, + lastDispatchAt: dispatchTimestamp, + paused: true, + pausedReason: "dispatch-oscillation", + status: freshTask.status ?? "queued", + error: oscillationError, + }); + await this.store.logEntry( + task.id, + `Dispatch oscillation auto-paused after ${nextDispatchStormCount} cycles within ${dispatchOscillationWindowMs}ms`, + ); + await this.store.appendAgentLog?.( + task.id, + "Dispatch oscillation detected — task auto-paused for operator review", + "text", + `cycleCount=${nextDispatchStormCount} windowMs=${dispatchOscillationWindowMs}`, + ); + await this.store.recordRunAuditEvent?.({ + taskId: task.id, + agentId: "scheduler", + runId: generateSyntheticRunId("scheduler-dispatch-oscillation", task.id), + domain: "database", + mutationType: "task:dispatch-oscillation-terminalized", + target: task.id, + metadata: { + taskId: task.id, + cycleCount: nextDispatchStormCount, + windowMs: dispatchOscillationWindowMs, + lastMoveSource: recentEngineTodoMovedAt ? "engine" : "scheduler", + }, + }); + schedulerLog.warn(`Task ${task.id} auto-paused after dispatch oscillation threshold ${dispatchOscillationThreshold} was exceeded (${nextDispatchStormCount} cycles)`); + continue; + } + schedulerLog.log(`Starting ${task.id}: ${task.title || task.id} (deps satisfied)`); await this.store.updateTask(task.id, { status: null, @@ -1902,9 +2007,15 @@ export class Scheduler { mergeRetries: 0, }); await this.store.moveTask(task.id, "in-progress", { + moveSource: "scheduler", allocateWorktree: (reservedNames) => this.planWorktreePath(task, settings.worktreeNaming, reservedNames, settings), }); + await this.store.updateTask(task.id, { + dispatchStormCount: nextDispatchStormCount, + lastDispatchAt: dispatchTimestamp, + }); + this.recentEngineTodoRequeues.delete(task.id); this.wasNodeBlocked.delete(task.id); this.wasNodeDispatchValidationBlocked.delete(task.id); this.wasPermanentAgentUnavailable.delete(task.id); diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index f506332a9a..709f631074 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -795,6 +795,100 @@ export class SelfHealingManager { log.log(`[${stage}] ${task.id}: triple-proof not satisfied — no action (operator-decides)`); } + private async listActiveHeartbeatTaskIds(): Promise> { + const activeTaskIds = new Set(); + if (!this.options.agentStore) { + return activeTaskIds; + } + + try { + const activeRuns = await this.options.agentStore.listActiveHeartbeatRuns(); + const activeWindowMs = RUNNING_ON_INACTIVE_TASK_STALE_RUN_MS; + const now = Date.now(); + for (const run of activeRuns) { + const startedAtMs = Date.parse(run.startedAt ?? ""); + if (!Number.isFinite(startedAtMs) || now - startedAtMs > activeWindowMs) continue; + const taskId = run.contextSnapshot && typeof run.contextSnapshot.taskId === "string" + ? run.contextSnapshot.taskId.toUpperCase() + : null; + if (taskId) activeTaskIds.add(taskId); + } + } catch (err: unknown) { + const message = err instanceof Error ? err.message : String(err); + log.warn(`Unable to enumerate active heartbeat runs: ${message}`); + } + + return activeTaskIds; + } + + private getFalsePositiveRequeueSignal(task: Task, options: { + executingIds?: Set; + activeHeartbeatTaskIds?: Set; + graceMs: number; + includeLiveWorktreeBoundBranch?: boolean; + includeCheckedOutLease?: boolean; + }): { reason: string; metadata: Record } | null { + const normalizedId = task.id.toUpperCase(); + const executionStartedAtMs = task.executionStartedAt ? Date.parse(task.executionStartedAt) : Number.NaN; + const executionAgeMs = Number.isFinite(executionStartedAtMs) ? Math.max(0, Date.now() - executionStartedAtMs) : null; + const liveWorktreeBoundBranch = Boolean( + task.worktree + && typeof task.branch === "string" + && task.branch.trim().length > 0 + && existsSync(task.worktree), + ); + const metadata = { + taskId: task.id, + branch: task.branch ?? null, + worktree: task.worktree ?? null, + checkedOutBy: task.checkedOutBy ?? null, + executionStartedAt: task.executionStartedAt ?? null, + executionAgeMs, + graceMs: options.graceMs, + liveWorktreeBoundBranch, + }; + + if (options.executingIds?.has(task.id)) { + return { reason: "executor-active", metadata }; + } + if (options.activeHeartbeatTaskIds?.has(normalizedId)) { + return { reason: "active-heartbeat-run", metadata }; + } + if ((options.includeCheckedOutLease ?? false) && task.checkedOutBy) { + return { reason: "checked-out-lease-active", metadata }; + } + if ((options.includeLiveWorktreeBoundBranch ?? true) && liveWorktreeBoundBranch) { + return { reason: "live-worktree-and-branch", metadata }; + } + if (executionAgeMs !== null && executionAgeMs <= options.graceMs) { + return { reason: "recent-execution-started", metadata }; + } + return null; + } + + private async emitFalsePositiveRequeueNoAction(task: Task, stage: string, mutationType: string, reason: string, metadata: Record): Promise { + try { + await createRunAuditor(this.store, { + runId: generateSyntheticRunId(`self-healing-${stage}`, task.id), + agentId: "self-healing", + taskId: task.id, + taskLineageId: task.lineageId, + phase: stage, + }).database({ + type: mutationType as DatabaseMutationType, + target: task.id, + metadata: { + ...metadata, + reason, + }, + }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + log.warn(`[${stage}] ${task.id}: false-positive no-action audit emission failed: ${message}`); + } + log.log(`[${stage}] ${task.id}: false-positive requeue suppressed (${reason})`); + } + // ── Lifecycle ─────────────────────────────────────────────────────── start(): void { @@ -1122,11 +1216,36 @@ export class SelfHealingManager { } const newCount = (task.stuckKillCount ?? 0) + 1; + const executingIds = this.options.getExecutingTaskIds?.() ?? new Set(); + const activeHeartbeatTaskIds = await this.listActiveHeartbeatTaskIds(); if (newCount > maxKills) { const hasIncompleteSteps = !!task.steps?.some((step) => NON_TERMINAL_STEP_STATUSES.has(step.status)); if (hasIncompleteSteps) { + const liveExecutionSignal = this.getFalsePositiveRequeueSignal(task, { + executingIds, + activeHeartbeatTaskIds, + graceMs: settings.taskStuckTimeoutMs ?? ORPHANED_EXECUTION_RECOVERY_GRACE_MS, + includeCheckedOutLease: true, + }); + if (liveExecutionSignal) { + await this.emitFalsePositiveRequeueNoAction( + task, + "stuck-loop-exhausted", + "task:stuck-loop-exhausted-no-action", + liveExecutionSignal.reason, + { + ...liveExecutionSignal.metadata, + lastReason: reason, + stuckKillCount: task.stuckKillCount ?? 0, + attemptedStuckKillCount: newCount, + maxStuckKills: maxKills, + }, + ); + return false; + } + log.warn(`${taskId} exceeded stuck kill budget (${newCount}/${maxKills}, reason=${reason}) with incomplete steps — re-queueing in todo with progress preserved`); await this.store.updateTask(taskId, { stuckKillCount: newCount }); try { @@ -2389,32 +2508,31 @@ export class SelfHealingManager { // explicit autoMerge:true tasks recover. const candidates = [...todoCandidates, ...inProgressCandidates, ...inReviewPausedCandidates] .filter((task) => allowsAutoMergeProcessing(task, settings)); - - const activeTaskIds = new Set(); - if (this.options.agentStore) { - try { - const activeRuns = await this.options.agentStore.listActiveHeartbeatRuns(); - const activeWindowMs = RUNNING_ON_INACTIVE_TASK_STALE_RUN_MS; - const now = Date.now(); - for (const run of activeRuns) { - const startedAtMs = Date.parse(run.startedAt ?? ""); - if (!Number.isFinite(startedAtMs) || now - startedAtMs > activeWindowMs) continue; - const taskId = run.contextSnapshot && typeof run.contextSnapshot.taskId === "string" - ? run.contextSnapshot.taskId.toUpperCase() - : null; - if (taskId) activeTaskIds.add(taskId); - } - } catch (err: unknown) { - const message = err instanceof Error ? err.message : String(err); - log.warn(`Unable to enumerate active heartbeat runs for self-owned branch reclaim sweep: ${message}`); - } - } + const executingIds = this.options.getExecutingTaskIds?.() ?? new Set(); + const activeTaskIds = await this.listActiveHeartbeatTaskIds(); let recovered = 0; const integrationBranch = await resolveIntegrationBranch(this.options.rootDir, settings); for (const task of candidates) { - if (task.checkedOutBy || activeTaskIds.has(task.id.toUpperCase()) || !task.branch || !task.worktree) continue; + if (!task.branch || !task.worktree) continue; if (task.userPaused) continue; + const liveExecutionSignal = this.getFalsePositiveRequeueSignal(task, { + executingIds, + activeHeartbeatTaskIds: activeTaskIds, + graceMs: STALE_ACTIVE_BRANCH_EXECUTION_GRACE_MS, + includeLiveWorktreeBoundBranch: false, + includeCheckedOutLease: true, + }); + if (liveExecutionSignal) { + await this.emitFalsePositiveRequeueNoAction( + task, + "reclaim-self-owned-branch-conflict", + "task:reclaim-self-owned-branch-conflict-no-action", + liveExecutionSignal.reason, + liveExecutionSignal.metadata, + ); + continue; + } if (task.column === "todo" && task.blockedBy) { log.log(`[self-healing] skipping blocked todo task ${task.id} during self-owned branch reclaim (blockedBy=${task.blockedBy})`); continue; @@ -7260,10 +7378,11 @@ export class SelfHealingManager { try { const tasks = await this.store.listTasks({ column: "in-progress", slim: true }); const executingIds = this.options.getExecutingTaskIds?.() ?? new Set(); + const activeHeartbeatTaskIds = await this.listActiveHeartbeatTaskIds(); const now = Date.now(); const stranded = tasks.filter((task) => { - if (task.column !== "in-progress" || task.paused || executingIds.has(task.id)) { + if (task.column !== "in-progress" || task.paused) { return false; } const hasMissingWorktreePath = typeof task.worktree === "string" && task.worktree.length > 0 && !existsSync(task.worktree); @@ -7290,13 +7409,68 @@ export class SelfHealingManager { let recovered = 0; for (const task of stranded) { try { - if (this.options.leaseManager && task.checkedOutBy) { - await this.options.leaseManager.recoverAbandonedLease( - task.id, - `in-progress limbo: ${describeWorktreeState(task)} + null branch`, - { preserveProgress: true }, + const liveExecutionSignal = this.getFalsePositiveRequeueSignal(task, { + executingIds, + activeHeartbeatTaskIds, + graceMs: ORPHANED_EXECUTION_RECOVERY_GRACE_MS, + }); + if (liveExecutionSignal) { + await this.emitFalsePositiveRequeueNoAction( + task, + "auto-recover-in-progress-limbo", + "task:auto-recover-in-progress-limbo-no-action", + liveExecutionSignal.reason, + liveExecutionSignal.metadata, ); - await this.options.leaseManager.reconcileLeaseRow(task.id); + continue; + } + + if (task.checkedOutBy) { + if (this.options.leaseManager) { + const leaseRecovered = await this.options.leaseManager.recoverAbandonedLease( + task.id, + `in-progress limbo: ${describeWorktreeState(task)} + null branch`, + { preserveProgress: true }, + ); + if (!leaseRecovered) { + await this.emitFalsePositiveRequeueNoAction( + task, + "auto-recover-in-progress-limbo", + "task:auto-recover-in-progress-limbo-no-action", + "checked-out-lease-active", + { + taskId: task.id, + branch: task.branch ?? null, + worktree: task.worktree ?? null, + checkedOutBy: task.checkedOutBy, + executionStartedAt: task.executionStartedAt ?? null, + executionAgeMs: task.executionStartedAt ? Math.max(0, Date.now() - Date.parse(task.executionStartedAt)) : null, + graceMs: ORPHANED_EXECUTION_RECOVERY_GRACE_MS, + liveWorktreeBoundBranch: false, + }, + ); + continue; + } + await this.options.leaseManager.reconcileLeaseRow(task.id); + } else { + await this.emitFalsePositiveRequeueNoAction( + task, + "auto-recover-in-progress-limbo", + "task:auto-recover-in-progress-limbo-no-action", + "checked-out-lease-active", + { + taskId: task.id, + branch: task.branch ?? null, + worktree: task.worktree ?? null, + checkedOutBy: task.checkedOutBy, + executionStartedAt: task.executionStartedAt ?? null, + executionAgeMs: task.executionStartedAt ? Math.max(0, Date.now() - Date.parse(task.executionStartedAt)) : null, + graceMs: ORPHANED_EXECUTION_RECOVERY_GRACE_MS, + liveWorktreeBoundBranch: false, + }, + ); + continue; + } } const stepStatuses = task.steps.map((step) => step.status);