FN-8453: unify concurrency accounting and indicators
Unify live-agent capacity accounting across engine and dashboard. - Derive Running and Waiting from workflow traits and durable agent liveness. - Apply unified limits to planner, executor, and merge admission while updating dashboard indicators. - Remove duplicate concurrency controls and document the unified operator model. Files changed: .changeset/fn-8453-unified-concurrency.md | 7 + docs/agent-tool-surface-full-loop.md | 4 +- docs/architecture.md | 2 +- docs/dashboard-guide.md | 4 +- docs/settings-reference.md | 4 +- .../skill/fusion/references/fusion-capabilities.md | 4 +- .../core/src/__tests__/live-agent-count.test.ts | 91 ++++---- packages/core/src/index.gate.ts | 6 + packages/core/src/index.ts | 6 + packages/core/src/live-agent-count.ts | 107 ++++++--- packages/dashboard/app/App.tsx | 28 ++- packages/dashboard/app/api/board-workflows.ts | 2 + packages/dashboard/app/components/Column.tsx | 6 +- .../dashboard/app/components/EngineControlMenu.tsx | 26 --- .../dashboard/app/components/ExecutorStatusBar.tsx | 38 ++- .../dashboard/app/components/SettingsModal.tsx | 1 - .../app/components/__tests__/Column.test.tsx | 6 +- .../__tests__/EngineControlMenu.test.tsx | 10 +- .../__tests__/ExecutorStatusBar.test.tsx | 32 ++- .../command-center/CommandCenterControls.tsx | 26 --- .../settings/sections/SchedulingSection.search.ts | 9 - .../settings/sections/SchedulingSection.tsx | 13 -- .../app/hooks/__tests__/useExecutorStats.test.ts | 12 +- packages/dashboard/app/hooks/useExecutorStats.ts | 50 ++-- .../src/__tests__/project-store-resolver.test.ts | 11 +- packages/dashboard/src/project-store-resolver.ts | 14 +- .../register-config-mcp-pi-settings-routes.ts | 3 +- packages/engine/src/__tests__/concurrency.test.ts | 123 +++++++++- .../engine/src/__tests__/project-engine.test.ts | 34 +++ packages/engine/src/__tests__/triage.test.ts | 7 +- packages/engine/src/concurrency.ts | 207 ++++++++++++++++- packages/engine/src/project-engine.ts | 151 ++++++++++-- packages/engine/src/scheduler.ts | 82 ++++++- packages/engine/src/triage.ts | 254 +++++++++++++-------- .../lib/dashboard-browser-safe-core-modules.json | 5 + 35 files changed, 991 insertions(+), 394 deletions(-) Fusion-Task-Id: FN-8453 Fusion-Task-Lineage: 12cfa5df-675d-4fce-b17e-932376544239 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-8453-unified-concurrency.md
Normal file
7
.changeset/fn-8453-unified-concurrency.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": minor
|
||||
---
|
||||
|
||||
summary: Unify max concurrency across planning/execution/review and simplify board capacity indicators.
|
||||
category: feature
|
||||
dev: maxConcurrent caps all top-level working agents per project; maxTriageConcurrent removed from UI (Settings, Command Center, Engine Control) and admission; free slots admit oldest createdAt via per-project atomic admission coordinator across lanes; footer Waiting/Running/Blocked; column headers active/total; nested runNested helpers remain parent-internal soft-breach by design.
|
||||
@@ -63,9 +63,9 @@ FR-07/FR-08 require one canonical operation beneath UI and tool callers. The par
|
||||
Fusion does not have a single-workflow-only scheduler by design. The scheduler can dispatch independent runnable tasks, but admission is bounded and serialized at specific safety gates:
|
||||
|
||||
- `packages/engine/src/scheduler.ts` computes dispatch capacity from **`maxConcurrent`**, **`maxWorktrees`**, and the shared **`semaphore`** in `computeConcurrencyGateDiagnostic()`. The settings default to `maxConcurrent` 2 and `maxWorktrees` 4 when unspecified (scheduler dispatch path).
|
||||
- `AgentSemaphore` in `packages/engine/src/concurrency.ts` gates all top-level triage, execution, and merge agents. Its priority queue serves merge before execute before specification (`PRIORITY_MERGE`, `PRIORITY_EXECUTE`, `PRIORITY_SPECIFY`). A full semaphore can therefore make work appear serialized even with multiple runnable cards.
|
||||
- `AgentSemaphore` in `packages/engine/src/concurrency.ts` gates all top-level planning, execution, and merge/review agents. Per-project admission ranks eligible candidates by `createdAt` then task ID across lanes before claiming the shared capacity; lane priority is not a free-slot rank key. Nested helpers remain parent-internal and intentionally soft-breach this displayed top-level cap to avoid parent/child deadlocks.
|
||||
- The workflow hold/release sweep in `scheduler.ts` reserves worktree and semaphore capacity with `tryAcquire()` **before** moving a task to `in-progress`, then transfers that pre-held slot to the executor. This prevents a race but means the available minimum of all gates is authoritative.
|
||||
- `maxWorktrees` counts only `in-progress` tasks; in-review worktrees do not consume that execution-worktree limit. `maxConcurrent` counts execution slots, while the global semaphore also accounts for planning and active review top-level holders.
|
||||
- `maxWorktrees` counts only `in-progress` tasks; in-review worktrees do not consume that execution-worktree limit. `maxConcurrent` caps per-project top-level working agents across planning, execution, and active review/merge, while the host semaphore remains the process-global pool.
|
||||
- Runnable candidates are additionally filtered for paused state, unmet dependencies, recovery backoff, workflow hold/release state, and file-scope overlap via `isRunnableQueuedOverlapCandidate()` / `pathsOverlap()` in `scheduler.ts`. This is coarse path-scope serialization, not FR-48 symbol locking.
|
||||
- `packages/engine/src/workflow-work-scheduler.ts` claims one due workflow work item per call with a lease. The surrounding scheduler’s repeated dispatch and capacity gates determine aggregate concurrency; this helper alone does not fan out a batch.
|
||||
- `packages/engine/src/verification-concurrency.ts` separately defaults expensive verification subprocesses to one concurrent project-wide slot. Task execution can be parallel while heavy E2E verification intentionally queues.
|
||||
|
||||
@@ -1134,7 +1134,7 @@ The run-audit system records every mutation performed by the engine across four
|
||||
|
||||
Events are tied to specific run IDs for end-to-end traceability.
|
||||
|
||||
For scheduler concurrency diagnostics, the queued reason now names the active limiter(s) and usage (for example `gate=maxConcurrent ...`). The reason includes the `bindingGates` (`maxConcurrent`/`maxWorktrees`/`semaphore`), per-gate `{ used, limit, slack }`, `holders`, and computed `available`. `holders.maxConcurrent` and `holders.maxWorktrees` are current `in-progress` task IDs; `holders.semaphore` mirrors that set but semaphore slots can also be consumed by triage/merge agents outside `in-progress`. So if `semaphore.used` exceeds the visible holder list, that usually indicates non-execution agents are legitimately consuming shared capacity (not stale accounting). `maxWorktrees` is also enforced inside `TaskStore.moveTaskInternal` when committing an allocated move into `in-progress`, making it a hard active execution worktree cap even when workflow WIP/`maxConcurrent` would allow more tasks. These queued-reason logs are transition-only: a newly emitted line indicates the limiter signature changed or the condition cleared and later reappeared, not that a poll loop simply observed the same blocked state again.
|
||||
For scheduler concurrency diagnostics, the queued reason names the active limiter(s) and usage (for example `gate=maxConcurrent ...`). The reason includes the `bindingGates` (`maxConcurrent`/`maxWorktrees`/`semaphore`), per-gate `{ used, limit, slack }`, `holders`, and computed `available`. `maxConcurrent` is a per-project cap on enriched live top-level planning, execution, and review/merge agents; the host semaphore is the separate process-global pool. Free project capacity is admitted oldest-first across lanes, rather than by lane priority. `maxWorktrees` is also enforced inside `TaskStore.moveTaskInternal` when committing an allocated move into `in-progress`, making it a hard active execution worktree cap even when workflow WIP/`maxConcurrent` would allow more tasks. These queued-reason logs are transition-only: a newly emitted line indicates the limiter signature changed or the condition cleared and later reappeared, not that a poll loop simply observed the same blocked state again.
|
||||
|
||||
**Run audit endpoints:**
|
||||
- `GET /api/agents/:id/runs/:runId/audit` — Returns audit trail for a specific agent run
|
||||
|
||||
@@ -1258,7 +1258,7 @@ Features:
|
||||
<!-- FNXC:CommandCenter 2026-06-27-10:03: Tokens detail charts must show every model bucket returned by analytics for accurate spend attribution; Overview remains a compact top-model summary because its copy explicitly frames those cards as top consumers/share. -->
|
||||
<!-- FNXC:CommandCenterActivity 2026-06-30-00:00: Activity active-agent counts include both durable-agent usage events and ephemeral task-worker execution runs from agentRuns, because task execution can be visible without a matching usage_events row. -->
|
||||
<!-- FNXC:CommandCenterActivity 2026-07-01-00:00: Graph-owned workflow step sessions publish active-to-terminal agentRuns lifecycle rows with task lineage and step metadata, so daily activity and Activity throughput charts include new workflow execution without dashboard-side recounting. -->
|
||||
- **Overview controls dashboard** sits at the top of the Overview landing surface on desktop and mobile. It includes AI engine stop/start backed by `globalPause`, live scheduler status from executor stats, the shared Global Max Concurrent slider backed by `/api/global-concurrency`, range sliders for `maxConcurrent`, `maxTriageConcurrent`, and `maxWorktrees` that persist through `/api/settings`, and a compact theme dropdown with the same color-chip swatches and Shadcn variant list as Settings → Appearance. The four concurrency sliders ask for confirmation after a changed value settles; confirming persists the new cap, while cancel, backdrop, or Escape dismissal reverts to the last persisted value without saving. The global and current-project max-concurrent sliders show running-agent counts plus a current-use dot on the track once utilization data loads; triage and worktree sliders remain cap-only. These controls reuse existing APIs and App-level theme setters; they do not add a new backend route or second theme owner.
|
||||
- **Overview controls dashboard** includes AI engine stop/start backed by `globalPause`, the shared Global Max Concurrent slider, and current-project **Max concurrency** plus **Max worktrees** controls. Max concurrency caps top-level working agents across planning, execution, and review/merge; free capacity is admitted oldest-first within the project. The footer reports **Waiting**, **Running (N/max)**, and **Blocked**; column headers report active/total. Nested helper agents remain parent-internal and may temporarily exceed the displayed top-level count.
|
||||
<!-- FNXC:TeamArea 2026-07-18-12:30: FN-8351 moves organization export and import to the Team tab so team-level portability controls are not presented as Overview dashboard controls. -->
|
||||
- **Team tab — Org export / import** lets an operator download a portable organization JSON bundle or paste one for a dry-run preview before confirming the apply step. Exports are secret-scrubbed by default: credentials and tokens are never included, while safe secret references can remain for setup in the destination project.
|
||||
- **Configuration versions** lives in **Settings → Project → Configuration Versions**. It lists recorded project-setting revisions newest first; select **Roll back** on any revision and confirm once to restore it. The restore is recorded as a new forward revision, so it can itself be undone without manually reconstructing settings.
|
||||
@@ -1451,7 +1451,7 @@ Use this panel when upgrading a project with pre-FN-6245/FN-6277 in-review rows
|
||||
<!-- FNXC:ExecutorStatusBar 2026-06-29-19:09: FN-7248 makes footer concurrency edits confirmation-gated like Command Center. Closing the popover, outside-clicking, pressing Escape, dismissing the backdrop, or unmounting must revert unconfirmed slider edits instead of saving them. -->
|
||||
<!-- FNXC:ExecutorStatusBar 2026-06-30-16:42: FN-7273 keeps the footer Engine Controls popover usable on mobile, narrow tablets, and tablet landscape by documenting that constrained screens use a full-width bottom panel above both fixed bottom bars instead of the compact desktop anchor. -->
|
||||
<!-- FNXC:GlobalConcurrencyControls 2026-07-15-17:30: FN-8007 keeps footer and dashboard current-use marker geometry aligned with the native range thumb, including its desktop and mobile thumb-size edge inset. -->
|
||||
The global AI engine stop/start control and triage pause/resume control live in the executor footer status bar rather than the header. Select the small engine-controls button beside the executor state badge, or select the state text such as **Running**, to open the footer popover. The popover includes **Stop AI engine** / **Start AI engine**, **Pause triage** / **Resume scheduling**, and live scheduler sliders for max concurrent tasks, max triage concurrency, and max worktrees. On mobile, narrow tablets, and tablet landscape, the same controls open as a full-width bottom panel above the executor footer and mobile navigation so the close button and sliders remain reachable. Use the visible **Close engine controls** X button, Escape, or outside-click to dismiss it. The global and current-project concurrency sliders also show how many agents are running, including actively-triaging planners (`triage` + `planning`, not paused), and a dot on the slider track for current use. The dot uses the same min-relative range coordinates and thumb-size edge inset as the native slider: it aligns to the cap-clamped running count, so one running agent at the slider minimum stays visible at the start and over-cap usage pins to the cap thumb instead of the expanded track end. Changed concurrency slider values ask for confirmation after the value settles. Confirming saves the global cap through `/api/global-concurrency` and project caps through `/api/settings`; cancel, backdrop dismissal, Escape, close, outside-click, or unmount reverts unconfirmed slider edits without saving. Multiple changed project sliders within one debounce window are summarized in one confirmation dialog, matching Command Center behavior.
|
||||
The global AI engine stop/start control and triage pause/resume control live in the executor footer status bar rather than the header. Select the small engine-controls button beside the executor state badge, or select the state text such as **Running**, to open the footer popover. The popover includes **Stop AI engine** / **Start AI engine**, **Pause triage** / **Resume scheduling**, and live scheduler sliders for **Max concurrency** and max worktrees. Max concurrency is the per-project top-level working-agent cap across planning, execution, and review/merge; nested helpers remain parent-internal and can temporarily exceed the displayed count. On mobile, narrow tablets, and tablet landscape, the same controls open as a full-width bottom panel above the executor footer and mobile navigation so the close button and sliders remain reachable. Use the visible **Close engine controls** X button, Escape, or outside-click to dismiss it. The global and current-project concurrency sliders show the shared live top-level agent count and a dot on the slider track for current use. The dot uses the same min-relative range coordinates and thumb-size edge inset as the native slider: it aligns to the cap-clamped running count, so one running agent at the slider minimum stays visible at the start and over-cap usage pins to the cap thumb instead of the expanded track end. Changed concurrency slider values ask for confirmation after the value settles. Confirming saves the global cap through `/api/global-concurrency` and project caps through `/api/settings`; cancel, backdrop dismissal, Escape, close, outside-click, or unmount reverts unconfirmed slider edits without saving. Multiple changed project sliders within one debounce window are summarized in one confirmation dialog, matching Command Center behavior.
|
||||
|
||||
<!-- FNXC:ExecutorStatusBar 2026-06-27-00:00: FN-7163 makes footer stats loading initial-only so routine heartbeat refreshes keep the populated footer and open concurrency popover mounted instead of blinking to the loading branch. -->
|
||||
Brief, single-poll executor stats fetch blips keep showing the last good footer stats instead of flashing **Connecting…**. Routine executor stats heartbeats also keep the populated footer mounted after initial load, so an open engine/concurrency popover stays open while counts refresh. The footer only switches to **Connecting…** for sustained suspension-like stats failures, or to an explicit error state for non-transient failures.
|
||||
|
||||
@@ -430,9 +430,9 @@ Security-sensitive file-browser escape hatches are project-only. `allowAbsoluteF
|
||||
| `globalPause` | `boolean` | `false` | Hard stop: terminate active engine sessions and pause scheduling immediately. |
|
||||
| `globalPauseReason` | `string` | `undefined` | Optional reason for `globalPause` (`"rate-limit"` for automatic pauses, `"manual"` for user-triggered pauses). Cleared on unpause. |
|
||||
| `enginePaused` | `boolean` | `false` | Soft pause: stop dispatching new work while letting active sessions finish. While paused (including shared pause windows with `globalPause`), stuck-task polling/timers are suspended so paused wall-clock time does not count against `taskStuckTimeoutMs`. Clearing pause state resumes runtime scheduling and gives tracked active sessions a fresh stuck-task grace window before normal detection resumes; when `autoMerge` is enabled, eligible `in-review` tasks are re-swept into the auto-merge queue (paused/blocked/failed review tasks remain skipped). |
|
||||
| `maxConcurrent` | `number` | `2` | Max concurrent task-lane AI agents (planning, executor, merge). Editable from Settings and the Command Center Overview controls dashboard. |
|
||||
| `maxConcurrent` | `number` | `2` | Max concurrency for top-level working agents per project across planning, execution, and review/merge. Nested helper agents remain parent-internal and may temporarily exceed this displayed count. Editable from Settings, Command Center, and Engine Control. |
|
||||
| `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` | Max concurrent planning agents. Editable from Settings and the Command Center Overview controls dashboard. |
|
||||
| `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. |
|
||||
| `pollIntervalMs` | `number` | `15000` | Scheduler poll interval (ms). |
|
||||
|
||||
@@ -155,8 +155,8 @@ Structured runtime metadata is authoritative in PostgreSQL. A retained `fusion.d
|
||||
|
||||
| Setting | Default | Description |
|
||||
|---------|---------|-------------|
|
||||
| `maxConcurrent` | 2 | Concurrent task execution lanes (executor + merge). Triage/specification is controlled by `maxTriageConcurrent`. |
|
||||
| `maxTriageConcurrent` | 2 | Concurrent triage/specification agents. Falls back to `maxConcurrent` when undefined. |
|
||||
| `maxConcurrent` | 2 | Per-project cap for top-level working agents across planning, execution, and review/merge. Free slots admit the oldest eligible task; nested helpers are parent-internal. |
|
||||
| `maxWorktrees` | 4 | Separate cap for execution worktree holders; it does not define the live Running count. |
|
||||
| `autoMerge` | true | Auto-merge completed tasks |
|
||||
| `requirePlanApproval` | false | Manual approval for specs |
|
||||
| `prCompletionMode` | direct | Completion mode: direct/pr-first |
|
||||
|
||||
@@ -1,64 +1,67 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { countRunningAgentTasks, deriveRunningAgentCounts, isRunningAgentTask } from "../live-agent-count.js";
|
||||
import type { Task } from "../types.js";
|
||||
import {
|
||||
countRunningAgentTasks,
|
||||
deriveRunningAgentCounts,
|
||||
enrichRunningAgentTaskShapeFromFlags,
|
||||
isRunningAgentTask,
|
||||
isWaitingAgentTask,
|
||||
} from "../live-agent-count.js";
|
||||
import type { RunningAgentTaskShape } from "../live-agent-count.js";
|
||||
|
||||
function task(overrides: Pick<Task, "column"> & Partial<Pick<Task, "status" | "paused">>): Pick<Task, "column" | "status" | "paused"> {
|
||||
return {
|
||||
column: overrides.column,
|
||||
status: overrides.status,
|
||||
paused: overrides.paused,
|
||||
};
|
||||
function task(overrides: Partial<RunningAgentTaskShape> & Pick<RunningAgentTaskShape, "column">): RunningAgentTaskShape {
|
||||
return { columnTerminalKind: "none", ...overrides };
|
||||
}
|
||||
|
||||
describe("live agent count predicates", () => {
|
||||
it("identifies tasks that hold top-level running-agent slots", () => {
|
||||
expect(isRunningAgentTask(task({ column: "in-progress" }))).toBe(true);
|
||||
expect(isRunningAgentTask(task({ column: "triage", status: "planning", paused: false }))).toBe(true);
|
||||
expect(isRunningAgentTask(task({ column: "triage", status: "planning", paused: true }))).toBe(false);
|
||||
it("counts live planners in every non-terminal workflow lane", () => {
|
||||
expect(isRunningAgentTask(task({ column: "todo", status: "planning" }))).toBe(true);
|
||||
expect(isRunningAgentTask(task({ column: "ideas", status: "planning" }))).toBe(true);
|
||||
expect(isRunningAgentTask(task({ column: "ideas", status: "planning", paused: true }))).toBe(false);
|
||||
expect(isRunningAgentTask(task({ column: "ideas", status: "planning", userPaused: true }))).toBe(false);
|
||||
});
|
||||
|
||||
it("requires durable liveness for WIP execution rather than an in-progress shell", () => {
|
||||
expect(isRunningAgentTask(task({ column: "in-progress", columnCountsTowardWip: true }))).toBe(false);
|
||||
expect(isRunningAgentTask(task({ column: "in-progress", columnCountsTowardWip: true, sessionFile: "/tmp/run" }))).toBe(true);
|
||||
expect(isRunningAgentTask(task({ column: "in-progress", columnCountsTowardWip: true, checkedOutBy: "agent-a" }))).toBe(true);
|
||||
expect(isRunningAgentTask(task({ column: "in-progress", columnCountsTowardWip: true, sessionFile: "/tmp/run", paused: true }))).toBe(false);
|
||||
});
|
||||
|
||||
it("counts only active review/merge statuses and excludes terminal columns", () => {
|
||||
for (const status of ["merging", "merging-pr", "merging-fix", "reviewing", "landing", "fixing"]) {
|
||||
expect(isRunningAgentTask(task({ column: "in-review", status, paused: false }))).toBe(true);
|
||||
expect(isRunningAgentTask(task({ column: "review", status, columnIsReviewOrMerge: true }))).toBe(true);
|
||||
}
|
||||
|
||||
expect(isRunningAgentTask(task({ column: "in-review", paused: false }))).toBe(false);
|
||||
expect(isRunningAgentTask(task({ column: "in-review", status: "pending", paused: false }))).toBe(false);
|
||||
expect(isRunningAgentTask(task({ column: "in-review", status: "reviewing", paused: true }))).toBe(false);
|
||||
expect(isRunningAgentTask(task({ column: "done" }))).toBe(false);
|
||||
expect(isRunningAgentTask(task({ column: "todo" }))).toBe(false);
|
||||
expect(isRunningAgentTask(task({ column: "archived" }))).toBe(false);
|
||||
expect(isRunningAgentTask(task({ column: "review", status: "pending", columnIsReviewOrMerge: true }))).toBe(false);
|
||||
expect(isRunningAgentTask(task({ column: "ideas", status: "merging", columnIsReviewOrMerge: false }))).toBe(false);
|
||||
expect(isRunningAgentTask(task({ column: "shipped", sessionFile: "/tmp/stale", columnCountsTowardWip: true, columnTerminalKind: "complete" }))).toBe(false);
|
||||
expect(isRunningAgentTask(task({ column: "working", sessionFile: "/tmp/live", columnCountsTowardWip: true, columnTerminalKind: "none" }))).toBe(true);
|
||||
});
|
||||
|
||||
it("counts only tasks that satisfy the shared running-agent predicate", () => {
|
||||
it("enriches terminal, waiting, and WIP traits from board flags", () => {
|
||||
const complete = enrichRunningAgentTaskShapeFromFlags(task({ column: "shipped", sessionFile: "/tmp/stale" }), { complete: true, countsTowardWip: true });
|
||||
expect(complete.columnTerminalKind).toBe("complete");
|
||||
expect(isRunningAgentTask(complete)).toBe(false);
|
||||
|
||||
const intake = enrichRunningAgentTaskShapeFromFlags(task({ column: "ideas" }), { intake: true });
|
||||
expect(isWaitingAgentTask(intake)).toBe(true);
|
||||
expect(isWaitingAgentTask({ ...intake, status: "planning" })).toBe(false);
|
||||
expect(isWaitingAgentTask(enrichRunningAgentTaskShapeFromFlags(task({ column: "hold" }), { hold: true }))).toBe(true);
|
||||
});
|
||||
|
||||
it("counts only the shared predicate", () => {
|
||||
expect(countRunningAgentTasks([
|
||||
task({ column: "in-progress", sessionFile: "/tmp/run" }),
|
||||
task({ column: "in-progress" }),
|
||||
task({ column: "triage", status: "planning", paused: false }),
|
||||
task({ column: "triage", status: "planning", paused: true }),
|
||||
task({ column: "in-review", status: "merging", paused: false }),
|
||||
task({ column: "in-review", status: "merging-pr", paused: false }),
|
||||
task({ column: "in-review", status: "merging-fix", paused: false }),
|
||||
task({ column: "in-review", status: "reviewing", paused: false }),
|
||||
task({ column: "in-review", status: "landing", paused: false }),
|
||||
task({ column: "in-review", status: "fixing", paused: false }),
|
||||
task({ column: "in-review", status: "fixing", paused: true }),
|
||||
task({ column: "todo" }),
|
||||
task({ column: "done" }),
|
||||
task({ column: "archived" }),
|
||||
])).toBe(8);
|
||||
task({ column: "triage", status: "planning" }),
|
||||
task({ column: "in-review", status: "merging", columnIsReviewOrMerge: true }),
|
||||
task({ column: "done", sessionFile: "/tmp/stale" }),
|
||||
])).toBe(3);
|
||||
});
|
||||
|
||||
it("normalizes display counts for zero, one, multi-project, unopened, and oversubscribed states", () => {
|
||||
expect(deriveRunningAgentCounts({})).toEqual({ currentlyActive: 0, projectsActive: {} });
|
||||
expect(deriveRunningAgentCounts({ proj_zero: 0, proj_one: 1 })).toEqual({
|
||||
it("normalizes aggregate display counts", () => {
|
||||
expect(deriveRunningAgentCounts({ proj_zero: 0, proj_one: 1, proj_nan: Number.NaN })).toEqual({
|
||||
currentlyActive: 1,
|
||||
projectsActive: { proj_one: 1 },
|
||||
});
|
||||
expect(deriveRunningAgentCounts({ proj_a: 2, proj_b: 4, proj_unopened: 0 })).toEqual({
|
||||
currentlyActive: 6,
|
||||
projectsActive: { proj_a: 2, proj_b: 4 },
|
||||
});
|
||||
expect(deriveRunningAgentCounts({ proj_over_limit: 12, proj_negative: -3, proj_nan: Number.NaN })).toEqual({
|
||||
currentlyActive: 12,
|
||||
projectsActive: { proj_over_limit: 12 },
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -566,7 +566,13 @@ export {
|
||||
getRunningAgentCountSource,
|
||||
deriveRunningAgentCounts,
|
||||
isRunningAgentTask,
|
||||
isWaitingAgentTask,
|
||||
countRunningAgentTasks,
|
||||
enrichRunningAgentTaskShape,
|
||||
enrichRunningAgentTaskShapeFromFlags,
|
||||
resolveColumnTerminalKind,
|
||||
type RunningAgentTaskShape,
|
||||
type ColumnTerminalKind,
|
||||
type RunningAgentCountSource,
|
||||
type RunningAgentCounts,
|
||||
} from "./live-agent-count.js";
|
||||
|
||||
@@ -656,7 +656,13 @@ export {
|
||||
getRunningAgentCountSource,
|
||||
deriveRunningAgentCounts,
|
||||
isRunningAgentTask,
|
||||
isWaitingAgentTask,
|
||||
countRunningAgentTasks,
|
||||
enrichRunningAgentTaskShape,
|
||||
enrichRunningAgentTaskShapeFromFlags,
|
||||
resolveColumnTerminalKind,
|
||||
type RunningAgentTaskShape,
|
||||
type ColumnTerminalKind,
|
||||
type RunningAgentCountSource,
|
||||
type RunningAgentCounts,
|
||||
} from "./live-agent-count.js";
|
||||
|
||||
@@ -1,13 +1,34 @@
|
||||
import { ACTIVE_MERGE_PIPELINE_STATUSES } from "./active-merge-status.js";
|
||||
import type { TraitFlags } from "./trait-types.js";
|
||||
import type { Task } from "./types.js";
|
||||
import type { WorkflowIr } from "./workflow-ir-types.js";
|
||||
import { columnHasFlag } from "./workflow-lifecycle-traits.js";
|
||||
|
||||
export type RunningAgentCountSource = (projectIds: readonly string[]) => Promise<Record<string, number>> | Record<string, number>;
|
||||
|
||||
type RunningAgentTaskShape = Pick<Task, "column" | "status" | "paused">;
|
||||
/** Terminal classification supplied by a workflow-IR or board-flags enricher. */
|
||||
export type ColumnTerminalKind = "none" | "complete" | "archived";
|
||||
|
||||
/**
|
||||
* The deliberately small, pure shape used by all top-level live-agent counts.
|
||||
* Store- and board-backed callers must attach trait-derived fields first.
|
||||
*/
|
||||
export type RunningAgentTaskShape = Pick<Task, "column" | "status" | "paused" | "userPaused" | "sessionFile" | "checkedOutBy"> & {
|
||||
columnTerminalKind?: ColumnTerminalKind;
|
||||
/** Trait-derived intake/hold membership, used by {@link isWaitingAgentTask}. */
|
||||
columnIsIntakeOrHold?: boolean;
|
||||
/** Trait-derived WIP membership; legacy fixtures fall back to in-progress. */
|
||||
columnCountsTowardWip?: boolean;
|
||||
/** Trait-derived review/merge membership; active merge statuses are live only here. */
|
||||
columnIsReviewOrMerge?: boolean;
|
||||
};
|
||||
|
||||
/*
|
||||
FNXC:MergeQueue 2026-07-15-10:40:
|
||||
In-review live agents include the full AI merge pipeline (merging/reviewing/landing) plus fix-pass and generic fixing statuses so utilization counts stay honest during clean-room review and land.
|
||||
FNXC:ConcurrencyIndicators 2026-08-03-12:00:
|
||||
FN-8453 / GitHub #2359 defines Running as a live top-level working agent, not a
|
||||
board-column or worktree-holder count. Every production store- or board-backed
|
||||
consumer enriches this pure shape from workflow traits before it counts; the
|
||||
literal terminal fallback exists only for legacy fixtures while no IR is loaded.
|
||||
*/
|
||||
const ACTIVE_IN_REVIEW_AGENT_STATUSES = new Set([
|
||||
...ACTIVE_MERGE_PIPELINE_STATUSES,
|
||||
@@ -16,17 +37,10 @@ const ACTIVE_IN_REVIEW_AGENT_STATUSES = new Set([
|
||||
|
||||
let runningAgentCountSource: RunningAgentCountSource | undefined;
|
||||
|
||||
/**
|
||||
* FNXC:GlobalConcurrencyControls 2026-06-26-17:22:
|
||||
* Live running-agent counts must come from side-effect-safe reads of `in-progress` task columns, not from stale slot or health bookkeeping. This DI seam lets dashboard, CLI, remote-node, and plugin consumers share one core path without starting project engines/runtimes, opening watchers, or mutating `globalConcurrency.currentlyActive`, `globalConcurrency.queuedCount`, or `projectHealth.inFlightAgentCount`.
|
||||
*/
|
||||
export function setRunningAgentCountSource(fn: RunningAgentCountSource | undefined): void {
|
||||
runningAgentCountSource = fn;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the registered side-effect-safe running-agent count source, if one has been wired by the host process.
|
||||
*/
|
||||
export function getRunningAgentCountSource(): RunningAgentCountSource | undefined {
|
||||
return runningAgentCountSource;
|
||||
}
|
||||
@@ -36,24 +50,65 @@ export interface RunningAgentCounts {
|
||||
projectsActive: Record<string, number>;
|
||||
}
|
||||
|
||||
/** Resolve the terminal classification of one column from its workflow IR. */
|
||||
export function resolveColumnTerminalKind(columnId: string, ir: WorkflowIr): ColumnTerminalKind {
|
||||
if (columnHasFlag(ir, columnId, "archived")) return "archived";
|
||||
if (columnHasFlag(ir, columnId, "complete")) return "complete";
|
||||
return "none";
|
||||
}
|
||||
|
||||
/** Attach the workflow traits required by the pure Running and Waiting predicates. */
|
||||
export function enrichRunningAgentTaskShape<T extends RunningAgentTaskShape>(task: T, ir: WorkflowIr): T & Required<Pick<RunningAgentTaskShape, "columnTerminalKind" | "columnIsIntakeOrHold" | "columnCountsTowardWip" | "columnIsReviewOrMerge">> {
|
||||
return {
|
||||
...task,
|
||||
columnTerminalKind: resolveColumnTerminalKind(task.column, ir),
|
||||
columnIsIntakeOrHold: columnHasFlag(ir, task.column, "intake") || columnHasFlag(ir, task.column, "hold"),
|
||||
columnCountsTowardWip: columnHasFlag(ir, task.column, "countsTowardWip"),
|
||||
columnIsReviewOrMerge: columnHasFlag(ir, task.column, "mergeOrchestration") || columnHasFlag(ir, task.column, "mergeBlocker"),
|
||||
};
|
||||
}
|
||||
|
||||
/** Attach the same traits from dashboard board-column flags without loading an IR. */
|
||||
export function enrichRunningAgentTaskShapeFromFlags<T extends RunningAgentTaskShape>(task: T, flags?: Pick<TraitFlags, "complete" | "archived" | "intake" | "hold" | "countsTowardWip" | "mergeOrchestration" | "mergeBlocker">): T & Required<Pick<RunningAgentTaskShape, "columnTerminalKind" | "columnIsIntakeOrHold" | "columnCountsTowardWip" | "columnIsReviewOrMerge">> {
|
||||
return {
|
||||
...task,
|
||||
columnTerminalKind: flags?.archived ? "archived" : flags?.complete ? "complete" : "none",
|
||||
columnIsIntakeOrHold: flags ? flags.intake === true || flags.hold === true : task.column === "triage" || task.column === "todo",
|
||||
columnCountsTowardWip: flags ? flags.countsTowardWip === true : task.column === "in-progress",
|
||||
// The literal fallback is fixture-only; board/store callers always supply flags/IR.
|
||||
columnIsReviewOrMerge: flags ? flags.mergeOrchestration === true || flags.mergeBlocker === true : task.column === "in-review",
|
||||
};
|
||||
}
|
||||
|
||||
function terminalKind(task: RunningAgentTaskShape): ColumnTerminalKind {
|
||||
// Legacy literals are intentionally fixture-only degradation when workflow IR is unavailable.
|
||||
return task.columnTerminalKind ?? (task.column === "done" ? "complete" : task.column === "archived" ? "archived" : "none");
|
||||
}
|
||||
|
||||
function hasDurableExecuteLiveness(task: RunningAgentTaskShape): boolean {
|
||||
return Boolean(task.sessionFile?.trim() || task.checkedOutBy?.trim());
|
||||
}
|
||||
|
||||
/**
|
||||
* FNXC:GlobalConcurrencyControls 2026-06-27-00:00:
|
||||
* FN-7160 defines live running-agent counts as top-level concurrency slot holders: in-progress executors, active unpaused triage planners, and active unpaused in-review reviewer/merger/fix agents, including PR/fix merge substates. Keep this pure predicate as the shared source of truth for engine slot accounting and all dashboard/CLI read-layer count surfaces so in-review agents cannot drift out of utilization displays again.
|
||||
* Returns true only for a live, unpaused top-level agent.
|
||||
* Planning may run in any non-terminal workflow column; execute needs durable
|
||||
* session/checkout evidence so idle worktree shells do not consume capacity.
|
||||
*/
|
||||
export function isRunningAgentTask(task: RunningAgentTaskShape): boolean {
|
||||
if (task.column === "in-progress") {
|
||||
return true;
|
||||
if (task.paused || task.userPaused || terminalKind(task) !== "none") return false;
|
||||
if (task.status === "planning") return true;
|
||||
// Review statuses are not globally live: a stale status in intake/WIP must not consume capacity.
|
||||
if (ACTIVE_IN_REVIEW_AGENT_STATUSES.has(String(task.status ?? ""))) {
|
||||
return task.columnIsReviewOrMerge ?? task.column === "in-review";
|
||||
}
|
||||
const isWip = task.columnCountsTowardWip ?? task.column === "in-progress";
|
||||
return isWip && hasDurableExecuteLiveness(task);
|
||||
}
|
||||
|
||||
if (task.column === "triage") {
|
||||
return task.status === "planning" && !task.paused;
|
||||
}
|
||||
|
||||
if (task.column === "in-review") {
|
||||
return ACTIVE_IN_REVIEW_AGENT_STATUSES.has(String(task.status ?? "")) && !task.paused;
|
||||
}
|
||||
|
||||
return false;
|
||||
/** Exact footer waiting membership: unpaused, non-terminal intake/hold work that is not live. */
|
||||
export function isWaitingAgentTask(task: RunningAgentTaskShape): boolean {
|
||||
if (task.paused || task.userPaused || terminalKind(task) !== "none" || isRunningAgentTask(task)) return false;
|
||||
return task.columnIsIntakeOrHold ?? (task.column === "triage" || task.column === "todo");
|
||||
}
|
||||
|
||||
export function countRunningAgentTasks(tasks: readonly RunningAgentTaskShape[]): number {
|
||||
@@ -63,14 +118,10 @@ export function countRunningAgentTasks(tasks: readonly RunningAgentTaskShape[]):
|
||||
export function deriveRunningAgentCounts(perProject: Record<string, number>): RunningAgentCounts {
|
||||
const projectsActive: Record<string, number> = {};
|
||||
let currentlyActive = 0;
|
||||
|
||||
for (const [projectId, rawCount] of Object.entries(perProject)) {
|
||||
const count = Number.isFinite(rawCount) ? Math.max(0, Math.trunc(rawCount)) : 0;
|
||||
currentlyActive += count;
|
||||
if (count > 0) {
|
||||
projectsActive[projectId] = count;
|
||||
}
|
||||
if (count > 0) projectsActive[projectId] = count;
|
||||
}
|
||||
|
||||
return { currentlyActive, projectsActive };
|
||||
}
|
||||
|
||||
@@ -29,6 +29,8 @@ import { useBackgroundSessions } from "./hooks/useBackgroundSessions";
|
||||
import { useGitHubStarPromptShown, markGitHubStarPromptShown } from "./hooks/useGitHubStarPrompt";
|
||||
import { useSessionBannersHidden } from "./hooks/useSessionBannerPref";
|
||||
import { useTasks } from "./hooks/useTasks";
|
||||
import { useBoardWorkflows } from "./hooks/useBoardWorkflows";
|
||||
import type { ExecutorColumnFlags } from "./hooks/useExecutorStats";
|
||||
import { useProjects } from "./hooks/useProjects";
|
||||
import { useAgents } from "./hooks/useAgents";
|
||||
import { useNodes } from "./hooks/useNodes";
|
||||
@@ -528,6 +530,29 @@ function AppInner() {
|
||||
sseEnabled: taskSseEnabled,
|
||||
}
|
||||
);
|
||||
const { boardWorkflows: footerBoardWorkflows } = useBoardWorkflows({ projectId: currentProject?.id });
|
||||
const footerTasks = isRemote && remoteData.tasks.length > 0 ? remoteData.tasks : tasks;
|
||||
const footerColumnFlagsByTaskId = useMemo(() => {
|
||||
const index = new Map<string, ExecutorColumnFlags>();
|
||||
// FNXC:ConcurrencyIndicators 2026-08-04-10:00: remote tasks belong to a
|
||||
// different store, so local board-workflow metadata must never be applied to
|
||||
// their ids. Until the remote node supplies its own traits, use only the
|
||||
// documented literal fallback rather than fabricate custom lifecycle state.
|
||||
if (isRemote || !footerBoardWorkflows) return index;
|
||||
const workflowsById = new Map(footerBoardWorkflows.workflows.map((workflow) => [workflow.id, workflow]));
|
||||
// Build traits for the exact local rows supplied to the footer.
|
||||
for (const task of footerTasks) {
|
||||
const workflow = workflowsById.get(footerBoardWorkflows.taskWorkflowIds[task.id] ?? footerBoardWorkflows.defaultWorkflowId);
|
||||
const flags = workflow?.columns.find((column) => column.id === task.column)?.flags;
|
||||
if (flags) index.set(task.id, flags);
|
||||
}
|
||||
return index;
|
||||
}, [footerBoardWorkflows, footerTasks, isRemote]);
|
||||
/*
|
||||
FNXC:ConcurrencyIndicators 2026-08-03-12:00:
|
||||
FN-8453 threads board workflow traits into the footer so custom intake,
|
||||
complete, WIP, and merge columns share the same live-agent predicate as the engine.
|
||||
*/
|
||||
|
||||
/*
|
||||
FNXC:Navigation 2026-06-22-00:00:
|
||||
@@ -1846,8 +1871,9 @@ function AppInner() {
|
||||
{rightDock.modal}
|
||||
{executorFooterVisible && currentProject && (
|
||||
<ExecutorStatusBar
|
||||
tasks={isRemote && remoteData.tasks.length > 0 ? remoteData.tasks : tasks}
|
||||
tasks={footerTasks}
|
||||
projectId={currentProject.id}
|
||||
columnFlagsByTaskId={footerColumnFlagsByTaskId}
|
||||
taskStuckTimeoutMs={taskStuckTimeoutMs}
|
||||
staleHighFanoutBlockerAgeThresholdMs={staleHighFanoutBlockerAgeThresholdMs}
|
||||
lastFetchTimeMs={lastFetchTimeMs}
|
||||
|
||||
@@ -42,6 +42,8 @@ export interface BoardWorkflowColumnFlags {
|
||||
hold?: boolean;
|
||||
intake?: boolean;
|
||||
mergeBlocker?: boolean;
|
||||
/** Merge/review lane membership used by the shared live-agent predicate. */
|
||||
mergeOrchestration?: boolean;
|
||||
humanReview?: boolean;
|
||||
[key: string]: boolean | undefined;
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import { useFlashOnIncrease } from "../hooks/useFlashOnIncrease";
|
||||
import { useConfirm } from "../hooks/useConfirm";
|
||||
import type { Task, TaskDetail, Column as ColumnType, ColumnId, TaskCreateInput, GithubIssueAction, MergeResult } from "@fusion/core";
|
||||
import { COLUMN_LABELS, COLUMN_DESCRIPTIONS, getErrorMessage } from "@fusion/core";
|
||||
import { enrichRunningAgentTaskShapeFromFlags, isRunningAgentTask } from "../../../core/src/live-agent-count";
|
||||
import { isNearDuplicateCanonicalInactive } from "../../../core/src/near-duplicate-canonical";
|
||||
import { TaskCard } from "./TaskCard";
|
||||
import { WorktreeGroup } from "./WorktreeGroup";
|
||||
@@ -286,6 +287,9 @@ function ColumnComponent({ column, tasks, projectId, maxConcurrent, showWorktree
|
||||
The project setting is an explicit show/hide control: worktree grouping and labels render only when enabled and only for the board's WIP/processing column. Turning it off must leave plain task cards with no legacy group shell in either legacy or workflow-mode columns.
|
||||
*/
|
||||
const showWorktreeGroups = showWorktreeGrouping === true && isWipProcessingColumn;
|
||||
const activeTaskCount = useMemo(() => tasks.filter((task) =>
|
||||
isRunningAgentTask(enrichRunningAgentTaskShapeFromFlags(task, columnFlags)),
|
||||
).length, [tasks, columnFlags]);
|
||||
// When search is active, skip pagination so all matching tasks are visible
|
||||
const shouldPaginate = !isArchived && !isSearchActive && !showWorktreeGroups && tasks.length > PAGINATED_COLUMN_THRESHOLD;
|
||||
|
||||
@@ -656,7 +660,7 @@ function ColumnComponent({ column, tasks, projectId, maxConcurrent, showWorktree
|
||||
<div className="column-header">
|
||||
<div className={`column-dot dot-${column}`} />
|
||||
<h2>{workflowMode ? (columnDisplayName ?? COLUMN_LABELS[column] ?? column) : COLUMN_LABELS[column]}</h2>
|
||||
<span className={`column-count${countFlashing ? " count-flash" : ""}`}>{tasks.length}</span>
|
||||
<span className={`column-count${countFlashing ? " count-flash" : ""}`}><span>{activeTaskCount}</span>/<span>{tasks.length}</span></span>
|
||||
{(workflowMode ? isReviewColumn : column === "in-review") && onToggleAutoMerge && (
|
||||
<label className="auto-merge-toggle" title={autoMerge ? t("column.autoMergeEnabled", "Auto-merge enabled") : t("column.autoMergeDisabled", "Auto-merge disabled")}>
|
||||
{/*
|
||||
|
||||
@@ -26,26 +26,22 @@ type AsyncState<T> =
|
||||
|
||||
type ConcurrencyValues = {
|
||||
maxConcurrent: number;
|
||||
maxTriageConcurrent: number;
|
||||
maxWorktrees: number;
|
||||
};
|
||||
|
||||
const CONCURRENCY_SAVE_DEBOUNCE_MS = 500;
|
||||
const DEFAULT_CONCURRENCY_VALUES: ConcurrencyValues = {
|
||||
maxConcurrent: DEFAULT_PROJECT_SETTINGS.maxConcurrent,
|
||||
maxTriageConcurrent: DEFAULT_PROJECT_SETTINGS.maxTriageConcurrent,
|
||||
maxWorktrees: DEFAULT_PROJECT_SETTINGS.maxWorktrees,
|
||||
};
|
||||
|
||||
const CONCURRENCY_SLIDER_LIMITS: Record<keyof ConcurrencyValues, { min: number; max: number }> = {
|
||||
maxConcurrent: { min: 1, max: 50 },
|
||||
maxTriageConcurrent: { min: 1, max: 50 },
|
||||
maxWorktrees: { min: 1, max: 50 },
|
||||
};
|
||||
|
||||
const CONCURRENCY_SETTING_LABEL_KEYS: Record<keyof ConcurrencyValues, { key: string; defaultValue: string }> = {
|
||||
maxConcurrent: { key: "commandCenter.controls.concurrency.maxConcurrent", defaultValue: "Max concurrent tasks" },
|
||||
maxTriageConcurrent: { key: "commandCenter.controls.concurrency.maxTriageConcurrent", defaultValue: "Max triage concurrent" },
|
||||
maxWorktrees: { key: "commandCenter.controls.concurrency.maxWorktrees", defaultValue: "Max worktrees" },
|
||||
};
|
||||
|
||||
@@ -218,7 +214,6 @@ export const EngineControlMenu = forwardRef<EngineControlMenuHandle, EngineContr
|
||||
if (!cancelled) {
|
||||
const persistedValues = {
|
||||
maxConcurrent: settings.maxConcurrent ?? config.maxConcurrent ?? DEFAULT_CONCURRENCY_VALUES.maxConcurrent,
|
||||
maxTriageConcurrent: settings.maxTriageConcurrent ?? DEFAULT_CONCURRENCY_VALUES.maxTriageConcurrent,
|
||||
maxWorktrees: settings.maxWorktrees ?? DEFAULT_CONCURRENCY_VALUES.maxWorktrees,
|
||||
};
|
||||
persistedProjectConcurrencyRef.current = persistedValues;
|
||||
@@ -563,27 +558,6 @@ export const EngineControlMenu = forwardRef<EngineControlMenuHandle, EngineContr
|
||||
) : null}
|
||||
</span>
|
||||
</label>
|
||||
<label className="engine-control-menu__slider" htmlFor="engine-control-max-triage-concurrent">
|
||||
<span className="engine-control-menu__slider-label">
|
||||
{t("commandCenter.controls.concurrency.maxTriageConcurrent", "Max triage concurrent")}
|
||||
<strong>{concurrencyValues.maxTriageConcurrent}</strong>
|
||||
</span>
|
||||
<input
|
||||
id="engine-control-max-triage-concurrent"
|
||||
className="engine-control-menu__range input"
|
||||
type="range"
|
||||
min={CONCURRENCY_SLIDER_LIMITS.maxTriageConcurrent.min}
|
||||
max={getConcurrencySliderMax("maxTriageConcurrent", concurrencyValues.maxTriageConcurrent)}
|
||||
value={concurrencyValues.maxTriageConcurrent}
|
||||
disabled={concurrencyState.status === "loading"}
|
||||
onChange={(event) => updateConcurrencyValue(
|
||||
"maxTriageConcurrent",
|
||||
event.target.value,
|
||||
CONCURRENCY_SLIDER_LIMITS.maxTriageConcurrent.min,
|
||||
getConcurrencySliderMax("maxTriageConcurrent", concurrencyValues.maxTriageConcurrent),
|
||||
)}
|
||||
/>
|
||||
</label>
|
||||
<label className="engine-control-menu__slider" htmlFor="engine-control-max-worktrees">
|
||||
<span className="engine-control-menu__slider-label">
|
||||
{t("commandCenter.controls.concurrency.maxWorktrees", "Max worktrees")}
|
||||
|
||||
@@ -10,7 +10,7 @@ import {
|
||||
} from "@fusion/core";
|
||||
import { AlertTriangle, Clock, Folder, MessageSquare, Pause, Play, Square, Zap } from "lucide-react";
|
||||
import { computeBlockerFanoutMap } from "../hooks/useBlockerFanout";
|
||||
import { useExecutorStats } from "../hooks/useExecutorStats";
|
||||
import { useExecutorStats, type ExecutorColumnFlags } from "../hooks/useExecutorStats";
|
||||
import { isLikelyTabSuspensionError } from "../hooks/visibilitySuspension";
|
||||
import { LoadingSpinner } from "./LoadingSpinner";
|
||||
import type { ExecutorState } from "../api";
|
||||
@@ -18,7 +18,7 @@ import { EngineControlMenu, type EngineControlMenuHandle } from "./EngineControl
|
||||
import { TerminalLauncher } from "./TerminalLauncher";
|
||||
import { useViewportMode } from "../hooks/useViewportMode";
|
||||
|
||||
type FooterStatId = "queued" | "running" | "stuck" | "blocked" | "review" | "fanout";
|
||||
type FooterStatId = "queued" | "running" | "stuck" | "blocked" | "fanout";
|
||||
|
||||
interface OpenStatTooltip {
|
||||
id: FooterStatId;
|
||||
@@ -61,6 +61,8 @@ interface ExecutorStatusBarProps {
|
||||
tasks: Task[];
|
||||
/** Project ID for fetching project-specific stats */
|
||||
projectId?: string;
|
||||
/** Optional task-scoped board trait index supplied by an embedding board. */
|
||||
columnFlagsByTaskId?: ReadonlyMap<string, ExecutorColumnFlags>;
|
||||
/** Project-level stuck task timeout in milliseconds (undefined = disabled) */
|
||||
taskStuckTimeoutMs?: number;
|
||||
/** Age threshold in milliseconds before high fan-out blockers escalate in dashboard surfaces. */
|
||||
@@ -141,17 +143,23 @@ function getStateDisplay(state: ExecutorState, t: TFunction<"app">): { label: st
|
||||
* - Executor state badge (idle/running/paused/stopped)
|
||||
* - Last activity timestamp
|
||||
*/
|
||||
export function ExecutorStatusBar({ tasks, projectId, taskStuckTimeoutMs, staleHighFanoutBlockerAgeThresholdMs, lastFetchTimeMs, currentProjectPath, onOpenProjectDirectory, keyboardOpen, hideWhenKeyboardOpen, onToggleTerminal, onOpenScripts, onRunScript, quickChatButtonMode = "off", onOpenQuickChat }: ExecutorStatusBarProps) {
|
||||
export function ExecutorStatusBar({ tasks, projectId, columnFlagsByTaskId: suppliedColumnFlagsByTaskId, taskStuckTimeoutMs, staleHighFanoutBlockerAgeThresholdMs, lastFetchTimeMs, currentProjectPath, onOpenProjectDirectory, keyboardOpen, hideWhenKeyboardOpen, onToggleTerminal, onOpenScripts, onRunScript, quickChatButtonMode = "off", onOpenQuickChat }: ExecutorStatusBarProps) {
|
||||
const { t } = useTranslation("app");
|
||||
const viewportMode = useViewportMode();
|
||||
const isMobile = viewportMode === "mobile";
|
||||
const showTerminalLauncher = !isMobile && Boolean(onToggleTerminal);
|
||||
const columnFlagsByTaskId = suppliedColumnFlagsByTaskId;
|
||||
/*
|
||||
FNXC:ConcurrencyIndicators 2026-08-03-12:00:
|
||||
FN-8453 receives a task-scoped board trait index from App before the shared
|
||||
live-agent predicate runs. Literal column ids are only a loading/legacy fallback.
|
||||
*/
|
||||
/*
|
||||
* FNXC:ChatLauncher 2026-06-22-15:18:
|
||||
* Settings can route Quick Chat to a footer launcher beside Terminal, keep the draggable floating FAB, or hide the launcher entirely. Footer launch stays desktop/tablet-only like Terminal while mobile opens from the floating path as a full-screen modal.
|
||||
*/
|
||||
const showQuickChatFooterLauncher = !isMobile && quickChatButtonMode === "footer" && Boolean(onOpenQuickChat);
|
||||
const { stats, loading, error } = useExecutorStats(tasks, projectId, taskStuckTimeoutMs, lastFetchTimeMs);
|
||||
const { stats, loading, error } = useExecutorStats(tasks, projectId, taskStuckTimeoutMs, lastFetchTimeMs, columnFlagsByTaskId);
|
||||
const [isProjectPathVisible, setIsProjectPathVisible] = useState(false);
|
||||
const [openStatTooltip, setOpenStatTooltip] = useState<OpenStatTooltip | null>(null);
|
||||
const engineControlMenuRef = useRef<EngineControlMenuHandle>(null);
|
||||
@@ -282,16 +290,16 @@ export function ExecutorStatusBar({ tasks, projectId, taskStuckTimeoutMs, staleH
|
||||
* executor statistics and launch controls only.
|
||||
*/}
|
||||
|
||||
{/* Queued tasks */}
|
||||
{/* Waiting intake/hold tasks */}
|
||||
<MobileStatSegment
|
||||
id="queued"
|
||||
isMobile={isMobile}
|
||||
isOpen={openStatTooltip?.id === "queued"}
|
||||
label={t("executor.queued", "Queued")}
|
||||
label={t("executor.waiting", "Waiting")}
|
||||
onToggle={toggleStatTooltip}
|
||||
>
|
||||
<span className="executor-status-bar__indicator executor-status-bar__indicator--queued" aria-hidden="true" />
|
||||
<span className="executor-status-bar__label">{t("executor.queued", "Queued")}</span>
|
||||
<span className="executor-status-bar__label">{t("executor.waiting", "Waiting")}</span>
|
||||
<span className="executor-status-bar__count">{stats.queuedTaskCount}</span>
|
||||
</MobileStatSegment>
|
||||
|
||||
@@ -356,22 +364,6 @@ export function ExecutorStatusBar({ tasks, projectId, taskStuckTimeoutMs, staleH
|
||||
</span>
|
||||
</MobileStatSegment>
|
||||
|
||||
{/* Separator */}
|
||||
<span className="executor-status-bar__divider" aria-hidden="true" />
|
||||
|
||||
{/* In review count */}
|
||||
<MobileStatSegment
|
||||
id="review"
|
||||
isMobile={isMobile}
|
||||
isOpen={openStatTooltip?.id === "review"}
|
||||
label={t("executor.inReview", "In Review")}
|
||||
onToggle={toggleStatTooltip}
|
||||
>
|
||||
<span className="executor-status-bar__indicator executor-status-bar__indicator--review" aria-hidden="true" />
|
||||
<span className="executor-status-bar__label">{t("executor.inReview", "In Review")}</span>
|
||||
<span className="executor-status-bar__count">{stats.inReviewCount}</span>
|
||||
</MobileStatSegment>
|
||||
|
||||
{highestOverlapBlocker && (
|
||||
<>
|
||||
<span className="executor-status-bar__divider" aria-hidden="true" />
|
||||
|
||||
@@ -1156,7 +1156,6 @@ export function SettingsModal({
|
||||
const [form, setForm] = useState<SettingsFormState>({
|
||||
maxConcurrent: 2,
|
||||
maxConcurrentVerifications: 1,
|
||||
maxTriageConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
pollIntervalMs: 15000,
|
||||
heartbeatMultiplier: 1,
|
||||
|
||||
@@ -109,7 +109,7 @@ describe("Column count-flash", () => {
|
||||
const tasks = [makeTask("FN-001")];
|
||||
render(<Column {...defaultProps} tasks={tasks} />);
|
||||
|
||||
const badge = screen.getByText("1");
|
||||
const badge = screen.getByText("1").parentElement!;
|
||||
expect(badge.className).toContain("column-count");
|
||||
expect(badge.className).not.toContain("count-flash");
|
||||
});
|
||||
@@ -121,7 +121,7 @@ describe("Column count-flash", () => {
|
||||
const moreTasks = [makeTask("FN-001"), makeTask("FN-002")];
|
||||
rerender(<Column {...defaultProps} tasks={moreTasks} />);
|
||||
|
||||
const badge = screen.getByText("2");
|
||||
const badge = screen.getByText("2").parentElement!;
|
||||
expect(badge.className).toContain("count-flash");
|
||||
});
|
||||
|
||||
@@ -998,7 +998,7 @@ describe("Column same-column drop", () => {
|
||||
// Dropping into "in-review" column (which has 0 tasks)
|
||||
render(<Column {...defaultProps} column="in-review" tasks={tasksInTargetColumn} onMoveTask={onMoveTask} addToast={addToast} />);
|
||||
|
||||
const columnEl = screen.getByText("0").closest(".column") as HTMLElement;
|
||||
const columnEl = screen.getAllByText("0")[0].closest(".column") as HTMLElement;
|
||||
const dataTransfer = {
|
||||
getData: vi.fn().mockReturnValue("FN-001"),
|
||||
dropEffect: "move",
|
||||
|
||||
@@ -304,20 +304,16 @@ describe("EngineControlMenu", () => {
|
||||
await openMenu();
|
||||
|
||||
const maxConcurrent = await screen.findByLabelText(/max concurrent tasks/i);
|
||||
const maxTriage = screen.getByLabelText(/max triage concurrent/i);
|
||||
const maxWorktrees = screen.getByLabelText(/max worktrees/i);
|
||||
|
||||
vi.useFakeTimers();
|
||||
|
||||
expect(maxConcurrent).toHaveAttribute("max", "60");
|
||||
expect(maxConcurrent).toHaveValue("60");
|
||||
expect(maxTriage).toHaveAttribute("max", "70");
|
||||
expect(maxTriage).toHaveValue("70");
|
||||
expect(maxWorktrees).toHaveAttribute("max", "80");
|
||||
expect(maxWorktrees).toHaveValue("80");
|
||||
|
||||
fireEvent.change(maxConcurrent, { target: { value: "9" } });
|
||||
fireEvent.change(maxTriage, { target: { value: "4" } });
|
||||
fireEvent.change(maxWorktrees, { target: { value: "8" } });
|
||||
|
||||
await act(async () => {
|
||||
@@ -325,7 +321,6 @@ describe("EngineControlMenu", () => {
|
||||
});
|
||||
|
||||
expect(screen.getByRole("dialog", { name: /confirm concurrency change/i })).toHaveTextContent("Max concurrent tasks from 60 to 9");
|
||||
expect(screen.getByRole("dialog", { name: /confirm concurrency change/i })).toHaveTextContent("Max triage concurrent from 70 to 4");
|
||||
expect(screen.getByRole("dialog", { name: /confirm concurrency change/i })).toHaveTextContent("Max worktrees from 80 to 8");
|
||||
expect(legacyMocks.updateSettings).not.toHaveBeenCalled();
|
||||
|
||||
@@ -335,7 +330,7 @@ describe("EngineControlMenu", () => {
|
||||
fireEvent.click(saveButton);
|
||||
|
||||
await waitFor(() => expect(legacyMocks.updateSettings).toHaveBeenCalledWith(
|
||||
{ maxConcurrent: 9, maxTriageConcurrent: 4, maxWorktrees: 8 },
|
||||
{ maxConcurrent: 9, maxWorktrees: 8 },
|
||||
"proj_123",
|
||||
));
|
||||
expect(apiMocks.fetchSettings).toHaveBeenCalledTimes(2);
|
||||
@@ -351,7 +346,6 @@ describe("EngineControlMenu", () => {
|
||||
await openMenu();
|
||||
|
||||
expect(await screen.findByLabelText(/max concurrent tasks/i)).toHaveAttribute("max", "50");
|
||||
expect(screen.getByLabelText(/max triage concurrent/i)).toHaveAttribute("max", "50");
|
||||
expect(screen.getByLabelText(/max worktrees/i)).toHaveAttribute("max", "50");
|
||||
});
|
||||
|
||||
@@ -699,7 +693,7 @@ describe("EngineControlMenu", () => {
|
||||
fireEvent.click(screen.getByRole("button", { name: /save change/i }));
|
||||
|
||||
await waitFor(() => expect(legacyMocks.updateSettings).toHaveBeenCalledWith(
|
||||
{ maxConcurrent: 50, maxTriageConcurrent: 1, maxWorktrees: 4 },
|
||||
{ maxConcurrent: 50, maxWorktrees: 4 },
|
||||
"proj_123",
|
||||
));
|
||||
});
|
||||
|
||||
@@ -143,8 +143,8 @@ describe("ExecutorStatusBar", () => {
|
||||
const statusBar = screen.getByRole("status");
|
||||
expect(statusBar).toHaveTextContent("Running");
|
||||
expect(statusBar).toHaveTextContent("Blocked");
|
||||
expect(statusBar).toHaveTextContent("Queued");
|
||||
expect(statusBar).toHaveTextContent("In Review");
|
||||
expect(statusBar).toHaveTextContent("Waiting");
|
||||
expect(statusBar).not.toHaveTextContent("In Review");
|
||||
expect(statusBar).not.toHaveTextContent("Done");
|
||||
expect(statusBar).not.toHaveTextContent("Escalated");
|
||||
});
|
||||
@@ -190,12 +190,12 @@ describe("ExecutorStatusBar", () => {
|
||||
);
|
||||
|
||||
const statusBar = screen.getByRole("status");
|
||||
expectSegmentCount("Queued", "9");
|
||||
expectSegmentCount("Waiting", "9");
|
||||
expectSegmentCount("Running", "2");
|
||||
expect(within(getSegmentByLabel("Running")).getByText("4")).toHaveClass("executor-status-bar__max");
|
||||
expectSegmentCount("Stuck", "1");
|
||||
expectSegmentCount("Blocked", "2");
|
||||
expectSegmentCount("In Review", "1");
|
||||
expect(statusBar).not.toHaveTextContent("In Review");
|
||||
expect(statusBar).toHaveTextContent("Overlap queue");
|
||||
expect(statusBar).toHaveTextContent("FN-010 · 5 todo");
|
||||
expect(statusBar).not.toHaveTextContent("Done");
|
||||
@@ -274,11 +274,10 @@ describe("ExecutorStatusBar", () => {
|
||||
expect(statusBar).toHaveTextContent("5");
|
||||
});
|
||||
|
||||
it("displays in-review count", () => {
|
||||
it("does not display the removed in-review footer segment", () => {
|
||||
render(<ExecutorStatusBar tasks={emptyTasks} />);
|
||||
|
||||
const statusBar = screen.getByRole("status");
|
||||
expect(statusBar).toHaveTextContent("3");
|
||||
expect(screen.getByRole("status")).not.toHaveTextContent("In Review");
|
||||
});
|
||||
|
||||
it("renders the terminal launcher in the footer on desktop and opens terminal from the preserved toggle test id", async () => {
|
||||
@@ -477,17 +476,16 @@ describe("ExecutorStatusBar", () => {
|
||||
});
|
||||
|
||||
describe("mobile stat tooltips", () => {
|
||||
const mobileStatIds = ["queued", "running", "blocked", "review"] as const;
|
||||
const mobileStatIds = ["queued", "running", "blocked"] as const;
|
||||
|
||||
beforeEach(() => {
|
||||
viewportModeMock.value = "mobile";
|
||||
});
|
||||
|
||||
it.each([
|
||||
["queued", "Queued"],
|
||||
["queued", "Waiting"],
|
||||
["running", "Running"],
|
||||
["blocked", "Blocked"],
|
||||
["review", "In Review"],
|
||||
] as const)("reveals the %s stat name on tap", async (id, label) => {
|
||||
const user = userEvent.setup();
|
||||
render(<ExecutorStatusBar tasks={emptyTasks} />);
|
||||
@@ -578,7 +576,7 @@ describe("ExecutorStatusBar", () => {
|
||||
|
||||
const desktopStatus = screen.getByRole("status");
|
||||
expect(desktopStatus).not.toHaveClass("executor-status-bar--mobile");
|
||||
expect(getSegmentByLabel("Queued").tagName).toBe("DIV");
|
||||
expect(getSegmentByLabel("Waiting").tagName).toBe("DIV");
|
||||
expect(desktopStatus.querySelectorAll("button.executor-status-bar__segment--stat")).toHaveLength(0);
|
||||
|
||||
viewportModeMock.value = "tablet";
|
||||
@@ -756,7 +754,7 @@ describe("ExecutorStatusBar", () => {
|
||||
|
||||
const statusBar = screen.getByRole("status");
|
||||
expect(statusBar).toHaveTextContent("Idle");
|
||||
expect(statusBar).toHaveTextContent("Queued");
|
||||
expect(statusBar).toHaveTextContent("Waiting");
|
||||
expect(statusBar).not.toHaveClass("executor-status-bar--loading");
|
||||
expect(screen.queryByText("Loading...")).not.toBeInTheDocument();
|
||||
expect(screen.getByTestId("engine-control-menu-trigger")).toBeInTheDocument();
|
||||
@@ -853,10 +851,10 @@ describe("ExecutorStatusBar", () => {
|
||||
render(<ExecutorStatusBar tasks={emptyTasks} />);
|
||||
|
||||
const statusBar = screen.getByRole("status");
|
||||
expect(statusBar).toHaveTextContent("Queued");
|
||||
expect(statusBar).toHaveTextContent("Waiting");
|
||||
expect(statusBar).toHaveTextContent("Running");
|
||||
expect(statusBar).toHaveTextContent("Blocked");
|
||||
expect(statusBar).toHaveTextContent("In Review");
|
||||
expect(statusBar).not.toHaveTextContent("In Review");
|
||||
expect(statusBar).not.toHaveClass("executor-status-bar--connecting");
|
||||
expect(statusBar.querySelector(".executor-status-bar--connecting")).toBeNull();
|
||||
expect(screen.queryByText("Connecting…")).not.toBeInTheDocument();
|
||||
@@ -984,13 +982,13 @@ describe("ExecutorStatusBar", () => {
|
||||
const tasks: any[] = [{ id: "FN-001" }];
|
||||
render(<ExecutorStatusBar tasks={tasks} projectId="proj_abc123" />);
|
||||
|
||||
expect(mockUseExecutorStats).toHaveBeenCalledWith(tasks, "proj_abc123", undefined, undefined);
|
||||
expect(mockUseExecutorStats).toHaveBeenCalledWith(tasks, "proj_abc123", undefined, undefined, undefined);
|
||||
});
|
||||
|
||||
it("passes tasks and undefined to useExecutorStats when projectId not provided", () => {
|
||||
render(<ExecutorStatusBar tasks={emptyTasks} />);
|
||||
|
||||
expect(mockUseExecutorStats).toHaveBeenCalledWith(emptyTasks, undefined, undefined, undefined);
|
||||
expect(mockUseExecutorStats).toHaveBeenCalledWith(emptyTasks, undefined, undefined, undefined, undefined);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1072,7 +1070,7 @@ describe("ExecutorStatusBar", () => {
|
||||
render(<ExecutorStatusBar tasks={tasks} />);
|
||||
|
||||
// useExecutorStats receives the tasks array as first argument
|
||||
expect(mockUseExecutorStats).toHaveBeenCalledWith(tasks, undefined, undefined, undefined);
|
||||
expect(mockUseExecutorStats).toHaveBeenCalledWith(tasks, undefined, undefined, undefined, undefined);
|
||||
});
|
||||
|
||||
it("renders stuck segment with correct count when stuck tasks detected", () => {
|
||||
|
||||
@@ -31,26 +31,22 @@ type AsyncState<T> =
|
||||
|
||||
type ConcurrencyValues = {
|
||||
maxConcurrent: number;
|
||||
maxTriageConcurrent: number;
|
||||
maxWorktrees: number;
|
||||
};
|
||||
|
||||
const CONCURRENCY_SAVE_DEBOUNCE_MS = 500;
|
||||
const DEFAULT_CONCURRENCY_VALUES: ConcurrencyValues = {
|
||||
maxConcurrent: DEFAULT_PROJECT_SETTINGS.maxConcurrent,
|
||||
maxTriageConcurrent: DEFAULT_PROJECT_SETTINGS.maxTriageConcurrent,
|
||||
maxWorktrees: DEFAULT_PROJECT_SETTINGS.maxWorktrees,
|
||||
};
|
||||
|
||||
const CONCURRENCY_SLIDER_LIMITS: Record<keyof ConcurrencyValues, { min: number; max: number }> = {
|
||||
maxConcurrent: { min: 1, max: 50 },
|
||||
maxTriageConcurrent: { min: 1, max: 50 },
|
||||
maxWorktrees: { min: 1, max: 50 },
|
||||
};
|
||||
|
||||
const CONCURRENCY_SETTING_LABEL_KEYS: Record<keyof ConcurrencyValues, { key: string; defaultValue: string }> = {
|
||||
maxConcurrent: { key: "commandCenter.controls.concurrency.maxConcurrent", defaultValue: "Max concurrent tasks" },
|
||||
maxTriageConcurrent: { key: "commandCenter.controls.concurrency.maxTriageConcurrent", defaultValue: "Max triage concurrent" },
|
||||
maxWorktrees: { key: "commandCenter.controls.concurrency.maxWorktrees", defaultValue: "Max worktrees" },
|
||||
};
|
||||
|
||||
@@ -129,7 +125,6 @@ export function CommandCenterControls({ projectId, colorTheme, themeMode, shadcn
|
||||
if (!cancelled) {
|
||||
const persistedValues = {
|
||||
maxConcurrent: settings.maxConcurrent ?? config.maxConcurrent ?? DEFAULT_CONCURRENCY_VALUES.maxConcurrent,
|
||||
maxTriageConcurrent: settings.maxTriageConcurrent ?? DEFAULT_CONCURRENCY_VALUES.maxTriageConcurrent,
|
||||
maxWorktrees: settings.maxWorktrees ?? DEFAULT_CONCURRENCY_VALUES.maxWorktrees,
|
||||
};
|
||||
persistedConcurrencyRef.current = persistedValues;
|
||||
@@ -489,27 +484,6 @@ export function CommandCenterControls({ projectId, colorTheme, themeMode, shadcn
|
||||
) : null}
|
||||
</span>
|
||||
</label>
|
||||
<label className="cc-controls-slider" htmlFor="cc-max-triage-concurrent">
|
||||
<span className="cc-controls-slider-label">
|
||||
{t("commandCenter.controls.concurrency.maxTriageConcurrent", "Max triage concurrent")}
|
||||
<strong>{concurrencyValues.maxTriageConcurrent}</strong>
|
||||
</span>
|
||||
<input
|
||||
id="cc-max-triage-concurrent"
|
||||
className="cc-controls-touch-slider"
|
||||
type="range"
|
||||
min={CONCURRENCY_SLIDER_LIMITS.maxTriageConcurrent.min}
|
||||
max={getConcurrencySliderMax("maxTriageConcurrent", concurrencyValues.maxTriageConcurrent)}
|
||||
value={concurrencyValues.maxTriageConcurrent}
|
||||
disabled={concurrencyState.status === "loading"}
|
||||
onChange={(event) => updateConcurrencyValue(
|
||||
"maxTriageConcurrent",
|
||||
event.target.value,
|
||||
CONCURRENCY_SLIDER_LIMITS.maxTriageConcurrent.min,
|
||||
getConcurrencySliderMax("maxTriageConcurrent", concurrencyValues.maxTriageConcurrent),
|
||||
)}
|
||||
/>
|
||||
</label>
|
||||
<label className="cc-controls-slider" htmlFor="cc-max-worktrees">
|
||||
<span className="cc-controls-slider-label">
|
||||
{t("commandCenter.controls.concurrency.maxWorktrees", "Max worktrees")}
|
||||
|
||||
@@ -30,15 +30,6 @@ export const schedulingSearchEntries: SettingsSearchEntry[] = [
|
||||
helpFallback: "Caps stacked typecheck/build verification across tasks. Default: 1. Range: 1–8.",
|
||||
keywords: ["parallelism", "tests", "cpu", "load"],
|
||||
},
|
||||
{
|
||||
sectionId: "scheduling",
|
||||
key: "maxTriageConcurrent",
|
||||
labelKey: "settings.scheduling.maxTriageConcurrent",
|
||||
labelFallback: "Max Triage Concurrent",
|
||||
helpKey: "settings.scheduling.maximumConcurrentPlanningAgents",
|
||||
helpFallback: "Maximum concurrent planning agents. Default: 2.",
|
||||
keywords: ["parallelism", "capacity", "spec"],
|
||||
},
|
||||
{
|
||||
sectionId: "scheduling",
|
||||
key: "executorToolFailureRetryCount",
|
||||
|
||||
@@ -77,19 +77,6 @@ export function SchedulingSection({ form, setForm, concurrencyLoading = false, o
|
||||
setForm((f) => ({ ...f, maxConcurrentVerifications: n } as SettingsFormState));
|
||||
}}
|
||||
/>
|
||||
<SettingsNumberRow
|
||||
descriptor={{
|
||||
key: "maxTriageConcurrent",
|
||||
label: t("settings.scheduling.maxTriageConcurrent", "Max Triage Concurrent"),
|
||||
help: t("settings.scheduling.maximumConcurrentPlanningAgents", "Maximum concurrent planning agents. Default: 2."),
|
||||
scope: "project",
|
||||
min: 1,
|
||||
max: 10,
|
||||
disabled: concurrencyLoading,
|
||||
}}
|
||||
value={form.maxTriageConcurrent ?? null}
|
||||
onChange={(v) => setForm((f) => ({ ...f, maxTriageConcurrent: v ?? undefined } as SettingsFormState))}
|
||||
/>
|
||||
<SettingsNumberRow
|
||||
descriptor={{
|
||||
key: "pollIntervalMs",
|
||||
|
||||
@@ -781,8 +781,8 @@ describe("useExecutorStats", () => {
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
});
|
||||
|
||||
expect(result.current.stats.queuedTaskCount).toBe(10); // triage/planning + todo, no done/archived/non-planning custom
|
||||
expect(result.current.stats.runningTaskCount).toBe(2); // in-progress only
|
||||
expect(result.current.stats.queuedTaskCount).toBe(8); // waiting intake/hold only; live planners are Running
|
||||
expect(result.current.stats.runningTaskCount).toBe(4); // live execute plus planning in any lane
|
||||
expect(result.current.stats.stuckTaskCount).toBe(1); // stuck is an in-progress subset
|
||||
expect(result.current.stats.blockedTaskCount).toBe(2); // actionable string/array blockedBy only
|
||||
expect(result.current.stats.inReviewCount).toBe(1); // in-review only
|
||||
@@ -790,7 +790,7 @@ describe("useExecutorStats", () => {
|
||||
expect(result.current.stats.executorState).toBe("running");
|
||||
});
|
||||
|
||||
it("counts planning/triage as queued but excludes done, archived, and unknown columns", async () => {
|
||||
it("counts intake waiting separately from live planning and excludes terminal/unknown columns", async () => {
|
||||
const tasks: Task[] = [
|
||||
createMockTask("FN-001", "triage"),
|
||||
{ ...createMockTask("FN-002", "triage"), status: "planning" } as Task,
|
||||
@@ -806,8 +806,8 @@ describe("useExecutorStats", () => {
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
});
|
||||
|
||||
expect(result.current.stats.runningTaskCount).toBe(0);
|
||||
expect(result.current.stats.queuedTaskCount).toBe(3);
|
||||
expect(result.current.stats.runningTaskCount).toBe(2);
|
||||
expect(result.current.stats.queuedTaskCount).toBe(1);
|
||||
expect(result.current.stats.inReviewCount).toBe(0);
|
||||
expect(result.current.stats.blockedTaskCount).toBe(0);
|
||||
expect(result.current.stats.stuckTaskCount).toBe(0);
|
||||
@@ -875,5 +875,7 @@ function createMockTask(id: string, column: Task["column"]): Task {
|
||||
log: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
// Fixture represents an active executor session, not an idle worktree shell.
|
||||
...(column === "in-progress" ? { sessionFile: `/tmp/${id}.json` } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { useState, useEffect, useCallback, useRef } from "react";
|
||||
import type { Task } from "@fusion/core";
|
||||
import type { Task, TraitFlags } from "@fusion/core";
|
||||
import { enrichRunningAgentTaskShapeFromFlags, isRunningAgentTask, isWaitingAgentTask } from "../../../core/src/live-agent-count";
|
||||
import { fetchExecutorStats } from "../api";
|
||||
import type { ExecutorStats, ExecutorState } from "../api";
|
||||
import { isTaskStuck } from "../utils/taskStuck";
|
||||
@@ -58,10 +59,15 @@ function deriveExecutorState(
|
||||
/**
|
||||
* Derive statistics from the task list.
|
||||
*
|
||||
* FNXC:ExecutorStatusBar 2026-07-03-00:18:
|
||||
* Footer task counters must mirror the board's operator-facing active work states: Queued includes todo plus planning/triage work, Running and Stuck stay scoped to in-progress execution, Done remains absent from the footer contract unless a labeled Done segment is introduced, and archived/completed/non-planning custom lanes never inflate active pressure counts.
|
||||
* FNXC:ExecutorStatusBar 2026-07-21-14:30:
|
||||
* FN-8453 / #2359 requires footer capacity indicators to share the live top-level
|
||||
* agent predicate with admission: Waiting is trait-derived intake/hold membership,
|
||||
* Running excludes paused, idle, and terminal cards, and custom columns require
|
||||
* task-scoped workflow flags rather than legacy column-id assumptions.
|
||||
*/
|
||||
function deriveStatsFromTasks(tasks: Task[], taskStuckTimeoutMs?: number, lastFetchTimeMs?: number): Pick<
|
||||
export type ExecutorColumnFlags = Pick<TraitFlags, "complete" | "archived" | "intake" | "hold" | "countsTowardWip" | "mergeOrchestration" | "mergeBlocker">;
|
||||
|
||||
export function deriveStatsFromTasks(tasks: Task[], taskStuckTimeoutMs?: number, lastFetchTimeMs?: number, columnFlagsById?: ReadonlyMap<string, ExecutorColumnFlags>, columnFlagsByTaskId?: ReadonlyMap<string, ExecutorColumnFlags>): Pick<
|
||||
ExecutorStats,
|
||||
"runningTaskCount" | "blockedTaskCount" | "stuckTaskCount" | "queuedTaskCount" | "inReviewCount"
|
||||
> {
|
||||
@@ -72,27 +78,15 @@ function deriveStatsFromTasks(tasks: Task[], taskStuckTimeoutMs?: number, lastFe
|
||||
let inReviewCount = 0;
|
||||
|
||||
for (const task of tasks) {
|
||||
switch (task.column) {
|
||||
case "in-progress":
|
||||
runningTaskCount++;
|
||||
if (isTaskStuck(task, taskStuckTimeoutMs, lastFetchTimeMs)) {
|
||||
stuckTaskCount++;
|
||||
}
|
||||
break;
|
||||
case "todo":
|
||||
case "triage":
|
||||
queuedTaskCount++;
|
||||
break;
|
||||
case "in-review":
|
||||
inReviewCount++;
|
||||
break;
|
||||
default:
|
||||
if (task.status === "planning" && !isTerminalOrActiveTaskColumn(task.column)) {
|
||||
queuedTaskCount++;
|
||||
}
|
||||
break;
|
||||
// Task-scoped flags preserve custom workflow meaning when aggregate boards reuse column ids.
|
||||
const enriched = enrichRunningAgentTaskShapeFromFlags(task, columnFlagsByTaskId?.get(task.id) ?? columnFlagsById?.get(task.column));
|
||||
if (isRunningAgentTask(enriched)) {
|
||||
runningTaskCount++;
|
||||
if (isTaskStuck(task, taskStuckTimeoutMs, lastFetchTimeMs)) stuckTaskCount++;
|
||||
}
|
||||
|
||||
if (isWaitingAgentTask(enriched)) queuedTaskCount++;
|
||||
// Kept in the API shape for compatibility; the footer no longer renders it.
|
||||
if (task.column === "in-review") inReviewCount++;
|
||||
if (hasActionableBlockedBy(task.blockedBy)) {
|
||||
blockedTaskCount++;
|
||||
}
|
||||
@@ -107,10 +101,6 @@ function deriveStatsFromTasks(tasks: Task[], taskStuckTimeoutMs?: number, lastFe
|
||||
};
|
||||
}
|
||||
|
||||
function isTerminalOrActiveTaskColumn(column: Task["column"]): boolean {
|
||||
return column === "in-progress" || column === "in-review" || column === "done" || column === "archived";
|
||||
}
|
||||
|
||||
function hasActionableBlockedBy(blockedBy: Task["blockedBy"] | string[] | null): boolean {
|
||||
if (Array.isArray(blockedBy)) {
|
||||
return blockedBy.some((id) => typeof id === "string" && id.trim().length > 0);
|
||||
@@ -140,7 +130,7 @@ const DEFAULT_API_DATA: Pick<ExecutorStats, "maxConcurrent" | "lastActivityAt">
|
||||
maxConcurrent: 2,
|
||||
};
|
||||
|
||||
export function useExecutorStats(tasks: Task[], projectId?: string, taskStuckTimeoutMs?: number, lastFetchTimeMs?: number): UseExecutorStatsResult {
|
||||
export function useExecutorStats(tasks: Task[], projectId?: string, taskStuckTimeoutMs?: number, lastFetchTimeMs?: number, columnFlagsByTaskId?: ReadonlyMap<string, ExecutorColumnFlags>): UseExecutorStatsResult {
|
||||
|
||||
const [apiDataState, setApiDataState] = useState<{
|
||||
projectId?: string;
|
||||
@@ -255,7 +245,7 @@ export function useExecutorStats(tasks: Task[], projectId?: string, taskStuckTim
|
||||
const effectiveLoading = loading || (!error && !currentProjectApiDataState);
|
||||
|
||||
// Derive stats from tasks and API data
|
||||
const taskStats = deriveStatsFromTasks(tasks, taskStuckTimeoutMs, lastFetchTimeMs);
|
||||
const taskStats = deriveStatsFromTasks(tasks, taskStuckTimeoutMs, lastFetchTimeMs, undefined, columnFlagsByTaskId);
|
||||
const executorState = deriveExecutorState(
|
||||
apiData.globalPause,
|
||||
apiData.enginePaused,
|
||||
|
||||
@@ -506,6 +506,7 @@ describe("countRunningAgentsInRegisteredProjectStores", () => {
|
||||
column: string;
|
||||
status?: string;
|
||||
paused?: boolean;
|
||||
sessionFile?: string;
|
||||
};
|
||||
|
||||
function installTaskList(store: TaskStore, seedTasks: Array<string | SeedTask>) {
|
||||
@@ -523,7 +524,7 @@ describe("countRunningAgentsInRegisteredProjectStores", () => {
|
||||
|
||||
it("counts in-progress tasks from already-open stores without opening unopened projects", async () => {
|
||||
const openStore = await getOrCreateProjectStore("proj_open");
|
||||
const listTasks = installTaskList(openStore, ["todo", "in-progress", "in-progress", "done"]);
|
||||
const listTasks = installTaskList(openStore, ["todo", { column: "in-progress", sessionFile: "/tmp/a" }, { column: "in-progress", sessionFile: "/tmp/b" }, "done"]);
|
||||
const openEntry = createdStores.find((entry) => entry.projectId === "proj_open");
|
||||
expect(openEntry).toBeDefined();
|
||||
vi.clearAllMocks();
|
||||
@@ -558,8 +559,8 @@ describe("countRunningAgentsInRegisteredProjectStores", () => {
|
||||
it("returns per-project counts for multiple already-open stores", async () => {
|
||||
const storeA = await getOrCreateProjectStore("proj_a");
|
||||
const storeB = await getOrCreateProjectStore("proj_b");
|
||||
const listTasksA = installTaskList(storeA, ["in-progress", "todo"]);
|
||||
const listTasksB = installTaskList(storeB, ["todo", "in-progress", "in-progress"]);
|
||||
const listTasksA = installTaskList(storeA, [{ column: "in-progress", sessionFile: "/tmp/a" }, "todo"]);
|
||||
const listTasksB = installTaskList(storeB, ["todo", { column: "in-progress", sessionFile: "/tmp/b" }, { column: "in-progress", sessionFile: "/tmp/c" }]);
|
||||
vi.clearAllMocks();
|
||||
|
||||
const counts = await countRunningAgentsInRegisteredProjectStores(["proj_a", "proj_b"]);
|
||||
@@ -584,7 +585,7 @@ describe("countRunningAgentsInRegisteredProjectStores", () => {
|
||||
it("sums in-progress executors, active triage agents, and active in-review agents while excluding inactive states", async () => {
|
||||
const store = await getOrCreateProjectStore("proj_mixed");
|
||||
installTaskList(store, [
|
||||
"in-progress",
|
||||
{ column: "in-progress", sessionFile: "/tmp/live" },
|
||||
{ column: "triage", status: "planning" },
|
||||
{ column: "triage", status: "planning", paused: true },
|
||||
{ column: "triage", status: "triaged" },
|
||||
@@ -606,7 +607,7 @@ describe("countRunningAgentsInRegisteredProjectStores", () => {
|
||||
const storeA = await getOrCreateProjectStore("proj_triage_a");
|
||||
const storeB = await getOrCreateProjectStore("proj_triage_b");
|
||||
const listTasksA = installTaskList(storeA, [
|
||||
"in-progress",
|
||||
{ column: "in-progress", sessionFile: "/tmp/live" },
|
||||
{ column: "triage", status: "planning" },
|
||||
{ column: "triage", status: "waiting" },
|
||||
]);
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
* const store = await getOrCreateProjectStore(projectId);
|
||||
*/
|
||||
|
||||
import { countRunningAgentTasks, type TaskStore } from "@fusion/core";
|
||||
import { countRunningAgentTasks, enrichRunningAgentTaskShape, resolveWorkflowIrForTask, type TaskStore } from "@fusion/core";
|
||||
|
||||
/**
|
||||
* Internal cache: projectId → TaskStore instance.
|
||||
@@ -277,7 +277,17 @@ export function listRegisteredProjectStores(): Array<{ projectId: string; store:
|
||||
*/
|
||||
export async function countRunningAgentsInStore(store: TaskStore): Promise<number> {
|
||||
const tasks = await store.listTasks({ slim: true });
|
||||
return countRunningAgentTasks(tasks);
|
||||
const irCache = new Map();
|
||||
/*
|
||||
FNXC:ConcurrencyIndicators 2026-08-03-12:00:
|
||||
FN-8453 requires dashboard store counts to enrich custom workflow traits before
|
||||
the pure predicate runs; otherwise a custom complete card with stale session
|
||||
metadata would falsely consume displayed/global top-level concurrency.
|
||||
*/
|
||||
const enriched = await Promise.all(tasks.map(async (task) =>
|
||||
enrichRunningAgentTaskShape(task, await resolveWorkflowIrForTask(store, task.id, irCache)),
|
||||
));
|
||||
return countRunningAgentTasks(enriched);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -108,13 +108,12 @@ export const registerConfigMcpPiSettingsRoutes: ApiRouteRegistrar = (ctx) => {
|
||||
const settings = await scopedStore.getSettingsFast();
|
||||
res.json({
|
||||
maxConcurrent: settings.maxConcurrent ?? options?.maxConcurrent ?? 2,
|
||||
maxTriageConcurrent: settings.maxTriageConcurrent ?? settings.maxConcurrent ?? 2,
|
||||
maxWorktrees: settings.maxWorktrees ?? 4,
|
||||
rootDir: scopedStore.getRootDir(),
|
||||
});
|
||||
} catch {
|
||||
const { store: scopedStore } = await getProjectContext(req);
|
||||
res.json({ maxConcurrent: options?.maxConcurrent ?? 2, maxTriageConcurrent: options?.maxConcurrent ?? 2, maxWorktrees: 4, rootDir: scopedStore.getRootDir() });
|
||||
res.json({ maxConcurrent: options?.maxConcurrent ?? 2, maxWorktrees: 4, rootDir: scopedStore.getRootDir() });
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
@@ -2,6 +2,8 @@ import { describe, it, expect, vi } from "vitest";
|
||||
import type { Task } from "@fusion/core";
|
||||
import {
|
||||
AgentSemaphore,
|
||||
ProjectAdmissionCoordinator,
|
||||
compareAdmissionCandidates,
|
||||
ScopedAgentSemaphore,
|
||||
PRIORITY_MERGE,
|
||||
PRIORITY_EXECUTE,
|
||||
@@ -396,7 +398,7 @@ describe("AgentSemaphore", () => {
|
||||
|
||||
it("counts persisted top-level slots with the shared in-review running-agent predicate", () => {
|
||||
const tasks = [
|
||||
{ column: "in-progress" },
|
||||
{ column: "in-progress", sessionFile: "/tmp/live" },
|
||||
{ column: "triage", status: "planning", paused: false },
|
||||
{ column: "triage", status: "planning", paused: true },
|
||||
{ column: "in-review", status: "reviewing", paused: false },
|
||||
@@ -411,9 +413,9 @@ describe("AgentSemaphore", () => {
|
||||
expect(persistedTopLevelAgentSlots(tasks)).toBe(7);
|
||||
});
|
||||
|
||||
it("claims top-level concurrency as max(live running agents, semaphore active, pending specify)", () => {
|
||||
it("claims only this project's live agents and pending planners", () => {
|
||||
const tasks = [
|
||||
{ column: "in-progress" },
|
||||
{ column: "in-progress", sessionFile: "/tmp/live" },
|
||||
{ column: "triage", status: "planning", paused: false },
|
||||
{ column: "triage", status: "planning", paused: false },
|
||||
{ column: "triage", status: "planning", paused: false },
|
||||
@@ -423,9 +425,10 @@ describe("AgentSemaphore", () => {
|
||||
|
||||
// 4 planning + 1 in-progress = 5 live holders (the reported over-cap symptom).
|
||||
expect(computeTopLevelConcurrencyClaimed({ tasks })).toBe(5);
|
||||
// Prefer the larger of live holders and in-memory activeCount.
|
||||
// Host semaphore activity belongs to the process-wide pool, not this
|
||||
// project's maxConcurrent accounting.
|
||||
expect(computeTopLevelConcurrencyClaimed({ tasks, semaphoreActiveCount: 2 })).toBe(5);
|
||||
expect(computeTopLevelConcurrencyClaimed({ tasks: [], semaphoreActiveCount: 3, pendingSpecifyCount: 2 })).toBe(3);
|
||||
expect(computeTopLevelConcurrencyClaimed({ tasks: [], semaphoreActiveCount: 3, pendingSpecifyCount: 2 })).toBe(2);
|
||||
expect(computeTopLevelConcurrencyClaimed({ tasks: [], semaphoreActiveCount: 1, pendingSpecifyCount: 2 })).toBe(2);
|
||||
});
|
||||
|
||||
@@ -580,8 +583,8 @@ describe("AgentSemaphore", () => {
|
||||
for (let i = 0; i < 6; i++) await sem.acquire();
|
||||
// Two genuinely running tasks persist; four held slots are leaked.
|
||||
const tasks = [
|
||||
{ column: "in-progress" },
|
||||
{ column: "in-progress" },
|
||||
{ column: "in-progress", sessionFile: "/tmp/live" },
|
||||
{ column: "in-progress", sessionFile: "/tmp/live" },
|
||||
] as Task[];
|
||||
|
||||
const first = recoverIdleSemaphoreLeakCandidate({
|
||||
@@ -627,7 +630,7 @@ describe("AgentSemaphore", () => {
|
||||
const sem = new AgentSemaphore(4);
|
||||
await sem.acquire();
|
||||
sem.acquireNestedSlot(); // legitimate nested overshoot: active=2, persisted=1
|
||||
const tasks = [{ column: "in-progress" }] as Task[];
|
||||
const tasks = [{ column: "in-progress", sessionFile: "/tmp/live" }] as Task[];
|
||||
|
||||
const candidate = recoverIdleSemaphoreLeakCandidate({
|
||||
semaphore: sem,
|
||||
@@ -666,7 +669,7 @@ describe("AgentSemaphore", () => {
|
||||
for (let i = 0; i < 5; i++) await sem.acquire(); // 5 persisted running tasks
|
||||
await sem.acquire(); // 1 leaked slot (no persisted task backs it)
|
||||
sem.acquireNestedSlot(); // live nested run starts: active=7, nested=1
|
||||
const tasks = Array.from({ length: 5 }, () => ({ column: "in-progress" })) as Task[];
|
||||
const tasks = Array.from({ length: 5 }, () => ({ column: "in-progress", sessionFile: "/tmp/live" })) as Task[];
|
||||
|
||||
const first = recoverIdleSemaphoreLeakCandidate({
|
||||
semaphore: sem,
|
||||
@@ -704,7 +707,7 @@ describe("AgentSemaphore", () => {
|
||||
// both held slots — no excess, no candidate.
|
||||
const result = recoverIdleSemaphoreLeakCandidate({
|
||||
semaphore: sem,
|
||||
tasks: [{ column: "in-progress" }] as Task[],
|
||||
tasks: [{ column: "in-progress", sessionFile: "/tmp/live" }] as Task[],
|
||||
candidateSinceMs: 999,
|
||||
inFlightCount: 1,
|
||||
nowMs: 700_000,
|
||||
@@ -1058,3 +1061,103 @@ describe("AgentSemaphore resilience (FN-978)", () => {
|
||||
expect(sem.activeCount).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
describe("ProjectAdmissionCoordinator", () => {
|
||||
it("admits the oldest same-project candidate atomically and partitions projects", async () => {
|
||||
const coordinator = new ProjectAdmissionCoordinator();
|
||||
const started: string[] = [];
|
||||
const candidates = [
|
||||
{ taskId: "FN-20", projectId: "a", createdAt: "2026-01-02T00:00:00.000Z", start: async () => { started.push("new"); } },
|
||||
{ taskId: "FN-10", projectId: "a", createdAt: "2026-01-01T00:00:00.000Z", start: async () => { started.push("old"); } },
|
||||
{ taskId: "FN-1", projectId: "b", createdAt: "2026-01-03T00:00:00.000Z", start: async () => { started.push("other-project"); } },
|
||||
];
|
||||
const sem = new AgentSemaphore(2);
|
||||
await Promise.all([
|
||||
coordinator.admitOldest({ projectId: "a", maxConcurrent: 1, claimed: () => 0, refresh: async () => candidates, semaphore: sem }),
|
||||
coordinator.admitOldest({ projectId: "a", maxConcurrent: 1, claimed: () => started.length, refresh: async () => candidates, semaphore: sem }),
|
||||
]);
|
||||
expect(started).toEqual(["old"]);
|
||||
sem.release();
|
||||
await coordinator.admitOldest({ projectId: "b", maxConcurrent: 1, claimed: () => 0, refresh: async () => candidates, semaphore: sem });
|
||||
expect(started).toEqual(["old", "other-project"]);
|
||||
});
|
||||
|
||||
it("releases a rejected handoff and retains an accepted reservation until lane transfer", async () => {
|
||||
const coordinator = new ProjectAdmissionCoordinator();
|
||||
const semaphore = new AgentSemaphore(1);
|
||||
const rejected = await coordinator.admitOldest({
|
||||
projectId: "project-a",
|
||||
maxConcurrent: 1,
|
||||
claimed: () => 0,
|
||||
semaphore,
|
||||
refresh: async () => [{
|
||||
taskId: "FN-1", projectId: "project-a", createdAt: "2026-01-01T00:00:00.000Z",
|
||||
start: async () => false,
|
||||
}],
|
||||
});
|
||||
expect(rejected).toBeUndefined();
|
||||
expect(semaphore.activeCount).toBe(0);
|
||||
|
||||
let releaseStart!: () => void;
|
||||
const startBlocked = new Promise<void>((resolve) => { releaseStart = resolve; });
|
||||
const first = coordinator.admitOldest({
|
||||
projectId: "project-a",
|
||||
maxConcurrent: 1,
|
||||
claimed: () => 0,
|
||||
semaphore,
|
||||
refresh: async () => [{
|
||||
taskId: "FN-2", projectId: "project-a", createdAt: "2026-01-01T00:00:00.000Z",
|
||||
start: async () => { await startBlocked; },
|
||||
}],
|
||||
});
|
||||
await Promise.resolve();
|
||||
const second = coordinator.admitOldest({
|
||||
projectId: "project-a",
|
||||
maxConcurrent: 1,
|
||||
claimed: () => 0,
|
||||
semaphore,
|
||||
refresh: async () => [{
|
||||
taskId: "FN-3", projectId: "project-a", createdAt: "2026-01-02T00:00:00.000Z",
|
||||
start: async () => true,
|
||||
}],
|
||||
});
|
||||
releaseStart();
|
||||
expect(await first).toBe("FN-2");
|
||||
expect(await second).toBeUndefined();
|
||||
coordinator.releaseReservation("FN-2");
|
||||
semaphore.release();
|
||||
});
|
||||
|
||||
it("refreshes every registered lane before selecting the cross-lane oldest task", async () => {
|
||||
const coordinator = new ProjectAdmissionCoordinator();
|
||||
const started: string[] = [];
|
||||
coordinator.registerProvider("planning", {
|
||||
projectId: "project-a",
|
||||
refresh: async () => [{
|
||||
taskId: "FN-20", projectId: "project-a", createdAt: "2026-01-02T00:00:00.000Z",
|
||||
start: async () => { started.push("planner"); },
|
||||
}],
|
||||
});
|
||||
coordinator.registerProvider("execute", {
|
||||
projectId: "project-a",
|
||||
refresh: async () => [{
|
||||
taskId: "FN-10", projectId: "project-a", createdAt: "2026-01-01T00:00:00.000Z",
|
||||
start: async () => { started.push("executor"); },
|
||||
}],
|
||||
});
|
||||
|
||||
await coordinator.admitOldest({ projectId: "project-a", maxConcurrent: 1, claimed: () => 0 });
|
||||
expect(started).toEqual(["executor"]);
|
||||
});
|
||||
|
||||
it("uses a stable total order for invalid timestamps and malformed ids", () => {
|
||||
const ordered = [
|
||||
{ taskId: "bad", createdAt: "not-a-date" },
|
||||
{ taskId: "FN-12", createdAt: "2026-01-01T00:00:00.000Z" },
|
||||
{ taskId: "FN-2", createdAt: "2026-01-01T00:00:00.000Z" },
|
||||
{ taskId: "also-bad" },
|
||||
].sort(compareAdmissionCandidates);
|
||||
expect(ordered.map((item) => item.taskId)).toEqual(["FN-2", "FN-12", "also-bad", "bad"]);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { Task } from "@fusion/core";
|
||||
import { ProjectEngine, __resetDeterministicMergerModeDeprecationWarned } from "../project-engine.js";
|
||||
import { AgentSemaphore } from "../concurrency.js";
|
||||
// Resolves to the vi.mock factory above (the mocked merger-ai exports the real-shaped
|
||||
// workspace land error classes so the dispatch's `instanceof` matching is exercised).
|
||||
import { WorkspacePartialLandError, WorkspaceRepoLandBusyError } from "../merger-ai.js";
|
||||
@@ -2445,6 +2446,8 @@ describe("ProjectEngine paused in-review auto-merge behavior", () => {
|
||||
const processPullRequestMerge = vi.fn(async () => "merged" as const);
|
||||
const engine = createEngine({ processPullRequestMerge, getMergeStrategy: () => "pull-request" });
|
||||
await engine.start();
|
||||
const semaphore = new AgentSemaphore(1);
|
||||
(engine as unknown as { runtime: { projectSemaphore?: AgentSemaphore } }).runtime.projectSemaphore = semaphore;
|
||||
engine.enqueueMerge("FN-pr");
|
||||
|
||||
await vi.waitFor(() => {
|
||||
@@ -2457,6 +2460,37 @@ describe("ProjectEngine paused in-review auto-merge behavior", () => {
|
||||
);
|
||||
});
|
||||
|
||||
expect(semaphore.activeCount).toBe(0);
|
||||
await engine.stop();
|
||||
});
|
||||
|
||||
it("runs a sole dequeued merge when coordinator capacity is available", async () => {
|
||||
const mockStore = createMockStore({ ...baseSettings, autoMerge: true, maxConcurrent: 1 });
|
||||
mockStore.store.getTask.mockResolvedValue({
|
||||
id: "FN-sole-merge",
|
||||
column: "in-review",
|
||||
paused: false,
|
||||
mergeRetries: 0,
|
||||
status: null,
|
||||
branch: "fusion/fn-sole-merge",
|
||||
createdAt: "2026-01-01T00:00:00.000Z",
|
||||
});
|
||||
mocks.currentStore = mockStore.store;
|
||||
|
||||
const engine = createEngine();
|
||||
await engine.start();
|
||||
// Exercise the production reservation path: this task has already been
|
||||
// dequeued, so it must add itself as the one-shot admission candidate.
|
||||
(engine as unknown as { runtime: { projectSemaphore?: AgentSemaphore } }).runtime.projectSemaphore = new AgentSemaphore(1);
|
||||
engine.enqueueMerge("FN-sole-merge");
|
||||
|
||||
await vi.waitFor(() => expect(mocks.runAiMerge).toHaveBeenCalledWith(
|
||||
mockStore.store,
|
||||
"/tmp/proj_test",
|
||||
"FN-sole-merge",
|
||||
expect.any(Object),
|
||||
));
|
||||
expect((engine as unknown as { mergeQueue: string[] }).mergeQueue).toEqual([]);
|
||||
await engine.stop();
|
||||
});
|
||||
|
||||
|
||||
@@ -1677,7 +1677,7 @@ Planner rewrote mission without the raw request.
|
||||
});
|
||||
|
||||
describe("poll ordering", () => {
|
||||
it("dispatches eligible triage tasks by priority desc then createdAt asc", async () => {
|
||||
it("dispatches eligible triage tasks by createdAt asc", async () => {
|
||||
const tasks: Task[] = [
|
||||
createTriageTask({
|
||||
id: "FN-100",
|
||||
@@ -1721,10 +1721,10 @@ Planner rewrote mission without the raw request.
|
||||
|
||||
expect(specifySpy).toHaveBeenCalledTimes(4);
|
||||
expect(specifySpy.mock.calls.map(([task]) => task.id)).toEqual([
|
||||
"FN-101",
|
||||
"FN-100",
|
||||
"FN-103",
|
||||
"FN-102",
|
||||
"FN-100",
|
||||
"FN-101",
|
||||
]);
|
||||
});
|
||||
|
||||
@@ -1802,6 +1802,7 @@ Planner rewrote mission without the raw request.
|
||||
...createTriageTask({ id: "FN-EXEC", priority: "normal" }),
|
||||
column: "in-progress",
|
||||
status: null,
|
||||
sessionFile: "/tmp/fusion-fn-exec-session.json",
|
||||
} as Task,
|
||||
];
|
||||
|
||||
|
||||
@@ -1,4 +1,11 @@
|
||||
import { countRunningAgentTasks, type Task } from "@fusion/core";
|
||||
import {
|
||||
compareTaskIdNumeric,
|
||||
countRunningAgentTasks,
|
||||
enrichRunningAgentTaskShape,
|
||||
resolveWorkflowIrForTask,
|
||||
type Task,
|
||||
type WorkflowIrResolverStore,
|
||||
} from "@fusion/core";
|
||||
import { createLogger } from "./logger.js";
|
||||
|
||||
const concurrencyLog = createLogger("concurrency");
|
||||
@@ -10,6 +17,155 @@ export const PRIORITY_EXECUTE = 1;
|
||||
/** Priority level for specification/triage agents — served last (default). */
|
||||
export const PRIORITY_SPECIFY = 0;
|
||||
|
||||
/** A task waiting to enter one of the top-level agent lanes. */
|
||||
export interface AdmissionCandidate {
|
||||
taskId: string;
|
||||
projectId: string;
|
||||
createdAt?: string;
|
||||
/** Records ownership of the host reservation before the lane starts. */
|
||||
reserve?: () => void;
|
||||
/**
|
||||
* Starts the owning lane after this coordinator has atomically reserved a slot.
|
||||
* Return `false` when the lane rejects the handoff before it has accepted the
|
||||
* reservation; this makes the coordinator release capacity in one place.
|
||||
*/
|
||||
start: () => Promise<boolean | void>;
|
||||
}
|
||||
|
||||
/** A lane contributes its current ready work on every project admission pass. */
|
||||
export interface AdmissionProvider {
|
||||
projectId: string;
|
||||
refresh: () => Promise<AdmissionCandidate[]>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Deterministic oldest-first ordering used for all task-lane admission.
|
||||
* Invalid/missing timestamps deliberately sort after valid timestamps; numeric
|
||||
* task ids break normal ties before lexical ids so a malformed fixture cannot
|
||||
* make Array.sort's NaN handling decide capacity admission.
|
||||
*/
|
||||
export function compareAdmissionCandidates(a: Pick<AdmissionCandidate, "taskId" | "createdAt">, b: Pick<AdmissionCandidate, "taskId" | "createdAt">): number {
|
||||
const aTime = a.createdAt ? Date.parse(a.createdAt) : Number.NaN;
|
||||
const bTime = b.createdAt ? Date.parse(b.createdAt) : Number.NaN;
|
||||
const aValid = Number.isFinite(aTime);
|
||||
const bValid = Number.isFinite(bTime);
|
||||
if (aValid !== bValid) return aValid ? -1 : 1;
|
||||
if (aValid && aTime !== bTime) return aTime - bTime;
|
||||
const numeric = compareTaskIdNumeric(a.taskId, b.taskId);
|
||||
return numeric !== 0 ? numeric : a.taskId.localeCompare(b.taskId);
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:ConcurrencyAdmission 2026-08-03-12:00:
|
||||
FN-8453 / #2359 requires a per-project oldest-first authority rather than
|
||||
independent triage/execute/merge polls or semaphore lane priority. Lanes refresh
|
||||
candidates then hand their starts here; nested runNested helpers are deliberately
|
||||
not candidates because they remain parent-internal soft breaches.
|
||||
*/
|
||||
export class ProjectAdmissionCoordinator {
|
||||
private draining = new Map<string, Promise<void>>();
|
||||
private providers = new Map<string, Map<string, AdmissionProvider>>();
|
||||
/**
|
||||
* Reservations bridge coordinator selection and durable task liveness. They
|
||||
* are deliberately project-scoped, so a prompt handoff cannot let a second
|
||||
* same-project admission observe stale persisted rows and exceed maxConcurrent.
|
||||
*/
|
||||
private reservations = new Map<string, Set<string>>();
|
||||
|
||||
private reserve(projectId: string, taskId: string): void {
|
||||
const tasks = this.reservations.get(projectId) ?? new Set<string>();
|
||||
tasks.add(taskId);
|
||||
this.reservations.set(projectId, tasks);
|
||||
}
|
||||
|
||||
releaseReservation(taskId: string): void {
|
||||
for (const [projectId, tasks] of this.reservations) {
|
||||
if (!tasks.delete(taskId)) continue;
|
||||
if (tasks.size === 0) this.reservations.delete(projectId);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
private reservationCount(projectId: string): number {
|
||||
return this.reservations.get(projectId)?.size ?? 0;
|
||||
}
|
||||
|
||||
/** Register a lane's refresh source. Re-registering replaces its prior source. */
|
||||
registerProvider(providerId: string, provider: AdmissionProvider): () => void {
|
||||
const projectProviders = this.providers.get(provider.projectId) ?? new Map<string, AdmissionProvider>();
|
||||
projectProviders.set(providerId, provider);
|
||||
this.providers.set(provider.projectId, projectProviders);
|
||||
return () => {
|
||||
const current = this.providers.get(provider.projectId);
|
||||
current?.delete(providerId);
|
||||
if (current?.size === 0) this.providers.delete(provider.projectId);
|
||||
};
|
||||
}
|
||||
|
||||
async admitOldest(params: {
|
||||
projectId: string;
|
||||
maxConcurrent: number;
|
||||
claimed: () => Promise<number> | number;
|
||||
/** One-shot source for callers that do not hold a durable lane registration. */
|
||||
refresh?: () => Promise<AdmissionCandidate[]>;
|
||||
semaphore?: Pick<AgentSemaphore, "tryAcquire" | "release">;
|
||||
}): Promise<string | undefined> {
|
||||
const existing = this.draining.get(params.projectId);
|
||||
if (existing) await existing;
|
||||
let admitted: string | undefined;
|
||||
const drain = (async () => {
|
||||
const providers = [...(this.providers.get(params.projectId)?.values() ?? [])];
|
||||
if (params.refresh) providers.push({ projectId: params.projectId, refresh: params.refresh });
|
||||
const candidates = (await Promise.all(providers.map((provider) => provider.refresh())))
|
||||
.flat()
|
||||
.filter((candidate) => candidate.projectId === params.projectId)
|
||||
.sort(compareAdmissionCandidates);
|
||||
// FNXC:ConcurrencyAdmission 2026-08-06-12:00: FN-8453/#2359 requires
|
||||
// in-memory handoffs to count until they either become live or are dropped.
|
||||
// Persisted task rows lag a fire-and-forget lane start, so omitting these
|
||||
// reservations lets a second coordinator pass over-admit one project.
|
||||
if (candidates.length === 0 || (await params.claimed()) + this.reservationCount(params.projectId) >= params.maxConcurrent) return;
|
||||
const winner = candidates[0];
|
||||
// Older test/runtime semaphore wrappers predate tryAcquire. They still
|
||||
// exercise project admission, while production semaphores atomically take
|
||||
// the host slot here.
|
||||
const hasReservableHostSlot = typeof params.semaphore?.tryAcquire === "function";
|
||||
const acquiredHostSlot = hasReservableHostSlot
|
||||
? params.semaphore!.tryAcquire()
|
||||
: true;
|
||||
if (!acquiredHostSlot) return;
|
||||
// Compatibility-only semaphore shims cannot hold a reservation. Their
|
||||
// lane tests provide claimed() synchronously, while real host semaphores
|
||||
// use this durable marker until take/drop below.
|
||||
if (hasReservableHostSlot) this.reserve(params.projectId, winner.taskId);
|
||||
try {
|
||||
winner.reserve?.();
|
||||
const accepted = await winner.start();
|
||||
if (accepted === false) {
|
||||
this.releaseReservation(winner.taskId);
|
||||
params.semaphore?.release();
|
||||
return;
|
||||
}
|
||||
admitted = winner.taskId;
|
||||
} catch (error) {
|
||||
this.releaseReservation(winner.taskId);
|
||||
params.semaphore?.release();
|
||||
throw error;
|
||||
}
|
||||
})();
|
||||
this.draining.set(params.projectId, drain);
|
||||
try {
|
||||
await drain;
|
||||
return admitted;
|
||||
} finally {
|
||||
if (this.draining.get(params.projectId) === drain) this.draining.delete(params.projectId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Shared coordinator instance used by lane polls in this engine process. */
|
||||
export const projectAdmissionCoordinator = new ProjectAdmissionCoordinator();
|
||||
|
||||
/** A waiter entry that tracks both the priority and the resolve callback. */
|
||||
interface PriorityWaiter {
|
||||
priority: number;
|
||||
@@ -70,12 +226,18 @@ export function registerPreHeldExecutorSlot(taskId: string): void {
|
||||
* Returns true when a slot was registered; the caller MUST release the underlying semaphore in its finally path.
|
||||
*/
|
||||
export function takePreHeldExecutorSlot(taskId: string): boolean {
|
||||
return preHeldExecutorSlots.delete(taskId);
|
||||
const taken = preHeldExecutorSlots.delete(taskId);
|
||||
if (taken) projectAdmissionCoordinator.releaseReservation(taskId);
|
||||
return taken;
|
||||
}
|
||||
|
||||
/** Drop a pre-held slot without transferring ownership (failed reserve / cancelled dispatch). Optionally releases the semaphore. */
|
||||
export function dropPreHeldExecutorSlot(taskId: string, semaphore?: { release(): void }): void {
|
||||
if (!preHeldExecutorSlots.delete(taskId)) return;
|
||||
// FNXC:ConcurrencyAdmission 2026-08-06-12:00: every rejection path funnels
|
||||
// through this helper, so releasing the matching coordinator marker here
|
||||
// prevents early scheduler/triage returns from permanently consuming a slot.
|
||||
projectAdmissionCoordinator.releaseReservation(taskId);
|
||||
semaphore?.release();
|
||||
}
|
||||
|
||||
@@ -97,20 +259,57 @@ export function persistedTopLevelAgentSlots(tasks: Task[]): number {
|
||||
return countRunningAgentTasks(tasks);
|
||||
}
|
||||
|
||||
/**
|
||||
* Store-backed claim counts must enrich workflow traits before using the pure
|
||||
* predicate; raw `persistedTopLevelAgentSlots` remains for pre-enriched tests
|
||||
* and callers that cannot resolve an IR.
|
||||
*/
|
||||
export async function persistedTopLevelAgentSlotsFromStore(store: WorkflowIrResolverStore, tasks: Task[]): Promise<number> {
|
||||
const irCache = new Map();
|
||||
const enriched = await Promise.all(tasks.map(async (task) => {
|
||||
const ir = await resolveWorkflowIrForTask(store, task.id, irCache);
|
||||
return enrichRunningAgentTaskShape(task, ir);
|
||||
}));
|
||||
return countRunningAgentTasks(enriched);
|
||||
}
|
||||
|
||||
/**
|
||||
* FNXC:GlobalConcurrencyControls 2026-07-14-18:30:
|
||||
* Admission control for new top-level agents must use the same running-agent predicate the dashboard shows next to the global/project caps. Prefer the larger of live task-based holders and in-memory semaphore activeCount so neither under-counts the other during the brief window between column/status writes and acquire/release.
|
||||
*/
|
||||
export function computeTopLevelConcurrencyClaimed(params: {
|
||||
tasks: readonly Task[];
|
||||
/**
|
||||
* Retained for compatibility but intentionally excluded from this
|
||||
* project-local claim. The host semaphore is a separate process-wide gate.
|
||||
*/
|
||||
semaphoreActiveCount?: number;
|
||||
/** specifyTask calls that have entered `processing` but not yet written status:"planning". */
|
||||
pendingSpecifyCount?: number;
|
||||
}): number {
|
||||
const persisted = countRunningAgentTasks(params.tasks);
|
||||
const pending = Math.max(0, Math.floor(params.pendingSpecifyCount ?? 0));
|
||||
const active = Math.max(0, Math.floor(params.semaphoreActiveCount ?? 0));
|
||||
return Math.max(active, persisted + pending);
|
||||
return persisted + pending;
|
||||
}
|
||||
|
||||
/**
|
||||
* Store-backed production counterpart to {@link computeTopLevelConcurrencyClaimed}.
|
||||
*
|
||||
* FNXC:ConcurrencyAdmission 2026-08-03-12:00:
|
||||
* FN-8453 forbids admission from raw task rows whenever workflow IR is available:
|
||||
* custom complete/archived columns can retain stale session metadata, so each row
|
||||
* must be trait-enriched before it is allowed to occupy a top-level capacity slot.
|
||||
*/
|
||||
export async function computeTopLevelConcurrencyClaimedFromStore(params: {
|
||||
store: WorkflowIrResolverStore;
|
||||
tasks: Task[];
|
||||
/** Host semaphore activity must never consume this project's capacity. */
|
||||
semaphoreActiveCount?: number;
|
||||
pendingSpecifyCount?: number;
|
||||
}): Promise<number> {
|
||||
const persisted = await persistedTopLevelAgentSlotsFromStore(params.store, params.tasks);
|
||||
const pending = Math.max(0, Math.floor(params.pendingSpecifyCount ?? 0));
|
||||
return persisted + pending;
|
||||
}
|
||||
|
||||
export interface IdleSemaphoreLeakRecoveryResult {
|
||||
|
||||
@@ -65,7 +65,7 @@ import type { RoutineRunner } from "./routine-runner.js";
|
||||
import { sweepStaleAutostashes, VerificationError } from "./merger.js";
|
||||
import { runAiMerge, landWorkspaceTask, WorkspacePartialLandError, WorkspaceRepoLandBusyError } from "./merger-ai.js";
|
||||
import { promoteBranchGroup, type BranchGroupPromotionResult, type CreateGroupPrFn, type SyncGroupPrFn } from "./group-merge-coordinator.js";
|
||||
import { PRIORITY_MERGE } from "./concurrency.js";
|
||||
import { computeTopLevelConcurrencyClaimedFromStore, projectAdmissionCoordinator } from "./concurrency.js";
|
||||
import { canStartNextMergeBody } from "./merge-reclaim-policy.js";
|
||||
import {
|
||||
registerProjectVerificationLimit,
|
||||
@@ -457,6 +457,9 @@ export class ProjectEngine {
|
||||
// ── Auto-merge state ──
|
||||
private mergeQueue: string[] = [];
|
||||
private mergeActive = new Set<string>();
|
||||
/** Merge ids selected by the shared coordinator but not yet handed to rawMerge. */
|
||||
private readonly coordinatorAdmittedMergeTaskIds = new Set<string>();
|
||||
private unregisterMergeAdmissionProvider?: () => void;
|
||||
private pausedReviewTaskIds = new Set<string>();
|
||||
private mergeRunning = false;
|
||||
private activeMergeSession: { dispose: () => void } | null = null;
|
||||
@@ -627,6 +630,42 @@ export class ProjectEngine {
|
||||
// eligibility gate (requestInterpreterMerge), NOT the human "merge now"
|
||||
// bypass, so a graph merge node can't override an autoMerge-off project.
|
||||
this.runtime.setMergeRequester?.((taskId, options) => this.requestInterpreterMerge(taskId, options));
|
||||
|
||||
const projectId = this.runtime.getTaskStore().getRootDir?.() ?? this.config.workingDirectory;
|
||||
/*
|
||||
FNXC:ConcurrencyAdmission 2026-08-06-16:20:
|
||||
FN-8453/#2359 requires the actual durable merge queue to refresh on every
|
||||
project admission pass. A one-shot candidate only exists after this pump
|
||||
dequeues it, which lets a newer planning/execute candidate overtake an older
|
||||
queued merge. Keep at most one selected merge reservation because this pump
|
||||
remains intentionally single-flight.
|
||||
*/
|
||||
this.unregisterMergeAdmissionProvider = projectAdmissionCoordinator.registerProvider(`merge:${projectId}`, {
|
||||
projectId,
|
||||
refresh: async () => {
|
||||
if (this.shuttingDown || !this.started || this.coordinatorAdmittedMergeTaskIds.size > 0) return [];
|
||||
const store = this.runtime.getTaskStore();
|
||||
const queuedTaskIds = [...this.mergeQueue];
|
||||
const tasks = await Promise.all(queuedTaskIds.map(async (taskId) => await store.getTask(taskId).catch(() => null)));
|
||||
return tasks.flatMap((task) => {
|
||||
if (!task || task.paused || task.userPaused || task.column !== "in-review") return [];
|
||||
return [{
|
||||
taskId: task.id,
|
||||
projectId,
|
||||
createdAt: task.createdAt,
|
||||
start: async () => {
|
||||
// Do not run merge work in the coordinator; hand the exact queued
|
||||
// id back to the single-flight pump, which will consume this marker.
|
||||
if (!this.mergeQueue.includes(task.id) || this.shuttingDown) return false;
|
||||
this.coordinatorAdmittedMergeTaskIds.add(task.id);
|
||||
void this.drainMergeQueue().catch((error: unknown) => {
|
||||
runtimeLog.error(`Coordinator-admitted merge drain failed: ${error instanceof Error ? error.message : String(error)}`);
|
||||
});
|
||||
},
|
||||
}];
|
||||
});
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
getActiveMergeTaskId(): string | null {
|
||||
@@ -1235,6 +1274,8 @@ export class ProjectEngine {
|
||||
|
||||
// FNXC:VerificationConcurrency 2026-07-15-09:05: Drop this project's cap so it no longer pins process min.
|
||||
unregisterProjectVerificationLimit(this.config.projectId);
|
||||
this.unregisterMergeAdmissionProvider?.();
|
||||
this.unregisterMergeAdmissionProvider = undefined;
|
||||
// Stop merge retry timer
|
||||
if (this.mergeRetryTimer) {
|
||||
clearTimeout(this.mergeRetryTimer);
|
||||
@@ -2548,6 +2589,11 @@ export class ProjectEngine {
|
||||
*/
|
||||
private async pickNextMergeTaskId(store: TaskStore): Promise<string | undefined> {
|
||||
if (this.mergeQueue.length === 0) return undefined;
|
||||
// A coordinator-selected merge must be dispatched before this queue's local
|
||||
// priority policy; otherwise the provider would reserve the old task but
|
||||
// this pump could start a newer one first.
|
||||
const admittedIndex = this.mergeQueue.findIndex((taskId) => this.coordinatorAdmittedMergeTaskIds.has(taskId));
|
||||
if (admittedIndex !== -1) return this.mergeQueue.splice(admittedIndex, 1)[0];
|
||||
// Fast path: with a single queued task there's nothing to reorder. Avoid an
|
||||
// extra getTask round-trip (and keep callers that mock getTask once happy).
|
||||
if (this.mergeQueue.length === 1) {
|
||||
@@ -3639,24 +3685,79 @@ export class ProjectEngine {
|
||||
// FNXC:MergeQueue 2026-07-15-10:05: Wait for any orphan body from a prior abort race before claiming the next generation.
|
||||
await this.awaitPriorMergeBodySettle();
|
||||
|
||||
const semaphore = (this.runtime as any).projectSemaphore ?? (this.runtime as any).globalSemaphore;
|
||||
const coordinatorReservedMerge = this.coordinatorAdmittedMergeTaskIds.delete(taskId);
|
||||
/*
|
||||
FNXC:ConcurrencyAdmission 2026-08-07-10:30:
|
||||
FN-8453/#2359 applies the same top-level slot reservation to direct and
|
||||
pull-request merge bodies. The current queue item is passed as a
|
||||
one-shot candidate after dequeue because durable merge providers only
|
||||
see remaining queue entries; without it a sole merge endlessly defers.
|
||||
*/
|
||||
const runWithMergeAdmission = async <T>(start: () => Promise<T>): Promise<T | undefined> => {
|
||||
if (!semaphore) return await start();
|
||||
if (coordinatorReservedMerge) {
|
||||
try {
|
||||
return await start();
|
||||
} finally {
|
||||
projectAdmissionCoordinator.releaseReservation(taskId);
|
||||
semaphore.release();
|
||||
}
|
||||
}
|
||||
let selected = false;
|
||||
let value: T | undefined;
|
||||
await projectAdmissionCoordinator.admitOldest({
|
||||
projectId: cwd,
|
||||
maxConcurrent: (await store.getSettings()).maxConcurrent ?? 2,
|
||||
semaphore,
|
||||
claimed: async () => computeTopLevelConcurrencyClaimedFromStore({
|
||||
store,
|
||||
tasks: await store.listTasks({ slim: true, includeArchived: false }),
|
||||
}),
|
||||
refresh: async () => [{
|
||||
taskId,
|
||||
projectId: cwd,
|
||||
createdAt: mergeCandidate?.createdAt,
|
||||
start: async () => {
|
||||
selected = true;
|
||||
value = await start();
|
||||
return true;
|
||||
},
|
||||
}],
|
||||
});
|
||||
if (!selected) return undefined;
|
||||
projectAdmissionCoordinator.releaseReservation(taskId);
|
||||
semaphore.release();
|
||||
return value;
|
||||
};
|
||||
|
||||
if (mergeStrategy === "pull-request" && this.options.processPullRequestMerge && !routeWorkspaceDirect) {
|
||||
/*
|
||||
FNXC:MergeQueue 2026-07-15-10:05:
|
||||
PR merge dispatch shares the single-flight pump. Race the PR body with abort so pause/reclaim unblocks drainMergeQueue even when processPullRequestMerge ignores cooperative abort.
|
||||
*/
|
||||
const abortSignal = this.claimActiveMerge(taskId);
|
||||
runtimeLog.log(`${hasManualResolver ? "Manual" : "Auto"}-merge processing PR flow for ${taskId}...`);
|
||||
const result = await this.runAbortableMergeBody(
|
||||
() =>
|
||||
this.options.processPullRequestMerge!(
|
||||
store,
|
||||
cwd,
|
||||
taskId,
|
||||
(this.runtime as any).worktreePool,
|
||||
),
|
||||
abortSignal,
|
||||
taskId,
|
||||
);
|
||||
const result = await runWithMergeAdmission(async () => {
|
||||
const abortSignal = this.claimActiveMerge(taskId);
|
||||
runtimeLog.log(`${hasManualResolver ? "Manual" : "Auto"}-merge processing PR flow for ${taskId}...`);
|
||||
return await this.runAbortableMergeBody(
|
||||
() =>
|
||||
this.options.processPullRequestMerge!(
|
||||
store,
|
||||
cwd,
|
||||
taskId,
|
||||
(this.runtime as any).worktreePool,
|
||||
),
|
||||
abortSignal,
|
||||
taskId,
|
||||
);
|
||||
});
|
||||
if (result === undefined) {
|
||||
// Another older lane won the shared capacity pass. Re-queue rather
|
||||
// than treating this deferral as a pull-request merge failure.
|
||||
this.mergeActive.delete(taskId);
|
||||
this.internalEnqueueMerge(taskId);
|
||||
continue;
|
||||
}
|
||||
if (result === "merged") {
|
||||
runtimeLog.log(`${hasManualResolver ? "Manual" : "Auto"}-merge PR merged: ${taskId}`);
|
||||
const mergedTask = await store.getTask(taskId).catch(() => null);
|
||||
@@ -3692,8 +3793,6 @@ export class ProjectEngine {
|
||||
// Direct merge via AI agent, gated by semaphore
|
||||
runtimeLog.log(`${hasManualResolver ? "Manual" : "Auto"}-merge merging ${taskId}...`);
|
||||
|
||||
const semaphore = (this.runtime as any).projectSemaphore ?? (this.runtime as any).globalSemaphore;
|
||||
|
||||
const pool = (this.runtime as any).worktreePool;
|
||||
|
||||
const agentStore = (this.runtime as any).agentStore;
|
||||
@@ -3810,11 +3909,14 @@ export class ProjectEngine {
|
||||
}, abortSignal, taskId);
|
||||
};
|
||||
|
||||
let result: MergeResult;
|
||||
if (semaphore) {
|
||||
result = await semaphore.run(rawMerge, PRIORITY_MERGE);
|
||||
} else {
|
||||
result = await rawMerge();
|
||||
const result = await runWithMergeAdmission(rawMerge);
|
||||
|
||||
if (!result) {
|
||||
// An older lane won this admission pass. Keep this merge queued;
|
||||
// treating the deferral as a merge failure would consume retries.
|
||||
this.mergeActive.delete(taskId);
|
||||
this.internalEnqueueMerge(taskId);
|
||||
continue;
|
||||
}
|
||||
|
||||
this.activeMergeSession = null;
|
||||
@@ -4591,6 +4693,13 @@ export class ProjectEngine {
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
// A selected queue entry can fail eligibility before reaching rawMerge.
|
||||
// Return its coordinator reservation rather than pinning a top-level slot.
|
||||
if (this.coordinatorAdmittedMergeTaskIds.delete(taskId)) {
|
||||
projectAdmissionCoordinator.releaseReservation(taskId);
|
||||
const semaphore = (this.runtime as any).projectSemaphore ?? (this.runtime as any).globalSemaphore;
|
||||
semaphore?.release();
|
||||
}
|
||||
this.clearActiveMergeClaim(taskId);
|
||||
this.mergeAbortController = null;
|
||||
this.mergeActive.delete(taskId);
|
||||
|
||||
@@ -21,8 +21,10 @@ import { existsSync } from "node:fs";
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import {
|
||||
computeTopLevelConcurrencyClaimed,
|
||||
computeTopLevelConcurrencyClaimedFromStore,
|
||||
dropPreHeldExecutorSlot,
|
||||
hasPreHeldExecutorSlot,
|
||||
projectAdmissionCoordinator,
|
||||
recoverIdleSemaphoreLeakCandidate,
|
||||
registerPreHeldExecutorSlot,
|
||||
type AgentSemaphore,
|
||||
@@ -630,6 +632,11 @@ export class Scheduler {
|
||||
private lastHeartbeatWriteMs = 0;
|
||||
private idleSemaphoreLeakCandidateSince: number | null = null;
|
||||
private readonly lastHighOverlapFanoutWarningKey = new Map<string, string>();
|
||||
/** Coordinator reservations survive the lane poll that selected them. */
|
||||
private readonly coordinatorAdmittedTaskIds = new Set<string>();
|
||||
/** Durable execute candidates refreshed on each scheduler pass for cross-lane ranking. */
|
||||
private readonly coordinatorReadyTasks = new Map<string, Task>();
|
||||
private unregisterAdmissionProvider?: () => void;
|
||||
|
||||
/**
|
||||
* Async listener guard convention:
|
||||
@@ -653,6 +660,29 @@ export class Scheduler {
|
||||
projectId: this.store.getRootDir(),
|
||||
logger: schedulerLog,
|
||||
});
|
||||
/*
|
||||
FNXC:ConcurrencyAdmission 2026-08-06-09:00:
|
||||
FN-8453's union must outlive a single scheduler poll. A temporary provider
|
||||
was gone before planning/merge asked for capacity, allowing newer work to
|
||||
overtake ready execute work. The refreshed map is the durable lane view.
|
||||
*/
|
||||
const projectId = this.store.getRootDir();
|
||||
this.unregisterAdmissionProvider = projectAdmissionCoordinator.registerProvider(`execute:${projectId}`, {
|
||||
projectId,
|
||||
refresh: async () => [...this.coordinatorReadyTasks.values()]
|
||||
.filter((task) => !this.coordinatorAdmittedTaskIds.has(task.id))
|
||||
.map((task) => ({
|
||||
taskId: task.id,
|
||||
projectId,
|
||||
createdAt: task.createdAt,
|
||||
reserve: () => { if (this.options.semaphore) registerPreHeldExecutorSlot(task.id); },
|
||||
start: async () => {
|
||||
this.coordinatorReadyTasks.delete(task.id);
|
||||
this.coordinatorAdmittedTaskIds.add(task.id);
|
||||
void this.schedule();
|
||||
},
|
||||
})),
|
||||
});
|
||||
/**
|
||||
* Event-driven scheduling: when a task is created, trigger a scheduling
|
||||
* pass immediately instead of waiting for the next poll interval.
|
||||
@@ -1075,6 +1105,9 @@ export class Scheduler {
|
||||
if (this.options.missionAutopilot) {
|
||||
this.options.missionAutopilot.stop();
|
||||
}
|
||||
this.unregisterAdmissionProvider?.();
|
||||
this.unregisterAdmissionProvider = undefined;
|
||||
this.coordinatorReadyTasks.clear();
|
||||
this.failedTaskIds.clear();
|
||||
this.wasNodeBlocked.clear();
|
||||
this.wasNodeDispatchValidationBlocked.clear();
|
||||
@@ -1451,6 +1484,16 @@ export class Scheduler {
|
||||
// When a semaphore is provided, factor in its available slots so we
|
||||
// don't schedule more tasks than the global limit allows.
|
||||
const inProgressTaskIds = inProgress.map((task) => task.id);
|
||||
/*
|
||||
FNXC:ConcurrencyAdmission 2026-08-03-13:00:
|
||||
FN-8453 capacity decisions must enrich task rows from their workflow IR.
|
||||
A custom complete column can retain stale session metadata, so raw task
|
||||
counting here would falsely exhaust executable capacity.
|
||||
*/
|
||||
const topLevelClaimedSlots = await computeTopLevelConcurrencyClaimedFromStore({
|
||||
store: this.store,
|
||||
tasks,
|
||||
});
|
||||
const computeDispatchCapacityDiagnostic = (startedThisTick: number): ConcurrencyGateDiagnostic => {
|
||||
const started = Math.max(0, Math.floor(startedThisTick));
|
||||
// U6 (KTD-10): report the default workflow's in-progress capacity as a
|
||||
@@ -1472,10 +1515,7 @@ export class Scheduler {
|
||||
maxWorktrees,
|
||||
semaphore: this.options.semaphore,
|
||||
inProgressTaskIds,
|
||||
topLevelClaimedSlots: computeTopLevelConcurrencyClaimed({
|
||||
tasks,
|
||||
semaphoreActiveCount: this.options.semaphore?.activeCount,
|
||||
}),
|
||||
topLevelClaimedSlots,
|
||||
startedThisTick: started,
|
||||
perColumnGates,
|
||||
});
|
||||
@@ -1548,6 +1588,28 @@ export class Scheduler {
|
||||
maxAutoMergeRetries,
|
||||
});
|
||||
todo = sortTasksByPriorityFanoutThenAgeAndId(todo, unblockWeights);
|
||||
/*
|
||||
FNXC:ConcurrencyAdmission 2026-08-03-14:00:
|
||||
FN-8453 makes the coordinator, rather than this lane's priority/fanout
|
||||
ordering, the authority for a free top-level slot. The scheduler exposes
|
||||
its ready tasks to the shared project registry and dispatches only the
|
||||
atomically admitted winner; other lanes refresh in the same admission pass.
|
||||
*/
|
||||
const projectId = this.store.getRootDir();
|
||||
this.coordinatorReadyTasks.clear();
|
||||
for (const task of todo) this.coordinatorReadyTasks.set(task.id, task);
|
||||
await projectAdmissionCoordinator.admitOldest({
|
||||
projectId,
|
||||
maxConcurrent,
|
||||
claimed: async () => computeTopLevelConcurrencyClaimedFromStore({ store: this.store, tasks: await this.store.listTasks({ slim: true, includeArchived: false }) }),
|
||||
semaphore: this.options.semaphore,
|
||||
});
|
||||
todo = todo.filter((task) => this.coordinatorAdmittedTaskIds.has(task.id));
|
||||
// FNXC:ConcurrencyAdmission 2026-08-04-10:00: coordinator IDs select one
|
||||
// handoff only. The pre-held semaphore slot remains the durable reservation;
|
||||
// retaining the ID after a retry would bypass oldest-first re-admission.
|
||||
for (const task of todo) this.coordinatorAdmittedTaskIds.delete(task.id);
|
||||
if (todo.length === 0) return;
|
||||
const topWeightedTask = todo.find((candidate) => (unblockWeights.get(candidate.id) ?? 0) >= 1);
|
||||
if (topWeightedTask) {
|
||||
schedulerLog.log(
|
||||
@@ -2904,10 +2966,9 @@ export class Scheduler {
|
||||
}
|
||||
}
|
||||
|
||||
const topLevelClaimedSlots = computeTopLevelConcurrencyClaimed({
|
||||
const topLevelClaimedSlots = await computeTopLevelConcurrencyClaimedFromStore({
|
||||
store: this.store,
|
||||
tasks,
|
||||
// Prior tryAcquire reservations in this sweep already bump activeCount.
|
||||
semaphoreActiveCount: this.options.semaphore?.activeCount,
|
||||
});
|
||||
const concurrencyDiagnostic = computeConcurrencyGateDiagnostic({
|
||||
agentSlots: reservedConcurrentSlots,
|
||||
@@ -2937,7 +2998,8 @@ export class Scheduler {
|
||||
}
|
||||
|
||||
const sem = this.options.semaphore;
|
||||
if (sem && !sem.tryAcquire()) {
|
||||
const coordinatorReserved = hasPreHeldExecutorSlot(task.id);
|
||||
if (sem && !coordinatorReserved && !sem.tryAcquire()) {
|
||||
if (reservedScope) {
|
||||
activeScopes.delete(task.id);
|
||||
activeScopeColumns.delete(task.id);
|
||||
@@ -2951,7 +3013,7 @@ export class Scheduler {
|
||||
await this.logDispatchQueuedReason(task.id, reason, formatConcurrencyLimitMemoKey(concurrencyDiagnostic));
|
||||
return null;
|
||||
}
|
||||
if (sem) {
|
||||
if (sem && !coordinatorReserved) {
|
||||
registerPreHeldExecutorSlot(task.id);
|
||||
}
|
||||
|
||||
|
||||
@@ -26,8 +26,6 @@ import {
|
||||
resolveTaskPlanningPrompt,
|
||||
resolveTaskSeamPrompt,
|
||||
resolvePersistAgentThinkingLog,
|
||||
compareTaskPriority,
|
||||
sortTasksByPriorityThenAgeAndId,
|
||||
compareTaskIdNumeric,
|
||||
resolveAgentMemoryInclusionMode,
|
||||
resolvePlanApprovalRequired,
|
||||
@@ -128,7 +126,11 @@ import { detectDanglingTaskDocReferences, formatDanglingDiagnostic } from "./spe
|
||||
import { buildSessionSkillContext } from "./session-skill-context.js";
|
||||
import {
|
||||
PRIORITY_SPECIFY,
|
||||
computeTopLevelConcurrencyClaimed,
|
||||
computeTopLevelConcurrencyClaimedFromStore,
|
||||
dropPreHeldExecutorSlot,
|
||||
projectAdmissionCoordinator,
|
||||
registerPreHeldExecutorSlot,
|
||||
takePreHeldExecutorSlot,
|
||||
recoverIdleSemaphoreLeakCandidate,
|
||||
type AgentSemaphore,
|
||||
} from "./concurrency.js";
|
||||
@@ -221,6 +223,10 @@ export class TriageProcessor {
|
||||
private processing = new Set<string>();
|
||||
/** Synchronous ownership fence shared with advanced-triage self-healing. */
|
||||
private advancedRecoveryReservations = new Set<string>();
|
||||
/** Prevent a selected planner from reappearing before specifyTask claims it. */
|
||||
private readonly coordinatorAdmittedTaskIds = new Set<string>();
|
||||
/** Durable planning provider keeps this lane visible to execute/merge polls. */
|
||||
private unregisterAdmissionProvider: (() => void) | null = null;
|
||||
/** Timestamps when tasks entered the `processing` set, for staleness detection. */
|
||||
private processingSince = new Map<string, number>();
|
||||
private wasGlobalPaused = false;
|
||||
@@ -367,6 +373,34 @@ export class TriageProcessor {
|
||||
private rootDir: string,
|
||||
private options: TriageProcessorOptions = {},
|
||||
) {
|
||||
this.unregisterAdmissionProvider = projectAdmissionCoordinator.registerProvider(`specify:${this.rootDir}`, {
|
||||
projectId: this.rootDir,
|
||||
refresh: async () => {
|
||||
const settings = await this.store.getSettings();
|
||||
// poll() supplies its own fresh candidates to the same admission pass;
|
||||
// do not duplicate them through this durable provider or a provider
|
||||
// handoff can bypass the poll's bounded refinement scheduling.
|
||||
if (!this.running || this.polling || settings.globalPause || settings.enginePaused) return [];
|
||||
const now = Date.now();
|
||||
// FNXC:ConcurrencyAdmission 2026-08-07-10:30:
|
||||
// FN-8453/#2359 requires coordinator refresh to use the identical
|
||||
// discovery predicate as poll(). A seed is ready before specifyTask
|
||||
// stamps status:"planning"; exposing only that durable status lets newer
|
||||
// execute/merge work overtake an older planner.
|
||||
const tasks = await this.discoverReadyPlanningTasks(
|
||||
await this.store.listTasks({ slim: true, includeArchived: false }),
|
||||
now,
|
||||
);
|
||||
return tasks.filter((task) => !this.coordinatorAdmittedTaskIds.has(task.id)).map((task) => ({
|
||||
taskId: task.id, projectId: this.rootDir, createdAt: task.createdAt,
|
||||
reserve: () => { if (this.options.semaphore) registerPreHeldExecutorSlot(task.id); },
|
||||
start: async () => {
|
||||
this.coordinatorAdmittedTaskIds.add(task.id);
|
||||
void this.specifyTask(task);
|
||||
},
|
||||
}));
|
||||
},
|
||||
});
|
||||
// When globalPause transitions from false → true, terminate all active triage sessions.
|
||||
store.on("settings:updated", ({ settings, previous }) => {
|
||||
if (settings.globalPause && !previous.globalPause) {
|
||||
@@ -508,6 +542,8 @@ export class TriageProcessor {
|
||||
|
||||
stop(): void {
|
||||
this.running = false;
|
||||
this.unregisterAdmissionProvider?.();
|
||||
this.unregisterAdmissionProvider = null;
|
||||
if (this.pollInterval) {
|
||||
clearInterval(this.pollInterval);
|
||||
this.pollInterval = null;
|
||||
@@ -1047,6 +1083,53 @@ export class TriageProcessor {
|
||||
* well before the dispatched tasks finish — so subsequent polls can discover
|
||||
* newly arrived triage tasks promptly.
|
||||
*/
|
||||
/**
|
||||
* Discover planner-ready work for both direct triage polling and coordinator
|
||||
* refresh. Keeping the seed-prompt checks here makes the cross-lane admission
|
||||
* union include cards before their planner writes status:"planning".
|
||||
*/
|
||||
private async discoverReadyPlanningTasks(allTasks: Task[], now: number): Promise<Task[]> {
|
||||
const eligibleTriageTasks = allTasks.filter(
|
||||
(t) => t.column === "triage" && isTaskStillInPlanningStage(t)
|
||||
&& !this.advancedRecoveryReservations.has(t.id)
|
||||
&& !this.processing.has(t.id) && !this.hasLivePlanningWork(t.id) && !t.paused
|
||||
&& t.status !== "awaiting-approval" && t.status !== "failed" && t.status !== "stuck-killed"
|
||||
&& !(t.nextRecoveryAt && new Date(t.nextRecoveryAt).getTime() > now),
|
||||
);
|
||||
const eligibleTodoTasksRaw = allTasks.filter(
|
||||
(t) => t.column === "todo" && !this.processing.has(t.id) && !this.hasLivePlanningWork(t.id) && !t.paused
|
||||
&& t.status !== "awaiting-approval" && t.status !== "failed" && t.status !== "stuck-killed"
|
||||
&& t.status !== "planning"
|
||||
&& !(t.nextRecoveryAt && new Date(t.nextRecoveryAt).getTime() > now),
|
||||
);
|
||||
const eligibleTodoTasks: Task[] = [];
|
||||
for (const todoTask of eligibleTodoTasksRaw) {
|
||||
if (todoTask.status === "needs-replan") {
|
||||
eligibleTodoTasks.push(todoTask);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
const promptPath = join(this.rootDir, ".fusion", "tasks", todoTask.id, "PROMPT.md");
|
||||
const content = await readFile(promptPath, "utf-8");
|
||||
if (isUnplannedSeedPrompt(content, todoTask.id, todoTask.title, todoTask.description)) {
|
||||
eligibleTodoTasks.push(todoTask);
|
||||
}
|
||||
} catch {
|
||||
// Missing/unreadable prompt — scheduler filesystem validation owns it.
|
||||
}
|
||||
}
|
||||
return [...eligibleTriageTasks, ...eligibleTodoTasks].sort((a, b) => {
|
||||
const aTime = Date.parse(a.createdAt);
|
||||
const bTime = Date.parse(b.createdAt);
|
||||
const aValid = Number.isFinite(aTime);
|
||||
const bValid = Number.isFinite(bTime);
|
||||
if (aValid !== bValid) return aValid ? -1 : 1;
|
||||
if (aValid && aTime !== bTime) return aTime - bTime;
|
||||
const numeric = compareTaskIdNumeric(a.id, b.id);
|
||||
return numeric !== 0 ? numeric : a.id.localeCompare(b.id);
|
||||
});
|
||||
}
|
||||
|
||||
private async poll(): Promise<void> {
|
||||
if (!this.running) return;
|
||||
if (this.polling) return;
|
||||
@@ -1099,84 +1182,15 @@ export class TriageProcessor {
|
||||
this.idleSemaphoreLeakCandidateSince = result.candidateSinceMs;
|
||||
}
|
||||
|
||||
const eligibleTriageTasks = allTasks.filter(
|
||||
(t) => t.column === "triage" && isTaskStillInPlanningStage(t)
|
||||
&& !this.advancedRecoveryReservations.has(t.id)
|
||||
&& !this.processing.has(t.id) && !this.hasLivePlanningWork(t.id) && !t.paused
|
||||
// Skip tasks awaiting manual plan approval — they should not be auto-discovered
|
||||
&& t.status !== "awaiting-approval"
|
||||
// Skip failed specifications until the user explicitly retries them.
|
||||
&& t.status !== "failed"
|
||||
&& t.status !== "stuck-killed"
|
||||
// Skip tasks with a recovery backoff that hasn't elapsed yet
|
||||
&& !(t.nextRecoveryAt && new Date(t.nextRecoveryAt).getTime() > now),
|
||||
);
|
||||
const triageTasks = await this.discoverReadyPlanningTasks(allTasks, now);
|
||||
|
||||
/*
|
||||
Workflows with a manual intake (e.g. Coding (Ideas)) merge the planner and capacity-hold stages into a single "todo" column. The triage service must also discover "todo" tasks whose PROMPT.md is still an unplanned seed — they have been promoted out of the manual intake but not yet planned in place. Planned todo tasks carry a real spec and are left for the scheduler. The seed-prompt file check is the ground-truth unplanned signal; it is false for every normal-workflow todo task because triage writes a real spec before it ever moves a card into todo.
|
||||
|
||||
FNXC:CodingIdeasWorkflow 2026-07-12-23:05:
|
||||
Two discovery gaps let plan-in-place workflow cards strand or misexecute in "todo":
|
||||
1. `needs-replan` todo tasks carry a REAL PROMPT.md (the failed plan under revision), so the seed check alone never rediscovers them. Workflows without a "triage" column keep replanning tasks in "todo" (the executor's workflow-aware replan rebound targets the planner column), so triage must pick up `needs-replan` todo cards regardless of prompt content — processTask already routes them through the isReplan path.
|
||||
2. Refinement seeds (`# {title}\n\n{description}`, no id prefix) previously failed the strict bootstrap-stub equality, so a promoted refinement skipped planning entirely; isUnplannedSeedPrompt accepts both seed shapes.
|
||||
FNXC:ConcurrencyAdmission 2026-08-03-12:00:
|
||||
FN-8453 removes the separate maxTriageConcurrent pool. Planning uses the
|
||||
same maxConcurrent live-agent claim as execute/review so a project cannot
|
||||
exceed its operator-facing top-level capacity in a different lane.
|
||||
*/
|
||||
const eligibleTodoTasksRaw = allTasks.filter(
|
||||
(t) => t.column === "todo" && !this.processing.has(t.id) && !this.hasLivePlanningWork(t.id) && !t.paused
|
||||
&& t.status !== "awaiting-approval"
|
||||
&& t.status !== "failed"
|
||||
&& t.status !== "stuck-killed"
|
||||
&& t.status !== "planning"
|
||||
&& !(t.nextRecoveryAt && new Date(t.nextRecoveryAt).getTime() > now),
|
||||
);
|
||||
const eligibleTodoTasks: Task[] = [];
|
||||
for (const todoTask of eligibleTodoTasksRaw) {
|
||||
if (todoTask.status === "needs-replan") {
|
||||
eligibleTodoTasks.push(todoTask);
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
const promptPath = join(this.rootDir, ".fusion", "tasks", todoTask.id, "PROMPT.md");
|
||||
const content = await readFile(promptPath, "utf-8");
|
||||
if (isUnplannedSeedPrompt(content, todoTask.id, todoTask.title, todoTask.description)) {
|
||||
eligibleTodoTasks.push(todoTask);
|
||||
}
|
||||
} catch {
|
||||
// Missing/unreadable prompt — skip; the scheduler's filesystem validation handles it.
|
||||
}
|
||||
}
|
||||
const triageTasks = sortTasksByPriorityThenAgeAndId([...eligibleTriageTasks, ...eligibleTodoTasks]).sort((a, b) => {
|
||||
const priorityCmp = compareTaskPriority(a.priority, b.priority);
|
||||
if (priorityCmp !== 0) {
|
||||
return priorityCmp;
|
||||
}
|
||||
|
||||
// Keep the global priority contract intact, but for same-priority tasks,
|
||||
// prefer refinements so follow-up work does not starve behind bulk triage imports.
|
||||
const aIsRefinement = a.sourceType === "task_refine";
|
||||
const bIsRefinement = b.sourceType === "task_refine";
|
||||
if (aIsRefinement !== bIsRefinement) {
|
||||
return aIsRefinement ? -1 : 1;
|
||||
}
|
||||
|
||||
if (a.createdAt !== b.createdAt) {
|
||||
return a.createdAt.localeCompare(b.createdAt);
|
||||
}
|
||||
|
||||
return compareTaskIdNumeric(a.id, b.id);
|
||||
});
|
||||
|
||||
// Respect both per-project maxTriageConcurrent and the global semaphore.
|
||||
// Only planning tasks count against the triage limit; execution is governed by maxConcurrent.
|
||||
/*
|
||||
FNXC:GlobalConcurrencyControls 2026-07-14-18:30:
|
||||
Live utilization counts in-progress executors and active planners toward the same global cap. Cap new triage starts by remaining room under that shared claim (not only semaphore.availableCount), so planning cannot fill the entire global max while an in-progress executor is already counted as running.
|
||||
*/
|
||||
const maxTriageConcurrent = settings.maxTriageConcurrent ?? settings.maxConcurrent ?? 2;
|
||||
const planning = allTasks.filter(
|
||||
(t) => (t.column === "triage" || t.column === "todo") && t.status === "planning" && !t.paused,
|
||||
).length;
|
||||
const activeAgents = planning;
|
||||
|
||||
const perProjectAvailable = Math.max(0, maxTriageConcurrent - activeAgents);
|
||||
const maxConcurrent = settings.maxConcurrent ?? 2;
|
||||
const semaphoreAvailable = this.options.semaphore
|
||||
? Math.max(0, this.options.semaphore.availableCount)
|
||||
: Infinity;
|
||||
@@ -1186,15 +1200,15 @@ export class TriageProcessor {
|
||||
const row = allTasks.find((t) => t.id === id);
|
||||
if (!row || row.status !== "planning") pendingSpecifyCount += 1;
|
||||
}
|
||||
const claimed = computeTopLevelConcurrencyClaimed({
|
||||
const claimed = await computeTopLevelConcurrencyClaimedFromStore({
|
||||
store: this.store,
|
||||
tasks: allTasks,
|
||||
semaphoreActiveCount: this.options.semaphore?.activeCount,
|
||||
pendingSpecifyCount,
|
||||
});
|
||||
const globalRoom = this.options.semaphore
|
||||
? Math.max(0, this.options.semaphore.limit - claimed)
|
||||
: Infinity;
|
||||
const maxToStart = Math.min(perProjectAvailable, semaphoreAvailable, globalRoom);
|
||||
// `claimed` is project-local. The scoped/global host semaphore remains a
|
||||
// distinct process-wide availability gate, so project A cannot spend B's cap.
|
||||
const projectRoom = Math.max(0, maxConcurrent - claimed);
|
||||
const maxToStart = Math.min(projectRoom, semaphoreAvailable);
|
||||
|
||||
if (maxToStart <= 0 && triageTasks.length > 0) {
|
||||
const semaphoreSnapshot = this.options.semaphore?.snapshot();
|
||||
@@ -1203,20 +1217,53 @@ export class TriageProcessor {
|
||||
: ", semaphore unavailable";
|
||||
const processingIds = [...this.processing].slice(0, 5);
|
||||
const eligibleIds = triageTasks.slice(0, 5).map((t) => t.id);
|
||||
const blockedBy = perProjectAvailable <= 0
|
||||
? "triage concurrency"
|
||||
: globalRoom <= 0
|
||||
? "global running-agent cap"
|
||||
: "global semaphore";
|
||||
const blockedBy = projectRoom <= 0 ? "running-agent cap" : "global semaphore";
|
||||
planLog.log(
|
||||
`Plan throttled by ${blockedBy}: eligible=${triageTasks.length} [${eligibleIds.join(", ")}], ` +
|
||||
`planning=${activeAgents}/${maxTriageConcurrent}, claimed=${claimed}, processing=${this.processing.size}` +
|
||||
`maxConcurrent=${maxConcurrent}, claimed=${claimed}, processing=${this.processing.size}` +
|
||||
`${processingIds.length > 0 ? ` [${processingIds.join(", ")}]` : ""}${semaphoreDetail}`,
|
||||
);
|
||||
}
|
||||
|
||||
// Keep handoff reservations visible even when a test/runtime wrapper delays
|
||||
// the planner's synchronous processing claim until after this poll returns.
|
||||
const admittedThisPoll = new Set<string>();
|
||||
for (let i = 0; i < Math.min(triageTasks.length, maxToStart); i++) {
|
||||
void this.specifyTask(triageTasks[i]);
|
||||
await projectAdmissionCoordinator.admitOldest({
|
||||
// rootDir is the stable per-project identity held by this processor.
|
||||
projectId: this.rootDir,
|
||||
maxConcurrent,
|
||||
claimed: async () => {
|
||||
const fresh = await this.store.listTasks({ slim: true, includeArchived: false });
|
||||
let pending = 0;
|
||||
for (const id of this.processing) {
|
||||
const row = fresh.find((task) => task.id === id);
|
||||
if (!row || row.status !== "planning") pending++;
|
||||
}
|
||||
return computeTopLevelConcurrencyClaimedFromStore({
|
||||
store: this.store,
|
||||
tasks: fresh,
|
||||
pendingSpecifyCount: pending,
|
||||
});
|
||||
},
|
||||
semaphore: this.options.semaphore,
|
||||
refresh: async () => triageTasks
|
||||
.filter((task) => !admittedThisPoll.has(task.id) && !this.coordinatorAdmittedTaskIds.has(task.id) && !this.processing.has(task.id) && !this.hasLivePlanningWork(task.id))
|
||||
.map((task) => ({
|
||||
taskId: task.id,
|
||||
projectId: this.rootDir,
|
||||
createdAt: task.createdAt,
|
||||
// FNXC:ConcurrencyAdmission 2026-08-05-10:00: the planner must
|
||||
// own the coordinator's real host reservation before it starts;
|
||||
// deferring to semaphore.run would reintroduce priority overtaking.
|
||||
reserve: () => { if (this.options.semaphore) registerPreHeldExecutorSlot(task.id); },
|
||||
start: async () => {
|
||||
admittedThisPoll.add(task.id);
|
||||
this.coordinatorAdmittedTaskIds.add(task.id);
|
||||
void this.specifyTask(task);
|
||||
},
|
||||
})),
|
||||
});
|
||||
}
|
||||
} catch (err) {
|
||||
planLog.error("Poll error:", err);
|
||||
@@ -1261,7 +1308,14 @@ export class TriageProcessor {
|
||||
this.advancedRecoveryReservations.has(task.id)
|
||||
|| this.processing.has(task.id)
|
||||
|| this.hasLivePlanningWork(task.id)
|
||||
) return;
|
||||
) {
|
||||
// FNXC:ConcurrencyAdmission 2026-08-06-09:00:
|
||||
// A coordinator winner owns a real pre-held host slot. A duplicate/stale
|
||||
// planner handoff must return it instead of pinning max concurrency.
|
||||
dropPreHeldExecutorSlot(task.id, this.options.semaphore);
|
||||
this.coordinatorAdmittedTaskIds.delete(task.id);
|
||||
return;
|
||||
}
|
||||
this.processing.add(task.id);
|
||||
this.processingSince.set(task.id, Date.now());
|
||||
|
||||
@@ -1879,7 +1933,15 @@ export class TriageProcessor {
|
||||
},
|
||||
});
|
||||
|
||||
if (this.options.semaphore) {
|
||||
if (this.options.semaphore && takePreHeldExecutorSlot(task.id)) {
|
||||
// Coordinator already owns this top-level slot; run directly so it
|
||||
// cannot join the priority queue after age-based admission.
|
||||
try {
|
||||
await retryableWork();
|
||||
} finally {
|
||||
this.options.semaphore.release();
|
||||
}
|
||||
} else if (this.options.semaphore) {
|
||||
await this.options.semaphore.run(retryableWork, PRIORITY_SPECIFY);
|
||||
} else {
|
||||
await retryableWork();
|
||||
@@ -2037,8 +2099,14 @@ export class TriageProcessor {
|
||||
this.options.onSpecifyError?.(task, err instanceof Error ? err : new Error(errorMessage));
|
||||
}
|
||||
} finally {
|
||||
// FNXC:ConcurrencyAdmission 2026-08-06-10:00: a coordinator reservation
|
||||
// can exist before planner setup reaches takePreHeldExecutorSlot(). Every
|
||||
// early setup failure must return that untransferred host slot; after a
|
||||
// successful transfer this is intentionally a no-op.
|
||||
dropPreHeldExecutorSlot(task.id, this.options.semaphore);
|
||||
this.processing.delete(task.id);
|
||||
this.processingSince.delete(task.id);
|
||||
this.coordinatorAdmittedTaskIds.delete(task.id);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -54,6 +54,11 @@
|
||||
"module": "detect-content-language",
|
||||
"reason": "2026-07-16: Language detection is pure shared string logic with an explicit Vite subpath alias.",
|
||||
"verifiedAt": "2026-07-16"
|
||||
},
|
||||
{
|
||||
"module": "live-agent-count",
|
||||
"reason": "2026-07-21: FN-8453 shares pure workflow-trait-based Running and Waiting predicates with dashboard capacity indicators; its dependency graph has no Node-only modules.",
|
||||
"verifiedAt": "2026-07-21"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user