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

feat(KB-152): make triage poll() dispatch specifyTask() calls concurrently

Browse Source 
- Change poll() to fire specifyTask() with `void` instead of `await`, enabling concurrent task specification bounded by AgentSemaphore
- Add JSDoc to poll() documenting concurrent dispatch and re-entrance guard behavior
- Add tests for concurrent dispatch, semaphore limiting, polling flag reset, and overlapping poll discovery
- Update existing re-entrance guard test to block on listTasks (discovery) rather than specifyTask
- Fix existing paused-tasks and globalPause tests for non-blocking dispatch timing
This commit is contained in:
Dustin Byrne
2026-03-28 02:31:41 -04:00
parent 40dacb49d3
commit 5e3cafe05b
2 changed files with 212 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

233
packages/engine/src/triage.test.ts
View File

@@ -250,42 +250,31 @@ describe("TriageProcessor poll re-entrance guard", () => {
vi.clearAllMocks();
});
it("prevents overlapping poll() calls — second call is a no-op", async () => {
const store = createMockStore([
{
id: "KB-001",
title: "Test",
description: "Test",
column: "triage",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
},
]);
it("prevents overlapping poll() calls while discovery is in progress", async () => {
let resolveListTasks: (() => void) | undefined;
const store = createMockStore([]);
// Make listTasks slow so the first poll is still in the discovery phase
// when the second poll fires
store.listTasks.mockImplementation(
() =>
new Promise<any[]>((resolve) => {
resolveListTasks = () => resolve([]);
}),
);
// Make specifyTask slow so the first poll is still running when the second fires
let resolveAgent: (() => void) | undefined;
mockedCreateHaiAgent.mockImplementation(async () => {
await new Promise<void>((r) => {
resolveAgent = r;
});
return {
session: {
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
},
} as any;
});
mockedCreateHaiAgent.mockResolvedValue({
session: {
prompt: vi.fn().mockResolvedValue(undefined),
dispose: vi.fn(),
},
} as any);
const triage = new TriageProcessor(store, "/tmp/test");
(triage as any).running = true;
// Start first poll (will block in createKbAgent)
// Start first poll (will block in listTasks)
const poll1 = (triage as any).poll();
// Allow microtasks to run so poll1 gets into specifyTask
// Allow microtasks to run so poll1 enters the try block
await new Promise((r) => setTimeout(r, 10));
// Start second poll — should return immediately due to guard
@@ -295,8 +284,8 @@ describe("TriageProcessor poll re-entrance guard", () => {
// listTasks should only have been called once (first poll)
expect(store.listTasks).toHaveBeenCalledTimes(1);
// Resolve the agent to let the first poll finish
resolveAgent?.();
// Resolve listTasks to let the first poll finish
resolveListTasks?.();
await poll1;
});
@@ -314,6 +303,182 @@ describe("TriageProcessor poll re-entrance guard", () => {
});
});
describe("TriageProcessor concurrent dispatch", () => {
beforeEach(() => {
vi.clearAllMocks();
});
it("dispatches multiple triage tasks concurrently in a single poll cycle", async () => {
const tasks = [
{ id: "KB-001", title: "T1", description: "T1", column: "triage" as const, dependencies: [], steps: [], currentStep: 0, log: [], createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() },
{ id: "KB-002", title: "T2", description: "T2", column: "triage" as const, dependencies: [], steps: [], currentStep: 0, log: [], createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() },
{ id: "KB-003", title: "T3", description: "T3", column: "triage" as const, dependencies: [], steps: [], currentStep: 0, log: [], createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() },
];
const store = createMockStore(tasks);
// Track which tasks are concurrently in-flight
let concurrent = 0;
let maxConcurrent = 0;
const resolvers: (() => void)[] = [];
mockedCreateHaiAgent.mockImplementation(async () => {
concurrent++;
maxConcurrent = Math.max(maxConcurrent, concurrent);
return {
session: {
prompt: vi.fn().mockImplementation(async () => {
await new Promise<void>((r) => resolvers.push(r));
concurrent--;
}),
dispose: vi.fn(),
},
} as any;
});
const sem = new AgentSemaphore(3); // Allow all 3 tasks to run concurrently
const triage = new TriageProcessor(store, "/tmp/test", { semaphore: sem });
(triage as any).running = true;
// poll() should return quickly (non-blocking dispatch)
await (triage as any).poll();
// All 3 tasks should have been dispatched concurrently
await new Promise((r) => setTimeout(r, 50));
expect(maxConcurrent).toBe(3);
expect(mockedCreateHaiAgent).toHaveBeenCalledTimes(3);
// Resolve all agents
resolvers.forEach((r) => r());
await new Promise((r) => setTimeout(r, 50));
});
it("polling flag resets before specifyTask calls finish", async () => {
const tasks = [
{ id: "KB-001", title: "T1", description: "T1", column: "triage" as const, dependencies: [], steps: [], currentStep: 0, log: [], createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() },
];
const store = createMockStore(tasks);
let resolveAgent: (() => void) | undefined;
mockedCreateHaiAgent.mockImplementation(async () => {
return {
session: {
prompt: vi.fn().mockImplementation(
() => new Promise<void>((r) => { resolveAgent = r; }),
),
dispose: vi.fn(),
},
} as any;
});
const triage = new TriageProcessor(store, "/tmp/test");
(triage as any).running = true;
// poll() should return quickly even though specifyTask is still running
await (triage as any).poll();
// polling flag should be false (reset) even though the agent is still working
expect((triage as any).polling).toBe(false);
// Resolve the agent to clean up
resolveAgent?.();
await new Promise((r) => setTimeout(r, 50));
});
it("semaphore still limits actual concurrency across concurrent dispatches", async () => {
const tasks = [
{ id: "KB-001", title: "T1", description: "T1", column: "triage" as const, dependencies: [], steps: [], currentStep: 0, log: [], createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() },
{ id: "KB-002", title: "T2", description: "T2", column: "triage" as const, dependencies: [], steps: [], currentStep: 0, log: [], createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() },
{ id: "KB-003", title: "T3", description: "T3", column: "triage" as const, dependencies: [], steps: [], currentStep: 0, log: [], createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() },
];
const store = createMockStore(tasks);
let concurrent = 0;
let maxConcurrent = 0;
const resolvers: (() => void)[] = [];
mockedCreateHaiAgent.mockImplementation(async () => {
concurrent++;
maxConcurrent = Math.max(maxConcurrent, concurrent);
return {
session: {
prompt: vi.fn().mockImplementation(async () => {
await new Promise<void>((r) => resolvers.push(r));
concurrent--;
}),
dispose: vi.fn(),
},
} as any;
});
const sem = new AgentSemaphore(1); // Only 1 slot
const triage = new TriageProcessor(store, "/tmp/test", { semaphore: sem });
(triage as any).running = true;
// poll() dispatches all 3 but semaphore only allows 1 at a time
await (triage as any).poll();
await new Promise((r) => setTimeout(r, 50));
// Only 1 agent should be running at a time
expect(maxConcurrent).toBe(1);
expect(mockedCreateHaiAgent).toHaveBeenCalledTimes(1);
// Resolve first, second gets its slot
resolvers[0]();
await new Promise((r) => setTimeout(r, 50));
expect(mockedCreateHaiAgent).toHaveBeenCalledTimes(2);
expect(maxConcurrent).toBe(1);
// Resolve second, third gets its slot
resolvers[1]();
await new Promise((r) => setTimeout(r, 50));
expect(mockedCreateHaiAgent).toHaveBeenCalledTimes(3);
expect(maxConcurrent).toBe(1);
// Resolve third
resolvers[2]();
await new Promise((r) => setTimeout(r, 50));
expect(sem.activeCount).toBe(0);
});
it("subsequent polls can discover new triage tasks while prior dispatches are still running", async () => {
const task1 = { id: "KB-001", title: "T1", description: "T1", column: "triage" as const, dependencies: [], steps: [], currentStep: 0, log: [], createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() };
const task2 = { id: "KB-002", title: "T2", description: "T2", column: "triage" as const, dependencies: [], steps: [], currentStep: 0, log: [], createdAt: new Date().toISOString(), updatedAt: new Date().toISOString() };
const store = createMockStore([task1]);
const resolvers: (() => void)[] = [];
mockedCreateHaiAgent.mockImplementation(async () => ({
session: {
prompt: vi.fn().mockImplementation(
() => new Promise<void>((r) => resolvers.push(r)),
),
dispose: vi.fn(),
},
} as any));
const sem = new AgentSemaphore(5);
const triage = new TriageProcessor(store, "/tmp/test", { semaphore: sem });
(triage as any).running = true;
// First poll dispatches KB-001
await (triage as any).poll();
await new Promise((r) => setTimeout(r, 20));
expect(mockedCreateHaiAgent).toHaveBeenCalledTimes(1);
// A new task arrives — second poll should discover it
store.listTasks.mockResolvedValue([task1, task2]);
// Second poll runs (polling flag already reset) — KB-001 is in processing set,
// so only KB-002 is dispatched
await (triage as any).poll();
await new Promise((r) => setTimeout(r, 20));
expect(mockedCreateHaiAgent).toHaveBeenCalledTimes(2);
// Clean up
resolvers.forEach((r) => r());
await new Promise((r) => setTimeout(r, 50));
});
});
describe("TriageProcessor dynamic poll interval", () => {
beforeEach(() => {
vi.clearAllMocks();
@@ -419,6 +584,8 @@ describe("TriageProcessor paused tasks", () => {
const triage = new TriageProcessor(store, "/tmp/test");
(triage as any).running = true;
await (triage as any).poll();
// Allow dispatched specifyTask to run (non-blocking dispatch)
await new Promise((r) => setTimeout(r, 50));
// Agent should be created for a non-paused task
expect(store.updateTask).toHaveBeenCalledWith("KB-002", { status: "specifying" });
@@ -518,6 +685,8 @@ describe("TriageProcessor globalPause", () => {
// Second poll — should process tasks
await (triage as any).poll();
// Allow dispatched specifyTask to run (non-blocking dispatch)
await new Promise((r) => setTimeout(r, 50));
expect(store.updateTask).toHaveBeenCalledWith("KB-002", { status: "specifying" });
});

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

@@ -267,6 +267,16 @@ export class TriageProcessor {
triageLog.log(`Poll interval updated to ${newIntervalMs}ms`);
}
/**
* Discover triage tasks and dispatch `specifyTask()` for each one.
*
* **Concurrent dispatch:** `specifyTask()` calls are fired without awaiting,
* so multiple triage tasks can be specified concurrently (bounded by the
* shared `AgentSemaphore`). The `polling` re-entrance guard prevents
* overlapping discovery cycles, but resets as soon as dispatch completes —
* well before the dispatched tasks finish — so subsequent polls can discover
* newly arrived triage tasks promptly.
*/
private async poll(): Promise<void> {
if (!this.running) return;
if (this.polling) return;
@@ -292,7 +302,7 @@ export class TriageProcessor {
);
for (const task of triageTasks) {
await this.specifyTask(task);
void this.specifyTask(task);
}
} catch (err) {
triageLog.error("Poll error:", err);