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

feat(FN-3249): apply run status icon color style (+3 more)

Browse Source 
Commits merged:
- fix(FN-3249): apply run status icon color style
- fix(FN-3249): address workflow revision dashboard accessibility feedback
- feat(FN-3249): complete Step 5 — document assigned execution ownership
- feat(FN-3249): complete Step 2 — durable assigned-agent execution ownership

Files changed:
.changeset/fn-3249-assigned-executor-ownership.md  |   5 +
 docs/agents.md                                     |  15 +++
 packages/core/src/__tests__/agent-store.test.ts    |  28 +++++
 packages/core/src/agent-store.ts                   |  34 ++++--
 .../dashboard/app/components/AgentDetailView.css   |   5 +
 .../dashboard/app/components/AgentDetailView.tsx   |   2 +-
 packages/dashboard/app/components/AgentsView.tsx   |   2 +
 .../runtimes/__tests__/in-process-runtime.test.ts  | 125 ++++++++++++++++++---
 packages/engine/src/runtimes/in-process-runtime.ts | 108 +++++++++++-------
 9 files changed, 260 insertions(+), 64 deletions(-)

Fusion-Task-Id: FN-3249
This commit is contained in:
Fusion
2026-05-04 22:36:04 -07:00
committed by gsxdsm
parent 583a8de019
commit 8565cdcff7
9 changed files with 260 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
packages/engine/src/runtimes/__tests__/in-process-runtime.test.ts
View File

@@ -596,7 +596,35 @@ describe("InProcessRuntime", () => {
});
}, 30000);
it("creates runtime task-worker agents with disabled heartbeat metadata and running state", async () => {
it("reuses assigned durable agent as execution owner without creating a task-worker", async () => {
await runtime.start();
const store = getAgentStore(runtime);
const durable = await store.createAgent({ name: "Durable Exec", role: "executor" });
const createAgentSpy = vi.spyOn(store, "createAgent");
const assignTaskSpy = vi.spyOn(store, "assignTask");
const syncLinkSpy = vi.spyOn(store, "syncExecutionTaskLink");
const executorOptions = mockExecutorCtor.mock.calls.at(-1)?.[0] as {
onStart?: (task: Task, worktreePath: string) => void;
};
executorOptions.onStart?.({ id: "FN-1661", assignedAgentId: durable.id } as Task, join(testDir, "worktree-FN-1661"));
await vi.waitFor(async () => {
const updated = await store.getAgent(durable.id);
expect(updated?.taskId).toBe("FN-1661");
expect(updated?.state).toBe("running");
});
expect(syncLinkSpy).toHaveBeenCalledWith(durable.id, "FN-1661");
expect(assignTaskSpy).not.toHaveBeenCalledWith(durable.id, "FN-1661");
expect(createAgentSpy).not.toHaveBeenCalledWith(expect.objectContaining({ name: "executor-FN-1661" }));
const agents = await store.listAgents({ includeEphemeral: true });
expect(agents.some((agent: Agent) => agent.name === "executor-FN-1661")).toBe(false);
}, 30000);
it("falls back to runtime task-worker agents for unassigned tasks", async () => {
await runtime.start();
const store = getAgentStore(runtime);
@@ -635,25 +663,29 @@ describe("InProcessRuntime", () => {
expect(assignTaskSpy.mock.invocationCallOrder[0]).toBeLessThan(updateStateSpy.mock.invocationCallOrder[0]);
}, 30000);
it("does not create duplicate task-worker agents when onStart fires twice for one task", async () => {
it("falls back to runtime task-worker when assignedAgentId points to ephemeral agent", async () => {
await runtime.start();
const store = getAgentStore(runtime);
const ephemeral = await store.createAgent({
name: "Spawned Child",
role: "executor",
metadata: { agentKind: "task-worker", managedBy: "task-executor" },
runtimeConfig: { enabled: false },
});
const executorOptions = mockExecutorCtor.mock.calls.at(-1)?.[0] as {
onStart?: (task: Task, worktreePath: string) => void;
};
executorOptions.onStart?.({ id: "FN-DUP-ONSTART" } as Task, join(testDir, "worktree-FN-DUP-ONSTART"));
executorOptions.onStart?.({ id: "FN-DUP-ONSTART" } as Task, join(testDir, "worktree-FN-DUP-ONSTART"));
executorOptions.onStart?.({ id: "FN-1662", assignedAgentId: ephemeral.id } as Task, join(testDir, "worktree-FN-1662"));
await vi.waitFor(async () => {
const agents = await store.listAgents({ includeEphemeral: true });
const matching = agents.filter((agent: Agent) => agent.name === "executor-FN-DUP-ONSTART");
expect(matching).toHaveLength(1);
expect(agents.some((agent: Agent) => agent.name === "executor-FN-1662")).toBe(true);
});
}, 30000);
it("does not wake executeHeartbeat for runtime task-worker assignment events", async () => {
it("does not wake executeHeartbeat for runtime ownership sync of durable assigned agents", async () => {
await runtime.start();
const monitor = runtime.getHeartbeatMonitor();
@@ -664,22 +696,56 @@ describe("InProcessRuntime", () => {
.spyOn(heartbeatMonitor, "executeHeartbeat")
.mockResolvedValue(executeResult);
const store = getAgentStore(runtime);
const durable = await store.createAgent({ name: "Owned Exec", role: "executor" });
const executorOptions = mockExecutorCtor.mock.calls.at(-1)?.[0] as {
onStart?: (task: Task, worktreePath: string) => void;
};
executorOptions.onStart?.({ id: "FN-2001" } as Task, join(testDir, "worktree-FN-2001"));
const store = getAgentStore(runtime);
executorOptions.onStart?.({ id: "FN-2001", assignedAgentId: durable.id } as Task, join(testDir, "worktree-FN-2001"));
await vi.waitFor(async () => {
const agents = await store.listAgents({ includeEphemeral: true });
expect(agents.some((agent: Agent) => agent.name === "executor-FN-2001")).toBe(true);
const updated = await store.getAgent(durable.id);
expect(updated?.taskId).toBe("FN-2001");
});
await new Promise((resolve) => setTimeout(resolve, 25));
expect(executeSpy).not.toHaveBeenCalled();
}, 30000);
it("cleans up durable execution owner on completion without deleting agent", async () => {
vi.useFakeTimers();
try {
await runtime.start();
const store = getAgentStore(runtime);
const durable = await store.createAgent({ name: "Durable Cleanup", role: "executor" });
const deleteAgentSpy = vi.spyOn(store, "deleteAgent").mockResolvedValue(undefined);
const executorOptions = mockExecutorCtor.mock.calls.at(-1)?.[0] as {
onStart?: (task: Task, worktreePath: string) => void;
onComplete?: (task: Task) => void;
};
executorOptions.onStart?.({ id: "FN-DURABLE-1", assignedAgentId: durable.id } as Task, join(testDir, "worktree-FN-DURABLE-1"));
await vi.waitFor(async () => {
const updated = await store.getAgent(durable.id);
expect(updated?.taskId).toBe("FN-DURABLE-1");
});
executorOptions.onComplete?.({ id: "FN-DURABLE-1" } as Task);
await vi.advanceTimersByTimeAsync(5000);
const updated = await store.getAgent(durable.id);
expect(updated?.state).toBe("terminated");
expect(updated?.taskId).toBeUndefined();
expect(deleteAgentSpy).not.toHaveBeenCalledWith(durable.id);
} finally {
vi.useRealTimers();
}
}, 30000);
it("auto-deletes task-worker agent on task completion after 5 second delay", async () => {
vi.useFakeTimers();
@@ -720,6 +786,39 @@ describe("InProcessRuntime", () => {
}
}, 30000);
it("cleans up durable execution owner on error without deleting agent", async () => {
vi.useFakeTimers();
try {
await runtime.start();
const store = getAgentStore(runtime);
const durable = await store.createAgent({ name: "Durable Error", role: "executor" });
const deleteAgentSpy = vi.spyOn(store, "deleteAgent").mockResolvedValue(undefined);
const executorOptions = mockExecutorCtor.mock.calls.at(-1)?.[0] as {
onStart?: (task: Task, worktreePath: string) => void;
onError?: (task: Task, error: Error) => void;
};
executorOptions.onStart?.({ id: "FN-DURABLE-2", assignedAgentId: durable.id } as Task, join(testDir, "worktree-FN-DURABLE-2"));
await vi.waitFor(async () => {
const updated = await store.getAgent(durable.id);
expect(updated?.taskId).toBe("FN-DURABLE-2");
});
executorOptions.onError?.({ id: "FN-DURABLE-2" } as Task, new Error("boom"));
await vi.advanceTimersByTimeAsync(5000);
const updated = await store.getAgent(durable.id);
expect(updated?.state).toBe("terminated");
expect(updated?.taskId).toBeUndefined();
expect(deleteAgentSpy).not.toHaveBeenCalledWith(durable.id);
} finally {
vi.useRealTimers();
}
}, 30000);
it("auto-deletes task-worker agent on task error after 5 second delay", async () => {
vi.useFakeTimers();

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

@@ -92,8 +92,8 @@ export class InProcessRuntime
private agentStore?: AgentStore;
private heartbeatMonitor?: HeartbeatMonitor;
private triggerScheduler?: HeartbeatTriggerScheduler;
/** Maps task IDs to agent IDs for lifecycle tracking */
private taskAgentMap = new Map<string, string>();
/** Maps task IDs to execution owner metadata for lifecycle tracking */
private taskAgentMap = new Map<string, { agentId: string; ephemeral: boolean }>();
private lastActivityAt: string = new Date().toISOString();
private pluginRunner?: PluginRunner;
private pluginStore?: PluginStore;
@@ -373,53 +373,77 @@ export class InProcessRuntime
onStart: (task, worktreePath) => {
this.recordActivity();
runtimeLog.log(`Started executing task ${task.id} in ${worktreePath}`);
// Create a runtime-managed task worker agent for lifecycle tracking.
// These workers are not heartbeat-managed dashboard agents, so mark them
// explicitly and disable heartbeat triggers/timers.
if (this.agentStore) {
if (this.taskAgentMap.has(task.id)) {
runtimeLog.warn(`Skipping task-worker creation for ${task.id}: agent already exists (${this.taskAgentMap.get(task.id)})`);
return;
}
if (!this.agentStore) return;
this.taskAgentMap.set(task.id, "creating");
this.agentStore.createAgent({
name: `executor-${task.id}`,
role: "executor",
metadata: {
agentKind: "task-worker",
taskWorker: true,
managedBy: "task-executor",
},
runtimeConfig: {
enabled: false,
},
}).then(async (agent: { id: string }) => {
this.taskAgentMap.set(task.id, agent.id);
void (async () => {
try {
const assignedAgentId = task.assignedAgentId;
if (assignedAgentId) {
const assignedAgent = await this.agentStore!.getAgent(assignedAgentId);
if (assignedAgent && !isEphemeralAgent(assignedAgent)) {
this.taskAgentMap.set(task.id, { agentId: assignedAgent.id, ephemeral: false });
await this.agentStore!.syncExecutionTaskLink(assignedAgent.id, task.id);
const currentState = assignedAgent.state;
if (currentState !== "running") {
if (currentState !== "active") {
await this.agentStore!.updateAgentState(assignedAgent.id, "active");
}
await this.agentStore!.updateAgentState(assignedAgent.id, "running");
}
return;
}
}
if (this.taskAgentMap.has(task.id)) {
runtimeLog.warn(`Skipping task-worker creation for ${task.id}: task already has execution owner`);
return;
}
// Create a runtime-managed task worker agent for lifecycle tracking.
// These workers are not heartbeat-managed dashboard agents, so mark them
// explicitly and disable heartbeat triggers/timers.
const agent = await this.agentStore!.createAgent({
name: `executor-${task.id}`,
role: "executor",
metadata: {
agentKind: "task-worker",
taskWorker: true,
managedBy: "task-executor",
},
runtimeConfig: {
enabled: false,
},
});
this.taskAgentMap.set(task.id, { agentId: agent.id, ephemeral: true });
await this.agentStore!.assignTask(agent.id, task.id);
await this.agentStore!.updateAgentState(agent.id, "active");
await this.agentStore!.updateAgentState(agent.id, "running");
}).catch((err: unknown) => {
this.taskAgentMap.delete(task.id);
runtimeLog.warn(`Failed to create agent for task ${task.id}:`, err);
});
}
} catch (err: unknown) {
runtimeLog.warn(`Failed to initialize execution owner for task ${task.id}:`, err);
}
})();
},
onComplete: (task) => {
this.recordActivity();
runtimeLog.log(`Completed task ${task.id}`);
this.recordTaskCompletion(task.id, true);
// Update agent state to terminated (completed)
const agentId = this.taskAgentMap.get(task.id);
if (agentId && this.agentStore) {
// Register pending deletion before flipping to terminated so
// agent:stateChanged listener doesn't schedule duplicate cleanup.
this.pendingEphemeralDeletions.add(agentId);
const owner = this.taskAgentMap.get(task.id);
if (owner && this.agentStore) {
const { agentId, ephemeral } = owner;
if (ephemeral) {
this.pendingEphemeralDeletions.add(agentId);
}
void this.agentStore.updateAgentState(agentId, "terminated").catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to update agent ${agentId} state to terminated (completion): ${msg}`);
});
void this.agentStore.syncExecutionTaskLink(agentId, undefined).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to clear execution task link for agent ${agentId} on completion: ${msg}`);
});
this.taskAgentMap.delete(task.id);
if (!ephemeral) return;
// Auto-delete the task-worker agent after a short delay so the UI
// can observe the terminal state before the agent is removed.
const timerId = setTimeout(async () => {
@@ -459,16 +483,22 @@ export class InProcessRuntime
}
// Update agent state to terminated (failed)
const agentId = this.taskAgentMap.get(task.id);
if (agentId && this.agentStore) {
// Register pending deletion before flipping to terminated so
// agent:stateChanged listener doesn't schedule duplicate cleanup.
this.pendingEphemeralDeletions.add(agentId);
const owner = this.taskAgentMap.get(task.id);
if (owner && this.agentStore) {
const { agentId, ephemeral } = owner;
if (ephemeral) {
this.pendingEphemeralDeletions.add(agentId);
}
void this.agentStore.updateAgentState(agentId, "terminated").catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to update agent ${agentId} state to terminated (error): ${msg}`);
});
void this.agentStore.syncExecutionTaskLink(agentId, undefined).catch((err: unknown) => {
const msg = err instanceof Error ? err.message : String(err);
runtimeLog.warn(`Failed to clear execution task link for agent ${agentId} on error: ${msg}`);
});
this.taskAgentMap.delete(task.id);
if (!ephemeral) return;
// Auto-delete the task-worker agent after a short delay so the UI
// can observe the terminal state before the agent is removed.
const timerId = setTimeout(async () => {