diff --git a/packages/core/src/__tests__/transition-parity.test.ts b/packages/core/src/__tests__/transition-parity.test.ts index 3587fe59fa..17bc033994 100644 --- a/packages/core/src/__tests__/transition-parity.test.ts +++ b/packages/core/src/__tests__/transition-parity.test.ts @@ -12,7 +12,7 @@ // - crash-mid-transition marker recovery (SQLite authoritative) // - unknown-column rejection // - guard rejection typed (flag-ON) vs legacy string (flag-OFF) -// - bypassGuards capacity pass-through (documenting; U6 fills enforcement) +// - in-txn capacity enforcement (U6; NEVER bypassable — KTD-10) import { describe, it, expect, beforeEach, afterEach } from "vitest"; import { VALID_TRANSITIONS } from "../types.js"; @@ -199,18 +199,69 @@ describe("transition-parity — store flag-ON scenarios", () => { expect(after?.worktree).toBe("/tmp/wt/seed-todo"); }); - it("bypassGuards capacity pass-through (U4 documenting test): engine move into in-progress is NOT blocked by capacity (U6 fills enforcement)", async () => { - // U4 intentionally leaves the per-(workflow,column) capacity check as a - // pass-through slot; capacity enforcement lands in U6. This test pins the - // U4 contract: no WIP-constrained scenario is enforced yet, and an engine - // move (bypassGuards) into a wip-flagged column commits. It must be UPDATED - // by U6 (capacity is NEVER bypassable, KTD-10) — not silently left green. + it("U6 in-txn capacity: default-workflow in-progress WIP reads through maxConcurrent and rejects the over-limit move", async () => { + // The default workflow's in-progress column has a `wip` trait whose limit + // reads through to settings.maxConcurrent (legacy parity). With limit 1, the + // first move into in-progress commits and a second rejects with the typed + // capacity-exhausted code. + await store.updateSettings({ maxConcurrent: 1 } as Parameters[0]); const t1 = await seedInColumn("todo"); const t2 = await seedInColumn("todo"); - const m1 = await store.moveTask(t1.id, "in-progress", { moveSource: "engine" }); - const m2 = await store.moveTask(t2.id, "in-progress", { moveSource: "engine" }); + const m1 = await store.moveTask(t1.id, "in-progress", { moveSource: "user" }); expect(m1.column).toBe("in-progress"); - expect(m2.column).toBe("in-progress"); + + let caught: unknown; + try { + await store.moveTask(t2.id, "in-progress", { moveSource: "user" }); + } catch (e) { + caught = e; + } + expect(caught).toBeInstanceOf(TransitionRejectionError); + expect((caught as TransitionRejectionError).rejection.code).toBe("capacity-exhausted"); + // The rejected card is untouched. + expect((await store.getTask(t2.id))?.column).toBe("todo"); + }); + + it("U6 capacity is NEVER bypassable (KTD-10): an engine/bypassGuards move into a full column still rejects", async () => { + await store.updateSettings({ maxConcurrent: 1 } as Parameters[0]); + const t1 = await seedInColumn("todo"); + const t2 = await seedInColumn("todo"); + await store.moveTask(t1.id, "in-progress", { moveSource: "user" }); + + let caught: unknown; + try { + // Engine-sourced + bypassGuards skips trait guards, but capacity is not a + // guard — it must still reject. + await store.moveTask(t2.id, "in-progress", { moveSource: "engine", bypassGuards: true }); + } catch (e) { + caught = e; + } + expect(caught).toBeInstanceOf(TransitionRejectionError); + expect((caught as TransitionRejectionError).rejection.code).toBe("capacity-exhausted"); + }); + + it("U6 capacity counts cards mid-transitionPending (they hold their slot from commit time)", async () => { + await store.updateSettings({ maxConcurrent: 1 } as Parameters[0]); + const t1 = await seedInColumn("todo"); + const t2 = await seedInColumn("todo"); + await store.moveTask(t1.id, "in-progress", { moveSource: "user" }); + // Simulate a crash before t1's marker clears: it is still mid-transition into + // in-progress, holding its slot. (Its column already equals in-progress, so + // this also independently holds the slot; this asserts the marker path does + // not under-count or double-count.) + const db = (store as unknown as { db: { prepare: (s: string) => { run: (...a: unknown[]) => unknown } } }).db; + db.prepare("UPDATE tasks SET transitionPending = ? WHERE id = ?").run( + JSON.stringify({ toColumn: "in-progress", hooksRemaining: ["default-workflow:postCommit"], startedAt: Date.now() }), + t1.id, + ); + let caught: unknown; + try { + await store.moveTask(t2.id, "in-progress", { moveSource: "user" }); + } catch (e) { + caught = e; + } + expect(caught).toBeInstanceOf(TransitionRejectionError); + expect((caught as TransitionRejectionError).rejection.code).toBe("capacity-exhausted"); }); }); diff --git a/packages/core/src/default-workflow-hooks.ts b/packages/core/src/default-workflow-hooks.ts index b257c98799..c3dc13495d 100644 --- a/packages/core/src/default-workflow-hooks.ts +++ b/packages/core/src/default-workflow-hooks.ts @@ -69,7 +69,7 @@ export interface DefaultWorkflowMoveContext { task: Task; fromColumn: string; toColumn: string; - moveSource: "user" | "engine"; + moveSource: "user" | "engine" | "scheduler"; /** True when guards + abort-on-exit are bypassed (engine/recovery, KTD-9). */ bypassGuards: boolean; movedAt: string; diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 2830d15e41..0a9fe5e003 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -132,6 +132,9 @@ export { } from "./workflow-transitions.js"; export type { ColumnAdjacency } from "./workflow-transitions.js"; export { isWorkflowColumnsEnabled } from "./workflow-columns-settings.js"; +// ── U6: workflow capacity (WIP) resolution shared by store + sweep ─────────── +export { resolveColumnCapacity } from "./workflow-capacity.js"; +export type { ColumnCapacity } from "./workflow-capacity.js"; // ── U5: workflow lifecycle reconciliation (switch / edit / delete) ─────────── export { OccupiedColumnsError, diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 8095ad72c7..5b9b51b49b 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -10,6 +10,7 @@ import { DEFAULT_PROJECT_SETTINGS } from "./settings-schema.js"; import { parseWorkflowIr, serializeWorkflowIr } from "./workflow-ir.js"; import { isWorkflowColumnsEnabled } from "./workflow-columns-settings.js"; import { resolveAllowedColumns, workflowHasColumn } from "./workflow-transitions.js"; +import { resolveColumnCapacity } from "./workflow-capacity.js"; import { OccupiedColumnsError, assertRehomeTargetValid, @@ -676,7 +677,7 @@ function deepMergeWithNullDelete( export interface TaskStoreEvents { "task:created": [task: Task]; - "task:moved": [data: { task: Task; from: Column; to: Column; source: "user" | "engine" }]; + "task:moved": [data: { task: Task; from: Column; to: Column; source: "user" | "engine" | "scheduler" }]; "task:updated": [task: Task]; "task:deleted": [task: Task, meta?: { githubIssueAction?: GithubIssueAction }]; "task:merged": [result: MergeResult]; @@ -1102,7 +1103,7 @@ interface MoveTaskOptions { preserveWorktree?: boolean; preserveStatus?: boolean; allocateWorktree?: (reservedNames: Set) => string | null; - moveSource?: "user" | "engine"; + moveSource?: "user" | "engine" | "scheduler"; skipMergeBlocker?: boolean; allowDirectInReviewMove?: boolean; /** @@ -1138,6 +1139,10 @@ interface MoveTaskInternalOptions { export class TaskStore extends EventEmitter { private static readonly ACTIVE_TASKS_WHERE = '"deletedAt" IS NULL'; + /** U6: sentinel effective-workflow id for default-workflow (null-selection) + * tasks, so they all share one per-column capacity pool (KTD-10). It is not a + * real workflow row id (no `builtin:`/custom collision possible). */ + private static readonly DEFAULT_WORKFLOW_POOL_ID = "__default-workflow__"; static async getOrCreateForProject( projectId?: string, @@ -5653,14 +5658,16 @@ export class TaskStore extends EventEmitter { // `getSettingsSync()` row would miss it — read merged settings (global + // project) via getSettingsFast(). This is an async read taken before the // lock-sensitive transaction; it does not touch the task lock. - const useWorkflow = isWorkflowColumnsEnabled(await this.getSettingsFast()); + const mergedSettingsForMove = await this.getSettingsFast(); + const useWorkflow = isWorkflowColumnsEnabled(mergedSettingsForMove); // bypassGuards (KTD-9): engine-sourced moves + the existing skipMergeBlocker // call sites map onto it. Capacity (KTD-10) is NEVER bypassed by this — the // capacity check is not a guard (U6 fills the enforcement; U4 leaves a // pass-through slot). An explicit option value wins; otherwise derive it. const bypassGuards = options?.recoveryRehome === true || - (options?.bypassGuards ?? (moveSource === "engine" || options?.skipMergeBlocker === true)); + (options?.bypassGuards ?? + (moveSource === "engine" || moveSource === "scheduler" || options?.skipMergeBlocker === true)); const workflowIr: WorkflowIr | undefined = useWorkflow ? this.resolveTaskWorkflowIrSync(id) : undefined; @@ -5974,6 +5981,41 @@ export class TaskStore extends EventEmitter { return; } + // ── U6: in-txn capacity enforcement (KTD-10) ────────────────────────── + // WIP limits are trait *config*; enforcement is a substrate capability + // that runs HERE, inside the move transaction, so two holds releasing into + // one slot serialize — exactly one commits, the other rejects and retries + // next sweep. It is NOT a guard: it runs regardless of bypassGuards / + // recoveryRehome / moveSource (engine/recovery/scheduler moves honor it + // too). Only a real column change into a capacity-bearing column is gated; + // same-column no-ops were returned earlier. The count is taken with the + // moving task EXCLUDED and the prospective slot it is about to occupy + // added back implicitly (it must fit alongside existing holders), so a + // full column (occupants == limit) rejects. + if (useWorkflow && workflowIr && fromColumn !== toColumn) { + const capacity = resolveColumnCapacity(workflowIr, toColumn, mergedSettingsForMove); + if (capacity.hasCapacity && Number.isFinite(capacity.limit)) { + const workflowId = this.resolveEffectiveWorkflowIdSync(id); + const occupants = this.countActiveInCapacitySlotSync({ + targetColumn: toColumn, + workflowId, + countPending: capacity.countPending, + excludeTaskId: id, + }); + if (occupants >= capacity.limit) { + throw new TransitionRejectionError( + makeTransitionRejection( + "capacity-exhausted", + "transition.rejected.capacityExhausted", + true, + `Column '${toColumn}' is at capacity (${occupants}/${capacity.limit})`, + ), + `Cannot move ${id} to '${toColumn}': column at capacity (${occupants}/${capacity.limit})`, + ); + } + } + } + this.upsertTaskWithFtsRecovery(task); this.insertRunAuditEventRow({ taskId: id, @@ -11880,6 +11922,81 @@ ${stepsSection}`; } } + /** + * U6 (KTD-10): the *effective workflow id* used to scope the per-(workflow, + * column) capacity count. A task with no selection (or a missing/empty + * selection row) resolves to the built-in default workflow, represented by a + * stable sentinel so all default-workflow tasks share one capacity pool. A + * selected workflow id (builtin or custom) is its own pool. Pure DB read; safe + * inside the move transaction. + */ + private resolveEffectiveWorkflowIdSync(taskId: string): string { + const selection = this.getTaskWorkflowSelection(taskId); + return selection?.workflowId ?? TaskStore.DEFAULT_WORKFLOW_POOL_ID; + } + + /** + * U6 (KTD-10): count cards currently occupying a (workflow, column) capacity + * slot, for the in-txn capacity check. Runs INSIDE `moveTaskInternal`'s + * transaction. A slot is held by a card that: + * - has committed its column to `targetColumn` (the steady-state holders), OR + * - (when `countPending`) has a `transitionPending` marker targeting + * `targetColumn` — it reserved the slot at commit time even though its + * post-commit hooks haven't finished yet. + * The moving task itself (`excludeTaskId`) is excluded so a same-column no-op + * or re-entry never counts itself. Only the candidates in the SAME effective + * workflow as the mover count (capacity is per-(workflow, column)). Soft-deleted + * tasks never hold a slot. + */ + private countActiveInCapacitySlotSync(params: { + targetColumn: string; + workflowId: string; + countPending: boolean; + excludeTaskId: string; + }): number { + const { targetColumn, workflowId, countPending, excludeTaskId } = params; + // Candidate rows: in the column now, or (optionally) mid-transition into it. + // LEFT JOIN the selection row so we can scope by effective workflow id in JS. + const rows = this.db + .prepare( + `SELECT t.id AS id, t."column" AS col, t.transitionPending AS tp, s.workflowId AS wid + FROM tasks t + LEFT JOIN task_workflow_selection s ON s.taskId = t.id + WHERE t.deletedAt IS NULL + AND t.id != ? + AND (t."column" = ? OR (t.transitionPending IS NOT NULL AND t.transitionPending != ''))`, + ) + .all(excludeTaskId, targetColumn) as Array<{ + id: string; + col: string; + tp: string | null; + wid: string | null; + }>; + + let count = 0; + for (const row of rows) { + const effectiveWorkflowId = row.wid ?? TaskStore.DEFAULT_WORKFLOW_POOL_ID; + if (effectiveWorkflowId !== workflowId) continue; + + if (row.col === targetColumn) { + count += 1; + continue; + } + // Not committed into the column — only counts if it has reserved the slot + // via a transitionPending marker targeting this column AND countPending. + if (!countPending || !row.tp) continue; + let toColumn: string | undefined; + try { + const parsed = JSON.parse(row.tp) as { toColumn?: unknown }; + if (typeof parsed.toColumn === "string") toColumn = parsed.toColumn; + } catch { + // Corrupt marker — treat as not holding this slot. + } + if (toColumn === targetColumn) count += 1; + } + return count; + } + getTaskWorkflowSelection(taskId: string): { workflowId: string; stepIds: string[] } | undefined { const row = this.db .prepare("SELECT workflowId, stepIds FROM task_workflow_selection WHERE taskId = ?") diff --git a/packages/core/src/workflow-capacity.ts b/packages/core/src/workflow-capacity.ts new file mode 100644 index 0000000000..9ef0c40851 --- /dev/null +++ b/packages/core/src/workflow-capacity.ts @@ -0,0 +1,116 @@ +/** + * Workflow capacity resolution (U6, KTD-10, R9 capacity half). + * + * WIP/capacity limits are trait *configuration*; their *enforcement* is a + * substrate capability that runs INSIDE `moveTaskInternal`'s transaction and is + * NEVER bypassable (not a guard — runs regardless of bypassGuards/recoveryRehome + * /moveSource). This module is the pure resolution layer shared by both the + * in-txn check (`store.ts`) and the hold/release sweep (`@fusion/engine` + * `hold-release.ts`): given a workflow IR + a column id + settings it answers + * - does this column have a `wip` (capacity) trait? + * - what is its effective limit (read-through to `settings.maxConcurrent` for + * the default workflow's in-progress column so the legacy knob keeps working + * — U6 scheduler-integration half)? + * - does its config opt into counting mid-`transitionPending` cards? + * + * It performs NO DB access and NO counting — the caller owns the count (the + * store counts in-txn; the sweep counts from a listTasks snapshot). Keeping the + * resolution pure means the two enforcement points can never disagree on what a + * limit *is*, only on the live count, which is exactly the serialization the + * in-txn check arbitrates (two holds, one slot → one wins). + */ + +import type { Settings } from "./types.js"; +import type { WorkflowIr, WorkflowIrV2, WorkflowIrColumn } from "./workflow-ir-types.js"; +import { DEFAULT_WORKFLOW_COLUMN_IDS } from "./workflow-ir.js"; +import { getTraitRegistry } from "./trait-registry.js"; + +/** The default-workflow column whose WIP limit read-through is + * `settings.maxConcurrent` (the legacy "N agents in-progress" gate). */ +const DEFAULT_WIP_COLUMN_ID = "in-progress"; + +/** Resolved capacity configuration for a single column. */ +export interface ColumnCapacity { + /** True when the column carries a capacity (`wip`/`countsTowardWip`) trait. */ + hasCapacity: boolean; + /** The effective max concurrent cards. `Infinity` means "no finite limit" + * (a capacity trait with no resolvable limit does not gate). */ + limit: number; + /** Whether mid-`transitionPending` cards (holding their destination slot from + * commit time) count toward the limit. Defaults true: a card that has + * committed its move into the column holds the slot even before its + * post-commit hooks finish (KTD-10). */ + countPending: boolean; +} + +const NO_CAPACITY: ColumnCapacity = { hasCapacity: false, limit: Infinity, countPending: true }; + +function findColumn(ir: WorkflowIr, columnId: string): WorkflowIrColumn | undefined { + const v2 = ir as WorkflowIrV2; + if (!Array.isArray(v2.columns)) return undefined; + return v2.columns.find((c) => c.id === columnId); +} + +/** True when the IR's column set is exactly the default-workflow column ids. */ +function isDefaultWorkflowColumns(ir: WorkflowIr): boolean { + const v2 = ir as WorkflowIrV2; + if (!Array.isArray(v2.columns)) return false; + const ids = v2.columns.map((c) => c.id); + if (ids.length !== DEFAULT_WORKFLOW_COLUMN_IDS.length) return false; + const set = new Set(ids); + return DEFAULT_WORKFLOW_COLUMN_IDS.every((id) => set.has(id)); +} + +/** + * Resolve the capacity configuration for `columnId` under `ir`. + * + * Limit resolution order: + * 1. An explicit numeric `limit` in the column's `wip` trait config wins. + * 2. Otherwise, for the DEFAULT workflow's `in-progress` column, read through + * to `settings.maxConcurrent` (default 2) so the legacy knob keeps working + * and flag-ON default-workflow scheduling matches flag-OFF (legacy parity). + * 3. Otherwise the column has a capacity trait but no resolvable finite limit + * → `Infinity` (does not gate; the trait is inert until configured). + */ +export function resolveColumnCapacity( + ir: WorkflowIr, + columnId: string, + settings?: Pick | undefined, +): ColumnCapacity { + const column = findColumn(ir, columnId); + if (!column) return NO_CAPACITY; + + const flags = getTraitRegistry().resolveColumnFlags(column); + if (!flags.countsTowardWip) return NO_CAPACITY; + + // The capacity trait config (the `wip` trait carries `limit` + `countPending`). + // Find the first trait config whose trait sets countsTowardWip. + let configLimit: number | undefined; + let countPending = true; + for (const ct of column.traits) { + const def = getTraitRegistry().getTrait(ct.trait); + if (!def?.flags.countsTowardWip) continue; + const cfg = ct.config ?? {}; + if (typeof cfg.limit === "number" && Number.isFinite(cfg.limit)) { + configLimit = cfg.limit; + } + if (typeof cfg.countPending === "boolean") { + countPending = cfg.countPending; + } + break; + } + + let limit: number; + if (configLimit !== undefined) { + limit = configLimit; + } else if (columnId === DEFAULT_WIP_COLUMN_ID && isDefaultWorkflowColumns(ir)) { + // Read-through: legacy maxConcurrent maps onto the default workflow's + // in-progress WIP limit (U6 scheduler integration). + const maxConcurrent = settings?.maxConcurrent; + limit = typeof maxConcurrent === "number" && Number.isFinite(maxConcurrent) ? maxConcurrent : 2; + } else { + limit = Infinity; + } + + return { hasCapacity: true, limit, countPending }; +} diff --git a/packages/engine/src/__tests__/hold-release.test.ts b/packages/engine/src/__tests__/hold-release.test.ts new file mode 100644 index 0000000000..a2810193c3 --- /dev/null +++ b/packages/engine/src/__tests__/hold-release.test.ts @@ -0,0 +1,402 @@ +// @vitest-environment node +// +// HOLD/RELEASE SWEEP SUITE (U6). +// +// Exercises the generalized scheduler sweep (`hold-release.ts`) against a REAL +// TaskStore so the in-txn capacity check (KTD-10) actually arbitrates races: +// - two holds, one slot → exactly one releases; other retries next sweep +// - timer release fires at its deadline under fake timers (no real sleeps) +// - manual release only on the explicit promote call +// - capacity release respects mid-transitionPending cards (in-txn authority) +// - cross-workflow dependency complete-flag unblocks + dual-accept diff logged +// - sweep release into a full column rejected by the in-txn check despite +// moveSource:"scheduler" bypassing trait guards (capacity is not a guard) +// - reservation-first: semaphore exhausted → no commit, card stays held +// - paused / recovery-backoff tasks skipped exactly as the legacy scheduler + +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"; +import { + runHoldReleaseSweep, + promoteHeldTask, + releaseHeldTaskByEvent, + type HoldReleaseDeps, + type SlotReservation, +} from "../hold-release.js"; + +function git(cwd: string, args: string): void { + execSync(`git ${args}`, { cwd, stdio: "ignore" }); +} + +/** Directly set a task's stored column (test setup helper — bypasses adjacency + * validation so a card can be placed at an arbitrary workflow column). */ +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, + ); +} + +/** Directly set a task's workflow selection row (bypasses step compilation). */ +function setSelection(store: TaskStore, taskId: string, workflowId: string): void { + const db = (store as unknown as { db: { prepare: (s: string) => { run: (...a: unknown[]) => unknown } } }).db; + db.prepare( + `INSERT INTO task_workflow_selection (taskId, workflowId, stepIds, updatedAt) + VALUES (?, ?, '[]', ?) + ON CONFLICT(taskId) DO UPDATE SET workflowId = excluded.workflowId, updatedAt = excluded.updatedAt`, + ).run(taskId, workflowId, new Date().toISOString()); +} + +/** Write a transitionPending marker directly (simulating a crash mid-transition). */ +function setTransitionPending(store: TaskStore, taskId: string, toColumn: string): void { + const db = (store as unknown as { db: { prepare: (s: string) => { run: (...a: unknown[]) => unknown } } }).db; + db.prepare("UPDATE tasks SET transitionPending = ? WHERE id = ?").run( + JSON.stringify({ toColumn, hooksRemaining: ["default-workflow:postCommit"], startedAt: Date.now() }), + taskId, + ); +} + +const noReserveDeps: HoldReleaseDeps = { now: () => Date.now() }; + +describe("hold-release sweep (U6)", () => { + let rootDir = ""; + let store: TaskStore; + + beforeEach(async () => { + rootDir = mkdtempSync(join(tmpdir(), "u6-hold-release-")); + 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.useRealTimers(); + vi.clearAllMocks(); + }); + + // A held card in the DEFAULT workflow: a task resting in `todo` + // (hold release: capacity), which releases into `in-progress` (wip). + async function seedTodoCard(): Promise { + const task = await store.createTask({ description: "card" }); + setColumn(store, task.id, "todo"); + return task.id; + } + + it("flag OFF: sweep is a no-op (legacy scheduler path untouched)", async () => { + await store.updateGlobalSettings({ experimentalFeatures: { workflowColumns: false } }); + const id = await seedTodoCard(); + const result = await runHoldReleaseSweep(store, noReserveDeps); + expect(result.released).toEqual([]); + expect((await store.getTask(id))?.column).toBe("todo"); + }); + + it("two holds, one slot: exactly one releases; the other releases next sweep after the slot frees", async () => { + await store.updateSettings({ maxConcurrent: 1 } as Parameters[0]); + const a = await seedTodoCard(); + const b = await seedTodoCard(); + + const r1 = await runHoldReleaseSweep(store, noReserveDeps); + expect(r1.released.length).toBe(1); + const released = r1.released[0]; + const stillHeld = released === a ? b : a; + expect((await store.getTask(released))?.column).toBe("in-progress"); + expect((await store.getTask(stillHeld))?.column).toBe("todo"); + + // Free the slot by moving the released card out of in-progress. + await store.moveTask(released, "in-review", { moveSource: "engine", allowDirectInReviewMove: true }); + const r2 = await runHoldReleaseSweep(store, noReserveDeps); + expect(r2.released).toContain(stillHeld); + expect((await store.getTask(stillHeld))?.column).toBe("in-progress"); + }); + + it("sweep release into a full column is rejected by the in-txn check (capacity is not a guard, scheduler bypasses guards)", async () => { + await store.updateSettings({ maxConcurrent: 1 } as Parameters[0]); + const occupant = await store.createTask({ description: "occupant" }); + setColumn(store, occupant.id, "in-progress"); + const held = await seedTodoCard(); + + const result = await runHoldReleaseSweep(store, noReserveDeps); + expect(result.released).not.toContain(held); + expect((await store.getTask(held))?.column).toBe("todo"); + }); + + it("capacity release respects cards mid-transitionPending (they hold the slot from commit time)", async () => { + await store.updateSettings({ maxConcurrent: 1 } as Parameters[0]); + // Occupant has committed into in-progress AND is mid-transitionPending — it + // holds the slot; the in-txn count must include it. + const occupant = await store.createTask({ description: "occupant" }); + setColumn(store, occupant.id, "in-progress"); + setTransitionPending(store, occupant.id, "in-progress"); + const held = await seedTodoCard(); + + const result = await runHoldReleaseSweep(store, noReserveDeps); + expect((await store.getTask(held))?.column).toBe("todo"); + expect(result.released).not.toContain(held); + }); + + it("paused and recovery-backoff tasks are skipped exactly as the legacy scheduler", async () => { + await store.updateSettings({ maxConcurrent: 5 } as Parameters[0]); + const paused = await seedTodoCard(); + await store.updateTask(paused, { paused: true }); + const backoff = await seedTodoCard(); + await store.updateTask(backoff, { nextRecoveryAt: new Date(Date.now() + 60_000).toISOString() }); + + const result = await runHoldReleaseSweep(store, { now: () => Date.now() }); + expect(result.released).not.toContain(paused); + expect(result.released).not.toContain(backoff); + expect((await store.getTask(paused))?.column).toBe("todo"); + expect((await store.getTask(backoff))?.column).toBe("todo"); + }); + + it("reservation-first: semaphore exhausted → no commit, card stays held", async () => { + await store.updateSettings({ maxConcurrent: 5 } as Parameters[0]); + const held = await seedTodoCard(); + // reserveSlot returns null (semaphore exhausted) for a processing-column + // release — the move must never be issued. + const deps: HoldReleaseDeps = { + now: () => Date.now(), + reserveSlot: (): SlotReservation | null => null, + }; + const result = await runHoldReleaseSweep(store, deps); + expect(result.released).not.toContain(held); + expect((await store.getTask(held))?.column).toBe("todo"); + }); + + it("reservation is RELEASED when the move rejects on capacity", async () => { + await store.updateSettings({ maxConcurrent: 1 } as Parameters[0]); + const occupant = await store.createTask({ description: "occupant" }); + setColumn(store, occupant.id, "in-progress"); + const held = await seedTodoCard(); + + const releases: number[] = []; + let reserveCount = 0; + const deps: HoldReleaseDeps = { + now: () => Date.now(), + reserveSlot: (): SlotReservation | null => { + reserveCount += 1; + return { release: () => releases.push(1) }; + }, + }; + const result = await runHoldReleaseSweep(store, deps); + expect(result.released).not.toContain(held); + // A reservation was taken (downstream pre-check passed since maxConcurrent + // read-through is evaluated against the snapshot) then released on the + // in-txn capacity rejection. If the pre-check already gated, reserveCount + // may be 0; if it reserved, it must have released exactly once. + if (reserveCount > 0) expect(releases.length).toBe(reserveCount); + }); +}); + +// ── Timer / manual / external-event holds (custom workflows) ────────────────── + +/** A custom workflow whose middle column is a hold with the given release kind. + * Columns: c-intake (intake) → c-hold (hold) → c-run (wip) → c-done (complete). */ +function customHoldWorkflowIr(release: string, holdConfig: Record = {}): WorkflowIr { + return { + version: "v2", + name: "custom-hold", + columns: [ + { id: "c-intake", name: "Intake", traits: [{ trait: "intake" }] }, + { id: "c-hold", name: "Hold", traits: [{ trait: "hold", config: { release, ...holdConfig } }] }, + { id: "c-run", name: "Run", traits: [{ trait: "wip", config: { limit: 5 } }] }, + { id: "c-done", name: "Done", traits: [{ trait: "complete" }] }, + ], + nodes: [ + { id: "start", kind: "start", column: "c-intake" }, + { id: "end", kind: "end", column: "c-done" }, + ], + edges: [{ from: "start", to: "end" }], + } as WorkflowIr; +} + +describe("hold-release sweep — timer / manual / external-event (U6)", () => { + let rootDir = ""; + let store: TaskStore; + + beforeEach(async () => { + rootDir = mkdtempSync(join(tmpdir(), "u6-hold-kinds-")); + 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.useRealTimers(); + }); + + async function seedCustomHold(release: string, holdConfig: Record = {}): Promise { + const def = await store.createWorkflowDefinition({ name: `wf-${release}`, ir: customHoldWorkflowIr(release, holdConfig) }); + const task = await store.createTask({ description: `hold-${release}` }); + setSelection(store, task.id, def.id); + setColumn(store, task.id, "c-hold"); + return task.id; + } + + it("timer release fires at the deadline under fake timers (no real sleeps)", async () => { + vi.useFakeTimers(); + const base = Date.now(); + const id = await seedCustomHold("timer", { durationMs: 10_000 }); + // Re-stamp columnMovedAt to the fake-clock base so the deadline is base+10s. + const db = (store as unknown as { db: { prepare: (s: string) => { run: (...a: unknown[]) => unknown } } }).db; + db.prepare('UPDATE tasks SET "columnMovedAt" = ? WHERE id = ?').run(new Date(base).toISOString(), id); + + // Before the deadline: not released. + const before = await runHoldReleaseSweep(store, { now: () => base + 5_000 }); + expect(before.released).not.toContain(id); + expect((await store.getTask(id))?.column).toBe("c-hold"); + + // At/after the deadline: released into the downstream run column. + const after = await runHoldReleaseSweep(store, { now: () => base + 10_000 }); + expect(after.released).toContain(id); + expect((await store.getTask(id))?.column).toBe("c-run"); + }); + + it("manual hold: the sweep never auto-releases; an explicit promote does", async () => { + const id = await seedCustomHold("manual"); + const swept = await runHoldReleaseSweep(store, { now: () => Date.now() }); + expect(swept.released).not.toContain(id); + expect((await store.getTask(id))?.column).toBe("c-hold"); + + const promoted = await promoteHeldTask(store, id); + expect(promoted.released).toBe(true); + expect(promoted.toColumn).toBe("c-run"); + expect((await store.getTask(id))?.column).toBe("c-run"); + }); + + it("external-event hold: the sweep never auto-releases; an event release does; a stray event on a manual hold is a no-op", async () => { + const eventId = await seedCustomHold("external-event"); + const swept = await runHoldReleaseSweep(store, { now: () => Date.now() }); + expect(swept.released).not.toContain(eventId); + + const released = await releaseHeldTaskByEvent(store, eventId, "webhook:approved"); + expect(released.released).toBe(true); + expect((await store.getTask(eventId))?.column).toBe("c-run"); + + // A manual hold is NOT releasable by an external event. + const manualId = await seedCustomHold("manual"); + const stray = await releaseHeldTaskByEvent(store, manualId, "webhook:approved"); + expect(stray.released).toBe(false); + expect((await store.getTask(manualId))?.column).toBe("c-hold"); + }); +}); + +// ── Dependency gating (KTD-5 + FN-5719 dual-accept) ─────────────────────────── + +/** A custom workflow with a hold(dependency) column. */ +function dependencyHoldWorkflowIr(): WorkflowIr { + return { + version: "v2", + name: "dep-hold", + columns: [ + { id: "d-intake", name: "Intake", traits: [{ trait: "intake" }] }, + { id: "d-hold", name: "Hold", traits: [{ trait: "hold", config: { release: "dependency" } }] }, + { id: "d-run", name: "Run", traits: [{ trait: "wip", config: { limit: 5 } }] }, + { id: "d-done", name: "Done", traits: [{ trait: "complete" }] }, + ], + nodes: [ + { id: "start", kind: "start", column: "d-intake" }, + { id: "end", kind: "end", column: "d-done" }, + ], + edges: [{ from: "start", to: "end" }], + } as WorkflowIr; +} + +/** A custom "producer" workflow whose terminal column carries the complete flag + * under a NON-legacy column id (so the complete-flag path differs from the + * legacy done/in-review/archived signal — used for the dual-accept diff). */ +function completeFlagWorkflowIr(): WorkflowIr { + return { + version: "v2", + name: "producer", + columns: [ + { id: "p-intake", name: "Intake", traits: [{ trait: "intake" }] }, + { id: "p-finished", name: "Finished", traits: [{ trait: "complete" }] }, + ], + nodes: [ + { id: "start", kind: "start", column: "p-intake" }, + { id: "end", kind: "end", column: "p-finished" }, + ], + edges: [{ from: "start", to: "end" }], + } as WorkflowIr; +} + +describe("hold-release sweep — dependency gating (KTD-5)", () => { + let rootDir = ""; + let store: TaskStore; + + beforeEach(async () => { + rootDir = mkdtempSync(join(tmpdir(), "u6-dep-")); + 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.restoreAllMocks(); + }); + + it("a dependency in another workflow's complete-flagged column unblocks the dependent; dual-accept logs a diff on disagreement", async () => { + const auditSpy = vi.spyOn(store, "recordRunAuditEvent"); + + const producerDef = await store.createWorkflowDefinition({ name: "producer", ir: completeFlagWorkflowIr() }); + const dep = await store.createTask({ description: "producer task" }); + setSelection(store, dep.id, producerDef.id); + // Producer NOT yet complete → dependent stays held. + setColumn(store, dep.id, "p-intake"); + + const depHoldDef = await store.createWorkflowDefinition({ name: "dep-hold", ir: dependencyHoldWorkflowIr() }); + const dependent = await store.createTask({ description: "dependent", dependencies: [dep.id] }); + setSelection(store, dependent.id, depHoldDef.id); + setColumn(store, dependent.id, "d-hold"); + + const r1 = await runHoldReleaseSweep(store, { now: () => Date.now() }); + expect(r1.released).not.toContain(dependent.id); + expect((await store.getTask(dependent.id))?.column).toBe("d-hold"); + + // Move the producer into its complete-flagged column (NON-legacy id). + setColumn(store, dep.id, "p-finished"); + auditSpy.mockClear(); + + const r2 = await runHoldReleaseSweep(store, { now: () => Date.now() }); + expect(r2.released).toContain(dependent.id); + expect((await store.getTask(dependent.id))?.column).toBe("d-run"); + + // Dual-accept disagreement: the complete-flag says satisfied, but the legacy + // signal (column p-finished is NOT done/in-review/archived, no marker) says + // NOT satisfied → an audit-diff event was logged. + const diffLogged = auditSpy.mock.calls.some( + (call) => (call[0] as { mutationType?: string })?.mutationType === "merge:dependency-parity-diff", + ); + expect(diffLogged).toBe(true); + }); +}); diff --git a/packages/engine/src/__tests__/scheduler.test.ts b/packages/engine/src/__tests__/scheduler.test.ts index 8889197fa4..7a431a3286 100644 --- a/packages/engine/src/__tests__/scheduler.test.ts +++ b/packages/engine/src/__tests__/scheduler.test.ts @@ -92,6 +92,12 @@ function createMockStore(overrides: Partial = {}): TaskStore { recordRunAuditEvent: vi.fn().mockResolvedValue(undefined), getRootDir: vi.fn().mockReturnValue("/test/project"), getTasksDir: vi.fn().mockReturnValue("/test/project/.fusion/tasks"), + // U6: the hold/release sweep consults workflow selection + completion markers + // when the workflowColumns flag is ON; default mocks keep flag-OFF behavior + // (sweep early-returns before touching these). + getTaskWorkflowSelection: vi.fn().mockReturnValue(undefined), + getWorkflowDefinition: vi.fn().mockResolvedValue(undefined), + getCompletionHandoffAcceptedMarker: vi.fn().mockReturnValue(null), on: vi.fn(), off: vi.fn(), ...overrides, @@ -530,6 +536,50 @@ describe("Scheduler", () => { }); }); + describe("U6 hold/release sweep integration (flag-gated)", () => { + function setupTodoStore(workflowColumns: boolean) { + vi.mocked(existsSync).mockReturnValue(true); + vi.mocked(readFile).mockResolvedValue("# Task\nDo something"); + const todo = createMockTask({ id: "FN-1", column: "todo", dependencies: [] }); + const store = createMockStore({ + listTasks: vi.fn().mockResolvedValue([todo]), + getTask: vi.fn().mockResolvedValue(todo), + getSettings: vi.fn().mockResolvedValue({ + maxConcurrent: 2, + maxWorktrees: 4, + experimentalFeatures: { workflowColumns }, + }), + }); + const scheduler = new Scheduler(store); + (scheduler as unknown as { running: boolean }).running = true; + return { store, scheduler }; + } + + it("flag-ON default-workflow pickup matches flag-OFF: same todo→in-progress dispatch", async () => { + // Flag-OFF baseline: the legacy pull-from-todo loop dispatches the card. + const off = setupTodoStore(false); + await off.scheduler.schedule(); + const offMoves = vi.mocked(off.store.moveTask).mock.calls.map((c) => [c[0], c[1]]); + expect(offMoves).toContainEqual(["FN-1", "in-progress"]); + + // Flag-ON: the sweep runs first (default-workflow todo is a capacity hold), + // then the legacy loop; the net dispatch is the SAME todo→in-progress move. + const on = setupTodoStore(true); + await on.scheduler.schedule(); + const onMoves = vi.mocked(on.store.moveTask).mock.calls.map((c) => [c[0], c[1]]); + expect(onMoves).toContainEqual(["FN-1", "in-progress"]); + }); + + it("flag-OFF: the sweep never issues a scheduler-sourced move (legacy path byte-identical)", 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); + }); + }); + describe("backlog pressure reporter integration", () => { it("invokes reporter from schedule when enabled", async () => { vi.useFakeTimers(); diff --git a/packages/engine/src/concurrency.ts b/packages/engine/src/concurrency.ts index 8d0ba1f1b3..6bdef4a97b 100644 --- a/packages/engine/src/concurrency.ts +++ b/packages/engine/src/concurrency.ts @@ -111,6 +111,22 @@ export class AgentSemaphore { }); } + /** + * Synchronously reserve a slot if one is immediately available, without + * queuing. Returns true (and bumps `activeCount`) when a slot was taken, + * false when the semaphore is full. Used by the U6 hold/release sweep's + * reservation-first ordering (KTD-10): reserve worktree + semaphore BEFORE + * issuing a release move, and {@link release} the reservation if the move + * rejects on capacity. Unlike {@link acquire} it never enqueues a waiter. + */ + tryAcquire(): boolean { + if (this._active < this.limit) { + this._active++; + return true; + } + return false; + } + /** * Release a previously acquired slot and unblock the next waiting caller * (if any). diff --git a/packages/engine/src/hold-release.ts b/packages/engine/src/hold-release.ts new file mode 100644 index 0000000000..46b9cad8c5 --- /dev/null +++ b/packages/engine/src/hold-release.ts @@ -0,0 +1,537 @@ +/** + * Hold/release sweep — the generalized scheduler (U6, KTD-10, R3 behavior half). + * + * Flag-ON, the scheduler's poll becomes a *hold/release sweep*: for each + * workflow in use by live tasks, it finds cards resting at `hold`-trait columns + * and evaluates their release condition: + * + * - `manual` — released ONLY by an explicit {@link promoteHeldTask} + * call (U9's promote endpoint / CLI). The sweep never + * auto-releases a manual hold. + * - `external-event` — released ONLY by {@link releaseHeldTaskByEvent} (a + * webhook/API release, same shape as manual + an event + * tag). + * - `timer` — released when the injected clock passes the hold's + * deadline (`columnMovedAt + durationMs`, or an explicit + * `deadlineAt`). Fake-timer friendly (FN-5048): the clock + * is injected, never `Date.now()` baked in. + * - `capacity` — released when a downstream capacity (`wip`) column has a + * free slot (same counting rules as the in-txn check). + * - `dependency` — released when the card's dependencies are satisfied + * (KTD-5: dependency task's column has the `complete` + * trait flag in ITS resolved workflow; FN-5719 dual-accept + * also honors the legacy completion signal, logging an + * audit-diff when the two disagree). + * + * Eligible cards move via `store.moveTask(..., { moveSource: "scheduler" })`. + * A scheduler move bypasses trait guards (it is substrate-driven) but the in-txn + * capacity check is NOT a guard — it still runs (KTD-10), so two holds racing + * into one slot serialize: exactly one commits, the other rejects with + * `capacity-exhausted` and retries next sweep. + * + * Reservation ordering (KTD-10): for releases into a processing (capacity) + * column, the sweep reserves worktree + semaphore slots BEFORE issuing the move + * and releases the reservation if the move rejects on capacity — a card is never + * moved into a column it cannot actually start in, and a semaphore-exhausted + * interleaving leaves the card held with no commit. + */ + +import { + isWorkflowColumnsEnabled, + resolveColumnCapacity, + resolveColumnFlags, + resolveColumnAdjacency, + TransitionRejectionError, + BUILTIN_CODING_WORKFLOW_IR, + getBuiltinWorkflow, + isBuiltinWorkflowId, + parseWorkflowIr, + type TaskStore, + type Task, + type Settings, + type WorkflowIr, + type WorkflowIrV2, + type WorkflowIrColumn, +} from "@fusion/core"; +import { schedulerLog } from "./logger.js"; + +const DEFAULT_WORKFLOW_POOL_ID = "__default-workflow__"; + +/** A reservation handle returned by {@link HoldReleaseDeps.reserveSlot}. The + * sweep calls `release()` if the subsequent move rejects on capacity. */ +export interface SlotReservation { + release(): void; +} + +/** Injected dependencies so the sweep stays unit-testable with fake timers and + * without real worktree/session allocation. */ +export interface HoldReleaseDeps { + /** Monotonic clock (ms). Inject a fake-timer-driven clock in tests; production + * passes `() => Date.now()`. */ + now: () => number; + /** + * Reserve a worktree + semaphore slot for a card about to be released into a + * processing column (KTD-10 reservation-first). Returns `null` when no slot + * could be reserved (e.g. semaphore exhausted) — the sweep then leaves the + * card held without issuing a move. Returns a {@link SlotReservation} whose + * `release()` the sweep calls if the move rejects on capacity. + * + * Optional: when absent, releases into processing columns proceed without a + * reservation (the in-txn capacity check still arbitrates), which is the + * default-workflow legacy parity path where the scheduler dispatch loop owns + * worktree allocation via `allocateWorktree`. + */ + reserveSlot?: (task: Task, targetColumn: string) => SlotReservation | null; + /** Allocate a worktree path for a release into a processing column (passed + * through to `moveTask`'s `allocateWorktree`). */ + allocateWorktree?: (task: Task, reservedNames: Set) => string | null; +} + +/** Outcome of one sweep pass (for tests + observability). */ +export interface HoldReleaseResult { + released: string[]; + /** taskId → reason it stayed held this pass. */ + held: Array<{ taskId: string; reason: string }>; +} + +// ── Workflow IR resolution (read-only, mirrors store + merge-trait) ─────────── + +async function resolveTaskWorkflowIr(store: TaskStore, taskId: string): Promise { + let workflowId: string | undefined; + try { + workflowId = store.getTaskWorkflowSelection(taskId)?.workflowId; + } catch { + workflowId = undefined; + } + if (!workflowId) return BUILTIN_CODING_WORKFLOW_IR; + if (isBuiltinWorkflowId(workflowId)) { + const builtin = getBuiltinWorkflow(workflowId); + return builtin?.ir ?? BUILTIN_CODING_WORKFLOW_IR; + } + try { + const def = await store.getWorkflowDefinition(workflowId); + if (!def) return BUILTIN_CODING_WORKFLOW_IR; + return typeof def.ir === "string" ? parseWorkflowIr(def.ir) : def.ir; + } catch { + return BUILTIN_CODING_WORKFLOW_IR; + } +} + +function effectiveWorkflowId(store: TaskStore, taskId: string): string { + try { + return store.getTaskWorkflowSelection(taskId)?.workflowId ?? DEFAULT_WORKFLOW_POOL_ID; + } catch { + return DEFAULT_WORKFLOW_POOL_ID; + } +} + +function findColumn(ir: WorkflowIr, columnId: string): WorkflowIrColumn | undefined { + if (ir.version !== "v2") return undefined; + return (ir as WorkflowIrV2).columns.find((c) => c.id === columnId); +} + +/** The hold trait config on a column, if any. */ +function resolveHoldConfig(column: WorkflowIrColumn): Record | undefined { + const flags = resolveColumnFlags(column); + if (!flags.hold) return undefined; + const ct = column.traits.find((t) => t.trait === "hold"); + return ct?.config ?? {}; +} + +/** True when the card currently rests at a hold column. */ +function isHeldTask(ir: WorkflowIr, task: Task): boolean { + const column = findColumn(ir, task.column); + if (!column) return false; + return resolveColumnFlags(column).hold === true; +} + +/** + * Resolve the release target column for a held card. + * + * For `capacity` holds, the target is the nearest downstream column (by the + * workflow's column adjacency, breadth-first from the hold column) that carries + * a capacity (`wip`) trait — for the default workflow this is `in-progress`. + * For other release kinds the target is the first adjacency neighbor that is not + * the hold column itself (the forward step out of the hold). + */ +function resolveReleaseTarget(ir: WorkflowIr, fromColumn: string, preferCapacity: boolean): string | undefined { + const v2 = ir as WorkflowIrV2; + const orderedIds = Array.isArray(v2.columns) ? v2.columns.map((c) => c.id) : []; + const fromIdx = orderedIds.indexOf(fromColumn); + const adjacency = resolveColumnAdjacency(ir); + const neighbors = adjacency.get(fromColumn) ?? []; + + if (preferCapacity) { + // Walk FORWARD in declared order for the nearest capacity-bearing column; + // the hold releases downstream, never backward. + for (let i = fromIdx + 1; i < orderedIds.length; i++) { + const col = findColumn(ir, orderedIds[i]); + if (col && resolveColumnFlags(col).countsTowardWip && neighbors.includes(orderedIds[i])) { + return orderedIds[i]; + } + } + // No directly-adjacent capacity column: fall back to the nearest forward + // capacity column reachable via adjacency BFS. + const seen = new Set([fromColumn]); + const queue = [...neighbors]; + while (queue.length > 0) { + const candidate = queue.shift()!; + if (seen.has(candidate)) continue; + seen.add(candidate); + const col = findColumn(ir, candidate); + if (col && resolveColumnFlags(col).countsTowardWip) return candidate; + for (const next of adjacency.get(candidate) ?? []) { + if (!seen.has(next)) queue.push(next); + } + } + } + + // Forward neighbor (declared-order next) if it is adjacent; else any neighbor + // that is forward in declared order; else the first neighbor. + const forwardId = fromIdx >= 0 ? orderedIds[fromIdx + 1] : undefined; + if (forwardId && neighbors.includes(forwardId)) return forwardId; + const forwardNeighbor = neighbors.find((n) => orderedIds.indexOf(n) > fromIdx); + if (forwardNeighbor) return forwardNeighbor; + return neighbors.find((n) => n !== fromColumn); +} + +// ── Dependency satisfaction (KTD-5 + FN-5719 dual-accept) ───────────────────── + +/** Legacy completion signal: dependency's column is a terminal/handoff column. */ +function legacyDependencySatisfied(dep: Task): boolean { + return dep.column === "done" || dep.column === "in-review" || dep.column === "archived"; +} + +/** + * KTD-5 dependency satisfaction: the dependency task's current column has the + * `complete` trait flag in ITS resolved workflow. Dual-accept (FN-5719): the + * legacy completion signal (done/in-review/archived column, or an accepted + * completion-handoff marker) is also honored; when the two disagree an + * audit-diff event is logged. + */ +async function dependencySatisfied(store: TaskStore, dep: Task): Promise { + const ir = await resolveTaskWorkflowIr(store, dep.id); + const column = findColumn(ir, dep.column); + const completeFlag = column ? resolveColumnFlags(column).complete === true : false; + + let markerAccepted = false; + try { + markerAccepted = store.getCompletionHandoffAcceptedMarker(dep.id) !== null; + } catch { + markerAccepted = false; + } + const legacy = legacyDependencySatisfied(dep) || markerAccepted; + + if (completeFlag !== legacy) { + try { + void store.recordRunAuditEvent?.({ + taskId: dep.id, + agentId: "scheduler", + runId: `hold-release:${dep.id}`, + domain: "database", + mutationType: "merge:dependency-parity-diff", + target: dep.id, + metadata: { + depId: dep.id, + completeFlagResult: completeFlag, + legacyResult: legacy, + source: "hold-release.dependency", + }, + }); + } catch { + // Audit is best-effort. + } + } + // Dual-accept: satisfied if EITHER signal says so (the dual-accept window + // closes at graduation per U12; until then both are accepted). + return completeFlag || legacy; +} + +async function allDependenciesSatisfied(store: TaskStore, task: Task, allTasks: Task[]): Promise { + for (const depId of task.dependencies ?? []) { + const dep = allTasks.find((t) => t.id === depId); + if (!dep) continue; // missing dep does not block (matches scheduler posture) + if (!(await dependencySatisfied(store, dep))) return false; + } + return true; +} + +// ── Timer release ───────────────────────────────────────────────────────────── + +/** Resolve the timer deadline (ms epoch) for a timer hold, or `undefined` if not + * resolvable. Supports an explicit `deadlineAt` (ISO or ms) or a relative + * `durationMs`/`timerMs` measured from `columnMovedAt`. */ +function resolveTimerDeadline(holdConfig: Record, task: Task): number | undefined { + const deadlineAt = holdConfig.deadlineAt; + if (typeof deadlineAt === "number" && Number.isFinite(deadlineAt)) return deadlineAt; + if (typeof deadlineAt === "string") { + const parsed = Date.parse(deadlineAt); + if (Number.isFinite(parsed)) return parsed; + } + const duration = + (typeof holdConfig.durationMs === "number" ? holdConfig.durationMs : undefined) ?? + (typeof holdConfig.timerMs === "number" ? holdConfig.timerMs : undefined); + if (typeof duration === "number" && Number.isFinite(duration)) { + const base = Date.parse(task.columnMovedAt ?? task.createdAt); + if (Number.isFinite(base)) return base + duration; + } + return undefined; +} + +// ── Capacity availability (same counting rule as the in-txn check) ──────────── + +/** + * Count cards occupying the (workflow, column) capacity slot from a task + * snapshot, mirroring the store's in-txn count: cards in the column now, plus + * (when countPending) cards mid-`transitionPending` targeting it, scoped to the + * SAME effective workflow. This is the sweep's *pre-check* — the authoritative + * arbitration is still the in-txn check, which rejects a losing racer. + */ +function countCapacitySlot( + store: TaskStore, + allTasks: Task[], + targetColumn: string, + workflowId: string, + countPending: boolean, +): number { + let count = 0; + for (const t of allTasks) { + if (effectiveWorkflowId(store, t.id) !== workflowId) continue; + if (t.column === targetColumn) { + count += 1; + continue; + } + if (!countPending) continue; + const tp = (t as Task & { transitionPending?: { toColumn?: string } | null }).transitionPending; + if (tp && typeof tp === "object" && tp.toColumn === targetColumn) count += 1; + } + return count; +} + +// ── The sweep ───────────────────────────────────────────────────────────────── + +/** + * Run one hold/release sweep pass. No-op (returns empty) when the workflowColumns + * flag is OFF — flag-OFF scheduler behavior is byte-identical (the legacy + * pull-from-todo loop is untouched). + */ +export async function runHoldReleaseSweep( + store: TaskStore, + deps: HoldReleaseDeps, +): Promise { + const result: HoldReleaseResult = { released: [], held: [] }; + + const settings = await store.getSettings(); + if (!isWorkflowColumnsEnabled(settings)) return result; + + const allTasks = await store.listTasks({ includeArchived: false }); + + for (const task of allTasks) { + // Skip paused / recovery-backoff tasks exactly as the legacy scheduler does. + if (task.paused || task.userPaused) { + continue; + } + if (task.nextRecoveryAt && Date.parse(task.nextRecoveryAt) > deps.now()) { + continue; + } + + const ir = await resolveTaskWorkflowIr(store, task.id); + if (!isHeldTask(ir, task)) continue; + + const column = findColumn(ir, task.column); + const holdConfig = column ? resolveHoldConfig(column) : undefined; + if (!column || !holdConfig) continue; + const release = typeof holdConfig.release === "string" ? holdConfig.release : "manual"; + + // manual / external-event are NEVER auto-released by the sweep. + if (release === "manual" || release === "external-event") { + result.held.push({ taskId: task.id, reason: `${release}-only` }); + continue; + } + + let shouldRelease = false; + if (release === "timer") { + const deadline = resolveTimerDeadline(holdConfig, task); + shouldRelease = deadline !== undefined && deps.now() >= deadline; + if (!shouldRelease) { + result.held.push({ taskId: task.id, reason: "timer-not-elapsed" }); + continue; + } + } else if (release === "dependency") { + shouldRelease = await allDependenciesSatisfied(store, task, allTasks); + if (!shouldRelease) { + result.held.push({ taskId: task.id, reason: "deps-unsatisfied" }); + continue; + } + } else if (release === "capacity") { + // Capacity holds release into the nearest downstream capacity column when a + // slot is free (pre-check); the in-txn check is the authority. + const target = resolveReleaseTarget(ir, task.column, true); + if (!target) { + result.held.push({ taskId: task.id, reason: "no-downstream-capacity-column" }); + continue; + } + const capacity = resolveColumnCapacity(ir, target, settings); + if (capacity.hasCapacity && Number.isFinite(capacity.limit)) { + const workflowId = effectiveWorkflowId(store, task.id); + const occupants = countCapacitySlot(store, allTasks, target, workflowId, capacity.countPending); + if (occupants >= capacity.limit) { + result.held.push({ taskId: task.id, reason: "downstream-full" }); + continue; + } + } + shouldRelease = true; + } + + if (!shouldRelease) continue; + + const target = resolveReleaseTarget(ir, task.column, release === "capacity"); + if (!target) { + result.held.push({ taskId: task.id, reason: "no-release-target" }); + continue; + } + + const released = await issueRelease(store, deps, task, target, ir); + if (released) { + result.released.push(task.id); + } else { + result.held.push({ taskId: task.id, reason: "move-rejected-or-no-slot" }); + } + } + + return result; +} + +/** + * Issue a single release move (`moveSource: "scheduler"`). For releases into a + * processing (capacity) column the reservation-first ordering (KTD-10) reserves + * worktree + semaphore before the move and releases the reservation if the move + * rejects on capacity. Returns true on a committed move, false otherwise (the + * card stays held). + */ +async function issueRelease( + store: TaskStore, + deps: HoldReleaseDeps, + task: Task, + target: string, + ir: WorkflowIr, +): Promise { + const targetColumn = findColumn(ir, target); + const targetIsProcessing = targetColumn ? resolveColumnFlags(targetColumn).countsTowardWip === true : false; + + let reservation: SlotReservation | null = null; + if (targetIsProcessing && deps.reserveSlot) { + reservation = deps.reserveSlot(task, target); + if (!reservation) { + // Semaphore/worktree exhausted — reservation-first means no move at all. + schedulerLog.log(`Hold release for ${task.id} deferred — no reservable slot for ${target}`); + return false; + } + } + + try { + await store.moveTask(task.id, target, { + moveSource: "scheduler", + allocateWorktree: + targetIsProcessing && deps.allocateWorktree + ? (reservedNames) => deps.allocateWorktree!(task, reservedNames) + : undefined, + }); + return true; + } catch (error) { + if (error instanceof TransitionRejectionError && error.rejection.code === "capacity-exhausted") { + // Lost the in-txn race for the slot — release the reservation, stay held. + reservation?.release(); + schedulerLog.log(`Hold release for ${task.id} rejected on capacity for ${target} — staying held`); + return false; + } + // Any other failure: release the reservation and let the card stay held. + reservation?.release(); + schedulerLog.warn( + `Hold release for ${task.id} into ${target} failed: ${error instanceof Error ? error.message : String(error)}`, + ); + return false; + } +} + +// ── Explicit (manual / external-event) releases ─────────────────────────────── + +/** + * Manually promote a held card out of its hold column (U9's promote endpoint / + * CLI calls this). Releases regardless of the hold's release kind — a manual + * promote is the explicit operator action the `manual` release kind waits for, + * and it is also accepted for other kinds as an operator override. The move + * still serializes through the in-txn capacity check (KTD-10): a promote into a + * full column rejects with `capacity-exhausted`, surfaced to the caller. + */ +export async function promoteHeldTask( + store: TaskStore, + taskId: string, + deps: Pick = {}, +): Promise<{ released: boolean; toColumn?: string; rejection?: string }> { + const task = await store.getTask(taskId); + if (!task) return { released: false, rejection: "task-not-found" }; + + const ir = await resolveTaskWorkflowIr(store, taskId); + if (!isHeldTask(ir, task)) { + return { released: false, rejection: "not-held" }; + } + const target = resolveReleaseTarget(ir, task.column, true); + if (!target) return { released: false, rejection: "no-release-target" }; + + const released = await issueRelease( + store, + { now: () => Date.now(), reserveSlot: deps.reserveSlot, allocateWorktree: deps.allocateWorktree }, + task, + target, + ir, + ); + return released ? { released: true, toColumn: target } : { released: false, rejection: "capacity-exhausted-or-no-slot" }; +} + +/** + * Release a held card on an external event (webhook/API). Same shape as + * {@link promoteHeldTask} plus an `eventTag` recorded in the audit; only acts on + * `external-event` holds (a no-op otherwise so a stray webhook can't release a + * manual/timer/capacity hold). + */ +export async function releaseHeldTaskByEvent( + store: TaskStore, + taskId: string, + eventTag: string, + deps: Pick = {}, +): Promise<{ released: boolean; toColumn?: string; rejection?: string }> { + const task = await store.getTask(taskId); + if (!task) return { released: false, rejection: "task-not-found" }; + + const ir = await resolveTaskWorkflowIr(store, taskId); + const column = findColumn(ir, task.column); + const holdConfig = column ? resolveHoldConfig(column) : undefined; + if (!column || !holdConfig || holdConfig.release !== "external-event") { + return { released: false, rejection: "not-external-event-hold" }; + } + try { + void store.recordRunAuditEvent?.({ + taskId, + agentId: "scheduler", + runId: `hold-release:event:${taskId}`, + domain: "database", + mutationType: "task:hold-release-event", + target: taskId, + metadata: { eventTag, fromColumn: task.column }, + }); + } catch { + // best-effort + } + const target = resolveReleaseTarget(ir, task.column, true); + if (!target) return { released: false, rejection: "no-release-target" }; + + const released = await issueRelease( + store, + { now: () => Date.now(), reserveSlot: deps.reserveSlot, allocateWorktree: deps.allocateWorktree }, + task, + target, + ir, + ); + return released ? { released: true, toColumn: target } : { released: false, rejection: "capacity-exhausted-or-no-slot" }; +} diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index c0c4b3b93c..eccf5c3ccb 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -32,6 +32,8 @@ import type { AutoClaimSnapshotManager } from "./auto-claim-snapshot.js"; import { StaleTaskReporter } from "./stale-task-reporter.js"; import { BacklogPressureReporter } from "./backlog-pressure-reporter.js"; import { createRunAuditor, generateSyntheticRunId } from "./run-audit.js"; +import { isWorkflowColumnsEnabled } from "@fusion/core"; +import { runHoldReleaseSweep, type SlotReservation } from "./hold-release.js"; /** * Check whether two sets of file scope paths overlap. @@ -278,6 +280,20 @@ interface ConcurrencyGateSnapshot { slack: number; } +/** + * U6 (KTD-10): a per-(workflow, column) capacity gate, the generalization of the + * three legacy gates to workflow-defined WIP columns. Additive — the three-gate + * report shape (maxConcurrent/maxWorktrees/semaphore) is preserved verbatim; this + * is an optional extra field populated only when the workflowColumns flag is ON. + */ +interface PerColumnCapacityGate { + workflowId: string; + columnId: string; + used: number; + limit: number; + slack: number; +} + interface ConcurrencyGateDiagnostic { available: number; bindingGates: ConcurrencyGateName[]; @@ -289,6 +305,9 @@ interface ConcurrencyGateDiagnostic { maxWorktrees: string[]; semaphore?: string[]; }; + /** U6: additive per-column capacity gates (flag-ON only; omitted otherwise so + * the legacy three-gate report shape is byte-identical when the flag is OFF). */ + perColumnGates?: PerColumnCapacityGate[]; } function computeConcurrencyGateDiagnostic(params: { @@ -299,6 +318,9 @@ function computeConcurrencyGateDiagnostic(params: { semaphore?: AgentSemaphore; inProgressTaskIds: string[]; available: number; + /** U6: additive per-column capacity gates (flag-ON only). Omitted → the legacy + * three-gate report is byte-identical. */ + perColumnGates?: PerColumnCapacityGate[]; }): ConcurrencyGateDiagnostic { const maxConcurrentGate: ConcurrencyGateSnapshot = { used: params.agentSlots, @@ -334,6 +356,8 @@ function computeConcurrencyGateDiagnostic(params: { maxWorktrees: [...params.inProgressTaskIds], semaphore: semaphoreGate ? [...params.inProgressTaskIds] : undefined, }, + // U6: additive only — present when flag-ON, omitted otherwise. + ...(params.perColumnGates ? { perColumnGates: params.perColumnGates } : {}), }; } @@ -938,6 +962,8 @@ export class Scheduler { semaphore: diagnostic.semaphoreGate, holders: diagnostic.holders, available: diagnostic.available, + // U6: additive per-column capacity gates (present only flag-ON). + ...(diagnostic.perColumnGates ? { perColumnGates: diagnostic.perColumnGates } : {}), }, }); } catch (error) { @@ -1171,6 +1197,18 @@ export class Scheduler { } this.wasEnginePaused = false; + // ── U6: hold/release sweep (flag-ON only) ────────────────────────────── + // Flag OFF: this is skipped entirely — the legacy pull-from-todo loop + // below is byte-identical. Flag ON: the sweep evaluates hold-column + // release conditions (manual/timer/capacity/dependency/external-event) and + // releases eligible cards via moveSource:"scheduler", serializing through + // the in-txn capacity check. For the DEFAULT workflow the legacy loop below + // still drives todo→in-progress pickup (parity); the sweep adds custom- + // workflow hold handling and the generalized capacity-release path. + if (isWorkflowColumnsEnabled(settings)) { + await this.runHoldReleaseSweepPass(); + } + // Count only in-progress tasks toward the worktree limit. // In-review tasks with worktrees are idle (waiting to merge) and // should not block new tasks from starting. @@ -1207,6 +1245,19 @@ export class Scheduler { semaphoreAvailable, ); const inProgressTaskIds = inProgress.map((task) => task.id); + // U6 (KTD-10): when the workflowColumns flag is ON, report the default + // workflow's in-progress capacity as a per-column gate — the generalization + // of the legacy maxConcurrent gate (which reads through to the same value). + // Additive: omitted flag-OFF so the three-gate report shape is unchanged. + const perColumnGates = isWorkflowColumnsEnabled(settings) + ? [{ + workflowId: "__default-workflow__", + columnId: "in-progress", + used: agentSlots, + limit: maxConcurrent, + slack: maxConcurrent - agentSlots, + }] + : undefined; const concurrencyGateDiagnostic = computeConcurrencyGateDiagnostic({ agentSlots, maxConcurrent, @@ -1215,6 +1266,7 @@ export class Scheduler { semaphore: this.options.semaphore, inProgressTaskIds, available, + perColumnGates, }); if (available <= 0) return; @@ -1892,6 +1944,37 @@ export class Scheduler { } } + /** + * U6: run one hold/release sweep pass, wiring the scheduler's semaphore + + * worktree allocation into the reservation-first ordering (KTD-10). Failures + * are isolated so a sweep error never breaks the scheduling pass. + */ + private async runHoldReleaseSweepPass(): Promise { + try { + await runHoldReleaseSweep(this.store, { + now: () => Date.now(), + reserveSlot: this.options.semaphore + ? (): SlotReservation | null => { + const sem = this.options.semaphore!; + if (!sem.tryAcquire()) return null; + let released = false; + return { + release: () => { + if (released) return; + released = true; + sem.release(); + }, + }; + } + : undefined, + allocateWorktree: (task, reservedNames) => + planTaskWorktreePath(task, this.store.getRootDir(), undefined, reservedNames, {}), + }); + } catch (error) { + schedulerLog.error("Hold/release sweep failed:", error); + } + } + /** * Handle a mission-linked task column move. * Keeps feature state synchronized with task columns across the full task