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

feat(FN-1254): add inbox-lite task selection for heartbeat agents

Browse Source 
- Add InboxTask typing and TaskStore.selectNextTaskForAgent() with priority ordering, dependency checks, paused filtering, and FIFO selection
- Wire heartbeat execution to auto-select and assign inbox work when no task is set, with optional checkout attempts and graceful conflict fallback
- Add POST /api/agents/:id/inbox to expose next-task selection details (task, priority, reason) and return task:null when no work is available
- Expand core, engine, and dashboard tests to cover selection priorities, heartbeat precedence/metadata, checkout-conflict handling, and route behavior with type-safe mocks
This commit is contained in:
gsxdsm
2026-04-08 17:57:16 -07:00
parent 86e3869f75
commit 61117c8a2b
8 changed files with 532 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

192
packages/engine/src/agent-heartbeat.test.ts
View File

@@ -960,8 +960,12 @@ describe("HeartbeatMonitor", () => {
};
}
type MockTaskStoreOverrides = Partial<TaskStore> & {
checkoutTask?: (taskId: string, agentId: string) => Promise<unknown>;
};
// Helper: create a basic mock task store
function createMockTaskStore(overrides: Partial<TaskStore> = {}): TaskStore {
function createMockTaskStore(overrides: MockTaskStoreOverrides = {}): TaskStore {
return {
getTask: vi.fn().mockResolvedValue({
id: "FN-001",
@@ -976,6 +980,7 @@ describe("HeartbeatMonitor", () => {
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
} as unknown as TaskDetail),
selectNextTaskForAgent: vi.fn().mockResolvedValue(null),
createTask: vi.fn().mockResolvedValue({
id: "FN-002",
description: "Created task",
@@ -1010,6 +1015,10 @@ describe("HeartbeatMonitor", () => {
updateAgentState: vi.fn().mockResolvedValue(undefined),
updateAgent: vi.fn().mockResolvedValue(undefined),
getAgent: vi.fn().mockResolvedValue(mockAgent),
assignTask: vi.fn().mockImplementation(async (_agentId: string, taskId: string | undefined) => {
mockAgent.taskId = taskId;
return mockAgent;
}),
startHeartbeatRun: vi.fn().mockResolvedValue({
id: "run-001",
agentId: "agent-001",
@@ -1116,6 +1125,187 @@ describe("HeartbeatMonitor", () => {
});
});
describe("executeHeartbeat - inbox selection", () => {
const makeInboxSelection = (taskId: string, priority: "in_progress" | "todo" | "blocked" = "todo") => {
const now = new Date().toISOString();
return {
task: {
id: taskId,
description: `Inbox task ${taskId}`,
column: priority === "in_progress" ? "in-progress" : "todo",
dependencies: [],
steps: [],
currentStep: 0,
log: [],
createdAt: now,
updatedAt: now,
},
priority,
reason: `selected:${priority}`,
} as any;
};
it("when agent has no taskId, inbox selects a todo task and assigns it", async () => {
const store = createStoreWithAgentForExec({ taskId: undefined });
const selectNextTaskForAgent = vi.fn().mockResolvedValue(makeInboxSelection("FN-INBOX", "todo"));
mockTaskStore = createMockTaskStore({
selectNextTaskForAgent,
getTask: vi.fn().mockResolvedValue({
id: "FN-INBOX",
title: "Inbox Task",
description: "Inbox-selected task",
prompt: "",
steps: [],
column: "todo",
dependencies: [],
log: [],
attachments: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
} as unknown as TaskDetail),
});
const mockSession = createMockAgentSession();
mockedCreateKbAgent.mockResolvedValue({ session: mockSession as any });
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
expect(selectNextTaskForAgent).toHaveBeenCalledWith("agent-001");
expect(store.assignTask).toHaveBeenCalledWith("agent-001", "FN-INBOX");
expect(mockTaskStore.getTask).toHaveBeenCalledWith("FN-INBOX");
});
it("explicit taskId override takes precedence over inbox selection", async () => {
const store = createStoreWithAgentForExec({ taskId: undefined });
const selectNextTaskForAgent = vi.fn().mockResolvedValue(makeInboxSelection("FN-INBOX", "todo"));
mockTaskStore = createMockTaskStore({ selectNextTaskForAgent });
const mockSession = createMockAgentSession();
mockedCreateKbAgent.mockResolvedValue({ session: mockSession as any });
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
await monitor.executeHeartbeat({
agentId: "agent-001",
source: "on_demand",
taskId: "FN-EXPLICIT",
});
expect(selectNextTaskForAgent).not.toHaveBeenCalled();
expect(mockTaskStore.getTask).toHaveBeenCalledWith("FN-EXPLICIT");
});
it("agent's existing taskId takes precedence over inbox selection", async () => {
const store = createStoreWithAgentForExec({ taskId: "FN-EXISTING" });
const selectNextTaskForAgent = vi.fn().mockResolvedValue(makeInboxSelection("FN-INBOX", "todo"));
mockTaskStore = createMockTaskStore({ selectNextTaskForAgent });
const mockSession = createMockAgentSession();
mockedCreateKbAgent.mockResolvedValue({ session: mockSession as any });
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
expect(selectNextTaskForAgent).not.toHaveBeenCalled();
expect(mockTaskStore.getTask).toHaveBeenCalledWith("FN-EXISTING");
});
it("when inbox returns null, heartbeat completes with no_assignment", async () => {
const store = createStoreWithAgentForExec({ taskId: undefined });
const selectNextTaskForAgent = vi.fn().mockResolvedValue(null);
mockTaskStore = createMockTaskStore({ selectNextTaskForAgent });
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
expect(selectNextTaskForAgent).toHaveBeenCalledWith("agent-001");
expect(result.resultJson).toEqual({ reason: "no_assignment" });
});
it("records inbox selection metadata in resultJson", async () => {
const store = createStoreWithAgentForExec({ taskId: undefined });
mockTaskStore = createMockTaskStore({
selectNextTaskForAgent: vi.fn().mockResolvedValue(makeInboxSelection("FN-INBOX", "todo")),
getTask: vi.fn().mockResolvedValue({
id: "FN-INBOX",
title: "Inbox Task",
description: "Inbox-selected task",
prompt: "",
steps: [],
column: "todo",
dependencies: [],
log: [],
attachments: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
} as unknown as TaskDetail),
});
const mockSession = createMockAgentSession();
mockedCreateKbAgent.mockResolvedValue({ session: mockSession as any });
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
expect(result.resultJson).toEqual(expect.objectContaining({
reason: "inbox_selected",
priority: "todo",
taskId: "FN-INBOX",
}));
});
it("supports in-progress inbox selections before todo", async () => {
const store = createStoreWithAgentForExec({ taskId: undefined });
mockTaskStore = createMockTaskStore({
selectNextTaskForAgent: vi.fn().mockResolvedValue(makeInboxSelection("FN-RESUME", "in_progress")),
getTask: vi.fn().mockResolvedValue({
id: "FN-RESUME",
title: "Resume task",
description: "Resume in-progress work",
prompt: "",
steps: [],
column: "in-progress",
dependencies: [],
log: [],
attachments: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
} as unknown as TaskDetail),
});
const mockSession = createMockAgentSession();
mockedCreateKbAgent.mockResolvedValue({ session: mockSession as any });
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
expect(mockTaskStore.getTask).toHaveBeenCalledWith("FN-RESUME");
expect(result.resultJson).toEqual(expect.objectContaining({
reason: "inbox_selected",
priority: "in_progress",
taskId: "FN-RESUME",
}));
});
it("gracefully skips inbox selection when checkoutTask throws", async () => {
const store = createStoreWithAgentForExec({ taskId: undefined });
const selectNextTaskForAgent = vi.fn().mockResolvedValue(makeInboxSelection("FN-CHECKOUT", "todo"));
const checkoutTask = vi.fn().mockRejectedValue(new Error("Task is already checked out"));
mockTaskStore = createMockTaskStore({
selectNextTaskForAgent,
checkoutTask: checkoutTask as any,
});
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
expect(selectNextTaskForAgent).toHaveBeenCalledWith("agent-001");
expect(checkoutTask).toHaveBeenCalledWith("FN-CHECKOUT", "agent-001");
expect(result.resultJson).toEqual({ reason: "no_assignment" });
expect(mockedCreateKbAgent).not.toHaveBeenCalled();
});
});
describe("execution", () => {
it("creates session with correct system prompt and tools", async () => {
const store = createStoreWithAgentForExec();

48
packages/engine/src/agent-heartbeat.ts
View File

@@ -17,7 +17,7 @@
* - onTerminated: Called when an unresponsive agent is terminated
*/
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent } from "@fusion/core";
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask } from "@fusion/core";
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
import { Type, type Static } from "@mariozechner/pi-ai";
import { createTaskCreateTool, createTaskLogTool, taskCreateParams } from "./agent-tools.js";
@@ -656,8 +656,38 @@ export class HeartbeatMonitor {
return (await this.store.getRunDetail(agentId, run.id))!;
}
// Resolve task assignment
const taskId = explicitTaskId ?? agent.taskId;
// Resolve task assignment (explicit override → existing assignment → inbox-lite selection)
let taskId = explicitTaskId ?? agent.taskId;
let inboxSelection: InboxTask | null = null;
if (!taskId) {
inboxSelection = await taskStore.selectNextTaskForAgent(agentId);
if (inboxSelection) {
taskId = inboxSelection.task.id;
heartbeatLog.log(`Inbox selected task ${taskId} (priority: ${inboxSelection.priority}) for agent ${agentId}`);
// Persist assignment to AgentStore so subsequent runs retain linkage.
if (agent.taskId !== taskId) {
await this.store.assignTask(agentId, taskId);
}
// FN-1253 compatibility: if checkout API is available on TaskStore,
// try to claim the lease. On conflict, skip this task gracefully.
const checkoutTask = (taskStore as TaskStore & {
checkoutTask?: (taskId: string, agentId: string) => Promise<unknown>;
}).checkoutTask;
if (typeof checkoutTask === "function") {
try {
await checkoutTask.call(taskStore, taskId, agentId);
} catch {
heartbeatLog.log(`Task ${taskId} already checked out — skipping`);
taskId = undefined;
inboxSelection = null;
}
}
}
}
if (taskId && run.contextSnapshot?.taskId !== taskId) {
const updatedRun: AgentHeartbeatRun = {
...run,
@@ -822,10 +852,20 @@ export class HeartbeatMonitor {
await flushAgentLogger();
// Complete run successfully
const completionResultJson: Record<string, unknown> = {
summary: heartbeatSummary,
toolCallCount,
};
if (inboxSelection) {
completionResultJson.reason = "inbox_selected";
completionResultJson.priority = inboxSelection.priority;
completionResultJson.taskId = taskId;
}
await this.completeRun(agentId, run.id, {
status: "completed",
usageJson: { inputTokens: 0, outputTokens: estimatedOutputTokens, cachedTokens: 0 },
resultJson: { summary: heartbeatSummary, toolCallCount },
resultJson: completionResultJson,
stdoutExcerpt: stdoutExcerpt || undefined,
});