From 3bf9bf5f74ecf43920d07e3371e1c6c0d0e00971 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Thu, 30 Jul 2026 00:31:27 -0700 Subject: [PATCH] collapse the plan-admission-throttle payload to one gate (+ AGENTS.md) (#2562) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) ## 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. --------- Co-authored-by: Claude Opus 5 (1M context) --- AGENTS.md | 2 +- ...iage-plan-admission-throttle-audit.test.ts | 121 ++++++++++++------ packages/engine/src/triage.ts | 51 +++++--- 3 files changed, 116 insertions(+), 58 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 40a6ae6e49..f233acb05f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -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: "`) 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. diff --git a/packages/engine/src/__tests__/triage-plan-admission-throttle-audit.test.ts b/packages/engine/src/__tests__/triage-plan-admission-throttle-audit.test.ts index 5ca1961a39..160150c276 100644 --- a/packages/engine/src/__tests__/triage-plan-admission-throttle-audit.test.ts +++ b/packages/engine/src/__tests__/triage-plan-admission-throttle-audit.test.ts @@ -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 } +/* +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 = {}): 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 { - 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 { + 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 }).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 }).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 }).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 }).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(); }); }); diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index 13aa5708dc..fc6fea9596 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -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(() => {