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

feat(FN-1085): align agent routing and runtime contracts

Browse Source 
- Harden core AgentStore lifecycle behavior and heartbeat runtime integration paths
- Align dashboard agent APIs, server routes, and agent UI flows with the updated contract
- Tighten CLI agent/message command routing and validate payload handling semantics
- Expand test coverage across core, dashboard, engine, and CLI for route, heartbeat, and instruction regressions
This commit is contained in:
gsxdsm
2026-04-08 00:44:33 -07:00
parent 92e6aa2b49
commit 07697b2f5b
19 changed files with 942 additions and 169 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
packages/dashboard/src/__tests__/routes-agent-runs.test.ts
View File

@@ -262,7 +262,6 @@ describe("Agent runs routes (without HeartbeatMonitor)", () => {
describe("Agent runs routes (with HeartbeatMonitor)", () => {
let store: MockStore;
let app: ReturnType<typeof import("../server.js").createServer>;
let mockStartRun: ReturnType<typeof vi.fn>;
let mockExecuteHeartbeat: ReturnType<typeof vi.fn>;
beforeEach(async () => {
@@ -271,14 +270,12 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
mockListAgents.mockResolvedValue([]);
mockGetActiveHeartbeatRun.mockResolvedValue(null);
mockStartRun = vi.fn();
mockExecuteHeartbeat = vi.fn();
store = new MockStore();
const { createServer } = await import("../server.js");
app = createServer(store as any, {
heartbeatMonitor: {
startRun: mockStartRun,
executeHeartbeat: mockExecuteHeartbeat,
},
});
@@ -289,9 +286,8 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
});
describe("POST /api/agents/:id/runs", () => {
it("delegates to heartbeatMonitor.startRun when available", async () => {
it("delegates to heartbeatMonitor.executeHeartbeat when available", async () => {
const mockRun = createMockRun({ invocationSource: "on_demand", triggerDetail: "Triggered from dashboard" });
mockStartRun.mockResolvedValue(mockRun);
mockExecuteHeartbeat.mockResolvedValue({ ...mockRun, status: "completed" });
const response = await request(
@@ -303,26 +299,20 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
);
expect(response.status).toBe(201);
expect(mockStartRun).toHaveBeenCalledWith("agent-001", {
source: "on_demand",
triggerDetail: "Triggered from dashboard",
contextSnapshot: {
wakeReason: "on_demand",
triggerDetail: "Triggered from dashboard",
},
});
// executeHeartbeat should be called fire-and-forget
expect(mockExecuteHeartbeat).toHaveBeenCalledWith({
agentId: "agent-001",
source: "on_demand",
triggerDetail: "Triggered from dashboard",
taskId: undefined,
contextSnapshot: {
wakeReason: "on_demand",
triggerDetail: "Triggered from dashboard",
},
});
});
it("passes custom source and triggerDetail to heartbeatMonitor", async () => {
const mockRun = createMockRun();
mockStartRun.mockResolvedValue(mockRun);
mockExecuteHeartbeat.mockResolvedValue(mockRun);
await request(
@@ -333,9 +323,11 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
{ "content-type": "application/json" },
);
expect(mockStartRun).toHaveBeenCalledWith("agent-001", {
expect(mockExecuteHeartbeat).toHaveBeenCalledWith({
agentId: "agent-001",
source: "timer",
triggerDetail: "Scheduled run",
taskId: undefined,
contextSnapshot: {
wakeReason: "timer",
triggerDetail: "Scheduled run",
@@ -349,7 +341,6 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
const mockEvent = { id: "evt-001", agentId: "agent-001", status: "ok", timestamp: "2026-01-01T00:00:00.000Z" };
mockRecordHeartbeat.mockResolvedValue(mockEvent);
const mockRun = createMockRun({ invocationSource: "on_demand" });
mockStartRun.mockResolvedValue(mockRun);
mockExecuteHeartbeat.mockResolvedValue(mockRun);
const response = await request(
@@ -361,8 +352,15 @@ describe("Agent runs routes (with HeartbeatMonitor)", () => {
);
expect(response.status).toBe(200);
expect(mockStartRun).toHaveBeenCalled();
expect(mockExecuteHeartbeat).toHaveBeenCalled();
expect(mockExecuteHeartbeat).toHaveBeenCalledWith({
agentId: "agent-001",
source: "on_demand",
triggerDetail: "Triggered from heartbeat",
contextSnapshot: {
wakeReason: "on_demand",
triggerDetail: "Triggered from heartbeat",
},
});
// Response should include both event and run
expect((response.body as any).event).toBeDefined();
expect((response.body as any).run).toBeDefined();

125
packages/dashboard/src/routes.test.ts
View File

@@ -7876,6 +7876,131 @@ describe("POST /workflow-step-templates/:id/create", () => {
});
});
describe("Agent create/update routes", () => {
let tempDir: string;
let fusionDir: string;
let agentId: string;
beforeEach(async () => {
tempDir = mkdtempSync(join(tmpdir(), "kb-routes-agents-fields-"));
fusionDir = join(tempDir, ".fusion");
mkdirSync(fusionDir, { recursive: true });
const { AgentStore } = await import("@fusion/core");
const agentStore = new AgentStore({ rootDir: fusionDir });
await agentStore.init();
const agent = await agentStore.createAgent({
name: "Initial Agent",
role: "executor",
});
agentId = agent.id;
});
afterEach(() => {
rmSync(tempDir, { recursive: true, force: true });
});
function buildAgentApp() {
const store = createMockStore({
getFusionDir: vi.fn().mockReturnValue(fusionDir),
} as any);
const app = express();
app.use(express.json());
app.use("/api", createApiRoutes(store));
return app;
}
it("POST /api/agents accepts all AgentCreateInput fields", async () => {
const res = await REQUEST(
buildAgentApp(),
"POST",
"/api/agents",
JSON.stringify({
name: "Full Agent",
role: "reviewer",
metadata: { team: "qa" },
title: "QA Reviewer",
icon: "🧪",
reportsTo: agentId,
runtimeConfig: { heartbeatIntervalMs: 60000 },
permissions: { read: true },
instructionsPath: "docs/reviewer.md",
instructionsText: "Check test quality.",
}),
{ "Content-Type": "application/json" },
);
expect(res.status).toBe(201);
expect(res.body).toMatchObject({
name: "Full Agent",
role: "reviewer",
metadata: { team: "qa" },
title: "QA Reviewer",
icon: "🧪",
reportsTo: agentId,
runtimeConfig: { heartbeatIntervalMs: 60000 },
permissions: { read: true },
instructionsPath: "docs/reviewer.md",
instructionsText: "Check test quality.",
});
});
it("PATCH /api/agents/:id accepts all AgentUpdateInput fields", async () => {
const res = await REQUEST(
buildAgentApp(),
"PATCH",
`/api/agents/${agentId}`,
JSON.stringify({
name: "Updated Agent",
role: "engineer",
metadata: { area: "infra" },
title: "Infra Engineer",
icon: "⚙️",
reportsTo: "agent-parent",
runtimeConfig: { heartbeatTimeoutMs: 120000 },
pauseReason: "manual",
permissions: { deploy: true },
totalInputTokens: 42,
totalOutputTokens: 21,
instructionsPath: "agents/infra.md",
instructionsText: "Focus on reliability.",
}),
{ "Content-Type": "application/json" },
);
expect(res.status).toBe(200);
expect(res.body).toMatchObject({
id: agentId,
name: "Updated Agent",
role: "engineer",
metadata: { area: "infra" },
title: "Infra Engineer",
icon: "⚙️",
reportsTo: "agent-parent",
runtimeConfig: { heartbeatTimeoutMs: 120000 },
pauseReason: "manual",
permissions: { deploy: true },
totalInputTokens: 42,
totalOutputTokens: 21,
instructionsPath: "agents/infra.md",
instructionsText: "Focus on reliability.",
});
});
it("POST /api/agents/:id/state returns 400 for invalid state transitions", async () => {
const res = await REQUEST(
buildAgentApp(),
"POST",
`/api/agents/${agentId}/state`,
JSON.stringify({ state: "terminated" }),
{ "Content-Type": "application/json" },
);
expect(res.status).toBe(400);
expect(res.body.error).toContain("Invalid state transition");
});
});
describe("POST /api/agents/:id/runs", () => {
let tempDir: string;
let fusionDir: string;

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

@@ -6893,14 +6893,69 @@ Output ONLY the prompt text (no markdown, no explanations).`;
}
});
function validateAgentInstructionsPayload(
res: Response,
instructionsPath: unknown,
instructionsText: unknown,
): boolean {
if (instructionsPath !== undefined && instructionsPath !== null && instructionsPath !== "") {
if (typeof instructionsPath !== "string") {
res.status(400).json({ error: "instructionsPath must be a string" });
return false;
}
if (instructionsPath.length > 500) {
res.status(400).json({ error: "instructionsPath must be at most 500 characters" });
return false;
}
if (instructionsPath.includes("..")) {
res.status(400).json({ error: "instructionsPath must not contain parent directory traversal (..)" });
return false;
}
const isAbsoluteUnix = instructionsPath.startsWith("/");
const isAbsoluteWindows = /^[A-Za-z]:[\\/]/.test(instructionsPath);
if (isAbsoluteUnix || isAbsoluteWindows) {
res.status(400).json({ error: "instructionsPath must be a project-relative path" });
return false;
}
if (!instructionsPath.endsWith(".md")) {
res.status(400).json({ error: "instructionsPath must end in .md" });
return false;
}
}
if (instructionsText !== undefined && instructionsText !== null && instructionsText !== "") {
if (typeof instructionsText !== "string") {
res.status(400).json({ error: "instructionsText must be a string" });
return false;
}
if (instructionsText.length > 50000) {
res.status(400).json({ error: "instructionsText must be at most 50,000 characters" });
return false;
}
}
return true;
}
/**
* POST /api/agents
* Create a new agent.
* Body: { name: string, role: string, metadata?: object }
*/
router.post("/agents", async (req, res) => {
try {
const { name, role, metadata } = req.body;
const {
name,
role,
metadata,
title,
icon,
reportsTo,
runtimeConfig,
permissions,
instructionsPath,
instructionsText,
} = req.body ?? {};
if (!name || typeof name !== "string") {
res.status(400).json({ error: "name is required" });
return;
@@ -6909,16 +6964,58 @@ Output ONLY the prompt text (no markdown, no explanations).`;
res.status(400).json({ error: "role is required" });
return;
}
if (metadata !== undefined && (typeof metadata !== "object" || metadata === null || Array.isArray(metadata))) {
res.status(400).json({ error: "metadata must be an object" });
return;
}
if (title !== undefined && title !== null && typeof title !== "string") {
res.status(400).json({ error: "title must be a string" });
return;
}
if (icon !== undefined && icon !== null && typeof icon !== "string") {
res.status(400).json({ error: "icon must be a string" });
return;
}
if (reportsTo !== undefined && reportsTo !== null && typeof reportsTo !== "string") {
res.status(400).json({ error: "reportsTo must be a string" });
return;
}
if (runtimeConfig !== undefined && (typeof runtimeConfig !== "object" || runtimeConfig === null || Array.isArray(runtimeConfig))) {
res.status(400).json({ error: "runtimeConfig must be an object" });
return;
}
if (permissions !== undefined && (typeof permissions !== "object" || permissions === null || Array.isArray(permissions))) {
res.status(400).json({ error: "permissions must be an object" });
return;
}
if (!validateAgentInstructionsPayload(res, instructionsPath, instructionsText)) {
return;
}
const scopedStore = await getScopedStore(req);
const { AgentStore } = await import("@fusion/core");
const agentStore = new AgentStore({ rootDir: scopedStore.getFusionDir() });
await agentStore.init();
const agent = await agentStore.createAgent({ name, role: role as import("@fusion/core").AgentCapability, metadata });
const agent = await agentStore.createAgent({
name,
role: role as import("@fusion/core").AgentCapability,
metadata,
title: title ?? undefined,
icon: icon ?? undefined,
reportsTo: reportsTo ?? undefined,
runtimeConfig,
permissions,
instructionsPath: instructionsPath ?? undefined,
instructionsText: instructionsText ?? undefined,
});
res.status(201).json(agent);
} catch (err: any) {
res.status(500).json({ error: err.message });
if (err.message?.includes("required") || err.message?.includes("cannot be empty")) {
res.status(400).json({ error: err.message });
} else {
res.status(500).json({ error: err.message });
}
}
});
@@ -7079,18 +7176,119 @@ Output ONLY the prompt text (no markdown, no explanations).`;
*/
router.patch("/agents/:id", async (req, res) => {
try {
const { name, role, metadata, runtimeConfig } = req.body;
const body = req.body ?? {};
const updates: import("@fusion/core").AgentUpdateInput = {};
if ("name" in body) {
if (body.name !== null && typeof body.name !== "string") {
res.status(400).json({ error: "name must be a string" });
return;
}
updates.name = body.name ?? undefined;
}
if ("role" in body) {
if (body.role !== null && typeof body.role !== "string") {
res.status(400).json({ error: "role must be a string" });
return;
}
updates.role = body.role ?? undefined;
}
if ("metadata" in body) {
if (body.metadata !== null && (typeof body.metadata !== "object" || Array.isArray(body.metadata))) {
res.status(400).json({ error: "metadata must be an object" });
return;
}
updates.metadata = body.metadata ?? undefined;
}
if ("title" in body) {
if (body.title !== null && typeof body.title !== "string") {
res.status(400).json({ error: "title must be a string" });
return;
}
updates.title = body.title ?? undefined;
}
if ("icon" in body) {
if (body.icon !== null && typeof body.icon !== "string") {
res.status(400).json({ error: "icon must be a string" });
return;
}
updates.icon = body.icon ?? undefined;
}
if ("reportsTo" in body) {
if (body.reportsTo !== null && typeof body.reportsTo !== "string") {
res.status(400).json({ error: "reportsTo must be a string" });
return;
}
updates.reportsTo = body.reportsTo ?? undefined;
}
if ("pauseReason" in body) {
if (body.pauseReason !== null && typeof body.pauseReason !== "string") {
res.status(400).json({ error: "pauseReason must be a string" });
return;
}
updates.pauseReason = body.pauseReason ?? undefined;
}
if ("runtimeConfig" in body) {
if (body.runtimeConfig !== null && (typeof body.runtimeConfig !== "object" || Array.isArray(body.runtimeConfig))) {
res.status(400).json({ error: "runtimeConfig must be an object" });
return;
}
updates.runtimeConfig = body.runtimeConfig ?? undefined;
}
if ("permissions" in body) {
if (body.permissions !== null && (typeof body.permissions !== "object" || Array.isArray(body.permissions))) {
res.status(400).json({ error: "permissions must be an object" });
return;
}
updates.permissions = body.permissions ?? undefined;
}
if ("totalInputTokens" in body) {
if (body.totalInputTokens !== null && typeof body.totalInputTokens !== "number") {
res.status(400).json({ error: "totalInputTokens must be a number" });
return;
}
updates.totalInputTokens = body.totalInputTokens ?? undefined;
}
if ("totalOutputTokens" in body) {
if (body.totalOutputTokens !== null && typeof body.totalOutputTokens !== "number") {
res.status(400).json({ error: "totalOutputTokens must be a number" });
return;
}
updates.totalOutputTokens = body.totalOutputTokens ?? undefined;
}
if (!validateAgentInstructionsPayload(res, body.instructionsPath, body.instructionsText)) {
return;
}
if ("instructionsPath" in body) {
updates.instructionsPath = body.instructionsPath ?? undefined;
}
if ("instructionsText" in body) {
updates.instructionsText = body.instructionsText ?? undefined;
}
const scopedStore = await getScopedStore(req);
const { AgentStore } = await import("@fusion/core");
const agentStore = new AgentStore({ rootDir: scopedStore.getFusionDir() });
await agentStore.init();
const agent = await agentStore.updateAgent(req.params.id, { name, role, metadata, runtimeConfig });
const agent = await agentStore.updateAgent(req.params.id, updates);
res.json(agent);
} catch (err: any) {
if (err.message?.includes("not found")) {
res.status(404).json({ error: err.message });
} else if (err.message?.includes("cannot be empty")) {
res.status(400).json({ error: err.message });
} else {
res.status(500).json({ error: err.message });
}
@@ -7104,38 +7302,9 @@ Output ONLY the prompt text (no markdown, no explanations).`;
*/
router.patch("/agents/:id/instructions", async (req, res) => {
try {
const { instructionsPath, instructionsText } = req.body;
// Validate instructionsPath if provided
if (instructionsPath !== undefined && instructionsPath !== "") {
if (typeof instructionsPath !== "string") {
res.status(400).json({ error: "instructionsPath must be a string" });
return;
}
if (instructionsPath.length > 500) {
res.status(400).json({ error: "instructionsPath must be at most 500 characters" });
return;
}
if (instructionsPath.includes("..")) {
res.status(400).json({ error: "instructionsPath must not contain parent directory traversal (..)" });
return;
}
if (!instructionsPath.endsWith(".md")) {
res.status(400).json({ error: "instructionsPath must end in .md" });
return;
}
}
// Validate instructionsText if provided
if (instructionsText !== undefined && instructionsText !== "") {
if (typeof instructionsText !== "string") {
res.status(400).json({ error: "instructionsText must be a string" });
return;
}
if (instructionsText.length > 50000) {
res.status(400).json({ error: "instructionsText must be at most 50,000 characters" });
return;
}
const { instructionsPath, instructionsText } = req.body ?? {};
if (!validateAgentInstructionsPayload(res, instructionsPath, instructionsText)) {
return;
}
const scopedStore = await getScopedStore(req);
@@ -7143,7 +7312,10 @@ Output ONLY the prompt text (no markdown, no explanations).`;
const agentStore = new AgentStore({ rootDir: scopedStore.getFusionDir() });
await agentStore.init();
const agent = await agentStore.updateAgent(req.params.id, { instructionsPath, instructionsText });
const agent = await agentStore.updateAgent(req.params.id, {
instructionsPath: instructionsPath ?? undefined,
instructionsText: instructionsText ?? undefined,
});
res.json(agent);
} catch (err: any) {
if (err.message?.includes("not found")) {
@@ -7177,7 +7349,7 @@ Output ONLY the prompt text (no markdown, no explanations).`;
} catch (err: any) {
if (err.message?.includes("not found")) {
res.status(404).json({ error: err.message });
} else if (err.message?.includes("Invalid state transition") || err.message?.includes("Cannot transition from terminated")) {
} else if (/invalid state transition/i.test(err.message ?? "")) {
res.status(400).json({ error: err.message });
} else {
res.status(500).json({ error: err.message });
@@ -7316,7 +7488,7 @@ Output ONLY the prompt text (no markdown, no explanations).`;
* Body: { status?: "ok"|"missed"|"recovered", triggerExecution?: boolean }
*
* When triggerExecution is true AND HeartbeatMonitor is available,
* also starts a heartbeat run after recording the heartbeat event.
* also executes a heartbeat run after recording the heartbeat event.
*/
router.post("/agents/:id/heartbeat", async (req, res) => {
try {
@@ -7332,18 +7504,14 @@ Output ONLY the prompt text (no markdown, no explanations).`;
// Optionally trigger execution
let run: import("@fusion/core").AgentHeartbeatRun | undefined;
if (triggerExecution && hasHeartbeatExecutor && heartbeatMonitor) {
run = await heartbeatMonitor.startRun(req.params.id, {
source: "on_demand",
triggerDetail: "Triggered from heartbeat",
});
// Fire-and-forget execution
void heartbeatMonitor.executeHeartbeat({
run = await heartbeatMonitor.executeHeartbeat({
agentId: req.params.id,
source: "on_demand",
triggerDetail: "Triggered from heartbeat",
}).catch((err: any) => {
console.error(`[heartbeat] Background execution failed for ${req.params.id}:`, err.message);
contextSnapshot: {
wakeReason: "on_demand",
triggerDetail: "Triggered from heartbeat",
},
});
}
@@ -7407,11 +7575,9 @@ Output ONLY the prompt text (no markdown, no explanations).`;
* Manually start a heartbeat run for an agent.
* Body: { source?: HeartbeatInvocationSource, triggerDetail?: string, taskId?: string }
*
* When HeartbeatMonitor is available, delegates to startRun() which enriches
* the run with execution context, transitions the agent to "running", and
* fires the onRunStarted event. The route returns the run immediately with
* "active" status while execution continues in the background via
* executeHeartbeat() fire-and-forget.
* When HeartbeatMonitor is available, delegates to executeHeartbeat() with
* a structured wake context snapshot. This ensures a single authoritative run
* record is created and fully completed without duplicate startRun calls.
*
* Returns 409 Conflict if the agent already has an active run.
*/
@@ -7443,21 +7609,13 @@ Output ONLY the prompt text (no markdown, no explanations).`;
return;
}
// Delegate to HeartbeatMonitor for enriched run creation
const run = await heartbeatMonitor.startRun(req.params.id, {
source: invocationSource,
triggerDetail: trigger,
contextSnapshot,
});
// Fire-and-forget execution in the background
void heartbeatMonitor.executeHeartbeat({
// Execute heartbeat end-to-end (single run record, no duplicate startRun call)
const run = await heartbeatMonitor.executeHeartbeat({
agentId: req.params.id,
source: invocationSource,
triggerDetail: trigger,
taskId,
}).catch((err: any) => {
console.error(`[heartbeat] Background execution failed for ${req.params.id}:`, err.message);
contextSnapshot,
});
res.status(201).json(run);
@@ -7479,7 +7637,7 @@ Output ONLY the prompt text (no markdown, no explanations).`;
// Enrich with invocation source, trigger detail, and context snapshot
(run as any).invocationSource = invocationSource;
(run as any).triggerDetail = triggerDetail;
(run as any).triggerDetail = trigger;
(run as any).contextSnapshot = contextSnapshot;
await agentStore.saveRun(run);

2
packages/dashboard/src/server.ts
View File

@@ -53,7 +53,7 @@ export interface ServerOptions {
/** Optional HeartbeatMonitor for triggering agent execution runs */
heartbeatMonitor?: {
startRun(agentId: string, options?: { source: import("@fusion/core").HeartbeatInvocationSource; triggerDetail?: string; contextSnapshot?: Record<string, unknown> }): Promise<import("@fusion/core").AgentHeartbeatRun>;
executeHeartbeat(options: { agentId: string; source: import("@fusion/core").HeartbeatInvocationSource; triggerDetail?: string; taskId?: string }): Promise<import("@fusion/core").AgentHeartbeatRun>;
executeHeartbeat(options: { agentId: string; source: import("@fusion/core").HeartbeatInvocationSource; triggerDetail?: string; taskId?: string; contextSnapshot?: Record<string, unknown> }): Promise<import("@fusion/core").AgentHeartbeatRun>;
};
}