FN-8822: prevent worktree capacity leaks
Keep live-task capacity admission visible and consistent across all lifecycle lanes. - Report canonical active holders when worktree or concurrent-task capacity is exhausted. - Persist deduplicated queue reasons for merge, triage, and workflow-continuation admission. - Cover retained worktree behavior and document the capacity model. Files changed: .changeset/fn-8822-worktree-capacity-leak.md | 7 +++ docs/architecture.md | 1 + docs/settings-reference.md | 4 +- .../src/__tests__/agent-heartbeat-worktree.test.ts | 11 ++++- .../engine/src/__tests__/project-engine.test.ts | 4 ++ ...ecutor-no-task-done-vs-worktree-reclaim.test.ts | 55 +++++++++++++++++++--- ...admission-worktree-ledger-renamed-lanes.test.ts | 4 ++ .../workflow-continuation-capacity.test.ts | 8 +++- .../src/__tests__/worktree-acquisition.test.ts | 6 +-- packages/engine/src/concurrency/concurrency.ts | 22 +++++++++ packages/engine/src/project-engine.ts | 34 ++++++++++++- packages/engine/src/runtimes/in-process-runtime.ts | 33 ++++++++++++- packages/engine/src/scheduler.ts | 53 ++++++++++----------- packages/engine/src/triage.ts | 18 +++++++ 14 files changed, 215 insertions(+), 45 deletions(-) Fusion-Task-Id: FN-8822 Fusion-Task-Lineage: 73eda70b-2b74-4b49-a0ff-29d41fd9aab8 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-8822-worktree-capacity-leak.md
Normal file
7
.changeset/fn-8822-worktree-capacity-leak.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Prevent inactive retained worktrees from exhausting live task capacity.
|
||||
category: fix
|
||||
dev: Shares worktree-capacity admission across execution, planning, merge, and workflow continuation lanes.
|
||||
@@ -638,6 +638,7 @@ See [Memory Plugin Contract](./memory-plugin-contract.md) for the full plan.
|
||||
|
||||
### Scheduling and execution
|
||||
- `Scheduler` (`scheduler.ts`) — dependency-aware task scheduling that dispatches eligible todo tasks by priority first, then dependency-unblock fanout within the same priority class (FN-4969), then FIFO (`createdAt` ascending) with task-id fallback. `urgent` always stays ahead of lower priorities, and overlap/file-scope blockers are excluded from fanout weighting.
|
||||
- **Worktree-capacity admission (FN-8822):** scheduler execute, triage specify, project-engine merge, and direct workflow-planning continuation handoffs share one serialized project coordinator. Its ceiling is `min(maxConcurrent, maxWorktrees)` when worktree limiting is enabled and counts only canonical live task claims plus transient reservations—not retained directories, stale metadata, paused/terminal tasks, or orphans. A genuinely full cap persists a deduplicated queued reason with the `maxWorktrees` gate, used/limit, and holder IDs through ordinary task status/log APIs. Retained worktrees are deliberately non-destructive: cleanup and pooling preserve active, dirty, or uniquely committed work.
|
||||
- `blockedBy` invariant (FN-3924/FN-4091): the field is only durable when it references a current unresolved explicit dependency (or, for dependency-free tasks, an active overlap blocker). Completion gating now validates `blockedBy` through live task resolution: missing blockers and blockers already in `done`/`archived` are treated as stale, while only still-active blockers continue to prevent `fn_task_done`. If no current blocker remains, scheduler/event reconciliation clears `blockedBy` to `null` and re-evaluates from live task state.
|
||||
- Dependency-cycle invariant (FN-5256): task dependency graphs are acyclic at write time (`DependencyCycleError` in `TaskStore` for `createTask`, `createTaskWithReservedId`, `updateTask`, and `applyReplicatedTaskCreate`) with `task:dependency-cycle-rejected` audit evidence. Self-healing batch 2 adds `reconcileDependencyCycles`, which emits `task:dependency-cycle-detected`, auto-repairs only bounded umbrella-back-edge loops via `task:auto-reconciled-dependency-cycle`, and leaves ambiguous cycles untouched with `task:dependency-cycle-unrepaired` for operator inspection.
|
||||
- Dependency-blocking lease invariant (FN-6292): an `in-progress` task with unmet scheduling dependencies must not contribute an active file-scope lease in scheduler lease maps. This prevents a holder from queueing its own dependency behind its lease and creating a circular wait.
|
||||
|
||||
@@ -467,7 +467,7 @@ Security-sensitive file-browser escape hatches are project-only. `allowAbsoluteF
|
||||
| `maxConcurrentVerifications` | `number` | `1` | Max concurrent verification subprocesses (`fn_run_verification`, merge test/build commands) process-wide. Caps stacked monorepo typecheck/build so concurrent tasks do not peg host CPU. Range **1–8** (clamped at runtime and in Settings). Editable from Settings → Scheduling. Each project engine registers its cap; the effective process limit is the **minimum** of registered project caps. |
|
||||
| `maxTriageConcurrent` | `number` | `2` | Legacy persisted value; ignored. Planning shares `maxConcurrent` and no Max Triage control is displayed. |
|
||||
| `globalMaxConcurrent` | `number` | `4` | System-wide max concurrent agents across all projects. |
|
||||
| `maxWorktrees` | `number` | `4` | Max git worktrees. Editable from Settings and the Command Center Overview controls dashboard. |
|
||||
| `maxWorktrees` | `number` | `4` | Max simultaneous **live task claims** when worktree limiting is enabled, shared by scheduler execution, triage planning, project-engine merge, and workflow-continuation admission. Retained worktree directories for queued, paused, terminal, or missing tasks do not consume this cap. A real full cap leaves the candidate queued with a task-log/status reason naming `maxWorktrees`, used/limit, and live holders. Editable from Settings and the Command Center Overview controls dashboard. |
|
||||
| `pollIntervalMs` | `number` | `15000` | Scheduler poll interval (ms). |
|
||||
| `heartbeatMultiplier` | `number` | `1` | Global multiplier applied to agent heartbeat timing: both heartbeat intervals and unresponsive timeout bases. Configured from the Agents screen (not Settings). |
|
||||
| `heartbeatScopeDiscipline` | `"strict" \| "lite" \| "off"` | `"strict"` | Heartbeat prompt procedure mode. `strict` keeps coordination-heavy scope discipline, `lite` restores pre-2026-05-11 wording, and `off` uses a minimal procedure. Per-agent `runtimeConfig.heartbeatScopeDiscipline` can override this default. |
|
||||
@@ -561,7 +561,7 @@ When `pushAfterMerge` is enabled, Fusion first tries a working-tree-independent
|
||||
| `worktreeCopyFiles` | `string[]` | `[]` | Repository-root-relative regular files to copy into each newly assigned non-resume task worktree. Configure from Settings → Worktrees with editable rows or Browse (useful for `.env`-style files). Fusion copies these files after fresh creation or pooled-worktree preparation and before `worktreeInitCommand`, secrets-env materialization, and task execution. Blank/duplicate entries are ignored; absolute paths, `..` traversal, missing files, directories, and unreadable/non-regular sources are skipped as non-fatal task-log/audit diagnostics without logging file contents. Resume/existing worktrees are not overwritten. |
|
||||
| `testCommand` | `string` | `undefined` | Merge-time test command (hard gate). When unset, Fusion auto-detects from lockfile. |
|
||||
| `buildCommand` | `string` | `undefined` | Merge-time build command (hard gate). |
|
||||
| `recycleWorktrees` | `boolean` | `false` | Default: off (opt-in). Reuse worktrees from a pool for faster startup. **Mutually exclusive with `worktreeNaming: "task-id"`** (task-pinned worktrees) — enabling both is rejected by the settings API/store, because pinning each task to its own directory is incompatible with the cross-task pool. Recycling is fully functional under `"random"` and `"task-title"` naming. |
|
||||
| `recycleWorktrees` | `boolean` | `false` | Default: off (opt-in). Reuse worktrees from a pool for faster startup. It does not define or reclaim scheduler capacity: admission counts live claims, while pooled/retained paths remain protected from destructive removal when active, dirty, or uniquely committed. **Mutually exclusive with `worktreeNaming: "task-id"`** (task-pinned worktrees) — enabling both is rejected by the settings API/store, because pinning each task to its own directory is incompatible with the cross-task pool. Recycling is fully functional under `"random"` and `"task-title"` naming. |
|
||||
| `showWorktreeGrouping` | `boolean` | `false` | Default: off. When off, WIP/processing columns render plain task cards without worktree group shells or worktree-name labels in both legacy and workflow-mode boards. When on, every WIP/processing column groups tasks by worktree and shows worktree names, including workflow-mode columns flagged as counting toward WIP. |
|
||||
| `openTasksInRightSidebar` | `boolean` | `false` | Default: off. When off, board task-card clicks keep the existing full-panel task detail that replaces the board. When on and the right dock is active on desktop/tablet, board task-card clicks open the task detail in the right sidebar so the board stays visible; mobile or hidden/inactive right-dock states automatically fall back to the full-panel behavior. Non-board task-open paths, including list split detail, floating pop-outs, graph/plugin opens, and deep `changes`/`retries`/`workflow` opens, keep their existing behavior; ordinary right-dock Tasks-list cards are governed by `openMobileTasksInPopup` first and otherwise use embedded dock detail. |
|
||||
| `openMobileTasksInPopup` | `boolean` | `false` | Default: off. When off, ordinary board task-card clicks keep the existing fallback behavior: the full-panel task detail, or the right dock when `openTasksInRightSidebar` is on and the dock is active; List row/card opens keep the desktop split-detail pane or the mobile/tablet docked detail; ordinary right-dock Tasks-list clicks open embedded dock detail with the normal back-to-list controls. When on, ordinary board task-card clicks, List row/card opens, and right-dock Tasks-list clicks open the task in the existing task popup/FloatingWindow surface on desktop, tablet, and mobile so the board, List view, or dock list remains visible; this popup route takes precedence over right-dock routing for those ordinary clicks. Desktop/tablet task popups restore the last saved popup size and position across task IDs and use the board/task-detail layer rather than the global utility layer, while their Activity dropdown stays above and attached during popup drag/resize; mobile task popups remain full-screen sheets. Board task-card deep `changes`/`retries`/`workflow` chips also open the popup with their requested tab; context-menu/refine/detail links, graph/plugin opens, nested task-detail opens, and explicit pop-out actions keep their existing behavior. |
|
||||
|
||||
@@ -49,11 +49,18 @@ describe("heartbeat worktree cwd", () => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
it("uses acquired worktree cwd for task-scoped runs", async () => {
|
||||
it("refreshes acquired worktree before creating task-scoped session", async () => {
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore, rootDir: "/repo" });
|
||||
await monitor.executeHeartbeat({ agentId: "a1", source: "on_demand" });
|
||||
expect(worktreeAcquisition.acquireTaskWorktree).toHaveBeenCalled();
|
||||
|
||||
expect(worktreeAcquisition.acquireTaskWorktree).toHaveBeenCalledWith(expect.objectContaining({
|
||||
task: expect.objectContaining({ id: "FN-1" }),
|
||||
refreshStaleBase: true,
|
||||
}));
|
||||
expect(piModule.createFnAgent).toHaveBeenCalledWith(expect.objectContaining({ cwd: "/tmp/wt" }));
|
||||
expect(worktreeAcquisition.acquireTaskWorktree.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
vi.mocked(piModule.createFnAgent).mock.invocationCallOrder[0],
|
||||
);
|
||||
});
|
||||
|
||||
it("uses rootDir for no-task runs", async () => {
|
||||
|
||||
@@ -2538,6 +2538,10 @@ describe("ProjectEngine paused in-review auto-merge behavior", () => {
|
||||
expect(privateEngine.mergeQueue).not.toContain("FN-MERGE-WAITING");
|
||||
expect(privateEngine.capacityDeferredMergeTaskIds.has("FN-MERGE-WAITING")).toBe(true);
|
||||
expect(engine.isMergePending("FN-MERGE-WAITING")).toBe(true);
|
||||
expect(mockStore.store.logEntry).toHaveBeenCalledWith(
|
||||
"FN-MERGE-WAITING",
|
||||
expect.stringContaining("maxWorktrees capacity exhausted: used=1/1"),
|
||||
);
|
||||
|
||||
// An unrelated queue wake must not make the deferred task runnable before its timer.
|
||||
mockStore.store.getTask.mockResolvedValue({
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { describe, it, expect, vi, beforeEach } from "vitest";
|
||||
import "../executor-test-helpers.js";
|
||||
import { TaskExecutor } from "../../executor.js";
|
||||
import * as worktreeAcquisition from "../../worktree/worktree-acquisition.js";
|
||||
import { mockedCreateFnAgent, createMockStore, resetExecutorMocks } from "../executor-test-helpers.js";
|
||||
|
||||
/*
|
||||
@@ -45,9 +46,51 @@ function makeSession() {
|
||||
};
|
||||
}
|
||||
|
||||
function makeExecutor(store: ReturnType<typeof createMockStore>): TaskExecutor {
|
||||
store.getRootDir = vi.fn().mockReturnValue("/tmp/test");
|
||||
return new TaskExecutor(store as any, "/tmp/test", {
|
||||
agentStore: {
|
||||
listAgents: vi.fn().mockResolvedValue([{
|
||||
id: "agent-executor",
|
||||
name: "Executor",
|
||||
role: "executor",
|
||||
roles: ["executor"],
|
||||
state: "active",
|
||||
runtimeConfig: {},
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
}]),
|
||||
getAgent: vi.fn().mockResolvedValue({
|
||||
id: "agent-executor",
|
||||
name: "Executor",
|
||||
role: "executor",
|
||||
roles: ["executor"],
|
||||
state: "active",
|
||||
runtimeConfig: {},
|
||||
}),
|
||||
checkoutTask: vi.fn().mockResolvedValue(undefined),
|
||||
acquireWorkflowSessionCapacity: vi.fn().mockResolvedValue("acquired"),
|
||||
releaseWorkflowSessionCapacity: vi.fn().mockResolvedValue(undefined),
|
||||
},
|
||||
} as any);
|
||||
}
|
||||
|
||||
describe("reliability interactions: executor no-fn_task_done vs worktree reclaim", () => {
|
||||
beforeEach(() => {
|
||||
resetExecutorMocks();
|
||||
/*
|
||||
FNXC:WorktreeBaseRefresh 2026-08-08-03:59:
|
||||
This retry/liveness suite owns the post-session lifecycle boundary, not Git reconciliation.
|
||||
Keep its synthetic /tmp checkout acquisition successful so refresh-enabled executor dispatch
|
||||
reaches the retry logic; dedicated temporary-Git tests prove the stale-base contract.
|
||||
*/
|
||||
vi.spyOn(worktreeAcquisition, "acquireTaskWorktree").mockResolvedValue({
|
||||
worktreePath: "/tmp/test/.worktrees/fn-4601",
|
||||
branch: "fusion/fn-4601",
|
||||
source: "existing",
|
||||
hydrated: false,
|
||||
isResume: true,
|
||||
});
|
||||
});
|
||||
|
||||
it("pre-retry liveness recheck aborts retry and silently requeues (FN-4806)", async () => {
|
||||
@@ -75,7 +118,7 @@ describe("reliability interactions: executor no-fn_task_done vs worktree reclaim
|
||||
|
||||
mockedCreateFnAgent.mockResolvedValue({ session: makeSession() } as any);
|
||||
|
||||
const executor = new TaskExecutor(store as any, "/tmp/test");
|
||||
const executor = makeExecutor(store);
|
||||
await executor.execute(state);
|
||||
|
||||
expect(mockedCreateFnAgent).toHaveBeenCalledTimes(1);
|
||||
@@ -113,7 +156,7 @@ describe("reliability interactions: executor no-fn_task_done vs worktree reclaim
|
||||
.mockResolvedValueOnce({ session: makeSession() } as any)
|
||||
.mockRejectedValueOnce(new Error("Refusing to start coding agent in missing worktree: /tmp/test/.worktrees/fn-4601"));
|
||||
|
||||
const executor = new TaskExecutor(store as any, "/tmp/test");
|
||||
const executor = makeExecutor(store);
|
||||
await executor.execute(state);
|
||||
|
||||
expect(store.updateTask).toHaveBeenCalledWith("FN-4601", expect.objectContaining({
|
||||
@@ -145,7 +188,7 @@ describe("reliability interactions: executor no-fn_task_done vs worktree reclaim
|
||||
.mockResolvedValueOnce({ session: makeSession() } as any)
|
||||
.mockRejectedValueOnce(new Error("boom"));
|
||||
|
||||
const executor = new TaskExecutor(store as any, "/tmp/test");
|
||||
const executor = makeExecutor(store);
|
||||
await executor.execute(state);
|
||||
|
||||
// FNXC:WorkflowLifecycle 2026-07-01-21:05: A non-recoverable execute error follows the terminal
|
||||
@@ -185,7 +228,7 @@ describe("reliability interactions: executor no-fn_task_done vs worktree reclaim
|
||||
|
||||
mockedCreateFnAgent.mockResolvedValue({ session: makeSession() } as any);
|
||||
|
||||
const executor = new TaskExecutor(store as any, "/tmp/test");
|
||||
const executor = makeExecutor(store);
|
||||
await executor.execute(state);
|
||||
|
||||
expect(store.moveTask).toHaveBeenCalledWith("FN-4601", "todo", { preserveProgress: true });
|
||||
@@ -226,7 +269,7 @@ describe("reliability interactions: executor no-fn_task_done vs worktree reclaim
|
||||
|
||||
mockedCreateFnAgent.mockResolvedValue({ session: makeSession() } as any);
|
||||
|
||||
const executor = new TaskExecutor(store as any, "/tmp/test");
|
||||
const executor = makeExecutor(store);
|
||||
await executor.execute(makeTask({ taskDoneRetryCount: 3 }));
|
||||
|
||||
// No resurrection: the park must abort before a second session spawns.
|
||||
@@ -252,7 +295,7 @@ describe("reliability interactions: executor no-fn_task_done vs worktree reclaim
|
||||
|
||||
mockedCreateFnAgent.mockResolvedValue({ session: makeSession() } as any);
|
||||
|
||||
const executor = new TaskExecutor(store as any, "/tmp/test");
|
||||
const executor = makeExecutor(store);
|
||||
await executor.execute(state);
|
||||
|
||||
expect(mockedCreateFnAgent.mock.calls.length).toBeGreaterThan(1);
|
||||
|
||||
@@ -216,6 +216,10 @@ describe("planning admission's worktree ledger on a renamed board", () => {
|
||||
const throttle = recorded.filter((event) => event.type === "task:plan-admission-throttled");
|
||||
expect(throttle.length, "a live task must still consume its worktree-capacity slot").toBeGreaterThan(0);
|
||||
expect(specifyTask).not.toHaveBeenCalled();
|
||||
expect(store.logEntry).toHaveBeenCalledWith(
|
||||
"FN-WAITING",
|
||||
expect.stringContaining("maxWorktrees capacity exhausted: used=1/1"),
|
||||
);
|
||||
});
|
||||
|
||||
it("counts a pending optional workflow-step lease in final planning admission", async () => {
|
||||
|
||||
@@ -42,6 +42,7 @@ function store(
|
||||
getTaskWorkflowSelection: vi.fn(() => undefined),
|
||||
getTaskWorkflowSelectionAsync: vi.fn(async () => undefined),
|
||||
getWorkflowDefinition: vi.fn(async () => undefined),
|
||||
logEntry: vi.fn(async () => undefined),
|
||||
} as unknown as TaskStore;
|
||||
}
|
||||
|
||||
@@ -78,9 +79,10 @@ describe("workflow continuation active-slot admission", () => {
|
||||
}],
|
||||
}));
|
||||
const dispatch = vi.fn(async () => {});
|
||||
const taskStore = store([...active, task(CONTINUATION_ID)]);
|
||||
|
||||
const admitted = await admitPlanningContinuation({
|
||||
store: store([...active, task(CONTINUATION_ID)]),
|
||||
store: taskStore,
|
||||
projectId: PROJECT_ID,
|
||||
task: task(CONTINUATION_ID),
|
||||
item,
|
||||
@@ -89,6 +91,10 @@ describe("workflow continuation active-slot admission", () => {
|
||||
|
||||
expect(admitted).toBe(false);
|
||||
expect(dispatch).not.toHaveBeenCalled();
|
||||
expect(taskStore.logEntry).toHaveBeenCalledWith(
|
||||
CONTINUATION_ID,
|
||||
expect.stringContaining("maxWorktrees capacity exhausted: used=9/9"),
|
||||
);
|
||||
});
|
||||
|
||||
it("starts the ninth task when only eight active tasks hold slots", async () => {
|
||||
|
||||
@@ -10,7 +10,7 @@ import * as desktopArtifacts from "../worktree/worktree-desktop-artifacts.js";
|
||||
import * as branchConflicts from "../execution/branch-conflicts.js";
|
||||
|
||||
vi.mock("../worktree/worktree-pool.js", async () => {
|
||||
const actual = await vi.importActual<any>("../worktree-pool.js");
|
||||
const actual = await vi.importActual<any>("../worktree/worktree-pool.js");
|
||||
return {
|
||||
...actual,
|
||||
classifyTaskWorktree: vi.fn().mockResolvedValue({ ok: true }),
|
||||
@@ -19,7 +19,7 @@ vi.mock("../worktree/worktree-pool.js", async () => {
|
||||
});
|
||||
|
||||
vi.mock("../execution/branch-conflicts.js", async () => {
|
||||
const actual = await vi.importActual<any>("../branch-conflicts.js");
|
||||
const actual = await vi.importActual<any>("../execution/branch-conflicts.js");
|
||||
return {
|
||||
...actual,
|
||||
classifyBootstrapMisbinding: vi.fn().mockResolvedValue({
|
||||
@@ -47,7 +47,7 @@ resolves git paths and throws there, which the pool catch treats as prepare fail
|
||||
through to fresh. No-op the guard so classification + pool wiring stay under test.
|
||||
*/
|
||||
vi.mock("../worktree/worktree-hooks.js", async () => {
|
||||
const actual = await vi.importActual<any>("../worktree-hooks.js");
|
||||
const actual = await vi.importActual<any>("../worktree/worktree-hooks.js");
|
||||
return {
|
||||
...actual,
|
||||
installTaskWorktreeIdentityGuard: vi.fn().mockResolvedValue(undefined),
|
||||
|
||||
@@ -36,6 +36,28 @@ export function resolveActiveTaskCapacityLimit(params: {
|
||||
: Math.min(params.maxConcurrent, maxWorktrees);
|
||||
}
|
||||
|
||||
/**
|
||||
* FNXC:WorktreeCapacity 2026-08-08-04:27:
|
||||
* Every production admission owner must persist the same operator-visible explanation when the
|
||||
* shared live-task ceiling is full. Retained directories are not holders: report only canonical
|
||||
* live task IDs, and name `maxWorktrees` only when it is the binding configured ceiling.
|
||||
*/
|
||||
export function formatAdmissionCapacityQueuedReason(params: {
|
||||
maxConcurrent: number;
|
||||
maxWorktrees: number;
|
||||
worktreeLimitEnabled?: boolean;
|
||||
claimed: number;
|
||||
holderTaskIds: Iterable<string>;
|
||||
}): string {
|
||||
const limit = resolveActiveTaskCapacityLimit(params);
|
||||
const worktreeLimit = resolveWorktreeCapacityLimit(params);
|
||||
const gate = worktreeLimit !== null && worktreeLimit <= params.maxConcurrent
|
||||
? "maxWorktrees"
|
||||
: "maxConcurrent";
|
||||
const holders = [...new Set(params.holderTaskIds)].sort();
|
||||
return `queued — ${gate} capacity exhausted: used=${params.claimed}/${limit}; holders=${holders.join(",") || "none"}`;
|
||||
}
|
||||
|
||||
/** Lifecycle lanes ordered by the project admission coordinator. */
|
||||
export type AdmissionLane = "review" | "execute" | "planning";
|
||||
|
||||
|
||||
@@ -78,6 +78,7 @@ import { sweepStaleAutostashes, VerificationError } from "./merger.js";
|
||||
import { runAiMerge, landWorkspaceTask, WorkspacePartialLandError, WorkspaceRepoLandBusyError } from "./merge/merger-ai.js";
|
||||
import { promoteBranchGroup, type BranchGroupPromotionResult, type CreateGroupPrFn, type SyncGroupPrFn } from "./merge/group-merge-coordinator.js";
|
||||
import {
|
||||
formatAdmissionCapacityQueuedReason,
|
||||
persistedTopLevelAgentTaskIdsFromStore,
|
||||
projectAdmissionCoordinator,
|
||||
resolveActiveTaskCapacityLimit,
|
||||
@@ -461,6 +462,8 @@ export class ProjectEngine {
|
||||
private mergeActive = new Set<string>();
|
||||
/** Capacity-deferred ids stay out of the runnable queue until their retry timer fires. */
|
||||
private readonly capacityDeferredMergeTaskIds = new Set<string>();
|
||||
/** Last persisted live-cap reason per merge; avoids rewriting the task log each poll. */
|
||||
private readonly capacityDeferredMergeReasons = new Map<string, string>();
|
||||
private readonly capacityDeferredMerges = new Map<string, {
|
||||
timer: ReturnType<typeof setTimeout>;
|
||||
resolvers: MergeResolver[];
|
||||
@@ -1356,6 +1359,7 @@ export class ProjectEngine {
|
||||
}
|
||||
this.capacityDeferredMerges.clear();
|
||||
this.capacityDeferredMergeTaskIds.clear();
|
||||
this.capacityDeferredMergeReasons.clear();
|
||||
this.stopPlannerOverseerPoll();
|
||||
|
||||
/*
|
||||
@@ -4008,7 +4012,35 @@ export class ProjectEngine {
|
||||
},
|
||||
}],
|
||||
});
|
||||
if (!selected) return undefined;
|
||||
if (!selected) {
|
||||
const snapshot = await getMergeClaimSnapshot();
|
||||
const limit = resolveActiveTaskCapacityLimit({
|
||||
maxConcurrent: admissionSettings.maxConcurrent ?? 2,
|
||||
maxWorktrees: admissionSettings.maxWorktrees ?? 4,
|
||||
worktreeLimitEnabled: admissionSettings.worktreeLimitEnabled,
|
||||
});
|
||||
if (snapshot.count >= limit) {
|
||||
/*
|
||||
FNXC:ConcurrencyAdmission 2026-08-08-04:27:
|
||||
A merge capacity defer used to be invisible because its queue is internal. Persist
|
||||
the shared live-cap reason on the task itself, but only when the fresh serialized
|
||||
snapshot proves exhaustion rather than a higher-priority candidate winning.
|
||||
*/
|
||||
const reason = formatAdmissionCapacityQueuedReason({
|
||||
maxConcurrent: admissionSettings.maxConcurrent ?? 2,
|
||||
maxWorktrees: admissionSettings.maxWorktrees ?? 4,
|
||||
worktreeLimitEnabled: admissionSettings.worktreeLimitEnabled,
|
||||
claimed: snapshot.count,
|
||||
holderTaskIds: snapshot.ids,
|
||||
});
|
||||
if (this.capacityDeferredMergeReasons.get(taskId) !== reason) {
|
||||
this.capacityDeferredMergeReasons.set(taskId, reason);
|
||||
await store.logEntry(taskId, reason);
|
||||
}
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
this.capacityDeferredMergeReasons.delete(taskId);
|
||||
try {
|
||||
return await start();
|
||||
} finally {
|
||||
|
||||
@@ -68,6 +68,7 @@ import { createRunAuditor, generateSyntheticRunId } from "../util/run-audit.js";
|
||||
import { setImmediate as setImmediateCb } from "node:timers";
|
||||
import { seedPreReleasePlanReviewContinuation } from "../plan-review-continuation.js";
|
||||
import {
|
||||
formatAdmissionCapacityQueuedReason,
|
||||
persistedTopLevelAgentTaskIdsFromStore,
|
||||
projectAdmissionCoordinator,
|
||||
resolveActiveTaskCapacityLimit,
|
||||
@@ -470,6 +471,7 @@ export async function drainDuePlanningContinuations(
|
||||
}
|
||||
|
||||
const planningContinuationRuns = new Set<string>();
|
||||
const planningContinuationCapacityReasons = new Map<string, string>();
|
||||
|
||||
export async function admitPlanningContinuation(input: {
|
||||
store: TaskStore;
|
||||
@@ -559,7 +561,36 @@ export async function admitPlanningContinuation(input: {
|
||||
},
|
||||
}],
|
||||
});
|
||||
return selected || duplicateHandled;
|
||||
if (selected || duplicateHandled) {
|
||||
planningContinuationCapacityReasons.delete(runKey);
|
||||
return true;
|
||||
}
|
||||
const snapshot = await getAdmissionSnapshot();
|
||||
const limit = resolveActiveTaskCapacityLimit({
|
||||
maxConcurrent: settings.maxConcurrent ?? 2,
|
||||
maxWorktrees: settings.maxWorktrees ?? 4,
|
||||
worktreeLimitEnabled: settings.worktreeLimitEnabled,
|
||||
});
|
||||
if (snapshot.count >= limit) {
|
||||
/*
|
||||
FNXC:ConcurrencyAdmission 2026-08-08-04:27:
|
||||
Direct workflow continuations bypass scheduler task-status handling. A full live-task cap must
|
||||
still be visible through the shared task log, using the same canonical-holder diagnostic as
|
||||
execute, triage, and merge admission; unchanged retries remain deduplicated.
|
||||
*/
|
||||
const reason = formatAdmissionCapacityQueuedReason({
|
||||
maxConcurrent: settings.maxConcurrent ?? 2,
|
||||
maxWorktrees: settings.maxWorktrees ?? 4,
|
||||
worktreeLimitEnabled: settings.worktreeLimitEnabled,
|
||||
claimed: snapshot.count,
|
||||
holderTaskIds: snapshot.ids,
|
||||
});
|
||||
if (planningContinuationCapacityReasons.get(runKey) !== reason) {
|
||||
planningContinuationCapacityReasons.set(runKey, reason);
|
||||
await input.store.logEntry(input.task.id, reason);
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
export function createPlanningContinuationDispatcher(input: {
|
||||
|
||||
@@ -2259,9 +2259,8 @@ export class Scheduler {
|
||||
isReviewColumnRole(columnFlagsForTask(task), task.column);
|
||||
/*
|
||||
FNXC:ConcurrencyIndicators 2026-08-01-19:22:
|
||||
Failed WIP is not a live holder (isRunningAgentTask). Keep the WIP id list aligned so
|
||||
diagnostic maxConcurrent holders and any WIP-only arithmetic do not re-count stranded failed
|
||||
parks that the worktree ledger already excludes.
|
||||
Failed WIP is not a live holder (isRunningAgentTask). Keep the WIP list aligned so
|
||||
same-sweep executor reservation arithmetic does not re-count stranded failed parks.
|
||||
*/
|
||||
const wipTaskIds = tasks
|
||||
.filter((task) => isWipColumnTask(task) && task.status !== "failed")
|
||||
@@ -2278,7 +2277,6 @@ export class Scheduler {
|
||||
const activeWorktreeTaskIds = await persistedTopLevelAgentTaskIdsFromStore(this.store, tasks);
|
||||
let reservedWorktreeSlots = activeWorktreeTaskIds.length;
|
||||
let reservedConcurrentSlots = wipTaskIds.length;
|
||||
const inProgressTaskIds = wipTaskIds;
|
||||
const dispatchPrepByTaskId = new Map<string, {
|
||||
baseBranch: string | null;
|
||||
dispatchStormCount: number;
|
||||
@@ -2818,23 +2816,6 @@ export class Scheduler {
|
||||
}
|
||||
}
|
||||
|
||||
const concurrencyDiagnostic = computeConcurrencyGateDiagnostic({
|
||||
agentSlots: reservedConcurrentSlots,
|
||||
maxConcurrent,
|
||||
activeWorktrees: reservedWorktreeSlots,
|
||||
maxWorktrees,
|
||||
worktreeHolderTaskIds: [...activeWorktreeTaskIds, ...dispatchPrepByTaskId.keys()],
|
||||
semaphore: this.options.semaphore,
|
||||
inProgressTaskIds,
|
||||
topLevelClaimedSlots: reservedWorktreeSlots,
|
||||
});
|
||||
/*
|
||||
The sweep diagnostic is intentionally descriptive only. Worktree preparation can outlive
|
||||
a holder, so using this older snapshot as an admission gate strands queued cards even after
|
||||
capacity frees. The serialized fresh reservation below is the sole capacity authority; the
|
||||
diagnostic remains available to explain a rejection from that authoritative check.
|
||||
*/
|
||||
|
||||
/*
|
||||
FNXC:WorktreeCapacity 2026-08-01-04:38:
|
||||
Serialize the workflow scheduler's direct hold release with planning and merge admission.
|
||||
@@ -2859,7 +2840,7 @@ export class Scheduler {
|
||||
return { count: ids.length, ids };
|
||||
})();
|
||||
let projectSlotReserved = false;
|
||||
await projectAdmissionCoordinator.admitNext({
|
||||
const admittedTaskId = await projectAdmissionCoordinator.admitNext({
|
||||
projectId: this.store.getRootDir(),
|
||||
maxConcurrent: activeTaskLimit,
|
||||
claimed: async () => (await getFinalClaimSnapshot()).count,
|
||||
@@ -2882,14 +2863,28 @@ export class Scheduler {
|
||||
activeScopes.delete(task.id);
|
||||
activeScopeColumns.delete(task.id);
|
||||
}
|
||||
const bindingGate: ConcurrencyGateName = maxWorktrees !== null && maxWorktrees <= maxConcurrent
|
||||
? "maxWorktrees"
|
||||
: "maxConcurrent";
|
||||
const reason = formatConcurrencyLimitReason({
|
||||
...concurrencyDiagnostic,
|
||||
available: 0,
|
||||
bindingGates: [...new Set([...concurrencyDiagnostic.bindingGates, bindingGate])],
|
||||
/*
|
||||
FNXC:WorktreeCapacity 2026-08-08-04:17:
|
||||
A scheduler candidate can lose one serialized admission pass because a higher-priority
|
||||
merge/planning provider was selected, even with a free slot. Only call it exhaustion
|
||||
after the coordinator admitted nobody and report the fresh canonical holder snapshot
|
||||
that made that decision; the pre-sweep snapshot is diagnostic-only and can be stale.
|
||||
*/
|
||||
const freshClaims = await getFinalClaimSnapshot();
|
||||
const exhausted = admittedTaskId === undefined;
|
||||
const freshDiagnostic = computeConcurrencyGateDiagnostic({
|
||||
agentSlots: freshClaims.count,
|
||||
maxConcurrent,
|
||||
activeWorktrees: freshClaims.count,
|
||||
maxWorktrees,
|
||||
worktreeHolderTaskIds: freshClaims.ids,
|
||||
semaphore: this.options.semaphore,
|
||||
inProgressTaskIds: freshClaims.ids,
|
||||
topLevelClaimedSlots: freshClaims.count,
|
||||
});
|
||||
const reason = exhausted
|
||||
? formatConcurrencyLimitReason(freshDiagnostic)
|
||||
: `queued — higher-priority lifecycle admission started: task=${admittedTaskId}`;
|
||||
await this.store.updateTask(task.id, { status: "queued" });
|
||||
await this.logDispatchQueuedReason(task.id, reason);
|
||||
return null;
|
||||
|
||||
@@ -153,6 +153,7 @@ import { buildSessionSkillContext } from "./cli-runtime/session-skill-context.js
|
||||
import {
|
||||
PRIORITY_SPECIFY,
|
||||
computeTopLevelConcurrencyClaimedFromStore,
|
||||
formatAdmissionCapacityQueuedReason,
|
||||
dropPreHeldExecutorSlot,
|
||||
persistedTopLevelAgentTaskIdsFromStore,
|
||||
projectAdmissionCoordinator,
|
||||
@@ -2161,6 +2162,13 @@ export class TriageProcessor {
|
||||
a payload with no reason field at all would read as "unknown".
|
||||
*/
|
||||
const blockedBy = worktreeRoom <= 0 && projectRoom > 0 ? "worktree cap" : "running-agent cap";
|
||||
const capacityReason = formatAdmissionCapacityQueuedReason({
|
||||
maxConcurrent,
|
||||
maxWorktrees: settings.maxWorktrees ?? 4,
|
||||
worktreeLimitEnabled: settings.worktreeLimitEnabled,
|
||||
claimed,
|
||||
holderTaskIds: await persistedTopLevelAgentTaskIdsFromStore(this.store, allTasks),
|
||||
});
|
||||
planLog.log(
|
||||
`Plan throttled by ${blockedBy}: eligible=${triageTasks.length} [${eligibleIds.join(", ")}], ` +
|
||||
`maxConcurrent=${maxConcurrent}, claimed=${claimed}, processing=${this.processing.size}` +
|
||||
@@ -2243,6 +2251,16 @@ export class TriageProcessor {
|
||||
put so the next poll retries.
|
||||
*/
|
||||
this.lastPlanThrottleSignature = throttleSignature;
|
||||
/*
|
||||
FNXC:ConcurrencyAdmission 2026-08-08-04:27:
|
||||
Triage used to emit capacity only to synthetic run-audit rows, leaving the task's
|
||||
shared board/API log silent. Mirror genuine live-cap exhaustion onto each queued
|
||||
candidate without awaiting it in the poll; the signature prevents poll spam.
|
||||
*/
|
||||
void Promise.all(eligibleIds.map((taskId) => this.store.logEntry(taskId, capacityReason)))
|
||||
.catch((logErr: unknown) => {
|
||||
planLog.warn(`Failed to write planning capacity reason: ${logErr instanceof Error ? logErr.message : String(logErr)}`);
|
||||
});
|
||||
})
|
||||
.catch((auditErr: unknown) => {
|
||||
planLog.warn(`Failed to write plan-admission-throttled run-audit event: ${auditErr instanceof Error ? auditErr.message : String(auditErr)}`);
|
||||
|
||||
Reference in New Issue
Block a user