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

dashboard: route per-project heartbeat execution via engineManager

Browse Source 
ProjectEngineManager runs one engine per registered project, but the
heartbeat handler in register-agent-runtime-routes.ts only used the
single global heartbeatMonitor passed via ServerOptions. That monitor
is bound to the cwd project's engine, so heartbeat triggers for any
secondary project silently no-op'd: the run id was returned but no
execution actually happened.

Add resolveHeartbeatMonitorFor(scopedStore) that walks
engineManager.getAllEngines() and returns the engine whose working
directory matches the request's scoped store, falling back to the
global monitor when its rootDir matches. Use the resolver at every
heartbeat call site (state-pause stop, /agents/:id/heartbeat,
/agents/:id/runs, /agents/:id/runs/stop).
This commit is contained in:
Semih
2026-05-09 15:17:06 +00:00
parent 32db77e4ef
commit accd19b18d
2 changed files with 59 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
packages/dashboard/src/routes.ts
View File

@@ -898,7 +898,49 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
// HeartbeatMonitor for triggering agent execution runs
const heartbeatMonitor = options?.heartbeatMonitor;
const hasHeartbeatExecutor = Boolean(heartbeatMonitor);
const engineManagerForHb = options?.engineManager;
const hasHeartbeatExecutor = Boolean(heartbeatMonitor || engineManagerForHb);
/**
* Resolve the heartbeat monitor that should service requests for the given
* scoped store. Multi-project setups have one engine per project and each
* engine carries its own monitor; the global `heartbeatMonitor` field on
* ServerOptions is bound to a single project (the cwd one). Without this
* resolver, requests for any non-cwd project's agents silently no-op.
*/
function resolveHeartbeatMonitorFor(
scopedStore: TaskStore,
): typeof heartbeatMonitor | undefined {
let storeRoot: string | undefined;
try {
storeRoot = resolve(scopedStore.getRootDir());
} catch {
storeRoot = undefined;
}
if (engineManagerForHb && storeRoot) {
for (const engine of engineManagerForHb.getAllEngines().values()) {
try {
if (resolve(engine.getWorkingDirectory()) !== storeRoot) continue;
} catch {
continue;
}
const hb = engine.getHeartbeatMonitor();
if (!hb) continue;
return {
rootDir: engine.getWorkingDirectory(),
startRun: hb.startRun.bind(hb),
executeHeartbeat: hb.executeHeartbeat.bind(hb),
stopRun: hb.stopRun.bind(hb),
};
}
}
if (heartbeatMonitor && isHeartbeatMonitorForProject(scopedStore)) {
return heartbeatMonitor;
}
return undefined;
}
/**
* Check whether the heartbeatMonitor is bound to the same project as scopedStore.
@@ -2760,6 +2802,7 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
hasHeartbeatExecutor,
heartbeatMonitor,
isHeartbeatMonitorForProject,
resolveHeartbeatMonitorFor,
runExcerptToAgentLogs,
parseRunAuditFilters,
normalizeRunAuditEvent,

36
packages/dashboard/src/routes/register-agent-runtime-routes.ts
View File

@@ -16,6 +16,9 @@ interface AgentRuntimeRouteDeps {
hasHeartbeatExecutor: boolean;
heartbeatMonitor: import("../server.js").ServerOptions["heartbeatMonitor"];
isHeartbeatMonitorForProject: (scopedStore: import("@fusion/core").TaskStore) => boolean;
resolveHeartbeatMonitorFor: (
scopedStore: import("@fusion/core").TaskStore,
) => import("../server.js").ServerOptions["heartbeatMonitor"] | undefined;
runExcerptToAgentLogs: (run: import("@fusion/core").AgentHeartbeatRun) => import("@fusion/core").AgentLogEntry[];
parseRunAuditFilters: (query: Record<string, unknown>) => {
taskId?: string;
@@ -42,6 +45,7 @@ export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRun
hasHeartbeatExecutor,
heartbeatMonitor,
isHeartbeatMonitorForProject,
resolveHeartbeatMonitorFor,
runExcerptToAgentLogs,
parseRunAuditFilters,
normalizeRunAuditEvent,
@@ -410,11 +414,7 @@ export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRun
throw notFound("Agent not found");
}
const projectHeartbeatMonitor = hasHeartbeatExecutor
&& heartbeatMonitor
&& isHeartbeatMonitorForProject(scopedStore)
? heartbeatMonitor
: null;
const projectHeartbeatMonitor = resolveHeartbeatMonitorFor(scopedStore) ?? null;
if (nextState === "paused" && projectHeartbeatMonitor) {
const activeRun = await agentStore.getActiveHeartbeatRun(agentId);
@@ -784,8 +784,9 @@ export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRun
// Optionally trigger execution
let run: import("@fusion/core").AgentHeartbeatRun | undefined;
if (triggerExecution && hasHeartbeatExecutor && heartbeatMonitor && isHeartbeatMonitorForProject(scopedStore)) {
run = await heartbeatMonitor.executeHeartbeat({
const projectHbForRun = resolveHeartbeatMonitorFor(scopedStore);
if (triggerExecution && projectHbForRun) {
run = await projectHbForRun.executeHeartbeat({
agentId: req.params.id,
source: "on_demand",
triggerDetail: "Triggered from heartbeat",
@@ -925,16 +926,9 @@ export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRun
contextSnapshot.triggeringCommentType = normalizedTriggeringCommentType;
}
if (hasHeartbeatExecutor && heartbeatMonitor) {
// Check for existing active run
const { store: scopedStore } = await getProjectContext(req);
// Guard: heartbeatMonitor is bound to a specific project root directory.
// Reject when the scoped store belongs to a different project.
if (!isHeartbeatMonitorForProject(scopedStore)) {
throw new ApiError(400, "Agent execution is only available for the server's primary project. The heartbeat monitor is not bound to this project.");
}
const { store: scopedStore } = await getProjectContext(req);
const projectHbForRun = resolveHeartbeatMonitorFor(scopedStore);
if (projectHbForRun) {
const { AgentStore: AgentStoreClass } = await import("@fusion/core");
const agentStore = new AgentStoreClass({ rootDir: scopedStore.getFusionDir() });
await agentStore.init();
@@ -950,7 +944,7 @@ export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRun
}
// Execute heartbeat end-to-end (single run record, no duplicate startRun call)
const run = await heartbeatMonitor.executeHeartbeat({
const run = await projectHbForRun.executeHeartbeat({
agentId: req.params.id,
source: invocationSource,
triggerDetail: trigger,
@@ -963,7 +957,6 @@ export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRun
res.status(201).json(run);
} else {
// Fallback: record-only behavior without HeartbeatMonitor
const { store: scopedStore } = await getProjectContext(req);
const { AgentStore } = await import("@fusion/core");
const agentStore = new AgentStore({ rootDir: scopedStore.getFusionDir() });
await agentStore.init();
@@ -1023,8 +1016,9 @@ export function registerAgentRuntimeRoutes(ctx: ApiRoutesContext, deps: AgentRun
return;
}
if (hasHeartbeatExecutor && heartbeatMonitor && isHeartbeatMonitorForProject(scopedStore)) {
await heartbeatMonitor.stopRun(req.params.id);
const projectHbForStop = resolveHeartbeatMonitorFor(scopedStore);
if (projectHbForStop) {
await projectHbForStop.stopRun(req.params.id);
} else {
const existingRun = await agentStore.getRunDetail(req.params.id, activeRun.id);
if (existingRun) {