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

collapse the plan-admission-throttle payload to one gate (+ AGENTS.md) (#2562)

Browse Source 
The cross-project semaphore is deleted, so
`task:plan-admission-throttled` was describing a gate that no longer
exists. Nothing wires `options.semaphore` any more, which left three
things dead-but-visible:

- `semaphoreAvailable` was permanently `Infinity`, so
`Math.min(projectRoom, …)` was a no-op keeping a deleted limiter in the
arithmetic
- `blockedBy` was a **discriminator** between `"running-agent cap"` and
`"global semaphore"`; only the first can occur
- four `semaphore*` metadata fields were always `undefined`, and two
more terms in the dedupe signature were constant

## `blockedBy` is kept, not dropped

Even though it is now a constant. The event exists (FN-8600) to answer
*“why did this card sit queued to plan?”* after the fact — a named
reason answers that even when there is one gate, whereas a payload with
**no** reason field reads as “unknown”. It costs nothing and preserves
the shape if a second gate is ever added.

The dedupe signature drops the two semaphore terms and keeps the
eligible task IDs — that term is what stops a **new** card’s stall being
swallowed when the counts land on an unchanged tuple, which is the
property the event depends on.

## AGENTS.md

It documented the removed field names verbatim, so it is updated in the
same commit. Leaving docs describing a payload the code cannot emit is
exactly the readable-but-wrong artifact this program keeps deleting.

## Verification

`pnpm lint` clean · engine `tsc` clean · `pnpm test:gate` green · triage
suites **234/234**.

---

**Correction I owe on `concurrency.ts`, measured rather than
estimated.** I earlier told the coordinator ~75% of its 886 lines could
go with the cross-project cap. That was line-range arithmetic and it was
wrong. With the cap now fully removed, `concurrency.ts` is **still 886
lines**, because `AgentSemaphore` has four consumers unrelated to it —
`verification-concurrency` (maxConcurrentVerifications),
`research-orchestrator` (research runs), `experiment-executor`
(maxConcurrentExperiments), `step-session-executor` (parallel steps) —
plus `ProjectAdmissionCoordinator`, which is FN-8453 oldest-first
**ordering**, not a limiter. The real remaining win there is the
pre-held-slot bookkeeping and the idle-semaphore leak recovery, which
existed to service the global instance; I will measure that as its own
slice rather than quote a fraction.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **Bug Fixes**
* Updated plan admission throttling to consistently use the project’s
running-agent capacity.
* Improved throttle audit events by reporting stable capacity details
and removing obsolete semaphore information.
* Preserved accurate deduplication for repeated throttling events,
including changes in stalled tasks.

* **Documentation**
* Updated run-audit guidance to match the revised throttling event
format.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-07-30 00:31:27 -07:00
committed by GitHub
parent 8393bba7dc
commit 3bf9bf5f74
3 changed files with 116 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
AGENTS.md
View File

@@ -293,7 +293,7 @@ Scoped exception (FN-5819): shared-branch-group members (`branchContext.assignme
- FN-8004: `agent:heartbeat-move-skipped-soft-delete` records a heartbeat move that races a soft-deleted task without parking the durable agent. Metadata remains ids/timestamps/source only (`agentId`, optional `taskId`/`deletedAt`, `moveAttemptedAt`, optional `source`); it never stores error prose.
- FN-8141: the executor's `fn_task_done(outcome="blocked", reason=..., blockedBy?=[...])` honest-blocked exit emits `task:execution-blocked-parked` when an executor parks a genuinely-impossible task `failed` (`error = "BLOCKED: <reason>"`) instead of laundering it to `done` by skipping steps. It bypasses the completion/verdict/bulk-completion gates (blocked is not a completion claim), leaves steps in their true statuses, preserves worktree/branch, records `blockedBy` as real `task.dependencies` edges so the task requeues behind the blocker, and does NOT hand off to review — the parked row is honored by the executor's `status === "failed"` post-loop branch and is not auto-recovered into in-review by `recoverStrandedCompletedTodoTasks` (steps are not all done/skipped and `task.error` is set). Metadata stays ids/outcomes-only (`taskId`, `blockedBy` ids, `hasReason` boolean — never the reason prose).
- FN-8305: durable symbol-lock operations emit `symbol-lock:acquired`, `symbol-lock:acquire-conflict`, `symbol-lock:renewed`, `symbol-lock:released`, `symbol-lock:reconcile-stale`, and deduplicated `symbol-lock:reconcile-stale-no-action`. Metadata is ids/counts/outcomes-only; normalized opaque symbol keys are permitted IDs, while raw symbol prose is not.
- FN-8600: triage emits `task:plan-admission-throttled` when planning admission is withheld while eligible cards are waiting, recording the binding gate (`blockedBy`: `"running-agent cap"` or `"global semaphore"`) plus `maxConcurrent`, `claimed`, `projectRoom`, `eligibleCount`, up to five `eligibleTaskIds`, `processingCount`, up to five `processingTaskIds`, and the semaphore `activeCount`/`limit`/`availableCount`/`waitingCount`. Metadata is ids/counts-only. Deduped on the gate signature INCLUDING the eligible task IDs, so a sustained stall collapses to one row while a new card's stall is never swallowed; the marker is set only after the write lands, so a failed write retries on the next poll. Purpose: before this event the binding gate existed only in a `planLog` line that is persisted nowhere, so "why did this card sit queued to plan?" was unanswerable after the fact. Reachable today by direct DB query only — the sole run-audit read route resolves through a durable agent's heartbeat run and this event uses a synthetic run id under `agentId:"triage"`.
- FN-8600: triage emits `task:plan-admission-throttled` when planning admission is withheld while eligible cards are waiting, recording the binding gate (`blockedBy`, now always `"running-agent cap"` — the cross-project semaphore that was the other value is deleted, and the four `semaphore*` fields went with it) plus `maxConcurrent`, `claimed`, `projectRoom`, `eligibleCount`, up to five `eligibleTaskIds`, `processingCount`, and up to five `processingTaskIds`. Metadata is ids/counts-only. Deduped on the gate signature INCLUDING the eligible task IDs, so a sustained stall collapses to one row while a new card's stall is never swallowed; the marker is set only after the write lands, so a failed write retries on the next poll. Purpose: before this event the binding gate existed only in a `planLog` line that is persisted nowhere, so "why did this card sit queued to plan?" was unanswerable after the fact. Reachable today by direct DB query only — the sole run-audit read route resolves through a durable agent's heartbeat run and this event uses a synthetic run id under `agentId:"triage"`.
- FN-8592: startup and periodic self-healing emit `task:reconcile-stranded-hold-continuation` when an idle hold-column card with a real spec is re-seeded at its pre-release Plan Review, and deduped `task:reconcile-stranded-hold-continuation-no-action` for a candidate guard or race loss. Metadata stays ids/counts/outcomes-only (`taskId`, `column`, node/workflow identifiers, staleness or reason); healthy non-candidates are silent. The repair is insert-only and uses the shared per-task advisory transaction lock; global/engine pause and `autoMerge:false` defer to the operator.
- FN-8492: the self-healing sweep `reconcile-orphaned-pending-step-results` (startup, right after legacy adoption, plus periodic maintenance) emits `task:reconcile-orphaned-pending-step-results` when it REWRITES `pending` workflow-step results with no live session behind them to `failed` (canonical liveness triple: `activeSessionRegistry` path, `executingTaskLock`, `isTaskActive`). It must never DELETE an orphaned entry — the merge gate blocks on pending/failed results, not on an enabled step with no result, so deletion silently satisfies the gate and the task merges with its review skipped; the `failed` rewrite keeps the gate closed and hands re-run/bypass to the failed-pre-merge-steps recovery and FN-7720 operator-bypass paths. `in-progress` rows are always skipped (executor-owned; resume is deferred at startup), the row is re-read immediately before the write, and user pauses are never disturbed. Metadata is ids/counts-only (`taskId`, `column`, `orphanedCount`, `resultCount`).
- FN-8356: self-healing emits `task:reconcile-stale-duplicate-decision` when it clears a triage-marker duplicate-decision pause against a missing, deleted, done, or archived canonical. Metadata is ids/outcomes-only (`taskId`, `canonicalId`, `canonicalColumn`, `canonicalDeleted`, `priorPausedReason`); active canonical decisions and user pauses remain untouched.

121
packages/engine/src/__tests__/triage-plan-admission-throttle-audit.test.ts
View File

@@ -1,6 +1,5 @@
import { describe, it, expect, vi, beforeEach } from "vitest";
import type { Settings, Task, TaskStore } from "@fusion/core";
import { AgentSemaphore } from "../concurrency.js";
import { TriageProcessor } from "../triage.js";
/*
@@ -54,7 +53,29 @@ function eligibleTodoTask(id: string): Task {
interface RecordedEvent { type: string; target: string; metadata?: Record<string, unknown> }
/*
FNXC:CapacityModel 2026-07-29-18:40 (PR #2562 review):
The card that CONSUMES the project's single agent slot. Previously an exhausted host
semaphore forced the withhold; that gate is deleted, so the binding gate is now the
per-project agent count and something has to be holding it.
*/
function runningTask(id: string): Task {
return {
id,
description: "already running",
column: "in-progress",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: "2026-07-26T15:00:00.000Z",
updatedAt: "2026-07-26T15:00:00.000Z",
} as Task;
}
function createStore(tasks: Task[], recorded: RecordedEvent[], settings: Partial<Settings> = {}): TaskStore {
// The running claimant is added here so every case exhausts the one project slot.
tasks = [runningTask("FN-RUNNING"), ...tasks];
return {
getTask: vi.fn().mockImplementation(async (id: string) => {
const task = tasks.find((candidate) => candidate.id === id);
@@ -62,7 +83,9 @@ function createStore(tasks: Task[], recorded: RecordedEvent[], settings: Partial
}),
listTasks: vi.fn().mockResolvedValue(tasks),
getSettings: vi.fn().mockResolvedValue({
maxConcurrent: 12,
// FNXC:CapacityModel 2026-07-29-18:40: one slot, consumed by the in-progress
// row below, so planning admission is withheld by the PROJECT agent count.
maxConcurrent: 1,
maxWorktrees: 4,
pollIntervalMs: 600_000,
groupOverlappingFiles: false,
@@ -92,9 +115,23 @@ function createStore(tasks: Task[], recorded: RecordedEvent[], settings: Partial
} as unknown as TaskStore;
}
/** Runs one real poll pass against an exhausted host semaphore. */
async function pollWithExhaustedSemaphore(store: TaskStore, semaphore: AgentSemaphore): Promise<TriageProcessor> {
const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", { semaphore });
/*
FNXC:CapacityModel 2026-07-29-18:40 (PR #2562 review — greptile P1):
Drive the throttle through the PROJECT AGENT COUNT, not an exhausted host semaphore.
These cases previously spent a 1-slot `AgentSemaphore` to force the withhold. That
gate is deleted with the cross-project cap, so the poll no longer consults it and no
throttle event was emitted — the suite exercised a limiter that no longer exists and
would have gone quietly non-firing.
The REQUIREMENT is unchanged and is what these cases still pin: when planning
admission is withheld while eligible cards wait, the binding gate must be recorded
durably (FN-8600 — an operator asked why a card sat "Queued to plan" for seven
minutes and it was unanswerable after the fact). Only the gate that can bind has
changed, so the setup exhausts `maxConcurrent` instead.
*/
async function pollWithExhaustedProjectCapacity(store: TaskStore): Promise<TriageProcessor> {
const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", {});
// poll() is a no-op unless the processor is running; these tests drive one pass directly rather
// than starting the interval timer, which would make them time-dependent.
(processor as unknown as { running: boolean }).running = true;
@@ -113,32 +150,25 @@ describe("plan admission throttle run-audit (FN-8600)", () => {
});
it("records the binding gate when planning is withheld with eligible work", async () => {
const semaphore = new AgentSemaphore(1);
expect(semaphore.tryAcquire()).toBe(true); // host capacity fully spent
const store = createStore([eligibleTodoTask("FN-8600")], recorded);
await pollWithExhaustedSemaphore(store, semaphore);
await pollWithExhaustedProjectCapacity(store);
const throttle = recorded.filter((event) => event.type === "task:plan-admission-throttled");
expect(throttle).toHaveLength(1);
// The gate the operator could not previously determine.
expect(throttle[0].metadata).toMatchObject({
blockedBy: "global semaphore",
maxConcurrent: 12,
blockedBy: "running-agent cap",
maxConcurrent: 1,
eligibleCount: 1,
eligibleTaskIds: ["FN-8600"],
semaphoreLimit: 1,
semaphoreAvailableCount: 0,
});
semaphore.release();
});
it("emits one row for a sustained stall instead of one per poll", async () => {
const semaphore = new AgentSemaphore(1);
expect(semaphore.tryAcquire()).toBe(true);
const store = createStore([eligibleTodoTask("FN-8600")], recorded);
const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", { semaphore });
const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", {});
(processor as unknown as { running: boolean }).running = true;
const poll = (processor as unknown as { poll: () => Promise<void> }).poll.bind(processor);
await poll();
@@ -147,7 +177,6 @@ describe("plan admission throttle run-audit (FN-8600)", () => {
await new Promise((resolve) => setImmediate(resolve));
expect(recorded.filter((event) => event.type === "task:plan-admission-throttled")).toHaveLength(1);
semaphore.release();
});
/*
@@ -157,25 +186,27 @@ describe("plan admission throttle run-audit (FN-8600)", () => {
whole job is answering "why is THIS card queued".
*/
it("emits again when a different card is the one stalling, even with identical counts", async () => {
const semaphore = new AgentSemaphore(1);
expect(semaphore.tryAcquire()).toBe(true);
const first = eligibleTodoTask("FN-8600");
const store = createStore([first], recorded);
const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", { semaphore });
const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", {});
(processor as unknown as { running: boolean }).running = true;
const poll = (processor as unknown as { poll: () => Promise<void> }).poll.bind(processor);
await poll();
// Same counts, different card: A leaves the queue as C enters.
/*
Same counts, different card: A leaves the queue as C enters. The running
claimant must be re-supplied — this mock REPLACES the list, and without it the
project slot frees up, the gate stops binding and no second row is emitted
(which is a correct outcome for a different scenario, not this one).
*/
(store.listTasks as unknown as { mockResolvedValue: (v: Task[]) => void })
.mockResolvedValue([eligibleTodoTask("FN-8601")]);
.mockResolvedValue([runningTask("FN-RUNNING"), eligibleTodoTask("FN-8601")]);
await poll();
const throttle = recorded.filter((event) => event.type === "task:plan-admission-throttled");
expect(throttle).toHaveLength(2);
expect(throttle[1].metadata).toMatchObject({ eligibleTaskIds: ["FN-8601"] });
semaphore.release();
});
/*
@@ -185,9 +216,7 @@ describe("plan admission throttle run-audit (FN-8600)", () => {
original unanswerable-stall problem.
*/
it("retries the audit write on the next poll when the first write fails", async () => {
const semaphore = new AgentSemaphore(1);
expect(semaphore.tryAcquire()).toBe(true);
const store = createStore([eligibleTodoTask("FN-8600")], recorded);
let attempts = 0;
(store as unknown as { recordRunAuditEvent: unknown }).recordRunAuditEvent = vi.fn()
@@ -197,7 +226,7 @@ describe("plan admission throttle run-audit (FN-8600)", () => {
recorded.push({ type: event.mutationType, target: event.target, metadata: event.metadata });
});
const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", { semaphore });
const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", {});
(processor as unknown as { running: boolean }).running = true;
const poll = (processor as unknown as { poll: () => Promise<void> }).poll.bind(processor);
@@ -209,28 +238,42 @@ describe("plan admission throttle run-audit (FN-8600)", () => {
await poll();
await new Promise((resolve) => setImmediate(resolve));
expect(recorded.filter((event) => event.type === "task:plan-admission-throttled")).toHaveLength(1);
semaphore.release();
});
it("stays silent when planning has capacity", async () => {
const semaphore = new AgentSemaphore(4);
const store = createStore([], recorded);
await pollWithExhaustedSemaphore(store, semaphore);
/*
FNXC:PlanAdmissionThrottle 2026-07-31-11:25 (PR #2562 review — coderabbit):
THIS TEST PROVED THE WRONG SILENCE. It used `createStore([])`, which seeds only
the running claimant and NO eligible card — so it asserted that an EMPTY QUEUE
emits no throttle, which is true of any implementation, including one that emits
a throttle on every poll where a card is waiting. The name says "has capacity";
the setup had no candidate for capacity to matter to.
Now the real shape: an eligible card AND room to start it. `maxConcurrent: 3`
against one running claimant leaves projectRoom > 0, so admission proceeds and
the throttle must stay silent for the reason the name claims.
`specifyTask` is stubbed because admission now actually dispatches — without it
the test would drive a real planning session.
*/
const store = createStore([eligibleTodoTask("FN-8600-ROOM")], recorded, { maxConcurrent: 3 });
const processor = new TriageProcessor(store, "/tmp/fn-8600-throttle-root", {});
vi.spyOn(processor, "specifyTask").mockResolvedValue(undefined);
(processor as unknown as { running: boolean }).running = true;
await (processor as unknown as { poll: () => Promise<void> }).poll();
await new Promise((resolve) => setImmediate(resolve));
expect(recorded.filter((event) => event.type === "task:plan-admission-throttled")).toHaveLength(0);
});
it("carries no prompt, title, or reason prose — ids and counts only", async () => {
const semaphore = new AgentSemaphore(1);
expect(semaphore.tryAcquire()).toBe(true);
const store = createStore([eligibleTodoTask("FN-8600")], recorded);
await pollWithExhaustedSemaphore(store, semaphore);
await pollWithExhaustedProjectCapacity(store);
const throttle = recorded.find((event) => event.type === "task:plan-admission-throttled");
expect(throttle).toBeDefined();
const serialized = JSON.stringify(throttle!.metadata ?? {});
expect(serialized).not.toContain("favorite projects");
semaphore.release();
});
});

51
packages/engine/src/triage.ts
View File

@@ -1858,9 +1858,6 @@ export class TriageProcessor {
exceed its operator-facing top-level capacity in a different lane.
*/
const maxConcurrent = settings.maxConcurrent ?? 2;
const semaphoreAvailable = this.options.semaphore
? Math.max(0, this.options.semaphore.availableCount)
: Infinity;
// processing entries that have not yet written status:"planning" still claim a future slot.
let pendingSpecifyCount = 0;
for (const id of this.processing) {
@@ -1872,23 +1869,40 @@ export class TriageProcessor {
tasks: allTasks,
pendingSpecifyCount,
});
// `claimed` is project-local. The scoped/global host semaphore remains a
// distinct process-wide availability gate, so project A cannot spend B's cap.
/*
FNXC:CapacityModel 2026-07-31-11:10 (PR #2562 review — coderabbit; CORRECTED):
Capacity is the PROJECT's agent count, full stop. The second term was the
cross-project host semaphore's AVAILABILITY, which no longer constrains
admission: permanently Infinity here, so `Math.min` was a no-op keeping a dead
limiter visible in the arithmetic.
REMOVED FROM THROTTLE ACCOUNTING, NOT DELETED. An earlier version of this note
said "nothing wires `options.semaphore` any more", which is false and dangerous
in a specific way: it reads as permission to delete live coordination code.
`options.semaphore` is still wired at five call sites — pre-held slot
registration (`reserve`), the release on drop, the leak-recovery path, and the
two admission-reservation hand-offs. Only its use as an ADMISSION LIMIT is gone.
*/
const projectRoom = Math.max(0, maxConcurrent - claimed);
const maxToStart = Math.min(projectRoom, semaphoreAvailable);
const maxToStart = projectRoom;
if (maxToStart <= 0 && triageTasks.length > 0) {
const semaphoreSnapshot = this.options.semaphore?.snapshot();
const semaphoreDetail = semaphoreSnapshot
? `, semaphore active=${semaphoreSnapshot.activeCount}/${semaphoreSnapshot.limit}, available=${semaphoreSnapshot.availableCount}, waiting=${semaphoreSnapshot.waitingCount}`
: ", semaphore unavailable";
const processingIds = [...this.processing].slice(0, 5);
const eligibleIds = triageTasks.slice(0, 5).map((t) => t.id);
const blockedBy = projectRoom <= 0 ? "running-agent cap" : "global semaphore";
/*
FNXC:CapacityModel 2026-07-29-10:20 (drop the cross-project cap — throttle payload):
`blockedBy` was a DISCRIMINATOR between two gates: "running-agent cap" and
"global semaphore". With the machine-wide cap deleted there is only one gate
left, so the field collapses to a constant. It is KEPT rather than dropped:
the event's whole purpose (FN-8600) is answering "why did this card sit
queued?", and a named reason answers it even when there is only one — while
a payload with no reason field at all would read as "unknown".
*/
const blockedBy = "running-agent cap";
planLog.log(
`Plan throttled by ${blockedBy}: eligible=${triageTasks.length} [${eligibleIds.join(", ")}], ` +
`maxConcurrent=${maxConcurrent}, claimed=${claimed}, processing=${this.processing.size}` +
`${processingIds.length > 0 ? ` [${processingIds.join(", ")}]` : ""}${semaphoreDetail}`,
`${processingIds.length > 0 ? ` [${processingIds.join(", ")}]` : ""}`,
);
/*
FNXC:ConcurrencyAdmission 2026-07-26-09:30:
@@ -1915,14 +1929,19 @@ export class TriageProcessor {
Live counts still jitter as unrelated lanes cycle, so this bounds write volume rather than
guaranteeing exactly one row.
*/
/*
FNXC:CapacityModel 2026-07-29-10:20: the two semaphore terms are dropped from
the dedupe signature with the gate they described. `blockedBy` is now
constant and contributes nothing, but stays for readability of the key; the
eligible task IDs remain the term that keeps a NEW card's stall from being
swallowed by an unchanged count tuple.
*/
const throttleSignature = [
blockedBy,
maxConcurrent,
claimed,
triageTasks.length,
this.processing.size,
semaphoreSnapshot?.activeCount ?? -1,
semaphoreSnapshot?.limit ?? -1,
eligibleIds.join(","),
].join("|");
if (this.lastPlanThrottleSignature !== throttleSignature) {
@@ -1950,10 +1969,6 @@ export class TriageProcessor {
eligibleTaskIds: eligibleIds,
processingCount: this.processing.size,
processingTaskIds: processingIds,
semaphoreActiveCount: semaphoreSnapshot?.activeCount,
semaphoreLimit: semaphoreSnapshot?.limit,
semaphoreAvailableCount: semaphoreSnapshot?.availableCount,
semaphoreWaitingCount: semaphoreSnapshot?.waitingCount,
},
})
.then(() => {