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

fix(engine): harden workflow runtime cutover

Browse Source
This commit is contained in:
gsxdsm
2026-06-22 21:45:05 -07:00
parent bf3276295c
commit 65c4dc5438
14 changed files with 1021 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
.changeset/workflow-runtime-cutover.md Normal file
View File

@@ -0,0 +1,9 @@
---
"@runfusion/fusion": patch
---
Graduate workflow columns and the workflow graph executor to the default runtime path.
Upgrade notes: stale persisted `experimentalFeatures.workflowColumns` and `experimentalFeatures.workflowGraphExecutor` values are ignored by the engine, so prior installs keep dispatching tasks through the workflow runtime after upgrade. `workflowInterpreterDualObserve` remains an internal diagnostic and defaults off.
If an upgraded project appears stalled, treat `todo` tasks with unmet dependencies, `paused`/`userPaused`, active checkout leases, unavailable assigned nodes, or file-scope overlap as intentionally parked. Eligible `todo` tasks without those blockers should be picked up by the workflow scheduler; eligible `in-progress` rows without a live executor are recovered through the normal orphan-resume/self-healing path. The old Experimental toggles are no longer a rollback switch; use a source rollback/downgrade to the previous release if the workflow runtime itself must be reverted.

191
docs/plans/2026-06-23-002-workflow-runtime-cutover-hardening-plan.md Normal file
View File

@@ -0,0 +1,191 @@
---
title: Workflow Runtime Cutover Hardening
type: fix
date: 2026-06-23
source_plan: docs/plans/2026-06-23-001-fix-workflow-runtime-cutover-plan.md
---
# Workflow Runtime Cutover Hardening
## Summary
Make workflow columns and graph execution safe as the default runtime by hardening the scheduler hold/release path, preserving executor recovery semantics, graduating stale workflow flags out of Experimental settings, and removing dead legacy dispatch only after reachability and rollback safety are proven.
---
## Problem Frame
The initial workflow runtime cutover made the workflow paths default, but review found the new path was not yet equivalent to legacy scheduler and executor invariants. The highest-risk gaps are capacity handling in the hold/release scheduler path, graph failure handling that can overwrite inner executor recovery, missing replacement tests after legacy test deletion, and incomplete flag graduation. This plan supersedes the earlier cutover plan with the document-review findings folded into executable scope.
---
## Requirements
**Branch and rollback**
- R1. Keep unrelated dashboard/cosmetic changes out of the workflow cutover PR.
- R2. Preserve rollback safety by keeping the cutover on an isolated branch and staging irreversible legacy-dispatch deletion behind reachability tests and validation evidence.
- R3. Users upgrading from a prior version must not have tasks stall because of stale workflow flag values, legacy columns, existing `todo`/`in-progress`/`in-review` rows, or persisted worktree/lease state.
- R4. The first published cutover release must have a verified operator rollback or downgrade path, including support guidance for users whose eligible tasks stop progressing after upgrade.
**Scheduler runtime**
- R5. The workflow hold/release scheduler path must preserve dispatch gates for dependencies, blocked missions, filesystem/spec staleness, pause states, checkout leases, node routing, permanent-agent availability, file-scope overlap, dispatch oscillation, `maxWorktrees`, `maxConcurrent`, and shared semaphore pressure.
- R6. Capacity failure must leave tasks queued and must not log `Starting`, clear status, or call `onSchedule` before all reservation checks pass.
- R7. Scheduler handoff failures after hold creation, release, `onSchedule`, or executor invocation must leave tasks recoverable and must not leak stale holds.
**Executor runtime**
- R8. `TaskExecutor.execute()` must use graph-default behavior even when stale persisted `workflowGraphExecutor=false` exists.
- R9. Graph-default execution must preserve legacy recovery semantics: inner executor requeues, mismatched store-row protection, pause aborts, duplicate execute protection, worktree liveness recovery, and no-`fn_task_done` handling.
**Flag graduation**
- R10. Workflow columns and workflow graph executor must no longer appear as user-facing Experimental kill switches.
- R11. Stale persisted workflow flag values must be ignored by runtime helpers and must not route old installations back to legacy behavior.
- R12. Hidden graduated workflow keys must have deterministic Settings save behavior when users save unrelated settings after upgrade.
- R13. Stale persisted `workflowInterpreterDualObserve=true` must either be ignored after graduation or remain controllable through a non-user operator mechanism; it must not stay enabled invisibly with no way to disable it.
**Test and review gate**
- R14. Every test referenced by `packages/engine/vitest.config.ts` must be tracked and committed.
- R15. Deleted legacy scheduler/executor tests must be replaced by targeted workflow-path coverage for the same live invariants before the PR removes the old files.
- R16. The branch must pass targeted engine/core tests, lint, typecheck, root test, build, and a follow-up `compound-engineering:ce-code-review`.
---
## Key Technical Decisions
- KTD1. Scheduler reservations stay non-mutating until all gates pass. The hold/release callback can inspect `maxConcurrent`, `maxWorktrees`, and `AgentSemaphore.availableCount`, but executor still owns the actual semaphore acquire so the scheduler does not double-acquire a slot.
- KTD2. Capacity tests must include race-shaped cases. Single-gate tests are not enough; coverage must prove same-sweep held-task releases under `maxConcurrent=1`, `maxWorktrees=1`, and saturated semaphore conditions.
- KTD3. Graph failure handling must treat the originally dispatched task ID as authoritative. If a minimal or stale store returns a different row from `getTask(task.id)`, graph recovery must preserve the inner executor result instead of mutating the wrong task.
- KTD4. Legacy dispatcher removal is last. The PR may harden graph-default and hold/release first; deleting unreachable legacy scheduler code happens only after stale-flag reachability and replacement tests prove no live entrypoint still depends on it.
- KTD5. Published rollback must be operational, not only source-control based. Before legacy deletion ships, the plan must prove either an operator-only fallback can restore scheduling without re-exposing Experimental user switches, or a documented downgrade/revert path works against cutover-era settings and task rows.
- KTD6. Flag graduation is a runtime and UI change. Defaults, helper semantics, Settings UI, and Settings save payloads must agree: stale persisted values are tolerated, but users cannot toggle these default runtime paths off from Experimental settings.
- KTD7. Upgrade safety is proved at persisted-state boundaries. Tests should use frozen prior-version fixtures or generate state with the previous released storage code, then exercise the real scheduler/executor entrypoints so task progression is verified after upgrade rather than inferred from helper behavior.
---
## Implementation Units
### U1. Branch Isolation And Diff Hygiene
- **Goal:** Keep the PR rollback boundary clean and exclude unrelated cosmetic work.
- **Files:** `docs/plans/2026-06-23-002-workflow-runtime-cutover-hardening-plan.md`
- **Approach:** Base the PR branch on `origin/main`, carry only workflow runtime, flag graduation, test, plan, and release metadata changes, and verify the diff does not include unrelated dashboard cosmetic files.
- **Test scenarios:** `git diff --name-only origin/main...HEAD` excludes cosmetic files such as `packages/dashboard/app/components/ScriptsModal.css` and `packages/dashboard/app/components/__tests__/ScheduledTasksModal.test.tsx`.
- **Verification:** Inspect branch history and PR file list before opening the PR.
### U2. Scheduler Hold/Release Dispatch Equivalence
- **Goal:** Make the workflow hold/release scheduler path equivalent to legacy live dispatch gates.
- **Files:** `packages/engine/src/scheduler.ts`, `packages/engine/src/hold-release.ts`, `packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts`, `packages/engine/vitest.config.ts`
- **Approach:** Move or share all live pre-dispatch checks into the hold/release reservation path. Run dependency, mission, filesystem/spec, pause, lease, node, permanent-agent, overlap, oscillation, and capacity checks before any status-clearing update or `Starting` log. Preserve `onSchedule` as a post-release effect only.
- **Test scenarios:** Cover dependency blocking, blocked mission, filesystem invalidation, stale prompt, global/engine/user pause, stale lease recovery failure, node validation block/fallback/handoff, no permanent executor, overlap lease, oscillation auto-pause, `maxConcurrent=1`, `maxWorktrees=1`, saturated semaphore, same-sweep multi-task race, prior-version stale workflow settings, pre-existing `todo` tasks, successful post-release `onSchedule`, and injected failures after hold creation, after release before executor invocation, after `onSchedule`, and after executor invocation throws before semaphore acquisition.
- **Verification:** `pnpm --filter @fusion/engine exec vitest run src/__tests__/scheduler-workflow-cutover.test.ts`
### U3. Executor Graph Entry And Recovery Equivalence
- **Goal:** Prove the production `TaskExecutor.execute()` entrypoint preserves legacy recovery behavior under graph-default execution.
- **Files:** `packages/engine/src/executor.ts`, `packages/engine/src/__tests__/workflow-graph-task-runner.test.ts`, `packages/engine/src/__tests__/executor-worktree.test.ts`, `packages/engine/src/__tests__/restart.integration.test.ts`, tests under `packages/engine/src/__tests__/reliability-interactions/`
- **Approach:** Keep the original dispatched task identity through graph runner setup and graph failure handling. Preserve inner executor recovery when the execute node requeues to `todo`. Ensure `prepareWorktree` returns an existing task worktree or an empty string, never the repo root.
- **Test scenarios:** Cover stale `workflowGraphExecutor=false`, unmet dependency pre-graph requeue, satisfied dependency graph dispatch, mismatched live row before runner start, mismatched live row in failure handling, inner executor `todo` requeue preservation, duplicate execute locking, pause/user-pause/global-pause abort behavior, worktree liveness requeue, and no-`fn_task_done` recovery final column parity.
- **Verification:** `pnpm --filter @fusion/engine exec vitest run src/__tests__/workflow-graph-task-runner.test.ts src/__tests__/executor-worktree.test.ts src/__tests__/restart.integration.test.ts src/__tests__/reliability-interactions/executor-liveness-gate.test.ts src/__tests__/reliability-interactions/executor-no-task-done-vs-worktree-reclaim.test.ts`
### U4. Workflow Flag Graduation
- **Goal:** Remove user-facing workflow kill switches while preserving stale persisted value compatibility.
- **Files:** `packages/core/src/workflow-columns-settings.ts`, `packages/core/src/experimental-features.ts`, `packages/core/src/settings-schema.ts`, `packages/core/src/__tests__/settings-defaults.test.ts`, `packages/core/src/__tests__/workflow-cutover.test.ts`, `packages/dashboard/app/components/settings/sections/ExperimentalSection.tsx`, `packages/dashboard/app/components/__tests__/WorkflowNodeEditor.test.tsx`, `packages/dashboard/app/components/__tests__/SettingsModal.test.tsx`, `packages/dashboard/app/__tests__/settings-sections.test.tsx`
- **Approach:** Keep runtime helpers always enabling workflow columns and graph execution regardless of stale false persisted values. Remove graduated workflow flags from defaults and from Experimental settings UI. Decide and test the Settings save-payload behavior for hidden graduated keys. Keep dual-observe off by default and hidden only if stale true values are ignored or an operator-only control remains.
- **Test scenarios:** Core settings defaults omit `workflowColumns` and `workflowGraphExecutor`; stale false values still produce enabled runtime helpers; upgraded prior-version settings retain harmless unknown experimental entries without disabling workflow runtime; Settings UI does not render workflow columns, workflow graph executor, or dual-observe controls in Experimental settings; opening Settings with stale hidden workflow keys, changing an unrelated Experimental toggle, and saving follows the documented payload behavior for those hidden keys; stale `workflowInterpreterDualObserve=true` is ignored or remains controllable through an operator-only mechanism.
- **Verification:** `pnpm --filter @fusion/core exec vitest run src/__tests__/settings-defaults.test.ts src/__tests__/workflow-cutover.test.ts` plus targeted dashboard settings tests.
### U5. Upgrade Progression Coverage
- **Goal:** Prove existing users' task queues keep progressing after upgrading into the cutover.
- **Files:** `packages/core/src/__tests__/workflow-cutover.test.ts`, `packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts`, targeted executor/reliability tests under `packages/engine/src/__tests__/`
- **Approach:** Seed representative prior-version persisted state from frozen fixtures for the previous released version or by generating fixtures with that version's storage code. Include stale experimental flags, legacy/custom workflow columns where applicable, existing `todo` rows, existing `in-progress` rows with worktrees, existing `in-review` rows, paused/user-paused rows, and checked-out rows with lease metadata. Run real scheduler/executor entrypoints and assert dispatchable tasks continue while intentionally paused/blocked tasks remain parked for the correct reason.
- **Test scenarios:** Upgraded `todo` tasks dispatch through hold/release; upgraded `in-progress` tasks are not duplicated or stolen; upgraded `in-review` tasks continue review/merge handling; paused/user-paused tasks do not auto-resume; stale leases follow existing recovery policy; stale workflow flags do not prevent any eligible task from progressing.
- **Verification:** Include these scenarios in `scheduler-workflow-cutover.test.ts`, `workflow-cutover.test.ts`, or focused reliability tests before deleting legacy dispatcher coverage.
### U6. Release Rollback Proof
- **Goal:** Prove users have a usable post-release recovery path if the cutover stalls eligible task progression.
- **Files:** `docs/plans/2026-06-23-002-workflow-runtime-cutover-hardening-plan.md`, `.changeset/workflow-runtime-cutover.md`, and rollback/downgrade tests or scripts if added
- **Approach:** Before legacy deletion ships, prove one rollback path: an operator-only runtime fallback that restores scheduling without re-exposing user-facing Experimental controls, or a documented downgrade/revert procedure that works against cutover-era settings and task rows. The support guidance should tell users how to identify intentionally parked tasks versus eligible tasks that should progress.
- **Test scenarios:** Seed cutover-era settings and task rows, run the chosen rollback/downgrade procedure, and assert eligible tasks resume scheduling while paused/dependency-blocked tasks remain correctly parked.
- **Verification:** Rollback proof is documented in the patch changeset Upgrade Notes or a linked support note before the PR is opened.
### U7. Legacy Dispatch Deletion And Reachability Proof
- **Goal:** Remove the unreachable legacy scheduler dispatcher without deleting a path that stale settings can still reach.
- **Files:** `packages/engine/src/scheduler.ts`, `packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts`, `packages/engine/vitest.config.ts`
- **Approach:** After U2 through U6 pass, delete or collapse the legacy todo dispatcher code that sits after the workflow sweep return. Preserve reporter emission and non-dispatch scheduler duties. Broaden reachability assertions beyond stale `workflowColumns=false`: prove stale graph false, legacy/custom columns, existing `todo`/`in-progress`/`in-review` rows, reporter-only scheduler duties, exported scheduler helpers, and plugin-facing entrypoints either enter hold/release or are explicitly removed with replacement tests. U5 upgrade-progression and U6 rollback proof are prerequisites for deleting legacy dispatcher code or removing legacy coverage.
- **Test scenarios:** Stale persisted `workflowColumns=false` still schedules through hold/release; stale graph false does not route to legacy execution; legacy/custom columns and existing task rows keep progressing or remain intentionally parked; no test, production, exported helper, or plugin-facing callsite references removed dispatcher helpers; engine-core gate includes tracked replacement workflow tests.
- **Verification:** `pnpm --filter @fusion/engine typecheck`, `pnpm --filter @fusion/engine test:core`, and `rg` checks for removed helper names if helpers are deleted.
### U8. Validation, Review, And PR
- **Goal:** Finish the branch with objective verification and a reviewable PR.
- **Files:** `packages/engine/vitest.config.ts`, `.changeset/workflow-runtime-cutover.md`
- **Approach:** Run targeted tests first, then root checks. Add a patch changeset for `@runfusion/fusion` because this default-runtime cutover affects published behavior. The changeset must include Upgrade Notes covering workflow columns and graph execution becoming default, stale workflow flag values being ignored, removed Experimental controls, expected behavior for eligible versus intentionally parked tasks, and the verified rollback/support path. Run code review after tests are green and fix actionable findings before PR.
- **Test scenarios:** Targeted tests prove the invariant matrix; root commands prove workspace integration.
- **Verification:** `pnpm lint`, `pnpm typecheck`, `pnpm test`, `pnpm build`, and `compound-engineering:ce-code-review mode:agent plan:docs/plans/2026-06-23-002-workflow-runtime-cutover-hardening-plan.md`.
---
## Acceptance Examples
- AE1. Given a ready `todo` task and `maxConcurrent=1` with another task already in progress, when the scheduler sweep runs, then the ready task remains queued, no `Starting` log is written, and `onSchedule` is not called.
- AE2. Given a saturated shared semaphore held by non-task work, when the scheduler sweep evaluates a ready `todo` task, then the task is not moved to `in-progress` and the queued reason names semaphore or concurrency pressure.
- AE3. Given stale persisted `workflowGraphExecutor=false`, when `TaskExecutor.execute()` runs a task with satisfied dependencies, then graph-default execution still runs and the legacy fallback path is not used.
- AE4. Given graph execute delegates to the inner executor and the inner executor requeues the task to `todo`, when the outer graph run reports execute failure, then the task remains available for normal scheduling and is not parked as failed or in review.
- AE5. Given stale persisted `workflowColumns=false`, when scheduler `schedule()` runs, then hold/release scheduling is used and the deleted legacy dispatcher is unreachable.
- AE6. Given Experimental settings render, when the workflow cutover is complete, then workflow columns, workflow graph executor, and dual-observe controls are absent.
- AE7. Given a user upgrades with existing `todo`, `in-progress`, and `in-review` tasks, when the scheduler and executor start after upgrade, then eligible tasks keep progressing and intentionally paused or dependency-blocked tasks remain parked with the correct reason.
- AE8. Given a user upgrades with stale workflow experimental settings, when tasks are scheduled or executed, then those stale settings are tolerated and do not disable workflow columns or graph execution.
- AE9. Given a user opens Settings after upgrade with stale hidden workflow keys, when they save an unrelated settings change, then the hidden workflow keys follow the documented payload behavior and cannot silently re-disable the default runtime.
- AE10. Given a published cutover release stalls eligible task progression, when an operator follows the documented rollback or downgrade path, then eligible tasks resume without corrupting persisted settings or task rows.
---
## Scope Boundaries
- In scope: scheduler hold/release equivalence, executor graph-default recovery equivalence, workflow flag graduation, replacement tests, legacy dispatcher removal once proven unreachable, and PR validation.
- In scope: upgrade-state tests for prior-version settings and existing task rows needed to prove tasks keep progressing.
- Out of scope: unrelated dashboard cosmetic fixes, new workflow editor UI behavior, new workflow engine features, and broad scheduler rewrites not required to preserve existing invariants.
- Deferred unless U6 proves branch/downgrade rollback is insufficient: a permanent operational feature flag. User-facing Experimental kill switches remain out of scope.
---
## System-Wide Impact
This change touches the task execution lifecycle, scheduler admission control, settings defaults, Settings UI, and the engine merge gate. Failures can block task execution across projects, so test coverage must prove behavior at the production entrypoints rather than only at helper seams.
---
## Risks And Dependencies
- Capacity handling can fail in two opposite ways: bypassing capacity entirely or double-acquiring a semaphore slot before executor runs. U2 must avoid both.
- Reservation handoff can leak if an exception lands between hold creation and executor ownership. U2 must inject these failures and prove later sweeps recover.
- Legacy test deletion can hide active invariants unless replacement tests are tracked and committed in the same PR.
- Removing user-facing flags before stale persisted values are ignored can strand existing installations on removed code paths.
- Hiding dual-observe without clearing or controlling stale true values can leave diagnostic behavior running invisibly.
- Deleting legacy dispatcher code without reachability proof makes rollback more expensive than branch revert alone.
---
## Verification
- `git diff --name-only origin/main...HEAD`
- `pnpm --filter @fusion/engine exec vitest run src/__tests__/scheduler-workflow-cutover.test.ts`
- `pnpm --filter @fusion/engine exec vitest run src/__tests__/workflow-graph-task-runner.test.ts src/__tests__/executor-worktree.test.ts src/__tests__/restart.integration.test.ts src/__tests__/reliability-interactions/executor-liveness-gate.test.ts src/__tests__/reliability-interactions/executor-no-task-done-vs-worktree-reclaim.test.ts`
- `pnpm --filter @fusion/core exec vitest run src/__tests__/settings-defaults.test.ts src/__tests__/workflow-cutover.test.ts`
- Targeted dashboard settings tests for Experimental settings visibility
- `pnpm lint`
- `pnpm typecheck`
- `pnpm smoke:boot`
- `pnpm test:gate`
- `pnpm test`
- `pnpm build`
- `compound-engineering:ce-code-review mode:agent plan:docs/plans/2026-06-23-002-workflow-runtime-cutover-hardening-plan.md`

1
packages/core/src/__tests__/settings-defaults.test.ts
View File

@@ -35,6 +35,7 @@ describe("settings defaults invariants", () => {
expect(isExperimentalFeatureEnabled(undefined, "workflowColumns")).toBe(false);
expect(isExperimentalFeatureEnabled(undefined, "workflowGraphExecutor")).toBe(false);
expect(isExperimentalFeatureEnabled(undefined, "workflowInterpreterDualObserve")).toBe(false);
expect(isExperimentalFeatureEnabled({ experimentalFeatures: { workflowInterpreterDualObserve: true } }, "workflowInterpreterDualObserve")).toBe(false);
expect(isWorkflowColumnsEnabled({ experimentalFeatures: { workflowColumns: false } })).toBe(true);
});

7
packages/core/src/experimental-features.ts
View File

@@ -7,8 +7,14 @@ const LEGACY_EXPERIMENTAL_FEATURE_ALIASES: Record<string, string> = {
/*
FNXC:WorkflowSettings 2026-06-22-18:00:
workflowGraphExecutor and workflowColumns graduated from Experimental. Runtime graph execution and workflow-defined columns are always on; stale persisted values are ignored by runtime helpers instead of acting as kill switches.
FNXC:WorkflowSettings 2026-06-23-21:55:
workflowInterpreterDualObserve is no longer user-controllable in Settings. Treat stale persisted true values as inert so upgraded users do not keep running hidden diagnostic shadow observation with no visible off switch.
*/
const DEFAULT_ON_EXPERIMENTAL_FEATURES = new Set<string>();
const RETIRED_EXPERIMENTAL_FEATURES = new Set<string>([
"workflowInterpreterDualObserve",
]);
export function isExperimentalFeatureEnabled(
settings: Pick<Settings, "experimentalFeatures"> | undefined,
@@ -16,6 +22,7 @@ export function isExperimentalFeatureEnabled(
): boolean {
const features = settings?.experimentalFeatures;
const canonicalKey = LEGACY_EXPERIMENTAL_FEATURE_ALIASES[key] ?? key;
if (RETIRED_EXPERIMENTAL_FEATURES.has(canonicalKey)) return false;
if (features?.[canonicalKey] === false) return false;
if (features?.[canonicalKey] === true) return true;

37
packages/dashboard/app/components/__tests__/SettingsModal.test.tsx
View File

@@ -4065,6 +4065,43 @@ describe("SettingsModal", () => {
expect(payload.experimentalFeatures.devServer).toBeUndefined();
});
it("hides graduated workflow flags while preserving stale persisted values on save", async () => {
mockFetchSettings.mockResolvedValue({
...defaultSettings,
experimentalFeatures: {
workflowColumns: false,
workflowGraphExecutor: false,
workflowInterpreterDualObserve: true,
insights: true,
},
});
renderModal();
await openExperimentalFeaturesSection();
expect(screen.queryByText("workflowColumns")).not.toBeInTheDocument();
expect(screen.queryByText("workflowGraphExecutor")).not.toBeInTheDocument();
expect(screen.queryByText(/dual-observe parity/i)).not.toBeInTheDocument();
await userEvent.click(screen.getByText("Save"));
await waitFor(() => {
expect(mockUpdateGlobalSettings).toHaveBeenCalledTimes(1);
});
/*
FNXC:SettingsExperimental 2026-06-23-21:20:
Workflow runtime flags are hidden because the graph engine and workflow columns are default runtime paths. Saving unrelated Settings changes must preserve stale persisted keys instead of rewriting or resurrecting them as UI-controlled toggles; runtime helpers ignore those stale values.
*/
expect(mockUpdateGlobalSettings.mock.calls[0][0].experimentalFeatures).toEqual({
workflowColumns: false,
workflowGraphExecutor: false,
workflowInterpreterDualObserve: true,
insights: true,
});
});
it("checks Left Sidebar Navigation by default when its flag is unset", async () => {
mockFetchSettings.mockResolvedValue({
...defaultSettings,

9
packages/engine/src/__tests__/executor-prompt.test.ts
View File

@@ -744,7 +744,7 @@ describe("TaskExecutor pause behavior", () => {
// Should move to todo, NOT mark as failed. This path (agent threw mid-
// execution while paused) explicitly nukes worktree+branch — work is
// discarded — so it must NOT flag preserveResumeState.
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo");
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", undefined);
expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", { status: "failed" });
});
@@ -2024,8 +2024,9 @@ describe("TaskExecutor global pause behavior", () => {
]);
// Global pause should move both tasks out of in-progress without marking failed.
expect(store.moveTask).toHaveBeenCalledWith("FN-002", expect.stringMatching(/^(todo|in-review)$/));
expect(store.moveTask).toHaveBeenCalledWith("FN-001", expect.stringMatching(/^(todo|in-review)$/));
const moveCalls = store.moveTask.mock.calls;
expect(moveCalls.some(([id, column]) => id === "FN-002" && /^(todo|in-review)$/.test(String(column)))).toBe(true);
expect(moveCalls.some(([id, column]) => id === "FN-001" && /^(todo|in-review)$/.test(String(column)))).toBe(true);
expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", { status: "failed" });
expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", { status: "failed" });
});
@@ -2053,7 +2054,7 @@ describe("TaskExecutor global pause behavior", () => {
createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(),
});
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo");
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "todo", undefined);
expect(store.updateTask).not.toHaveBeenCalledWith("FN-001", { status: "failed" });
});

4
packages/engine/src/__tests__/restart.integration.test.ts
View File

@@ -1225,11 +1225,11 @@ describe("Crash scenario edge cases", () => {
await executor.resumeOrphaned();
await waitForAsyncExpectation(() => {
expect(onError).toHaveBeenCalledWith(task, expect.any(Error));
expect(onError).toHaveBeenCalledWith(expect.objectContaining({ id: task.id }), expect.any(Error));
});
// onError should have been called
expect(onError).toHaveBeenCalledWith(task, expect.any(Error));
expect(onError).toHaveBeenCalledWith(expect.objectContaining({ id: task.id }), expect.any(Error));
// Semaphore slot should be released
expect(sem.activeCount).toBe(0);

220
packages/engine/src/__tests__/scheduler-workflow-cutover.test.ts Normal file
View File

@@ -0,0 +1,220 @@
import { beforeEach, describe, expect, it, vi } from "vitest";
import { makeTransitionRejection, TransitionRejectionError, type Task, type TaskStore } from "@fusion/core";
import { existsSync } from "node:fs";
import { readFile } from "node:fs/promises";
import { Scheduler } from "../scheduler.js";
import { AgentSemaphore } from "../concurrency.js";
vi.mock("node:fs", async (importOriginal) => {
const actual = await importOriginal<typeof import("node:fs")>();
return { ...actual, existsSync: vi.fn() };
});
vi.mock("node:fs/promises", async (importOriginal) => {
const actual = await importOriginal<typeof import("node:fs/promises")>();
return { ...actual, readFile: vi.fn() };
});
function task(overrides: Partial<Task> = {}): Task {
return {
id: "FN-100",
title: "Workflow task",
description: "",
column: "todo",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: "2026-06-23T00:00:00.000Z",
updatedAt: "2026-06-23T00:00:00.000Z",
...overrides,
} as Task;
}
function storeWith(tasks: Task[], settings: Record<string, unknown> = {}): TaskStore {
const byId = new Map(tasks.map((candidate) => [candidate.id, candidate]));
return {
listTasks: vi.fn(async () => [...byId.values()]),
getTask: vi.fn(async (id: string) => byId.get(id) ?? null),
getSettings: vi.fn(async () => ({
maxConcurrent: 2,
maxWorktrees: 4,
experimentalFeatures: { workflowColumns: false },
...settings,
})),
updateTask: vi.fn(async (id: string, patch: Partial<Task>) => {
const current = byId.get(id);
if (current) Object.assign(current, patch);
return current as Task;
}),
moveTask: vi.fn(async (id: string, column: Task["column"]) => {
const current = byId.get(id);
if (current) current.column = column;
return current as Task;
}),
parseFileScopeFromPrompt: vi.fn(async () => []),
logEntry: vi.fn(async () => undefined),
getRootDir: vi.fn(() => "/tmp/project"),
getTasksDir: vi.fn(() => "/tmp/project/.fusion/tasks"),
on: vi.fn(),
off: vi.fn(),
recordRunAuditEvent: vi.fn(async () => undefined),
getMissionStore: vi.fn(() => ({
listMissions: () => [],
listGoalIdsForMission: () => [],
})),
} as unknown as TaskStore;
}
describe("Scheduler workflow cutover", () => {
beforeEach(() => {
vi.clearAllMocks();
vi.mocked(existsSync).mockReturnValue(true);
vi.mocked(readFile).mockResolvedValue("# Task\nBody");
});
it("uses the workflow sweep for todo pickup even when stale workflowColumns=false is persisted", async () => {
const ready = task({ id: "FN-100" });
const store = storeWith([ready]);
const onSchedule = vi.fn();
const scheduler = new Scheduler(store, { onSchedule });
(scheduler as unknown as { running: boolean }).running = true;
await scheduler.schedule();
expect(store.moveTask).toHaveBeenCalledWith("FN-100", "in-progress", expect.objectContaining({
moveSource: "scheduler",
allocateWorktree: expect.any(Function),
}));
expect(store.updateTask).toHaveBeenCalledWith("FN-100", expect.objectContaining({
status: null,
blockedBy: null,
mergeRetries: 0,
effectiveNodeId: null,
effectiveNodeSource: "local",
}));
expect(onSchedule).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-100", column: "in-progress" }));
});
it("keeps dependency-blocked todo tasks queued on the workflow sweep path", async () => {
const blocker = task({ id: "FN-001", column: "todo" });
const dependent = task({ id: "FN-002", dependencies: ["FN-001"] });
const store = storeWith([blocker, dependent]);
const onBlocked = vi.fn();
const scheduler = new Scheduler(store, { onBlocked });
(scheduler as unknown as { running: boolean }).running = true;
await scheduler.schedule();
expect(store.updateTask).toHaveBeenCalledWith("FN-002", {
status: "queued",
blockedBy: "FN-001",
});
expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything());
expect(onBlocked).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-002" }), ["FN-001"]);
});
it("does not clear status or release work when maxConcurrent is full", async () => {
const active = task({ id: "FN-001", column: "in-progress" });
const ready = task({ id: "FN-002", status: "queued" });
const store = storeWith([active, ready], { maxConcurrent: 1, maxWorktrees: 4 });
const onSchedule = vi.fn();
const scheduler = new Scheduler(store, { onSchedule });
(scheduler as unknown as { running: boolean }).running = true;
await scheduler.schedule();
expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything());
expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", expect.objectContaining({ status: null }));
expect(onSchedule).not.toHaveBeenCalledWith(expect.objectContaining({ id: "FN-002" }));
expect(ready.column).toBe("todo");
});
it("does not clear status or release work when maxWorktrees is full", async () => {
const active = task({ id: "FN-001", column: "in-progress" });
const ready = task({ id: "FN-002", status: "queued" });
const store = storeWith([active, ready], { maxConcurrent: 4, maxWorktrees: 1 });
const onSchedule = vi.fn();
const scheduler = new Scheduler(store, { onSchedule });
(scheduler as unknown as { running: boolean }).running = true;
await scheduler.schedule();
expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything());
expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", expect.objectContaining({ status: null }));
expect(onSchedule).not.toHaveBeenCalledWith(expect.objectContaining({ id: "FN-002" }));
expect(ready.column).toBe("todo");
});
it("reserves same-sweep capacity so only one ready task is released into one slot", async () => {
const first = task({ id: "FN-001", status: "queued" });
const second = task({ id: "FN-002", status: "queued" });
const store = storeWith([first, second], { maxConcurrent: 1, maxWorktrees: 1 });
const onSchedule = vi.fn();
const scheduler = new Scheduler(store, { onSchedule });
(scheduler as unknown as { running: boolean }).running = true;
await scheduler.schedule();
expect(store.moveTask).toHaveBeenCalledTimes(1);
expect(store.moveTask).toHaveBeenCalledWith("FN-001", "in-progress", expect.anything());
expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything());
expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", expect.objectContaining({ status: null }));
expect(onSchedule).toHaveBeenCalledTimes(1);
expect(onSchedule).toHaveBeenCalledWith(expect.objectContaining({ id: "FN-001", column: "in-progress" }));
expect(second.column).toBe("todo");
expect(second.status).toBe("queued");
});
it("leaves a task queued when the authoritative release move rejects after reservation", async () => {
const ready = task({ id: "FN-002", status: "queued" });
const store = storeWith([ready], { maxConcurrent: 4, maxWorktrees: 4 });
vi.mocked(store.moveTask).mockRejectedValueOnce(
new TransitionRejectionError(
makeTransitionRejection(
"capacity-exhausted",
"transition.rejected.capacityExhausted",
true,
"Column is at capacity",
),
"Column is at capacity",
),
);
const onSchedule = vi.fn();
const scheduler = new Scheduler(store, { onSchedule });
(scheduler as unknown as { running: boolean }).running = true;
await scheduler.schedule();
expect(store.moveTask).toHaveBeenCalledWith("FN-002", "in-progress", expect.anything());
expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", expect.objectContaining({ status: null }));
expect(store.logEntry).not.toHaveBeenCalledWith(
"FN-002",
expect.stringContaining("Node routing resolved"),
);
expect(onSchedule).not.toHaveBeenCalled();
expect(ready.column).toBe("todo");
expect(ready.status).toBe("queued");
});
it("does not release work when the shared semaphore is saturated", async () => {
const ready = task({ id: "FN-002", status: "queued" });
const store = storeWith([ready], { maxConcurrent: 4, maxWorktrees: 4 });
const semaphore = new AgentSemaphore(1);
await semaphore.acquire();
const onSchedule = vi.fn();
const scheduler = new Scheduler(store, { onSchedule, semaphore });
(scheduler as unknown as { running: boolean }).running = true;
try {
await scheduler.schedule();
} finally {
semaphore.release();
}
expect(store.moveTask).not.toHaveBeenCalledWith("FN-002", "in-progress", expect.anything());
expect(store.updateTask).not.toHaveBeenCalledWith("FN-002", expect.objectContaining({ status: null }));
expect(onSchedule).not.toHaveBeenCalled();
expect(ready.column).toBe("todo");
});
});

15
packages/engine/src/__tests__/workflow-graph-task-runner.test.ts
View File

@@ -11,7 +11,10 @@ const flagOn = { experimentalFeatures: { workflowGraphExecutor: true } } as unkn
Settings,
"experimentalFeatures"
>;
const flagOff = { experimentalFeatures: {} } as unknown as Pick<Settings, "experimentalFeatures">;
const flagOff = { experimentalFeatures: { workflowGraphExecutor: false } } as unknown as Pick<
Settings,
"experimentalFeatures"
>;
/** start → lint(custom) → execute → review → merge → notify(custom) → end, with seam failure edges to end. */
function fullLifecycleIr(): WorkflowIr {
@@ -186,13 +189,19 @@ describe("WorkflowGraphTaskRunner (CU-U2)", () => {
});
it("ignores stale workflowGraphExecutor=false and still runs the graph", async () => {
const calls: string[] = [];
const runner = new WorkflowGraphTaskRunner({
store: storeWith(definition(fullLifecycleIr())),
seams: recordingSeams([]),
runCustomNode: async () => ({ outcome: "success" }),
seams: recordingSeams(calls),
runCustomNode: async (node) => {
calls.push(`custom:${node.id}`);
return { outcome: "success" };
},
});
const result = await runner.run(task, flagOff);
expect(result.disposition).toBe("completed");
expect(calls).toEqual(["custom:lint", "execute", "review", "merge", "custom:notify"]);
expect(result.visitedNodeIds).toEqual(["start", "lint", "execute", "review", "merge", "notify"]);
});
it("falls back when the task has no workflow selection", async () => {

90
packages/engine/src/executor.ts
View File

@@ -1549,6 +1549,11 @@ export class TaskExecutor {
private stuckAborted = new Map<string, boolean>();
/** Tasks explicitly canceled by user move (in-progress → todo). */
private userCanceledTaskIds = new Set<string>();
/*
FNXC:WorkflowLifecycle 2026-06-23-21:16:
During graph-owned execute nodes, the inner executor may intentionally self-requeue a task to `todo` for recoverable worktree/session repair. Persisted rows can be stale in tests or during store races, so keep a run-local marker that tells the outer graph failure sink not to overwrite that recovery with an in-review handoff.
*/
private graphExecuteSelfRequeued = new Set<string>();
/** In-memory loop recovery state per task. Keyed by taskId, not persisted.
* Tracks compact-and-resume attempt count per execute() lifecycle.
* Reset at execute() lifecycle end (finally block). */
@@ -1587,6 +1592,12 @@ export class TaskExecutor {
activeSessionRegistry.registerPath(worktreePath, { taskId, kind: "executor", ownerKey: taskId });
}
private markGraphExecuteSelfRequeued(taskId: string): void {
if (this.graphRouting.has(taskId)) {
this.graphExecuteSelfRequeued.add(taskId);
}
}
private deleteActiveSession(taskId: string, worktreePath?: string): void {
this.activeSessions.delete(taskId);
// U5: drop the effective column-agent principal for this task's session.
@@ -3381,6 +3392,7 @@ export class TaskExecutor {
nextRecoveryAt: decision.nextState.nextRecoveryAt,
sessionFile: null,
});
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveResumeState: true });
return true;
}
@@ -4129,8 +4141,15 @@ export class TaskExecutor {
*/
settings = { ...settings };
let selection: { workflowId: string; stepIds: string[] } | undefined;
if (typeof this.store.getTaskWorkflowSelection !== "function") {
/*
FNXC:WorkflowExecution 2026-06-23-22:01:
Graph execution is the default for production TaskStore implementations, which expose workflow-selection APIs. Minimal test stores and older embedded adapters can lack that API; fall back to the legacy executor instead of half-entering graph routing with no workflow persistence surface.
*/
return false;
}
try {
selection = this.store.getTaskWorkflowSelection?.(task.id);
selection = this.store.getTaskWorkflowSelection(task.id);
} catch (err) {
await this.handleGraphFailure(task, {
disposition: "failed",
@@ -4268,7 +4287,14 @@ export class TaskExecutor {
});
let result: WorkflowGraphTaskRunResult;
try {
const detail = await this.store.getTask(task.id);
const loadedDetail = await this.store.getTask(task.id);
/*
FNXC:WorkflowExecution 2026-06-23-11:36:
Graph dispatch must preserve the row identity that entered execute(). Minimal test stores and stale adapters can return an unrelated fallback task from getTask(); trusting that row would run the workflow under the wrong task id and bypass executor invariants. Use the refreshed row only when it matches the dispatch task.
*/
const detail: TaskDetail = loadedDetail?.id === task.id
? loadedDetail
: { ...task, prompt: task.prompt ?? task.description ?? "" };
result = await runner.run(detail, settings);
} catch (err) {
executorLog.error(
@@ -4325,6 +4351,7 @@ export class TaskExecutor {
this.graphColumnAgentResolver.delete(task.id);
this.graphUnattendedRuns.delete(task.id);
this.graphSeamGoverningNodeId.delete(task.id);
this.graphExecuteSelfRequeued.delete(task.id);
// Per-instance keys: clear every instance slot owned by this task.
const ctxPrefix = `${task.id}:`;
for (const key of this.graphStepActiveContext.keys()) {
@@ -5069,6 +5096,12 @@ export class TaskExecutor {
// completes; a step-review node (when present) decides done-ness instead.
try {
const live = await this.store.getTask(task.id);
if (!live || live.id !== task.id) {
return {
success: false,
error: `step ${stepIndex} live task unavailable after implementation pass`,
};
}
const active = this.foreachActiveForTask(task.id, instanceId);
const status = live.steps[stepIndex]?.status;
if (status === "done" || status === "skipped") return { success: true };
@@ -5123,9 +5156,13 @@ export class TaskExecutor {
return {
prepareWorktree: async (_ctx, task) => {
const live = await this.store.getTask(task.id);
/*
FNXC:WorkflowExecution 2026-06-23-11:49:
The workflow execute node must not perform a second worktree acquisition ahead of the authoritative executor. Passing the repo root as a prepared worktree makes the inner execute() reject a valid fresh-worktree task as repo-root reuse; pass only an existing task worktree and let execute() acquire when none exists.
*/
const prepared: PreparedWorktree = {
worktreePath: live.worktree || this.rootDir,
branchName: live.branch,
worktreePath: live.worktree || task.worktree || "",
branchName: live.branch || task.branch,
};
return { outcome: "success", value: "worktree-ready", data: prepared };
},
@@ -6713,7 +6750,17 @@ export class TaskExecutor {
this.clearCompletedTaskWatchdog(task.id);
this.options.stuckTaskDetector?.untrackTask(task.id);
try {
const live = await this.store.getTask(task.id);
const loadedLive = await this.store.getTask(task.id);
/*
FNXC:WorkflowLifecycle 2026-06-23-12:01:
Graph failure handling must never mutate a different task row than the one that entered execute(). Minimal stores can return fallback rows from getTask(); treat that as an unavailable live snapshot and leave the inner executor recovery result intact instead of handing off the wrong task.
*/
if (!loadedLive || loadedLive.id !== task.id) {
executorLog.warn(`${task.id}: graph failure live-state refetch returned ${loadedLive?.id ?? "null"} — preserving inner executor result`);
await this.persistTokenUsage(task.id);
return;
}
const live = loadedLive;
// A paused/aborted implementation is not a graph failure while the task
// is still in-progress — leave the pause machinery in charge instead of
// parking the task in review.
@@ -6976,6 +7023,21 @@ export class TaskExecutor {
const failedNode = result.visitedNodeIds[result.visitedNodeIds.length - 1];
const mergeGraphFailure = this.isMergeGraphFailure(failedNode);
const failureValue = this.graphFailureValue(result);
const executeNodeSelfRequeued = failedNode === "execute" && this.graphExecuteSelfRequeued.has(task.id);
if (failedNode === "execute" && (live.column === "todo" || executeNodeSelfRequeued)) {
/*
FNXC:WorkflowLifecycle 2026-06-23-12:03:
The graph execute node delegates to the authoritative executor. If that inner executor requeues the task to todo for self-heal/retry, the outer graph failure must not override it by parking the task in review.
FNXC:WorkflowLifecycle 2026-06-23-21:19:
Also honor the in-process self-requeue marker. Upgrade/restart races and minimal stores can return a stale `in-progress` live row even after the inner executor already moved the task to `todo`; stale reads must not strand progressing tasks in review.
*/
const benignMessage = `Workflow graph execute node ended after executor re-queued task to todo (${failureValue ?? "no-value"}) — executor recovery preserved`;
executorLog.log(`${task.id}: ${benignMessage}`);
await this.store.logEntry(task.id, benignMessage, undefined, this.getRunContextFor(task.id));
await this.persistTokenUsage(task.id);
return;
}
if (mergeGraphFailure && !this.isTerminalMergeGraphFailureValue(failureValue) && await this.routeGraphMergeFailureToRetry(live, result, abortProvenance)) {
return;
}
@@ -7651,6 +7713,7 @@ export class TaskExecutor {
undefined,
this.getRunContextFor(task.id),
);
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveProgress: true });
executorLog.log(`✗ ${task.id} worktree liveness failed — requeued to todo (${nextRequeueCount}/${MAX_TASK_DONE_REQUEUE_RETRIES})`);
} else {
@@ -7831,6 +7894,7 @@ export class TaskExecutor {
}
this.clearPausedAborted(task.id);
await this.store.logEntry(task.id, "Execution paused — step sessions terminated, moved to todo", undefined, this.getRunContextFor(task.id));
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveResumeState: true });
return;
}
@@ -8082,6 +8146,7 @@ export class TaskExecutor {
}
this.clearPausedAborted(task.id);
await this.store.logEntry(task.id, "Execution paused during step-session", undefined, this.getRunContextFor(task.id));
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveResumeState: true });
} else if (this.stuckAborted.has(task.id)) {
stuckRequeue = this.stuckAborted.get(task.id) ?? true;
@@ -8125,6 +8190,7 @@ export class TaskExecutor {
worktree: null,
branch: null,
});
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveProgress: true });
stuckRequeue = null; // Prevent outer finally from re-processing
return;
@@ -8218,6 +8284,7 @@ export class TaskExecutor {
branch: null,
});
if (latestTask.column !== "todo") {
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", preserveProgress ? { preserveProgress: true } : undefined);
executorLog.log(`${task.id} moved to todo for retry after stuck kill${preserveProgress ? " (progress preserved)" : ""}`);
}
@@ -8731,6 +8798,7 @@ export class TaskExecutor {
} else {
executorLog.log(`${task.id} paused (graceful session exit) — moving to todo`);
await this.store.logEntry(task.id, "Execution paused — session preserved for resume, moved to todo");
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveResumeState: true });
}
return;
@@ -9129,6 +9197,7 @@ export class TaskExecutor {
// the next pickup will re-anchor it on the fresh checkout.
await this.store.updateTask(task.id, { worktree: null, branch: null, baseCommitSha: null });
await this.persistTokenUsage(task.id);
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveProgress: true });
executorLog.log(silentMessage);
} else if (refusalHandled) {
@@ -9156,6 +9225,7 @@ export class TaskExecutor {
undefined,
this.getRunContextFor(task.id),
);
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveProgress: true });
executorLog.log(`✗ ${task.id} failed after ${MAX_TASK_DONE_SESSION_RETRIES} retries — requeued to todo (${nextRequeueCount}/${MAX_TASK_DONE_REQUEUE_RETRIES})`);
} else {
@@ -9337,6 +9407,7 @@ export class TaskExecutor {
hasResumableProgress ? { worktree: undefined } : { worktree: undefined, branch: undefined },
);
await this.store.logEntry(task.id, "Execution paused — agent terminated, moved to todo", undefined, this.getRunContextFor(task.id));
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", hasResumableProgress ? { preserveResumeState: true } : undefined);
}
} else if (this.stuckAborted.has(task.id)) {
@@ -9444,6 +9515,7 @@ export class TaskExecutor {
nextRecoveryAt: decision.nextState.nextRecoveryAt,
sessionFile: null,
});
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveResumeState: true });
return;
}
@@ -9626,6 +9698,7 @@ export class TaskExecutor {
// "worktree gone" from "pointer not yet repopulated". Matches sibling
// recovery paths in auto-recovery-handlers/contamination.ts,
// tryBootstrapMisbindingRecovery, and self-healing reclaim.
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveResumeState: true, preserveWorktree: true });
return;
}
@@ -9817,6 +9890,7 @@ export class TaskExecutor {
worktree: null,
branch: null,
});
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveProgress: true });
return;
}
@@ -9960,6 +10034,7 @@ export class TaskExecutor {
// the captured snapshot can be hours old and would race against
// any concurrent recovery (see comment above).
if (latestTask.column !== "todo") {
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", preserveProgress ? { preserveProgress: true } : undefined);
// Audit trail: record task move (FN-1404)
await audit.database({ type: "task:move", target: task.id, metadata: { to: "todo" } });
@@ -10768,6 +10843,7 @@ export class TaskExecutor {
undefined,
this.getRunContextFor(task.id),
);
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveProgress: true });
} else {
await this.store.updateTask(task.id, {
@@ -13235,6 +13311,7 @@ You have access to the file system to review changes.${verdictBlock}`;
paused: false,
pausedReason: null,
});
this.markGraphExecuteSelfRequeued(task.id);
await this.store.moveTask(task.id, "todo", { preserveResumeState: false, preserveWorktree: true });
return true;
} catch (error) {
@@ -13859,6 +13936,9 @@ You have access to the file system to review changes.${verdictBlock}`;
source: "executor-session-start",
auditor: audit,
});
if (recovery.outcome !== "escalate-exhausted") {
this.markGraphExecuteSelfRequeued(task.id);
}
await audit.git({
type: "worktree:auto-recovered",

16
packages/engine/src/hold-release.ts
View File

@@ -432,9 +432,10 @@ async function issueRelease(
const onMoved = (data: { task: object; to: string }): void => {
if (data.to === target) movedTaskObjects.add(data.task);
};
store.on("task:moved", onMoved);
store.on?.("task:moved", onMoved);
try {
const originalColumn = task.column;
const result = await store.moveTask(task.id, target, {
moveSource: "scheduler",
allocateWorktree:
@@ -442,7 +443,16 @@ async function issueRelease(
? (reservedNames) => deps.allocateWorktree!(task, reservedNames)
: undefined,
});
if (reservation && !movedTaskObjects.has(result)) {
/*
FNXC:WorkflowScheduling 2026-06-23-21:57:
The cutover scheduler uses hold/release in tests and older embedded stores that may not expose task:moved events. Treat a returned task that clearly moved from the original column to the target as the committed release so minimal stores do not leak reservations or falsely report a racing same-column no-op.
*/
const returnedMovedTask = movedTaskObjects.size === 0
&& (
result === undefined
|| (result.id === task.id && result.column === target && originalColumn !== target)
);
if (reservation && !movedTaskObjects.has(result) && !returnedMovedTask) {
// Same-column no-op: a racing sweep already moved this card to the target.
reservation.release();
schedulerLog.log(`Hold release for ${task.id} skipped — already at ${target} (racing sweep won)`);
@@ -463,7 +473,7 @@ async function issueRelease(
);
return false;
} finally {
store.off("task:moved", onMoved);
store.off?.("task:moved", onMoved);
}
}

22
packages/engine/src/merger.ts
View File

@@ -114,7 +114,7 @@ import { accumulateSessionTokenUsage } from "./session-token-usage.js";
import { createResolvedAgentSession, extractRuntimeHint, resolveMergerSessionModel } from "./agent-session-helpers.js";
import { createFallbackModelObserver } from "./fallback-model-observer.js";
import { buildSessionSkillContext } from "./session-skill-context.js";
import { classifyTaskWorktree, getRegisteredWorktreeBranches, RemovalReason, removeWorktree, type WorktreePool } from "./worktree-pool.js";
import { classifyTaskWorktree, getRegisteredWorktreeBranches, isRepoRootPath, RemovalReason, removeWorktree, type WorktreePool } from "./worktree-pool.js";
import { activeSessionRegistry } from "./active-session-registry.js";
import { AgentLogger } from "./agent-logger.js";
import { mergerLog } from "./logger.js";
@@ -7925,6 +7925,20 @@ export async function aiMergeTask(
diagnostics: Record<string, unknown>,
): Promise<void> => {
const priorWorktreePath = task.worktree ?? null;
if (priorWorktreePath && isRepoRootPath(projectRootDir, priorWorktreePath)) {
/*
* FNXC:WorkflowCutover 2026-06-23-04:45:
* Merge reuse handoff must reject a task worktree that equals the project root before acquisition fallback can clear the assignment. Executor resume may self-heal stale root assignments, but merge must not hide a handoff contract violation by creating a fresh task worktree.
*/
throw new MergeHandoffRefusedError("reuse-misconfigured", "worktree-equals-project-root", {
taskId,
projectRoot: projectRootDir,
worktreePath: priorWorktreePath,
requestedMode: requestedIntegrationMode,
reason,
diagnostics,
});
}
// FN-5345/FN-5377: consult existing registration of `fusion/<id>` before
// creating a fresh worktree. If the branch is already registered at a
@@ -8314,6 +8328,12 @@ export async function aiMergeTask(
gate: error.gate,
reason: error.reason,
});
} else if (isRepoRootPath(projectRootDir, reusableWorktreePath)) {
/*
* FNXC:WorkflowCutover 2026-06-23-04:45:
* Merge reuse handoff must reject a task worktree that equals the project root. Executor resume may self-heal stale root assignments, but merge must not turn this dangerous state into a fresh-worktree fallback because that hides a handoff contract violation.
*/
throw error;
} else {
const classification = await classifyTaskWorktree(projectRootDir, reusableWorktreePath);
if (!classification.ok) {

428
packages/engine/src/scheduler.ts
View File

@@ -874,6 +874,13 @@ export class Scheduler {
* @returns Object with `valid: true` if checks pass, or `valid: false` with a `reason` string if they fail
*/
private async validateTaskFilesystem(id: string): Promise<{ valid: boolean; reason?: string }> {
if (typeof this.store.getTasksDir !== "function") {
/*
FNXC:WorkflowScheduling 2026-06-23-11:38:
Scheduler test fakes and older embedded stores may not expose task-directory helpers. The production TaskStore still enforces task-dir and PROMPT.md validation, but minimal stores should not abort the workflow sweep before lease recovery and node-routing guards run.
*/
return { valid: true };
}
const taskDir = join(this.store.getTasksDir(), id);
// Check if task directory exists
@@ -2066,7 +2073,18 @@ export class Scheduler {
private async runHoldReleaseSweepPass(tasks: Task[], settings: Settings): Promise<void> {
try {
const maxWorktrees = settings.maxWorktrees ?? this.options.maxWorktrees ?? 4;
const maxConcurrent = settings.maxConcurrent ?? this.options.maxConcurrent ?? 2;
let reservedWorktreeSlots = tasks.filter((task) => task.column === "in-progress").length;
let reservedConcurrentSlots = reservedWorktreeSlots;
const inProgressTaskIds = tasks.filter((task) => task.column === "in-progress").map((task) => task.id);
const dispatchPrepByTaskId = new Map<string, {
baseBranch: string | null;
dispatchStormCount: number;
dispatchTimestamp: string;
effectiveNodeId: string | null;
effectiveNodeSource: string;
task: Task;
}>();
const activeScopes = new Map<string, string[]>();
const activeScopeColumns = new Map<string, Task["column"]>();
const overlapIgnorePaths = settings.overlapIgnorePaths ?? [];
@@ -2123,10 +2141,340 @@ export class Scheduler {
}
}
await runHoldReleaseSweep(this.store, {
const result = await runHoldReleaseSweep(this.store, {
now: () => Date.now(),
reserveSlot: async (task): Promise<SlotReservation | null> => {
let reservedScope = false;
const unmetDeps = getUnmetSchedulingDependencies(task, tasks, schedulingDependencyOptions);
if (unmetDeps.length > 0) {
await this.store.updateTask(task.id, {
status: "queued",
blockedBy: unmetDeps[0],
});
await this.logDispatchQueuedReason(task.id, `queued — unmet dependencies: ${unmetDeps.join(", ")}`);
this.options.onBlocked?.(task, unmetDeps);
return null;
}
if (this.options.missionStore && task.sliceId) {
try {
const slice = this.options.missionStore.getSlice(task.sliceId);
const milestone = slice ? this.options.missionStore.getMilestone(slice.milestoneId) : undefined;
const mission = milestone ? this.options.missionStore.getMission(milestone.missionId) : undefined;
if (mission?.status === "blocked") {
await this.store.updateTask(task.id, { status: "queued" });
await this.logDispatchQueuedReason(task.id, "queued — mission is blocked");
return null;
}
} catch (err: unknown) {
const errorMessage = err instanceof Error ? err.message : String(err);
schedulerLog.warn(
`Mission/slice lookup failed during workflow scheduling (task ${task.id}): ${errorMessage} — proceeding without blocked-slice check`,
);
}
}
/*
FNXC:WorkflowScheduling 2026-06-23-11:12:
The workflow sweep is the only dispatcher, so the scheduler-only pre-dispatch gates must run before a capacity hold moves to an execution column. Keep dependency, filesystem, node-routing, permanent-agent, and oscillation checks on this path instead of relying on the retired todo loop.
*/
const validation = await this.validateTaskFilesystem(task.id);
if (!validation.valid) {
schedulerLog.warn(`Task ${task.id} filesystem validation failed: ${validation.reason}`);
await this.store.moveTask(task.id, "triage");
await this.store.logEntry(task.id, "Task moved to triage — filesystem validation failed", validation.reason);
return null;
}
if (typeof this.store.getTasksDir === "function") {
const promptPath = getPromptPath(this.store.getTasksDir(), task.id);
const staleness = await evaluateSpecStaleness({ settings, promptPath, task });
if (staleness.isStale) {
schedulerLog.warn(`Task ${task.id} specification is stale — ${staleness.reason}`);
await this.store.moveTask(task.id, "triage");
await this.store.updateTask(task.id, { status: "needs-replan" });
await this.store.logEntry(task.id, staleness.reason);
return null;
}
}
const freshTask = await this.store.getTask(task.id);
if (!freshTask || freshTask.column !== task.column || freshTask.paused || freshTask.userPaused) {
if (freshTask?.userPaused === true && freshTask.status !== "queued") {
await this.store.updateTask(task.id, { status: "queued" });
await this.logDispatchQueuedReason(task.id, "queued — user paused (manual move to todo)");
}
return null;
}
if (freshTask.checkedOutBy && this.options.leaseManager) {
const recovered = await this.options.leaseManager.recoverAbandonedLease(
freshTask.id,
"scheduler detected stale todo lease",
{ preserveProgress: true },
);
if (!recovered) {
await this.options.leaseManager.reconcileLeaseRow(freshTask.id);
await this.store.updateTask(freshTask.id, { status: "queued" });
await this.logDispatchQueuedReason(freshTask.id, "queued — checkout lease recovery blocked dispatch");
return null;
}
}
const latestSettings = await this.store.getSettings();
if (latestSettings.globalPause) {
schedulerLog.log(`Task ${task.id} dispatch aborted — globalPause became active mid-pass`);
return null;
}
if (latestSettings.enginePaused) {
schedulerLog.log(`Task ${task.id} dispatch aborted — enginePaused became active mid-pass`);
return null;
}
let effectiveNode = resolveEffectiveNode(freshTask, settings);
schedulerLog.log(`Task ${task.id} routed to node=${effectiveNode.nodeId ?? "local"} (source=${effectiveNode.source})`);
if (effectiveNode.nodeId !== undefined && this.options.validateNodeDispatch) {
const nodeValidation = await this.options.validateNodeDispatch(effectiveNode.nodeId);
if (!nodeValidation.allowed) {
if (!this.wasNodeDispatchValidationBlocked.has(task.id)) {
this.wasNodeDispatchValidationBlocked.add(task.id);
schedulerLog.log(`Task ${task.id} dispatch blocked — ${nodeValidation.reason}`);
await this.store.logEntry(task.id, nodeValidation.reason);
}
return null;
}
this.wasNodeDispatchValidationBlocked.delete(task.id);
}
if (effectiveNode.nodeId !== undefined && this.options.nodeHealthMonitor) {
const localNodeId = this.options.localNodeId ?? "local";
if (freshTask.checkoutNodeId && freshTask.checkedOutBy && freshTask.checkoutNodeId !== localNodeId) {
const ownerNodeHealth = this.options.nodeHealthMonitor.getNodeHealth(freshTask.checkoutNodeId);
const handoffDecision = decideOwningNodeHandoff({
task: freshTask,
ownerNodeId: freshTask.checkoutNodeId,
ownerNodeHealth,
localNodeId,
handoffPolicy: settings.owningNodeHandoffPolicy,
});
if (handoffDecision.action === "park") {
if (!this.wasNodeBlocked.has(task.id)) {
this.wasNodeBlocked.add(task.id);
if (ownerNodeHealth === "offline" || ownerNodeHealth === "error" || ownerNodeHealth === "online") {
await this.emitNodeUnreachableRecoveryAudit(freshTask, {
ownerNodeId: freshTask.checkoutNodeId,
ownerNodeHealth,
handoffAction: handoffDecision.action,
handoffReason: handoffDecision.reason,
decisionPath: "scheduler-handoff-park",
newColumn: freshTask.column,
dispatchNodeBefore: effectiveNode.nodeId,
dispatchNodeAfter: effectiveNode.nodeId,
});
}
const reason = `Owning-node handoff parked dispatch: ${handoffDecision.reason}`;
schedulerLog.log(`Task ${task.id} dispatch blocked — ${reason}`);
await this.store.logEntry(task.id, reason);
try {
await this.store.recordRunAuditEvent?.({
taskId: freshTask.id,
agentId: "scheduler",
runId: generateSyntheticRunId("scheduler", freshTask.id),
domain: "database",
mutationType: "node:handoff:parked",
target: freshTask.id,
metadata: {
taskId: freshTask.id,
ownerNodeId: freshTask.checkoutNodeId,
ownerNodeHealth:
ownerNodeHealth === "offline" || ownerNodeHealth === "error" || ownerNodeHealth === "online"
? ownerNodeHealth
: "unknown",
localNodeId,
handoffPolicy: settings.owningNodeHandoffPolicy,
decisionReason: handoffDecision.reason,
source: "scheduler.dispatch",
},
});
} catch (error) {
schedulerLog.warn(`Task ${task.id} failed to emit node:handoff:parked audit: ${error instanceof Error ? error.message : String(error)}`);
}
}
return null;
}
await this.store.logEntry(task.id, `Owning-node handoff applied: ${handoffDecision.reason}`);
try {
await this.store.recordRunAuditEvent?.({
taskId: freshTask.id,
agentId: "scheduler",
runId: generateSyntheticRunId("scheduler", freshTask.id),
domain: "database",
mutationType: handoffDecision.action === "reassign-local" ? "node:handoff:reassign-local" : "node:handoff:reassign-any",
target: freshTask.id,
metadata: {
taskId: freshTask.id,
ownerNodeId: freshTask.checkoutNodeId,
ownerNodeHealth:
ownerNodeHealth === "offline" || ownerNodeHealth === "error" || ownerNodeHealth === "online"
? ownerNodeHealth
: "unknown",
localNodeId,
handoffPolicy: settings.owningNodeHandoffPolicy,
decisionReason: handoffDecision.reason,
source: "scheduler.dispatch",
},
});
} catch (error) {
schedulerLog.warn(`Task ${task.id} failed to emit node:handoff audit: ${error instanceof Error ? error.message : String(error)}`);
}
const dispatchNodeBefore = effectiveNode.nodeId;
if (handoffDecision.action === "reassign-local") {
effectiveNode = { nodeId: undefined, source: "local" };
}
if (ownerNodeHealth === "offline" || ownerNodeHealth === "error" || ownerNodeHealth === "online") {
await this.emitNodeUnreachableRecoveryAudit(freshTask, {
ownerNodeId: freshTask.checkoutNodeId,
ownerNodeHealth,
handoffAction: handoffDecision.action,
handoffReason: handoffDecision.reason,
decisionPath:
handoffDecision.action === "reassign-local"
? "scheduler-handoff-reassign-local"
: "scheduler-handoff-reassign-any",
newColumn: freshTask.column,
dispatchNodeBefore,
dispatchNodeAfter: effectiveNode.nodeId,
});
}
}
if (effectiveNode.nodeId !== undefined) {
const nodeHealth = this.options.nodeHealthMonitor.getNodeHealth(effectiveNode.nodeId);
const decision = applyUnavailableNodePolicy({
effectiveNode,
nodeHealth,
policy: settings.unavailableNodePolicy,
});
if (!decision.allowed) {
if (!this.wasNodeBlocked.has(task.id)) {
this.wasNodeBlocked.add(task.id);
schedulerLog.log(`Task ${task.id} dispatch blocked — ${decision.reason}`);
await this.store.logEntry(task.id, decision.reason);
}
return null;
}
this.wasNodeBlocked.delete(task.id);
if (decision.fallbackToLocal) {
schedulerLog.log(`Task ${task.id} falling back to local — ${decision.reason}`);
await this.store.logEntry(task.id, decision.reason);
effectiveNode = { nodeId: undefined, source: "local" };
}
}
}
if (latestSettings.ephemeralAgentsEnabled === false && !freshTask.assignedAgentId && this.options.agentStore) {
const selectedAgent = await selectPermanentAgentForTask({
task: freshTask,
agentStore: this.options.agentStore,
taskStore: this.store,
});
if (!selectedAgent) {
await this.store.updateTask(task.id, { status: "queued" });
if (!this.wasPermanentAgentUnavailable.has(task.id)) {
await this.logDispatchQueuedReason(
task.id,
"queued — no permanent executor available (ephemeral agents disabled)",
);
this.wasPermanentAgentUnavailable.add(task.id);
}
return null;
}
await this.store.updateTask(task.id, { assignedAgentId: selectedAgent.id });
await this.store.logEntry(
task.id,
`Auto-assigned to permanent agent ${selectedAgent.id} (ephemeral agents disabled)`,
);
this.wasPermanentAgentUnavailable.delete(task.id);
} else {
this.wasPermanentAgentUnavailable.delete(task.id);
}
const oscillationSettings = latestSettings as Settings & {
dispatchOscillationSettleMs?: number;
dispatchOscillationThreshold?: number;
dispatchOscillationWindowMs?: number;
};
const dispatchSettleMs = oscillationSettings.dispatchOscillationSettleMs
?? DEFAULT_DISPATCH_OSCILLATION_SETTLE_MS;
const dispatchOscillationThreshold = oscillationSettings.dispatchOscillationThreshold
?? DEFAULT_DISPATCH_OSCILLATION_THRESHOLD;
const dispatchOscillationWindowMs = oscillationSettings.dispatchOscillationWindowMs
?? DEFAULT_DISPATCH_OSCILLATION_WINDOW_MS;
const recentEngineTodoMovedAt = this.recentEngineTodoRequeues.get(task.id);
if (recentEngineTodoMovedAt) {
if (freshTask.columnMovedAt !== recentEngineTodoMovedAt) {
this.recentEngineTodoRequeues.delete(task.id);
} else {
const movedAtMs = Date.parse(recentEngineTodoMovedAt);
const settleAgeMs = Number.isFinite(movedAtMs) ? Math.max(0, Date.now() - movedAtMs) : dispatchSettleMs;
if (settleAgeMs < dispatchSettleMs) {
schedulerLog.log(`Task ${task.id} was engine-requeued ${settleAgeMs}ms ago — waiting ${dispatchSettleMs}ms settle window before redispatch`);
return null;
}
this.recentEngineTodoRequeues.delete(task.id);
}
}
const dispatchTimestamp = new Date().toISOString();
const lastDispatchAtMs = freshTask.lastDispatchAt ? Date.parse(freshTask.lastDispatchAt) : Number.NaN;
const priorDispatchWithinWindow = Number.isFinite(lastDispatchAtMs)
&& Date.now() - lastDispatchAtMs <= dispatchOscillationWindowMs;
const nextDispatchStormCount = priorDispatchWithinWindow
? (freshTask.dispatchStormCount ?? 0) + 1
: 1;
if (nextDispatchStormCount > dispatchOscillationThreshold) {
const oscillationError = freshTask.error
?? `DISPATCH_OSCILLATION: detected ${nextDispatchStormCount} todo↔in-progress cycles within ${dispatchOscillationWindowMs}ms. Task auto-paused for operator review.`;
await this.store.updateTask(task.id, {
dispatchStormCount: nextDispatchStormCount,
lastDispatchAt: dispatchTimestamp,
paused: true,
pausedReason: "dispatch-oscillation",
status: freshTask.status ?? "queued",
error: oscillationError,
});
await this.store.logEntry(
task.id,
`Dispatch oscillation auto-paused after ${nextDispatchStormCount} cycles within ${dispatchOscillationWindowMs}ms`,
);
await this.store.appendAgentLog?.(
task.id,
"Dispatch oscillation detected — task auto-paused for operator review",
"text",
`cycleCount=${nextDispatchStormCount} windowMs=${dispatchOscillationWindowMs}`,
);
await this.store.recordRunAuditEvent?.({
taskId: task.id,
agentId: "scheduler",
runId: generateSyntheticRunId("scheduler-dispatch-oscillation", task.id),
domain: "database",
mutationType: "task:dispatch-oscillation-terminalized",
target: task.id,
metadata: {
taskId: task.id,
cycleCount: nextDispatchStormCount,
windowMs: dispatchOscillationWindowMs,
lastMoveSource: recentEngineTodoMovedAt ? "engine" : "scheduler",
},
});
schedulerLog.warn(`Task ${task.id} auto-paused after dispatch oscillation threshold ${dispatchOscillationThreshold} was exceeded (${nextDispatchStormCount} cycles)`);
return null;
}
if (settings.groupOverlappingFiles) {
const taskScope = await getFilteredFileScope(task.id);
if (taskScope.length > 0 && !isCoordinationOnlyTask(task, taskScope)) {
@@ -2148,32 +2496,58 @@ export class Scheduler {
return null;
}
if (task.overlapBlockedBy) {
await this.store.updateTask(task.id, { overlapBlockedBy: null });
}
activeScopes.set(task.id, taskScope);
activeScopeColumns.set(task.id, "in-progress");
reservedScope = true;
} else if (task.overlapBlockedBy) {
await this.store.updateTask(task.id, { overlapBlockedBy: null });
if (isCoordinationOnlyTask(task, taskScope)) {
await this.store.logEntry(
task.id,
"coordination/no-commit task bypassed non-implementation overlap lease",
);
}
}
}
if (Number.isFinite(maxWorktrees) && reservedWorktreeSlots >= maxWorktrees) {
const concurrencyDiagnostic = computeConcurrencyGateDiagnostic({
agentSlots: reservedConcurrentSlots,
maxConcurrent,
activeWorktrees: reservedWorktreeSlots,
maxWorktrees,
semaphore: this.options.semaphore,
inProgressTaskIds,
});
/*
FNXC:WorkflowScheduling 2026-06-23-20:58:
The workflow hold/release sweep is the only todo pickup path, so it must honor the same maxConcurrent, maxWorktrees, and shared semaphore pressure before releasing a task to in-progress. This is deliberately a non-mutating preflight: executor owns the actual semaphore acquire, and the scheduler only prevents capacity-obvious over-release without double-acquiring slots.
*/
if (concurrencyDiagnostic.available <= 0) {
if (reservedScope) {
activeScopes.delete(task.id);
activeScopeColumns.delete(task.id);
}
const reason = formatConcurrencyLimitReason(concurrencyDiagnostic);
await this.store.updateTask(task.id, { status: "queued" });
await this.logDispatchQueuedReason(task.id, reason, formatConcurrencyLimitMemoKey(concurrencyDiagnostic));
return null;
}
const sem = this.options.semaphore;
if (sem && !sem.tryAcquire()) {
if (reservedScope) {
activeScopes.delete(task.id);
activeScopeColumns.delete(task.id);
}
return null;
}
dispatchPrepByTaskId.set(task.id, {
baseBranch: this.resolveBaseBranch(freshTask, tasks),
dispatchStormCount: nextDispatchStormCount,
dispatchTimestamp,
effectiveNodeId: effectiveNode.nodeId ?? null,
effectiveNodeSource: effectiveNode.source,
task: freshTask,
});
reservedWorktreeSlots += 1;
reservedConcurrentSlots += 1;
let released = false;
return {
release: () => {
@@ -2184,13 +2558,45 @@ export class Scheduler {
activeScopeColumns.delete(task.id);
}
reservedWorktreeSlots = Math.max(0, reservedWorktreeSlots - 1);
sem?.release();
reservedConcurrentSlots = Math.max(0, reservedConcurrentSlots - 1);
dispatchPrepByTaskId.delete(task.id);
},
};
},
allocateWorktree: (task, reservedNames) =>
planTaskWorktreePath(task, this.store.getRootDir(), undefined, reservedNames, {}),
});
for (const taskId of result.released) {
const prep = dispatchPrepByTaskId.get(taskId);
if (!prep) continue;
/*
FNXC:WorkflowScheduling 2026-06-23-21:49:
A workflow hold release is not a committed dispatch until moveTask succeeds and appears in result.released. Only then may the scheduler emit "Starting" and clear queued state. Call onSchedule before best-effort metadata/log writes so a post-release store/log failure does not strand an in-progress task without executor handoff.
*/
schedulerLog.log(`Starting ${taskId}: ${prep.task.title || taskId} (deps satisfied)`);
const latest = await this.store.getTask(taskId).catch(() => null);
try {
this.options.onSchedule?.(latest ?? prep.task);
} catch (error) {
schedulerLog.error(`onSchedule failed for ${taskId}:`, error);
}
await this.store.updateTask(taskId, {
status: null,
blockedBy: null,
executionStartBranch: prep.baseBranch ?? undefined,
effectiveNodeId: prep.effectiveNodeId,
effectiveNodeSource: prep.effectiveNodeSource,
mergeRetries: 0,
dispatchStormCount: prep.dispatchStormCount,
lastDispatchAt: prep.dispatchTimestamp,
});
this.recentEngineTodoRequeues.delete(taskId);
this.wasNodeBlocked.delete(taskId);
this.wasNodeDispatchValidationBlocked.delete(taskId);
this.wasPermanentAgentUnavailable.delete(taskId);
this.clearDispatchQueuedReasonMemo(taskId);
await this.store.logEntry(taskId, `Node routing resolved: ${prep.effectiveNodeId ?? "local"} (source: ${prep.effectiveNodeSource})`);
}
} catch (error) {
schedulerLog.error("Hold/release sweep failed:", error);
}

1
packages/engine/vitest.config.ts
View File

@@ -77,6 +77,7 @@ export default defineConfig({
"src/__tests__/hold-release.test.ts",
"src/__tests__/workflow-graph-task-runner.test.ts",
"src/__tests__/workflow-graph-executor-parity.test.ts",
"src/__tests__/scheduler-workflow-cutover.test.ts",
"src/__tests__/executor-base-commit-capture.test.ts",
"src/__tests__/executor-capture-modified-files-attribution.test.ts",
"src/__tests__/triage-preflight.test.ts",