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

feat(FN-4401): complete remaining auto-claim snapshot integration

Browse Source 
Fusion-Task-Id: FN-4401
Fusion-Task-Lineage: c1b6c497-b22c-48d5-b1c8-299877bf09ac
This commit is contained in:
Fusion
2026-05-14 00:58:06 -07:00
committed by gsxdsm
parent b8f5a237bd
commit f860618657
16 changed files with 299 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
packages/engine/src/__tests__/auto-claim-snapshot.test.ts
View File

@@ -12,15 +12,15 @@ function makeTask(overrides: Partial<Task> & Pick<Task, "id">): Task {
createdAt: overrides.createdAt ?? "2026-01-01T00:00:00.000Z",
updatedAt: overrides.updatedAt ?? "2026-01-01T00:00:00.000Z",
dependencies: overrides.dependencies ?? [],
labels: overrides.labels ?? [],
comments: overrides.comments ?? [],
activityLog: overrides.activityLog ?? [],
metadata: overrides.metadata ?? {},
steps: overrides.steps ?? [],
currentStep: overrides.currentStep ?? 0,
log: overrides.log ?? [],
assignedAgentId: overrides.assignedAgentId,
checkedOutBy: overrides.checkedOutBy,
paused: overrides.paused,
columnMovedAt: overrides.columnMovedAt,
} as Task;
} as unknown as Task;
}
describe("AutoClaimSnapshotManager", () => {

45
packages/engine/src/__tests__/heartbeat-executor.test.ts
View File

@@ -896,6 +896,51 @@ describe("executeHeartbeat", () => {
expect(store.claimTaskForAgent).not.toHaveBeenCalled();
});
it("reuses one snapshot rebuild across concurrent no-task heartbeats", async () => {
const listTasks = vi.fn().mockResolvedValue([
{
id: "FN-CANDIDATE",
description: "executor reliability follow-up",
title: "Executor reliability",
prompt: "",
steps: [],
column: "todo",
dependencies: [],
log: [],
attachments: [],
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
} as unknown as TaskDetail,
]);
mockTaskStore = createMockTaskStore({ listTasks });
const store = createStoreWithAgentForExec({ taskId: undefined, soul: "executor reliability owner" });
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
await Promise.all([
monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" }),
monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" }),
]);
expect(listTasks).toHaveBeenCalledTimes(1);
});
it("omits candidate section when autoClaimCandidatesInPrompt resolves to zero", async () => {
const store = createStoreWithAgentForExec({
taskId: undefined,
soul: "executor reliability owner",
runtimeConfig: { autoClaimRelevantTasks: true, autoClaimCandidatesInPrompt: 0 },
});
const mockSession = createMockAgentSession();
mockedCreateFnAgent.mockResolvedValue({ session: mockSession as any });
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
await monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" });
const executionPrompt = mockSession.prompt.mock.calls.at(-1)?.[0] as string;
expect(executionPrompt).toContain("auto-claim relevant tasks: disabled (prompt-suppressed)");
expect(executionPrompt).not.toContain("Open unowned tasks you may auto-claim");
});
it("agent WITH instructionsText but no task creates session and completes successfully", async () => {
const store = createStoreWithAgentForExec({ taskId: undefined, instructionsText: "Monitor task board and create follow-up tasks" });
const mockSession = createMockAgentSession();

53
packages/engine/src/__tests__/scheduler-auto-claim-invalidation.test.ts Normal file
View File

@@ -0,0 +1,53 @@
import { describe, expect, it, vi } from "vitest";
import type { TaskStore } from "@fusion/core";
import { Scheduler } from "../scheduler.js";
function createStore() {
const listeners = new Map<string, ((payload: unknown) => void)[]>();
const on = vi.fn((event: string, listener: (payload: unknown) => void) => {
const existing = listeners.get(event) ?? [];
existing.push(listener);
listeners.set(event, existing);
});
const store = {
on,
off: vi.fn(),
} as unknown as TaskStore;
const emit = (event: string, payload: unknown) => {
for (const listener of listeners.get(event) ?? []) {
listener(payload);
}
};
return { store, emit };
}
describe("Scheduler auto-claim snapshot invalidation", () => {
it("invalidates on task:created and task:updated", () => {
const invalidate = vi.fn();
const { store, emit } = createStore();
new Scheduler(store, { snapshotManager: { invalidate } as any });
emit("task:created", { task: { id: "FN-1" } });
emit("task:updated", { id: "FN-1" });
expect(invalidate).toHaveBeenCalledWith("task:created");
expect(invalidate).toHaveBeenCalledWith("task:updated");
});
it("invalidates task:moved only when todo is source or destination", () => {
const invalidate = vi.fn();
const { store, emit } = createStore();
new Scheduler(store, { snapshotManager: { invalidate } as any });
emit("task:moved", { task: { id: "FN-1" }, from: "todo", to: "in-progress" });
emit("task:moved", { task: { id: "FN-2" }, from: "in-progress", to: "todo" });
emit("task:moved", { task: { id: "FN-3" }, from: "in-review", to: "done" });
expect(invalidate).toHaveBeenCalledWith("task:moved:todo->in-progress");
expect(invalidate).toHaveBeenCalledWith("task:moved:in-progress->todo");
expect(invalidate).toHaveBeenCalledTimes(2);
});
});

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

@@ -18,6 +18,7 @@
*/
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask, RunMutationContext, Settings, AgentConfigRevision, ReflectionStore, ChatStore, ChatRoom, ChatRoomMessage, AgentMemoryInclusionMode } from "@fusion/core";
import { AutoClaimSnapshotManager, type AutoClaimCandidate } from "./auto-claim-snapshot.js";
import { ApprovalRequestStore, buildExecutionMemoryInstructions, isEphemeralAgent, hasAgentIdentity, resolveEffectiveAgentPermissionPolicy, canAgentTakeImplementationTask, canAgentTakeImplementationTaskForExplicitRouting, resolvePersistAgentThinkingLog, resolveAgentMemoryInclusionMode } from "@fusion/core";
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
import { Type, type Static } from "@mariozechner/pi-ai";
@@ -89,6 +90,8 @@ export interface HeartbeatMonitorOptions {
onRunStarted?: (agentId: string, run: AgentHeartbeatRun) => void;
/** Callback when a run completes */
onRunCompleted?: (agentId: string, run: AgentHeartbeatRun) => void;
/** Project-wide auto-claim snapshot manager. */
snapshotManager?: AutoClaimSnapshotManager;
/** TaskStore for fn_task_create and fn_task_log tools during heartbeat execution.
* When not provided, executeHeartbeat() will throw. */
taskStore?: TaskStore;
@@ -232,7 +235,16 @@ function isAutoClaimRelevantTasksEnabled(agent: Agent): boolean {
return runtimeConfig.autoClaimRelevantTasks !== false;
}
type RelevanceScorableTask = Pick<TaskDetail, "title" | "description">;
function resolveAutoClaimCandidatesInPromptLimit(agent: Agent, settings?: Settings): number {
const runtimeConfig = (agent.runtimeConfig ?? {}) as AgentHeartbeatConfig;
const perAgent = runtimeConfig.autoClaimCandidatesInPrompt;
const projectValue = settings?.autoClaimCandidatesInPrompt;
const raw = typeof perAgent === "number" ? perAgent : (typeof projectValue === "number" ? projectValue : 5);
const integer = Number.isFinite(raw) ? Math.trunc(raw) : 5;
return Math.max(0, Math.min(10, integer));
}
type RelevanceScorableTask = { title?: string | null; description: string };
const agentSoulWordsCache = new Map<string, { soulSnapshot: string; words: readonly string[] }>();
@@ -649,6 +661,7 @@ export class HeartbeatMonitor {
private reflectionService?: AgentReflectionService;
private selfImproveService?: SelfImproveServiceLike;
private approvalRequestStore?: ApprovalRequestStore;
private snapshotManager?: AutoClaimSnapshotManager;
private trackedAgents: Map<string, TrackedAgent> = new Map();
private agentStartLocks: Map<string, Promise<unknown>> = new Map();
@@ -677,6 +690,7 @@ export class HeartbeatMonitor {
this.reflectionStore = options.reflectionStore;
this.reflectionService = options.reflectionService;
this.selfImproveService = options.selfImproveService;
this.snapshotManager = options.snapshotManager ?? (this.taskStore ? new AutoClaimSnapshotManager({ taskStore: this.taskStore }) : undefined);
}
getChatStore(): ChatStore | undefined {
@@ -1727,58 +1741,38 @@ export class HeartbeatMonitor {
engineRunContext.taskId = taskId;
}
let autoClaimCandidates: TaskDetail[] = [];
let autoClaimCandidates: AutoClaimCandidate[] = [];
const autoClaimEnabled = isAutoClaimRelevantTasksEnabled(agent);
if (!taskId && canRunNoTaskHeartbeat && autoClaimEnabled) {
const listTasks = (taskStore as TaskStore & { listTasks?: (options?: { slim?: boolean }) => Promise<TaskDetail[]> }).listTasks;
if (typeof listTasks === "function") {
try {
const allTasks = await listTasks.call(taskStore, { slim: true });
const tasksById = new Map(allTasks.map((candidate) => [candidate.id, candidate]));
const openCandidates = allTasks
.filter((candidate) => (
candidate.column === "todo"
&& candidate.paused !== true
&& !candidate.assignedAgentId
&& !candidate.checkedOutBy
&& candidate.dependencies.every((dependencyId) => {
const dependency = tasksById.get(dependencyId);
return dependency?.column === "done" || dependency?.column === "archived";
})
))
.sort((a, b) => {
const aSortAt = a.columnMovedAt ?? a.createdAt;
const bSortAt = b.columnMovedAt ?? b.createdAt;
return aSortAt.localeCompare(bSortAt);
})
.slice(0, 10);
const roleCompatibleCandidates = openCandidates.filter((candidate) => canAgentTakeImplementationTask(agent, candidate));
const skippedIncompatibleCount = openCandidates.length - roleCompatibleCandidates.length;
if (skippedIncompatibleCount > 0) {
heartbeatLog.log(
`Agent ${agentId} (role=${agent.role}) skipped auto-claim of ${skippedIncompatibleCount} implementation task(s) — only executor agents may claim implementation work`,
);
}
autoClaimCandidates = roleCompatibleCandidates;
const ranked = roleCompatibleCandidates
.map((candidate) => ({ candidate, score: taskRelevanceScore(agent, candidate as TaskDetail) }))
.filter((entry) => entry.score > 0)
.sort((a, b) => b.score - a.score || (a.candidate.columnMovedAt ?? a.candidate.createdAt).localeCompare(b.candidate.columnMovedAt ?? b.candidate.createdAt));
if (ranked.length > 0) {
const claimResult = await this.store.claimTaskForAgent(agentId, ranked[0].candidate.id, runContext);
if (claimResult.ok) {
taskId = ranked[0].candidate.id;
heartbeatLog.log(`Agent ${agentId} auto-claimed relevant task ${taskId}`);
} else {
heartbeatLog.log(`Agent ${agentId} auto-claim skipped (${claimResult.reason})`);
}
}
} catch (autoClaimError) {
heartbeatLog.warn(`Auto-claim scan failed for ${agentId}: ${autoClaimError instanceof Error ? autoClaimError.message : String(autoClaimError)}`);
if (!taskId && canRunNoTaskHeartbeat && autoClaimEnabled && this.snapshotManager) {
try {
const snapshot = await this.snapshotManager.getSnapshot();
const roleCompatibleCandidates = snapshot.tasks.filter((candidate) => canAgentTakeImplementationTask(agent, candidate));
const skippedIncompatibleCount = snapshot.tasks.length - roleCompatibleCandidates.length;
if (skippedIncompatibleCount > 0) {
heartbeatLog.log(
`Agent ${agentId} (role=${agent.role}) skipped auto-claim of ${skippedIncompatibleCount} implementation task(s) — only executor agents may claim implementation work`,
);
}
autoClaimCandidates = roleCompatibleCandidates;
const ranked = roleCompatibleCandidates
.map((candidate) => ({ candidate, score: candidate.baseScore + taskRelevanceScore(agent, candidate) }))
.filter((entry) => entry.score > 0)
.sort((a, b) => b.score - a.score || (a.candidate.columnMovedAt ?? a.candidate.createdAt).localeCompare(b.candidate.columnMovedAt ?? b.candidate.createdAt));
if (ranked.length > 0) {
const winnerId = ranked[0].candidate.id;
const winner = await taskStore.getTask(winnerId);
const claimResult = await this.store.claimTaskForAgent(agentId, winner.id, runContext);
if (claimResult.ok) {
taskId = winner.id;
heartbeatLog.log(`Agent ${agentId} auto-claimed relevant task ${taskId}`);
} else {
heartbeatLog.log(`Agent ${agentId} auto-claim skipped (${claimResult.reason})`);
}
}
} catch (autoClaimError) {
heartbeatLog.warn(`Auto-claim scan failed for ${agentId}: ${autoClaimError instanceof Error ? autoClaimError.message : String(autoClaimError)}`);
}
}
if (!taskId) {
@@ -2319,13 +2313,16 @@ export class HeartbeatMonitor {
);
}
const candidateLines = autoClaimCandidates.length > 0
const promptCandidateLimit = resolveAutoClaimCandidatesInPromptLimit(agent, heartbeatModelSettings);
const candidateLines = promptCandidateLimit > 0
? [
"",
"Open Task Candidates (auto-claim scan):",
...autoClaimCandidates.slice(0, 10).map((candidate) => `- ${candidate.id}: ${candidate.title ?? candidate.description.slice(0, 80)}`),
...autoClaimCandidates
.slice(0, promptCandidateLimit)
.map((candidate) => `- ${candidate.id}: ${candidate.title ?? candidate.descriptionFirstLine}`),
]
: ["", "Open Task Candidates (auto-claim scan): none found"];
: [];
executionPrompt = [
`Heartbeat execution for agent "${agent.name}" (ID: ${agent.id})`,
@@ -2343,7 +2340,7 @@ export class HeartbeatMonitor {
`- assigned task: none`,
`- pending messages: ${pendingMessages.length}`,
`- pending room messages: ${pendingRoomMessages.total}`,
`- auto-claim relevant tasks: ${autoClaimEnabled ? "enabled" : "disabled"}`,
`- auto-claim relevant tasks: ${autoClaimEnabled ? (promptCandidateLimit === 0 ? "disabled (prompt-suppressed)" : "enabled") : "disabled"}`,
"",
"Treat this wake delta as the highest-priority change for this heartbeat.",
"This is an autonomous heartbeat run (manual or automatic): re-anchor on",
@@ -2385,6 +2382,7 @@ export class HeartbeatMonitor {
"",
"Call fn_heartbeat_done when finished.",
].join("\n");
heartbeatLog.log(`[auto-claim-prompt] agent=${agentId} chars=${executionPrompt.length} count=${Math.min(promptCandidateLimit, autoClaimCandidates.length)}`);
} else {
// Task-scoped heartbeat: agent has an assigned task
const taskTitle = taskDetail!.title ?? taskDetail!.description.slice(0, 100);

5
packages/engine/src/runtimes/in-process-runtime.ts
View File

@@ -19,6 +19,7 @@ import { TaskExecutor, type TaskExecutorOptions } from "../executor.js";
import { WorktreePool, isGitRepository } from "../worktree-pool.js";
import { AgentSemaphore } from "../concurrency.js";
import { HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext } from "../agent-heartbeat.js";
import { AutoClaimSnapshotManager } from "../auto-claim-snapshot.js";
import { RoutineRunner, type RoutineRunnerOptions } from "../routine-runner.js";
import { RoutineScheduler } from "../routine-scheduler.js";
import { createAiPromptExecutor } from "../cron-runner.js";
@@ -309,6 +310,8 @@ export class InProcessRuntime
getExecutingTaskIds: () => this.executor?.getExecutingTaskIds() ?? new Set<string>(),
});
const autoClaimSnapshotManager = new AutoClaimSnapshotManager({ taskStore: this.taskStore });
this.scheduler = new Scheduler(this.taskStore, {
maxConcurrent: this.config.maxConcurrent,
maxWorktrees: this.config.maxWorktrees,
@@ -332,6 +335,7 @@ export class InProcessRuntime
const mappedPath = await this.centralCore.getProjectNodePath(this.config.projectId, nodeId);
return validateProjectNodeMapping({ nodeId, mappedPath });
},
snapshotManager: autoClaimSnapshotManager,
});
@@ -464,6 +468,7 @@ export class InProcessRuntime
reflectionStore: reflectionStoreForService,
reflectionService,
selfImproveService,
snapshotManager: autoClaimSnapshotManager,
onMissed: (agentId, reason) => {
runtimeLog.warn(`Agent ${agentId} missed heartbeat: ${reason}`);
},

8
packages/engine/src/scheduler.ts
View File

@@ -23,6 +23,7 @@ import { applyUnavailableNodePolicy } from "./node-routing-policy.js";
import type { NodeDispatchValidationResult } from "./node-dispatch-validation.js";
import type { MeshLeaseManager } from "./mesh-lease-manager.js";
import { selectPermanentAgentForTask } from "./agent-assignment.js";
import type { AutoClaimSnapshotManager } from "./auto-claim-snapshot.js";
/**
* Check whether two sets of file scope paths overlap.
@@ -145,6 +146,8 @@ export interface SchedulerOptions {
nodeHealthMonitor?: import("./node-health-monitor.js").NodeHealthMonitor;
/** Optional dispatch validator used to block dispatch on configuration issues before health policy checks. */
validateNodeDispatch?: (nodeId: string) => Promise<NodeDispatchValidationResult>;
/** Optional shared auto-claim snapshot manager for invalidation on task mutations. */
snapshotManager?: AutoClaimSnapshotManager;
}
/**
@@ -202,6 +205,7 @@ export class Scheduler {
* This reduces latency from up to 15 seconds to near-instant.
*/
this.store.on("task:created", () => {
this.options.snapshotManager?.invalidate("task:created");
schedulerLog.log("Task created — triggering scheduling");
this.schedule();
});
@@ -241,6 +245,9 @@ export class Scheduler {
* update feature status and potentially activate next pending slice.
*/
this.store.on("task:moved", async ({ task, from, to }) => {
if (from === "todo" || to === "todo") {
this.options.snapshotManager?.invalidate(`task:moved:${from}->${to}`);
}
// PR Monitoring
if (this.options.prMonitor) {
if (to === "in-review" && task.prInfo) {
@@ -351,6 +358,7 @@ export class Scheduler {
* Also detects task-level unpause transitions and triggers immediate scheduling.
*/
this.store.on("task:updated", (task) => {
this.options.snapshotManager?.invalidate("task:updated");
// Track mission failure signals before moveTask clears failure metadata.
if (task.sliceId && task.column === "in-progress" && task.status === "failed") {
this.failedTaskIds.add(task.id);