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

FN-8671: isolate triage admission state in tests

Browse Source 
Prevent leaked singleton admission state from affecting later triage polling tests.

- Add test-only coordinator reset and inspection seams for all admission categories.
- Stop tracked triage processors before clearing shared reservation and pre-held-slot state.
- Cover teardown behavior and document singleton-state isolation guidance.

Files changed:
 docs/testing.md                                   |  6 ++
 packages/engine/src/__tests__/concurrency.test.ts | 56 ++++++++++++++++
 packages/engine/src/__tests__/triage.test.ts      | 82 ++++++++++++++++++++---
 packages/engine/src/concurrency.ts                | 30 +++++++++
 4 files changed, 166 insertions(+), 8 deletions(-)

Fusion-Task-Id: FN-8671

Fusion-Task-Lineage: 41217db4-1fd6-45d6-a846-c57fc3c7052e

Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
2026-08-01 00:14:32 -07:00
parent 3b680cce04
commit dd569395e8
4 changed files with 166 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
docs/testing.md
View File

@@ -614,6 +614,12 @@ Prefer `it.each` over copy-pasted `it()` blocks. When trimming, keep: first case
- Integration tests exercising real SQLite, real worker pool, or spawned processes.
- Lean core/engine unit tests with low mock burden.
## Test isolation for module-singleton state
<!-- FNXC:ConcurrencyAdmission 2026-08-01-06:57: Module-singleton admission state can survive mocked lane starts and unstopped processors, silently consuming capacity in later tests. FN-8671 fixes that root cause without quarantine: stop tracked owners first, then clear shared state in a finally block and assert the result through read-only inspection seams. -->
When a test owns a process-wide singleton that has asynchronous owners (timers, processors, or lane starts), use the same teardown in `beforeEach` and `afterEach`: await every tracked owner’s `stop()` with `Promise.allSettled`, then clear all shared state in a `finally` block. Do not clear first: a pending stop or callback can repopulate the singleton after the apparent reset. Test reset mutators establish cleanup; read-only inspection seams must prove reservations, mutex/draining state, registrations, and companion module-global slots are actually empty. Fix the isolation seam at the root rather than adding retries, wider timeouts, weakened assertions, or a quarantine entry.
## Standing Rule: Do Not Add Slow Tests (FN-5048)
- Default new tests to narrow seams, in-memory fakes, shared harnesses, and targeted assertions.

56
packages/engine/src/__tests__/concurrency.test.ts
View File

@@ -10,6 +10,7 @@ import {
PRIORITY_SPECIFY,
clearPreHeldExecutorSlotsForTests,
computeTopLevelConcurrencyClaimed,
getPreHeldExecutorSlotsForTests,
dropPreHeldExecutorSlot,
hasPreHeldExecutorSlot,
persistedTopLevelAgentSlots,
@@ -1068,6 +1069,61 @@ describe("AgentSemaphore resilience (FN-978)", () => {
describe("ProjectAdmissionCoordinator", () => {
it("clears test-only coordinator and pre-held state across every shared category", async () => {
const coordinator = new ProjectAdmissionCoordinator();
const projectId = "project-reset";
let resolveClaim!: (value: number) => void;
const pendingClaim = new Promise<number>((resolve) => { resolveClaim = resolve; });
const drainingReservation = coordinator.reserveIfAvailable({
projectId,
taskId: "FN-DRAINING",
maxConcurrent: 4,
claimed: () => pendingClaim,
});
await Promise.resolve();
expect(coordinator.inspectProjectStateForTests(projectId).draining).toBe(true);
resolveClaim(0);
await drainingReservation;
expect(coordinator.inspectProjectStateForTests(projectId)).toMatchObject({
reservedCount: 1,
draining: false,
});
coordinator.registerProvider("specify:project-reset", {
projectId,
refresh: async () => [],
});
expect(coordinator.inspectProjectStateForTests(projectId).providerIds)
.toContain("specify:project-reset");
coordinator.clearReservationsForTests();
expect(coordinator.inspectProjectStateForTests(projectId)).toEqual({
reservedCount: 0,
draining: false,
providerIds: [],
});
expect(await coordinator.reserveIfAvailable({
projectId,
taskId: "FN-AFTER-DRAINING-RESET",
maxConcurrent: 1,
claimed: () => 0,
})).toBe(true);
coordinator.clearReservationsForTests();
coordinator.registerProvider("specify:project-reset", {
projectId,
refresh: async () => [],
});
expect(coordinator.inspectProjectStateForTests(projectId).providerIds)
.toEqual(["specify:project-reset"]);
registerPreHeldExecutorSlot("FN-PREHELD-RESET");
expect(getPreHeldExecutorSlotsForTests()).toContain("FN-PREHELD-RESET");
clearPreHeldExecutorSlotsForTests();
expect(getPreHeldExecutorSlotsForTests()).toEqual([]);
});
it("shares the final active-task slot across planning, execution, and merge lanes", async () => {
const coordinator = new ProjectAdmissionCoordinator();
const started: string[] = [];

82
packages/engine/src/__tests__/triage.test.ts
View File

@@ -11,6 +11,7 @@ import {
import {
AgentSemaphore,
clearPreHeldExecutorSlotsForTests,
getPreHeldExecutorSlotsForTests,
hasPreHeldExecutorSlot,
projectAdmissionCoordinator,
registerPreHeldExecutorSlot,
@@ -1570,13 +1571,41 @@ describe("TriageProcessor", () => {
let store: TaskStore;
let processor: TriageProcessor;
const rootDir = "/fake/root";
const trackedProcessors: TriageProcessor[] = [];
beforeEach(() => {
const trackProcessor = (instance: TriageProcessor): TriageProcessor => {
trackedProcessors.push(instance);
return instance;
};
/*
FNXC:ConcurrencyAdmission 2026-08-01-06:42:
Stubbed specifyTask calls never release the singleton reservation, and
unstopped processors retain specify providers. Stop every tracked processor
before clearing in finally: clearing first allows stop/timer callbacks to
repopulate shared state after the apparent reset.
*/
const resetTriageAdmissionState = async (): Promise<void> => {
try {
await Promise.allSettled(trackedProcessors.map(async (instance) => instance.stop()));
} finally {
projectAdmissionCoordinator.clearReservationsForTests();
clearPreHeldExecutorSlotsForTests();
trackedProcessors.length = 0;
}
};
beforeEach(async () => {
await resetTriageAdmissionState();
store = createMockStore();
processor = new TriageProcessor(store, rootDir);
processor = trackProcessor(new TriageProcessor(store, rootDir));
mockReviewStep.mockReset();
});
afterEach(async () => {
await resetTriageAdmissionState();
});
it("creates processor with default options", () => {
expect(processor).toBeInstanceOf(TriageProcessor);
});
@@ -1743,6 +1772,43 @@ Planner rewrote mission without the raw request.
expect(store.on).toHaveBeenCalledWith("settings:updated", expect.any(Function));
});
it("clears poll admission state even when a tracked processor stop throws", async () => {
const projectId = "/fake/triage-teardown";
const task = createTriageTask({ id: "FN-TEARDOWN" });
const triageStore = createMockStore({
listTasks: vi.fn().mockResolvedValue([task]),
getSettings: vi.fn().mockResolvedValue({
maxConcurrent: 10,
pollIntervalMs: 10_000,
groupOverlappingFiles: false,
autoMerge: true,
}),
});
const leakingProcessor = trackProcessor(new TriageProcessor(triageStore, projectId, {
semaphore: new AgentSemaphore(10),
}));
vi.spyOn(leakingProcessor, "specifyTask").mockResolvedValue(undefined);
(leakingProcessor as any).running = true;
await (leakingProcessor as any).poll();
expect(projectAdmissionCoordinator.inspectProjectStateForTests(projectId).reservedCount).toBe(1);
expect(getPreHeldExecutorSlotsForTests()).toContain(task.id);
vi.spyOn(leakingProcessor, "stop").mockImplementation(() => {
throw new Error("stop failed");
});
await resetTriageAdmissionState();
expect(projectAdmissionCoordinator.inspectProjectStateForTests(projectId)).toEqual({
reservedCount: 0,
draining: false,
providerIds: [],
});
expect(projectAdmissionCoordinator.inspectProjectStateForTests(projectId).providerIds)
.not.toContain(`specify:${projectId}`);
expect(getPreHeldExecutorSlotsForTests()).toEqual([]);
});
describe("poll ordering", () => {
it("dispatches eligible triage tasks by createdAt asc", async () => {
const tasks: Task[] = [
@@ -1777,7 +1843,7 @@ Planner rewrote mission without the raw request.
autoMerge: true,
}),
});
const triageProcessor = new TriageProcessor(triageStore, rootDir);
const triageProcessor = trackProcessor(new TriageProcessor(triageStore, rootDir));
const specifySpy = vi
.spyOn(triageProcessor, "specifyTask")
.mockResolvedValue(undefined);
@@ -1814,7 +1880,7 @@ Planner rewrote mission without the raw request.
autoMerge: true,
}),
});
const triageProcessor = new TriageProcessor(triageStore, rootDir);
const triageProcessor = trackProcessor(new TriageProcessor(triageStore, rootDir));
const specifySpy = vi
.spyOn(triageProcessor, "specifyTask")
.mockResolvedValue(undefined);
@@ -1851,7 +1917,7 @@ Planner rewrote mission without the raw request.
autoMerge: true,
}),
});
const triageProcessor = new TriageProcessor(triageStore, rootDir);
const triageProcessor = trackProcessor(new TriageProcessor(triageStore, rootDir));
const specifySpy = vi
.spyOn(triageProcessor, "specifyTask")
.mockResolvedValue(undefined);
@@ -1896,7 +1962,7 @@ Planner rewrote mission without the raw request.
limit: 4,
snapshot: vi.fn(() => ({ activeCount: 0, waitingCount: 0, availableCount: 4, limit: 4 })),
};
const triageProcessor = new TriageProcessor(triageStore, rootDir, { semaphore: semaphore as any });
const triageProcessor = trackProcessor(new TriageProcessor(triageStore, rootDir, { semaphore: semaphore as any }));
const specifySpy = vi
.spyOn(triageProcessor, "specifyTask")
.mockResolvedValue(undefined);
@@ -1933,7 +1999,7 @@ Planner rewrote mission without the raw request.
autoMerge: true,
} as Settings),
});
const triageProcessor = new TriageProcessor(triageStore, rootDir);
const triageProcessor = trackProcessor(new TriageProcessor(triageStore, rootDir));
const { promptWithFallback } = await import("../pi.js");
mockCreateFnAgent.mockClear();
@@ -1995,7 +2061,7 @@ Planner rewrote mission without the raw request.
autoMerge: true,
}),
});
const triageProcessor = new TriageProcessor(triageStore, rootDir);
const triageProcessor = trackProcessor(new TriageProcessor(triageStore, rootDir));
const specifySpy = vi
.spyOn(triageProcessor, "specifyTask")
.mockResolvedValue(undefined);

30
packages/engine/src/concurrency.ts
View File

@@ -105,6 +105,31 @@ export class ProjectAdmissionCoordinator {
}
}
/*
FNXC:ConcurrencyAdmission 2026-08-01-06:42:
The process-wide coordinator outlives tests whose stubbed lane start never
releases a reservation and whose processor is never stopped. Test-only reset
and inspection seams clear every shared category and prove cleanliness
observably, preventing silent project-capacity exhaustion in later tests.
*/
clearReservationsForTests(): void {
this.reservations.clear();
this.draining.clear();
this.providers.clear();
}
inspectProjectStateForTests(projectId: string): {
reservedCount: number;
draining: boolean;
providerIds: string[];
} {
return {
reservedCount: this.reservationCount(projectId),
draining: this.draining.has(projectId),
providerIds: [...(this.providers.get(projectId)?.keys() ?? [])].sort(),
};
}
private reservationCount(projectId: string): number {
return this.reservations.get(projectId)?.size ?? 0;
}
@@ -412,6 +437,11 @@ export function clearPreHeldExecutorSlotsForTests(): void {
preHeldAdmissionReservations.clear();
}
/** Test-only read seam for asserting no pre-held executor handoffs survive teardown. */
export function getPreHeldExecutorSlotsForTests(): string[] {
return [...preHeldExecutorSlots].sort();
}
/**
* FNXC:GlobalConcurrencyControls 2026-06-27-00:00:
* Persisted semaphore repair must use the same top-level slot predicate as dashboard and CLI live counts, including active in-review agents, so read-layer utilization and engine recovery cannot drift.