diff --git a/packages/core/src/__tests__/migration-workflow-columns.test.ts b/packages/core/src/__tests__/migration-workflow-columns.test.ts index 6cedf60893..c92259fcf7 100644 --- a/packages/core/src/__tests__/migration-workflow-columns.test.ts +++ b/packages/core/src/__tests__/migration-workflow-columns.test.ts @@ -219,24 +219,18 @@ describe("U12 rollback safety — flag OFF after flag ON keeps legacy behavior", await store.selectTaskWorkflowAndReconcile(task.id, wf.id); expect((await store.getTask(task.id)).column).toBe("intake"); - // Toggle the flag OFF — the card stays in the custom "intake" column. + // Toggle the flag OFF — #1409: the ON→OFF evacuation re-homes the card from + // the custom "intake" column to the nearest legacy column (the default + // workflow's entry column, triage) so it is not stranded on the legacy path. await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: false } }); - expect((await store.getTask(task.id)).column).toBe("intake"); + expect((await store.getTask(task.id)).column).toBe("triage"); - // listTasks must not throw with a task sitting in an unknown column. + // listTasks stays healthy. await expect(store.listTasks()).resolves.toBeDefined(); - // A move attempt degrades to the legacy "Invalid transition" error rather - // than a TypeError on the undefined VALID_TRANSITIONS lookup. - let caught: unknown; - try { - await store.moveTask(task.id, "in-progress", { moveSource: "user" }); - } catch (e) { - caught = e; - } - expect(caught).toBeInstanceOf(Error); - expect((caught as Error).message).toMatch(/Invalid transition/); - expect((caught as Error)).not.toBeInstanceOf(TypeError); + // The evacuated card now moves legacy-style: triage → todo works. + await store.moveTask(task.id, "todo", { moveSource: "user" }); + expect((await store.getTask(task.id)).column).toBe("todo"); }); }); diff --git a/packages/core/src/__tests__/transition-pending-recovery.test.ts b/packages/core/src/__tests__/transition-pending-recovery.test.ts new file mode 100644 index 0000000000..064393f800 --- /dev/null +++ b/packages/core/src/__tests__/transition-pending-recovery.test.ts @@ -0,0 +1,195 @@ +// @vitest-environment node +// +// #1401 + #1409: store-level recovery / evacuation passes for the workflow +// columns feature. +// +// #1401 — transitionPending recovery sweep: +// * a crash-simulated stale marker is recovered (cleared) by the sweep, +// * the phantom capacity slot the marker reserved is released so a fresh +// card can re-enter a full (capacity=1) column afterwards, +// * the sweep is idempotent (a second run finds nothing). +// +// #1409 — flag ON→OFF evacuation: +// * toggling workflowColumns OFF with a card in a custom column re-homes it +// to a legacy column, the board stays listable, and legacy moves work. +// * a flag-OFF store init evacuates a card left in a custom column. + +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import { createTaskStoreTestHarness } from "./store-test-helpers.js"; +import type { WorkflowIr } from "../workflow-ir-types.js"; +import { makeTransitionPending, serializeTransitionPending } from "../transition-types.js"; + +/** A custom workflow whose middle column carries a WIP capacity limit of 1. */ +function cappedIr(): WorkflowIr { + return { + version: "v2", + name: "capped", + columns: [ + { id: "intake", name: "intake", traits: [{ trait: "intake" }] }, + { + id: "build", + name: "build", + traits: [{ trait: "wip", config: { limit: 1, countPending: true } }], + }, + { id: "ship", name: "ship", traits: [] }, + ], + nodes: [ + { id: "start", kind: "start", column: "intake" }, + { id: "work", kind: "prompt", column: "build", config: { prompt: "do" } }, + { id: "end", kind: "end", column: "ship" }, + ], + edges: [ + { from: "start", to: "work", condition: "success" }, + { from: "work", to: "end", condition: "success" }, + ], + }; +} + +function simpleCustomIr(): WorkflowIr { + return { + version: "v2", + name: "simple-custom", + columns: [ + { id: "intake", name: "intake", traits: [{ trait: "intake" }] }, + { id: "build", name: "build", traits: [] }, + { id: "ship", name: "ship", traits: [] }, + ], + nodes: [ + { id: "start", kind: "start", column: "intake" }, + { id: "work", kind: "prompt", column: "build", config: { prompt: "do" } }, + { id: "end", kind: "end", column: "ship" }, + ], + edges: [ + { from: "start", to: "work", condition: "success" }, + { from: "work", to: "end", condition: "success" }, + ], + }; +} + +describe("#1401 transitionPending recovery sweep", () => { + const harness = createTaskStoreTestHarness(); + let store: ReturnType; + + beforeEach(async () => { + await harness.beforeEach(); + store = harness.store(); + await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: true } }); + }); + afterEach(async () => { + await harness.afterEach(); + }); + + function rawDb(): { + prepare: (s: string) => { run: (...a: unknown[]) => unknown; get: (...a: unknown[]) => unknown }; + } { + return (store as unknown as { db: ReturnType }).db; + } + + function readMarkerColumn(taskId: string): string | null { + const row = rawDb() + .prepare(`SELECT transitionPending FROM tasks WHERE id = ?`) + .get(taskId) as { transitionPending: string | null } | undefined; + return row?.transitionPending ?? null; + } + + it("recovers a crash-simulated stale marker and is idempotent", async () => { + const t = await store.createTask({ description: "stale-marker" }); + // Simulate a crash that left a transitionPending marker set forever. + const marker = serializeTransitionPending( + makeTransitionPending("build", ["default-workflow:postCommit"], Date.now() - 60_000), + ); + rawDb().prepare(`UPDATE tasks SET transitionPending = ? WHERE id = ?`).run(marker, t.id); + expect(readMarkerColumn(t.id)).not.toBeNull(); + + const first = await store.recoverStaleTransitionPending(); + expect(first.scanned).toBeGreaterThanOrEqual(1); + expect(first.recovered).toBe(1); + // Marker cleared → capacity slot released. + expect(readMarkerColumn(t.id)).toBeNull(); + + // Idempotent: nothing left to recover. + const second = await store.recoverStaleTransitionPending(); + expect(second.recovered).toBe(0); + }); + + it("releases the phantom capacity slot a stale marker reserved (count returns to normal)", async () => { + const wf = await store.createWorkflowDefinition({ name: "capped", ir: cappedIr() }); + + // A "ghost" task crashed mid-transition into the capacity-1 "build" column: + // its marker reserves the only slot even though it never committed there. + const ghost = await store.createTask({ description: "ghost" }); + await store.selectTaskWorkflowAndReconcile(ghost.id, wf.id); + const ghostMarker = serializeTransitionPending( + makeTransitionPending("build", ["default-workflow:postCommit"], Date.now() - 60_000), + ); + rawDb().prepare(`UPDATE tasks SET transitionPending = ? WHERE id = ?`).run(ghostMarker, ghost.id); + + // A fresh card in the same workflow cannot enter "build": the phantom marker + // is counted as occupying the single capacity slot. + const fresh = await store.createTask({ description: "fresh" }); + await store.selectTaskWorkflowAndReconcile(fresh.id, wf.id); + expect((await store.getTask(fresh.id)).column).toBe("intake"); + + let blocked: unknown; + try { + await store.moveTask(fresh.id, "build", { moveSource: "user" }); + } catch (e) { + blocked = e; + } + expect(blocked).toBeInstanceOf(Error); + expect((await store.getTask(fresh.id)).column).toBe("intake"); + + // Recovery clears the stale marker, releasing the slot. + const result = await store.recoverStaleTransitionPending(); + expect(result.recovered).toBeGreaterThanOrEqual(1); + + // Now the fresh card can enter the capacity column. + await store.moveTask(fresh.id, "build", { moveSource: "user" }); + expect((await store.getTask(fresh.id)).column).toBe("build"); + }); +}); + +describe("#1409 flag ON→OFF evacuation", () => { + const harness = createTaskStoreTestHarness(); + let store: ReturnType; + + beforeEach(async () => { + await harness.beforeEach(); + store = harness.store(); + }); + afterEach(async () => { + await harness.afterEach(); + }); + + it("toggling OFF re-homes a card from a custom column to a legacy column; moves work", async () => { + await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: true } }); + const wf = await store.createWorkflowDefinition({ name: "simple-custom", ir: simpleCustomIr() }); + const task = await store.createTask({ description: "evac" }); + await store.selectTaskWorkflowAndReconcile(task.id, wf.id); + expect((await store.getTask(task.id)).column).toBe("intake"); + + // Toggle OFF — evacuation re-homes the card to the nearest legacy column + // (the default workflow's entry column, triage). + await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: false } }); + expect((await store.getTask(task.id)).column).toBe("triage"); + + // Board listable; legacy moves work from the evacuated column. + await expect(store.listTasks()).resolves.toBeDefined(); + await store.moveTask(task.id, "todo", { moveSource: "user" }); + await store.moveTask(task.id, "in-progress", { moveSource: "user" }); + expect((await store.getTask(task.id)).column).toBe("in-progress"); + }); + + it("evacuateCustomColumnsToLegacy is idempotent (a second run is a no-op)", async () => { + await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: true } }); + const wf = await store.createWorkflowDefinition({ name: "simple-custom-2", ir: simpleCustomIr() }); + const task = await store.createTask({ description: "evac2" }); + await store.selectTaskWorkflowAndReconcile(task.id, wf.id); + await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: false } }); + + // First explicit run already evacuated (via the toggle); a fresh run is a no-op. + const again = await store.evacuateCustomColumnsToLegacy("flag-off-init"); + expect(again.evacuated).toBe(0); + expect((await store.getTask(task.id)).column).toBe("triage"); + }); +}); diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index d519411fe6..da180873cc 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -5,7 +5,7 @@ import { join } from "node:path"; import { existsSync, watch, type FSWatcher } from "node:fs"; import type { Task, TaskDetail, TaskCreateInput, TaskAttachment, AgentLogEntry, BoardConfig, Column, ColumnId, CheckoutClaimPrecondition, MergeResult, Settings, GlobalSettings, ProjectSettings, ActivityLogEntry, ActivityEventType, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, InboxTask, TaskLogEntry, RunMutationContext, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, ArchivedTaskEntry, ArchiveAgentLogMode, TaskPriority, SourceType, WorkflowStepTemplate, Agent, AutostashOrphanRecord, TaskCommitAssociation, TaskCommitAssociationMatchSource, TaskCommitAssociationConfidence, GithubIssueAction, MergeQueueEntry, MergeQueueEnqueueOptions, MergeQueueAcquireOptions, MergeQueueReleaseOutcome, HandoffToReviewOptions, GoalCitation, GoalCitationFilter, GoalCitationInput, GoalCitationSurface, BranchGroup, BranchGroupCreateInput, BranchGroupUpdate, TaskBranchAssignmentMode, MergeRequestRecord, MergeRequestState, CompletionHandoffMarker } from "./types.js"; import { createActivityLogSnapshot, createRunAuditSnapshot, createTaskMetadataSnapshot, toTaskMetadataRecord, validateSnapshotEnvelope, type ActivityLogSnapshot, type RunAuditSnapshot, type TaskMetadataSnapshot } from "./shared-mesh-state.js"; -import { VALID_TRANSITIONS, DEFAULT_SETTINGS, isGlobalOnlySettingsKey, WORKFLOW_STEP_TEMPLATES, validateDocumentKey } from "./types.js"; +import { VALID_TRANSITIONS, COLUMNS, DEFAULT_SETTINGS, isGlobalOnlySettingsKey, WORKFLOW_STEP_TEMPLATES, validateDocumentKey } from "./types.js"; import { DEFAULT_PROJECT_SETTINGS } from "./settings-schema.js"; import { parseWorkflowIr, serializeWorkflowIr } from "./workflow-ir.js"; import { isWorkflowColumnsEnabled } from "./workflow-columns-settings.js"; @@ -36,7 +36,12 @@ import { makeTransitionRejection, makeTransitionPending, } from "./transition-types.js"; -import { writeTransitionPending, clearTransitionPending } from "./transition-pending.js"; +import { + writeTransitionPending, + clearTransitionPending, + readTransitionPending, + reconcileHooksRemaining, +} from "./transition-pending.js"; import { BUILTIN_CODING_WORKFLOW_IR } from "./builtin-coding-workflow-ir.js"; import type { WorkflowIr, WorkflowIrColumn } from "./workflow-ir-types.js"; // Side-effect import: registers the 14 built-in trait DEFINITIONS into the @@ -1604,6 +1609,15 @@ export class TaskStore extends EventEmitter { const settings = await this.getSettingsFast(); if (isWorkflowColumnsEnabled(settings)) { await this.runWorkflowColumnsIntegrityPass(); + // #1401: recover any transitionPending markers stranded by a crash + // between the in-txn write and the post-commit clear (they otherwise + // permanently inflate capacity counts for their target column). + await this.recoverStaleTransitionPending(); + } else { + // #1409: flag-OFF init — evacuate any card stuck in a non-legacy column + // (e.g. the flag was toggled OFF out-of-process while a card sat in a + // custom column) so the board stays listable and moves work. + await this.evacuateCustomColumnsToLegacy("flag-off-init"); } } catch (err) { storeLog.warn("workflowColumns integrity pass failed during init", { @@ -3369,6 +3383,20 @@ export class TaskStore extends EventEmitter { const updatedMerged: Settings = { ...DEFAULT_SETTINGS, ...globalSettings, ...updatedProjectSettings } as Settings; this.emit("settings:updated", { settings: updatedMerged, previous: previousMerged }); + // #1409: if this update flipped workflowColumns ON→OFF, evacuate any card + // stranded in a custom (non-legacy) column back to a legacy column so the + // board stays listable / movable on the legacy path. + if (isWorkflowColumnsEnabled(previousMerged) && !isWorkflowColumnsEnabled(updatedMerged)) { + try { + await this.evacuateCustomColumnsToLegacy("flag-toggled-off"); + } catch (err) { + storeLog.warn("workflowColumns ON→OFF evacuation failed", { + phase: "evacuate-custom-columns", + error: err instanceof Error ? err.message : String(err), + }); + } + } + // Bootstrap project memory file when memory is toggled on if (updatedMerged.memoryEnabled !== false && previousMerged.memoryEnabled === false) { try { @@ -3469,6 +3497,21 @@ export class TaskStore extends EventEmitter { // Emit settings:updated so SSE listeners pick up the change this.emit("settings:updated", { settings: merged, previous }); + + // #1409: workflowColumns lives in experimentalFeatures (a global key), so the + // ON→OFF toggle flows through here. Evacuate any card stranded in a custom + // column when the flag flips off. + if (isWorkflowColumnsEnabled(previous) && !isWorkflowColumnsEnabled(merged)) { + try { + await this.evacuateCustomColumnsToLegacy("flag-toggled-off"); + } catch (err) { + storeLog.warn("workflowColumns ON→OFF evacuation failed", { + phase: "evacuate-custom-columns", + error: err instanceof Error ? err.message : String(err), + }); + } + } + return merged; } @@ -5839,7 +5882,18 @@ export class TaskStore extends EventEmitter { if (useWorkflow && workflowIr) { // ── Flag-ON validation + sync guards (typed rejections, KTD-3/R13) ───── // 1. Target column must exist in the task's workflow → unknown-column. - if (!workflowHasColumn(workflowIr, toColumn)) { + // #1411: a recoveryRehome move to a LEGACY column (todo/archived/…) is + // the engine's self-healing rescue path — those targets are guaranteed + // safe landing columns even when a custom workflow never defined them. + // recoveryRehome already skips adjacency (below); it must likewise skip + // the unknown-column rejection for legacy recovery targets, otherwise a + // custom-workflow card could never be rescued to todo/archived and would + // stay stuck — the exact bug #1411 describes. Non-legacy unknown targets + // still reject (a genuine programming error), and normal (non-recovery) + // moves are unaffected. + const recoveryToLegacy = + options?.recoveryRehome === true && (COLUMNS as readonly string[]).includes(toColumn); + if (!workflowHasColumn(workflowIr, toColumn) && !recoveryToLegacy) { throw new TransitionRejectionError( makeTransitionRejection( "unknown-column", @@ -5932,12 +5986,26 @@ export class TaskStore extends EventEmitter { // A task can sit in a custom column when the flag was toggled ON→OFF; // `VALID_TRANSITIONS` only keys the legacy columns, so a missing entry // degrades to the legacy "Invalid transition" error instead of a TypeError. - const validTargets = VALID_TRANSITIONS[task.column as Column] ?? []; - if (!validTargets.includes(toColumn)) { - throw new Error( - `Invalid transition: '${task.column}' → '${toColumn}'. ` + - `Valid targets: ${validTargets.join(", ") || "none"}`, - ); + // #1409: flag-OFF evacuation. A recoveryRehome move OUT of a non-legacy + // (custom) column into a legacy target is the ON→OFF evacuation path — + // `VALID_TRANSITIONS` never keys a custom source column, so the legacy + // check below would strand the card forever. Allow it through (bypassing + // only the adjacency check; this is unreachable for normal flag-OFF moves, + // which never set recoveryRehome and always start from a legacy column, so + // characterization behavior is byte-identical). + const sourceIsLegacy = (COLUMNS as readonly string[]).includes(task.column); + const isEvacuation = + options?.recoveryRehome === true && + !sourceIsLegacy && + (COLUMNS as readonly string[]).includes(toColumn); + if (!isEvacuation) { + const validTargets = VALID_TRANSITIONS[task.column as Column] ?? []; + if (!validTargets.includes(toColumn)) { + throw new Error( + `Invalid transition: '${task.column}' → '${toColumn}'. ` + + `Valid targets: ${validTargets.join(", ") || "none"}`, + ); + } } if (fromColumn === "in-review" && toColumn === "done" && !options?.skipMergeBlocker) { @@ -12249,6 +12317,201 @@ ${stepsSection}`; return { scanned, rehomed, skippedTerminal }; } + // ── #1401: transitionPending recovery sweep ─────────────────────────────── + // + // A crash between the in-txn `transitionPending` marker write and the + // post-commit `clearTransitionPending` leaves the marker set forever. Because + // `countActiveInCapacitySlotSync` counts a pending marker as occupying a + // capacity slot for its `toColumn`, a stale marker permanently inflates that + // (workflow, column) capacity count. This sweep is the backstop the comments + // across store.ts / merge-trait.ts / transition-pending.ts reference: it scans + // every task carrying a non-null marker, reconciles `hooksRemaining` against + // the currently-known hook set, re-runs the surviving idempotent post-commit + // hooks via the same runner the live path uses, audits the recovery, and + // clears the marker so the reserved capacity slot is released. + // + // Idempotent: the default-workflow field effects already committed in-lock, so + // re-running them is a no-op, and a second sweep finds no markers. Plugin hooks + // are re-derived from the resolved IR (so an uninstalled-plugin hook simply + // drops, surfaced as an audit warning) and are expected to be idempotent per + // KTD-2. Runs at store init (alongside the integrity pass) and periodically + // from the flag-ON sweep cadence. + async recoverStaleTransitionPending(): Promise<{ scanned: number; recovered: number; degradedHooks: number }> { + let scanned = 0; + let recovered = 0; + let degradedHooks = 0; + + const rows = this.db + .prepare( + `SELECT id FROM tasks WHERE transitionPending IS NOT NULL AND transitionPending != '' AND deletedAt IS NULL`, + ) + .all() as Array<{ id: string }>; + + // The set of hook ids the current process can still honor: the always-present + // default-workflow post-commit marker plus every registered plugin trait's + // onEnter/onExit hook. A marker entry not in this set belongs to an + // uninstalled plugin and is dropped (audited) rather than re-run. + const registry = getTraitRegistry(); + const knownHookIds = new Set(["default-workflow:postCommit"]); + for (const def of registry.listTraits()) { + if (def.hooks?.onEnter) knownHookIds.add(`${def.id}:onEnter`); + if (def.hooks?.onExit) knownHookIds.add(`${def.id}:onExit`); + } + + for (const { id } of rows) { + scanned += 1; + const marker = readTransitionPending(this.db, id); + // null = nothing pending (corrupt/empty marker degrades to settled); we + // still clear the stored column so the slot is released. undefined = row + // vanished mid-sweep — skip. + if (marker === undefined) continue; + + await this.withTaskLock(id, async () => { + // Re-read inside the lock: another path may have cleared it already. + const live = readTransitionPending(this.db, id); + if (live == null) { + // Corrupt/empty marker — clear the stored value defensively so it stops + // counting against capacity, then move on. + if (live === null) { + try { + clearTransitionPending(this.db, id); + } catch { + // best-effort + } + } + return; + } + + const { hooksRemaining, warnings } = reconcileHooksRemaining(live.hooksRemaining, knownHookIds); + degradedHooks += warnings.length; + + // Re-run the surviving idempotent post-commit hooks. The default-workflow + // field effects already committed in-lock pre-crash, so the only work that + // can still be owed is the plugin trait hook runner, which re-derives its + // pending set from the resolved IR and is idempotent (KTD-2). We invoke it + // only when a plugin hook entry survived (a marker carrying just + // `default-workflow:postCommit` needs no re-run — just a clear). + const hasSurvivingPluginHook = hooksRemaining.some((h) => h !== "default-workflow:postCommit"); + if (hasSurvivingPluginHook) { + const task = this.readTaskFromDb(id, { includeDeleted: false }); + if (task) { + const ir = this.resolveTaskWorkflowIrSync(id); + // fromColumn is unknown post-crash; the marker only records toColumn. + // The hook runner keys onEnter off toColumn (and onExit off fromColumn); + // re-running onEnter for the destination is the recoverable, idempotent + // half. Use the task's current column as fromColumn (it committed to + // toColumn at marker-write time, so current == toColumn and onExit is a + // no-op, which is correct — we never re-fire an exit we may have run). + try { + await this.runPluginColumnTransitionHooks(id, ir, task.column, live.toColumn); + } catch (err) { + storeLog.warn("transitionPending recovery: hook re-run faulted (degraded)", { + phase: "recover-stale-transition-pending", + taskId: id, + error: err instanceof Error ? err.message : String(err), + }); + } + } + } + + for (const warning of warnings) { + storeLog.warn(warning, { + phase: "recover-stale-transition-pending", + taskId: id, + }); + } + + // Clear the marker — releases the reserved capacity slot. + try { + clearTransitionPending(this.db, id); + } catch { + // best-effort; a later sweep retries. + } + + this.recordRunAuditEvent({ + taskId: id, + agentId: "system", + runId: `transition-pending-recovery-${id}-${Date.now()}`, + domain: "database", + mutationType: "task:transition-pending-recovered", + target: id, + metadata: { + toColumn: live.toColumn, + hooksReran: hooksRemaining, + droppedHooks: warnings.length, + startedAt: live.startedAt, + }, + }); + recovered += 1; + }); + } + + if (recovered > 0 || degradedHooks > 0) { + storeLog.log("transitionPending recovery sweep completed", { + phase: "recover-stale-transition-pending", + scanned, + recovered, + degradedHooks, + }); + } + return { scanned, recovered, degradedHooks }; + } + + // ── #1409: flag ON→OFF evacuation ───────────────────────────────────────── + // + // When `workflowColumns` is disabled (or at flag-OFF store init), the board + // reverts to the legacy enum/`VALID_TRANSITIONS` path, where only the legacy + // {@link COLUMNS} are valid. Any card sitting in a CUSTOM (non-legacy) column + // would be stuck: it can't be listed/moved through the legacy path. This pass + // detects those cards and re-homes each to the nearest legacy column — the + // default workflow's entry column (`todo`) — via the existing recovery-rehome + // path (engine source + bypassGuards + recoveryRehome, capacity-honoring), + // auditing one event per card. Terminal cards (done/archived) are left put. + // + // Idempotent: a second run finds every card in a legacy column and is a no-op. + async evacuateCustomColumnsToLegacy( + trigger: "flag-off-init" | "flag-toggled-off", + ): Promise<{ scanned: number; evacuated: number }> { + let scanned = 0; + let evacuated = 0; + + const legacyColumns = new Set(COLUMNS); + // Nearest legacy landing column: the default workflow's entry column + // (triage). Falls back to "triage" defensively if the IR can't be resolved. + const targetColumn = resolveEntryColumnId(BUILTIN_CODING_WORKFLOW_IR) ?? "triage"; + + const rows = this.db + .prepare(`SELECT id, "column" AS col FROM tasks WHERE deletedAt IS NULL`) + .all() as Array<{ id: string; col: string }>; + + for (const { id, col } of rows) { + scanned += 1; + // Already in a legacy column (the common case) — nothing to evacuate. + if (legacyColumns.has(col)) continue; + // Never disturb terminal cards (legacy terminal semantics — these column + // ids are never legacy here, but guard defensively for parity with the + // integrity pass). + if (col === "done" || col === "archived") continue; + + await this.rehomeOccupant(id, targetColumn, "workflow-edit-rehome", { + evacuation: true, + trigger, + invalidColumn: col, + }); + evacuated += 1; + } + + if (evacuated > 0) { + storeLog.log("workflowColumns ON→OFF evacuation completed", { + phase: "evacuate-custom-columns", + trigger, + scanned, + evacuated, + }); + } + return { scanned, evacuated }; + } + // ── Workflow selection (resolves a workflow to enabledWorkflowSteps) ──── // // Selection never touches the engine's scheduler/executor/merger. It compiles diff --git a/packages/engine/src/__tests__/self-healing-custom-workflow-recovery.test.ts b/packages/engine/src/__tests__/self-healing-custom-workflow-recovery.test.ts new file mode 100644 index 0000000000..8fc16d54e0 --- /dev/null +++ b/packages/engine/src/__tests__/self-healing-custom-workflow-recovery.test.ts @@ -0,0 +1,124 @@ +// @vitest-environment node +// +// #1411: self-healing recovery/backward moves on CUSTOM workflows must pass +// `recoveryRehome: true` (not rely on `bypassGuards`, which skips trait guards +// but NOT order-derived column-graph adjacency). A custom workflow whose +// order-derived adjacency lacks the custom-column → todo edge would otherwise +// reject the recovery move and strand the card. +// +// This exercises a REAL TaskStore (flag-ON) so the in-lock adjacency check +// (resolveAllowedColumns) actually runs: +// - a backward recovery move WITHOUT recoveryRehome (engine source + +// bypassGuards) is rejected by adjacency, proving bypassGuards alone is +// insufficient (the bug), +// - the SAME move WITH recoveryRehome: true succeeds (the fix self-healing +// now applies at its moveTask call sites). + +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { execSync } from "node:child_process"; +import { TaskStore, type WorkflowIr } from "@fusion/core"; + +function git(cwd: string, args: string): void { + execSync(`git ${args}`, { cwd, stdio: "ignore" }); +} + +function setColumn(store: TaskStore, taskId: string, column: string): void { + const db = (store as unknown as { db: { prepare: (s: string) => { run: (...a: unknown[]) => unknown } } }).db; + db.prepare('UPDATE tasks SET "column" = ?, "columnMovedAt" = ? WHERE id = ?').run( + column, + new Date().toISOString(), + taskId, + ); +} + +/** + * A custom workflow whose linear order is intake → build → done. Its + * order-derived adjacency has NO edge build → todo (todo is not even a column), + * so a recovery move build → todo is only reachable via recoveryRehome. + */ +function customIr(): WorkflowIr { + return { + version: "v2", + name: "linear-custom", + columns: [ + { id: "intake", name: "intake", traits: [{ trait: "intake" }] }, + { id: "build", name: "build", traits: [] }, + { id: "done", name: "done", traits: [{ trait: "complete" }] }, + ], + nodes: [ + { id: "start", kind: "start", column: "intake" }, + { id: "work", kind: "prompt", column: "build", config: { prompt: "do" } }, + { id: "end", kind: "end", column: "done" }, + ], + edges: [ + { from: "start", to: "work", condition: "success" }, + { from: "work", to: "end", condition: "success" }, + ], + } as WorkflowIr; +} + +describe("#1411 self-healing recovery move on custom workflows", () => { + let rootDir = ""; + let store: TaskStore; + + beforeEach(async () => { + rootDir = mkdtempSync(join(tmpdir(), "fn-1411-")); + 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"); + store = new TaskStore(rootDir, undefined, { inMemoryDb: false }); + await store.init(); + await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: true } }); + }); + + afterEach(() => { + try { store?.close(); } catch { /* ignore */ } + if (rootDir) rmSync(rootDir, { recursive: true, force: true }); + vi.clearAllMocks(); + }); + + async function seedCardInBuild(): Promise { + const wf = await store.createWorkflowDefinition({ name: "linear-custom", ir: customIr() }); + const task = await store.createTask({ description: "stuck-in-build" }); + await store.selectTaskWorkflowAndReconcile(task.id, wf.id); + setColumn(store, task.id, "build"); + expect((await store.getTask(task.id)).column).toBe("build"); + return task.id; + } + + it("bypassGuards alone is rejected by order-derived adjacency (build → todo)", async () => { + const id = await seedCardInBuild(); + let caught: unknown; + try { + // Mirrors a self-healing backward move BEFORE the fix: engine source + + // bypassGuards, but no recoveryRehome. Adjacency (build → todo) has no edge. + await store.moveTask(id, "todo", { moveSource: "engine", bypassGuards: true, preserveProgress: true }); + } catch (e) { + caught = e; + } + expect(caught).toBeInstanceOf(Error); + expect((await store.getTask(id)).column).toBe("build"); + }); + + it("recoveryRehome: true lets the recovery move reach todo (the fix)", async () => { + const id = await seedCardInBuild(); + await store.moveTask(id, "todo", { + moveSource: "engine", + recoveryRehome: true, + preserveProgress: true, + }); + expect((await store.getTask(id)).column).toBe("todo"); + }); + + it("recoveryRehome: true also reaches a terminal recovery target (archived)", async () => { + const id = await seedCardInBuild(); + await store.moveTask(id, "archived", { moveSource: "engine", recoveryRehome: true }); + expect((await store.getTask(id)).column).toBe("archived"); + }); +}); diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 500edfe0fc..55c603a4d9 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -28,7 +28,7 @@ import { promisify } from "node:util"; import { setImmediate as setImmediateCb } from "node:timers"; import { existsSync, mkdirSync, readdirSync, readFileSync, realpathSync, rmSync, statSync, writeFileSync } from "node:fs"; import { isAbsolute, join, relative, resolve } from "node:path"; -import { IN_REVIEW_STALL_DEADLOCK_LOG_PREFIX, IN_REVIEW_STALL_LOG_PREFIX, allowsAutoMergeProcessing, countRecentIdenticalStallEntries, detectDependencyCycle, detectSelfDefeatingDependency, getInReviewStalledSignal, getInReviewStallReason, getPrimaryPrInfo, getStalePausedReviewSignal, getStalePausedTodoSignal, getTaskHardMergeBlocker, getTaskMergeBlocker, isEphemeralAgent, isMergeRequestContractShadowEnabled, isSharedBranchGroupMemberIntegration, parseExplicitDuplicateMarker, type AgentStore, type ChatStore, type MessageStore, type TaskStore, type Settings, type Task, type MergeDetails, type TaskPriority, type MergeResult } from "@fusion/core"; +import { IN_REVIEW_STALL_DEADLOCK_LOG_PREFIX, IN_REVIEW_STALL_LOG_PREFIX, allowsAutoMergeProcessing, countRecentIdenticalStallEntries, detectDependencyCycle, detectSelfDefeatingDependency, getInReviewStalledSignal, getInReviewStallReason, getPrimaryPrInfo, getStalePausedReviewSignal, getStalePausedTodoSignal, getTaskHardMergeBlocker, getTaskMergeBlocker, isEphemeralAgent, isMergeRequestContractShadowEnabled, isWorkflowColumnsEnabled, isSharedBranchGroupMemberIntegration, parseExplicitDuplicateMarker, type AgentStore, type ChatStore, type MessageStore, type TaskStore, type Settings, type Task, type MergeDetails, type TaskPriority, type MergeResult } from "@fusion/core"; import type { MeshLeaseManager } from "./mesh-lease-manager.js"; import { createLogger, schedulerLog } from "./logger.js"; import { RemovalReason, classifyTaskWorktree, getRegisteredWorktreeBranchMap, getRegisteredWorktreePaths, isUsableTaskWorktree, removeWorktree, resolveWorktreeBackend, scanIdleWorktrees, scanOrphanedBranches } from "./worktree-pool.js"; @@ -110,7 +110,9 @@ export async function archiveAsGhostBug( findings: decision.findings.slice(0, 10), }, }); - await store.moveTask(taskId, "archived"); + // #1411: recovery/terminal move — recoveryRehome skips order-derived adjacency + // so a custom-workflow card can always reach the terminal column. + await store.moveTask(taskId, "archived", { moveSource: "engine", recoveryRehome: true }); } async function classifyOwnedLandedEvidenceForSelfHealing(rootDir: string, task: Task, mergeTargetBranch: string): Promise { @@ -437,9 +439,10 @@ export async function autoRecoverWorktreeSessionStartFailure( : `Auto-recovered: retry/verification session targeted unusable worktree${staleWorktree ? ` (${staleWorktree})` : ""} — cleared stale session metadata and requeued to todo (attempt ${nextCount}/${MAX_WORKTREE_SESSION_RETRIES}, failure: ${failureExcerpt})`, ); if (noProgress) { - await store.moveTask(task.id, "todo"); + // #1411: backward recovery move — recoveryRehome skips order-derived adjacency. + await store.moveTask(task.id, "todo", { moveSource: "engine", recoveryRehome: true }); } else { - await store.moveTask(task.id, "todo", { preserveProgress: true }); + await store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine", recoveryRehome: true }); } return { outcome: "requeue-todo", retries: nextCount, classification }; } @@ -1112,6 +1115,9 @@ export class SelfHealingManager { await this.store.moveTask(taskId, "todo", { preserveProgress: true, preserveStatus: true, + // #1411: backward recovery — skip order-derived adjacency. + moveSource: "engine", + recoveryRehome: true, }); } catch (moveErr: unknown) { const moveErrMessage = moveErr instanceof Error ? moveErr.message : String(moveErr); @@ -1768,6 +1774,10 @@ export class SelfHealingManager { { name: "auto-archive-meta-resolved", fn: () => this.autoArchiveResolvedMetaTasks() }, { name: "auto-archive-meta-stalled", fn: () => this.autoArchiveStalledMetaTasks() }, { name: "board-stall-auto-recovery", fn: () => this.runBoardStallAutoRecoverySweep() }, + // #1401: periodically recover transitionPending markers stranded by a + // crash between the in-txn write and the post-commit clear (flag-ON + // only; a no-op when there are no markers). + { name: "recover-stale-transition-pending", fn: () => this.runStaleTransitionPendingSweep() }, { name: "reconcile-self-defeating-deps", fn: () => this.reconcileSelfDefeatingDependencies() }, { name: "reconcile-dependency-cycles", fn: () => this.reconcileDependencyCycles().then(() => undefined) }, { name: "reclaim-pr-conflicts", fn: () => this.reclaimPrConflicts() }, @@ -2200,6 +2210,8 @@ export class SelfHealingManager { }); await this.store.moveTask(task.id, "todo", { moveSource: "engine", + // #1411: backward recovery — skip order-derived adjacency. + recoveryRehome: true, preserveWorktree: true, preserveProgress: true, preserveResumeState: true, @@ -2467,6 +2479,8 @@ export class SelfHealingManager { } else { await this.store.moveTask(task.id, "todo", { moveSource: "engine", + // #1411: backward recovery — skip order-derived adjacency. + recoveryRehome: true, preserveProgress: true, preserveResumeState: true, }); @@ -2570,6 +2584,8 @@ export class SelfHealingManager { } else { await this.store.moveTask(task.id, "todo", { moveSource: "engine", + // #1411: backward recovery — skip order-derived adjacency. + recoveryRehome: true, preserveProgress: true, preserveResumeState: true, }); @@ -2637,6 +2653,8 @@ export class SelfHealingManager { const idleMs = Number.isFinite(idleAnchorMs) ? Math.max(0, Date.now() - idleAnchorMs) : null; await this.store.moveTask(task.id, "todo", { moveSource: "engine", + // #1411: backward recovery — skip order-derived adjacency. + recoveryRehome: true, preserveWorktree: true, preserveProgress: true, preserveResumeState: true, @@ -2703,6 +2721,8 @@ export class SelfHealingManager { } else { await this.store.moveTask(task.id, "todo", { moveSource: "engine", + // #1411: backward recovery — skip order-derived adjacency. + recoveryRehome: true, preserveWorktree: true, preserveProgress: true, preserveResumeState: true, @@ -3621,6 +3641,8 @@ export class SelfHealingManager { preserveWorktree: true, preserveResumeState: true, moveSource: "engine", + // #1411: backward recovery — skip order-derived adjacency. + recoveryRehome: true, }); await this.store.logEntry( task.id, @@ -3886,6 +3908,19 @@ export class SelfHealingManager { return archived; } + /** + * #1401: periodic transitionPending recovery sweep. Flag-ON only — when + * `workflowColumns` is OFF the legacy path never writes markers, so there is + * nothing to recover. Delegates to the store's idempotent recovery method + * (a no-op when no stale markers exist), keeping capacity counts honest after + * a crash between the in-txn marker write and the post-commit clear. + */ + async runStaleTransitionPendingSweep(): Promise { + const settings = await this.store.getSettings(); + if (!isWorkflowColumnsEnabled(settings)) return; + await this.store.recoverStaleTransitionPending(); + } + async runBoardStallAutoRecoverySweep(): Promise<{ holders: string[]; recovered: number; unrecovered: boolean }> { const settings = await this.store.getSettings(); const windowMs = Number(settings.boardStallSweepWindowMs ?? 2 * 60 * 60_000); @@ -4616,7 +4651,8 @@ export class SelfHealingManager { await this.emitBackwardMoveNoAction(task, "finalize-no-op-review", "task:finalize-no-op-review-no-action", proof); continue; } - await this.store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine" }); + // #1411: backward recovery — skip order-derived adjacency. + await this.store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine", recoveryRehome: true }); continue; } @@ -4666,7 +4702,8 @@ export class SelfHealingManager { classification: "proven-no-op", baseRef: classification.baseRef, }); - await this.store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine" }); + // #1411: backward recovery — skip order-derived adjacency. + await this.store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine", recoveryRehome: true }); recovered++; continue; } @@ -5102,7 +5139,8 @@ export class SelfHealingManager { task.id, "Auto-recovered: in-review task still had incomplete steps — moved back to todo for retry", ); - await this.store.moveTask(task.id, "todo", { preserveProgress: true }); + // #1411: backward recovery — skip order-derived adjacency. + await this.store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine", recoveryRehome: true }); log.log(`Recovered stale incomplete review task ${task.id}: moved back to todo`); recovered++; } catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err); @@ -5506,7 +5544,8 @@ export class SelfHealingManager { task.id, "Auto-recovered: in-review task idle past stuck-task timeout — kicked back to todo", ); - await this.store.moveTask(task.id, "todo", { preserveProgress: true }); + // #1411: backward recovery — skip order-derived adjacency. + await this.store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine", recoveryRehome: true }); log.log(`Kicked ghost review task ${task.id} back to todo`); recovered++; } catch (err: unknown) { @@ -7248,7 +7287,8 @@ export class SelfHealingManager { stepStatuses, }, }); - await this.store.moveTask(task.id, "todo", { preserveProgress: true }); + // #1411: backward recovery — skip order-derived adjacency. + await this.store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine", recoveryRehome: true }); recovered++; } catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err); @@ -7785,7 +7825,8 @@ export class SelfHealingManager { task.id, "Auto-recovered no-progress no-task_done failure — clean worktree, moved back to todo", ); - await this.store.moveTask(task.id, "todo"); + // #1411: backward recovery — skip order-derived adjacency. + await this.store.moveTask(task.id, "todo", { moveSource: "engine", recoveryRehome: true }); recovered++; } catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err); log.error(`Failed to recover no-progress no-task_done failure ${task.id}: ${errorMessage}`); @@ -7952,7 +7993,8 @@ export class SelfHealingManager { task.id, `Auto-retry ${nextCount}/${MAX_TASK_DONE_RETRIES}: agent finished without fn_task_done — requeuing to todo to resume partial work`, ); - await this.store.moveTask(task.id, "todo", { preserveProgress: true }); + // #1411: backward recovery — skip order-derived adjacency. + await this.store.moveTask(task.id, "todo", { preserveProgress: true, moveSource: "engine", recoveryRehome: true }); recovered++; } catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err);