feat(FN-1058): add heartbeat execution with agent tool support
- Extract shared agent tools (bash, read, write, edit) from executor into agent-tools.ts for reuse - Implement heartbeat execution in HeartbeatMonitor with configurable interval and task_done support - Wire heartbeat execution into InProcessRuntime lifecycle (start/stop) - Add comprehensive test suite covering execution, timeouts, concurrency, and error handling - Add heartbeat-related log constants to logger
This commit is contained in:
gsxdsm
@@ -1,6 +1,31 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import { HeartbeatMonitor, type AgentSession } from "./agent-heartbeat.js";
|
||||
import type { AgentStore } from "@fusion/core";
|
||||
import { HeartbeatMonitor, type AgentSession, type HeartbeatExecutionOptions, HEARTBEAT_SYSTEM_PROMPT } from "./agent-heartbeat.js";
|
||||
import type { AgentStore, AgentHeartbeatRun, TaskStore, TaskDetail, Agent } from "@fusion/core";
|
||||
|
||||
// Mock logger to suppress noise in test output
|
||||
vi.mock("./logger.js", () => {
|
||||
const createMockLogger = () => ({
|
||||
log: vi.fn(),
|
||||
warn: vi.fn(),
|
||||
error: vi.fn(),
|
||||
});
|
||||
return {
|
||||
createLogger: vi.fn(() => createMockLogger()),
|
||||
heartbeatLog: createMockLogger(),
|
||||
};
|
||||
});
|
||||
|
||||
// Mock pi.ts for executeHeartbeat tests
|
||||
vi.mock("./pi.js", () => ({
|
||||
createKbAgent: vi.fn(),
|
||||
promptWithFallback: vi.fn(async (session: any, prompt: string) => {
|
||||
await session.prompt(prompt);
|
||||
}),
|
||||
}));
|
||||
|
||||
// Import the mocked functions for test control
|
||||
import { createKbAgent } from "./pi.js";
|
||||
const mockedCreateKbAgent = vi.mocked(createKbAgent);
|
||||
|
||||
// Mock store factory
|
||||
function createMockStore(overrides: Partial<AgentStore> = {}): AgentStore {
|
||||
@@ -719,4 +744,575 @@ describe("HeartbeatMonitor", () => {
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
// ── Heartbeat Execution Tests ──────────────────────────────────────────
|
||||
|
||||
describe("executeHeartbeat", () => {
|
||||
let mockTaskStore: TaskStore;
|
||||
let mockAgent: Agent;
|
||||
|
||||
// Helper: create a mock session returned by createKbAgent
|
||||
function createMockAgentSession() {
|
||||
return {
|
||||
prompt: vi.fn().mockResolvedValue(undefined),
|
||||
dispose: vi.fn(),
|
||||
subscribe: vi.fn(),
|
||||
model: { provider: "mock", id: "mock-model" },
|
||||
};
|
||||
}
|
||||
|
||||
// Helper: create a basic mock task store
|
||||
function createMockTaskStore(overrides: Partial<TaskStore> = {}): TaskStore {
|
||||
return {
|
||||
getTask: vi.fn().mockResolvedValue({
|
||||
id: "FN-001",
|
||||
title: "Test Task",
|
||||
description: "Test task description",
|
||||
prompt: "# Test PROMPT.md\nSome content",
|
||||
steps: [],
|
||||
column: "todo",
|
||||
dependencies: [],
|
||||
log: [],
|
||||
attachments: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
} as unknown as TaskDetail),
|
||||
createTask: vi.fn().mockResolvedValue({
|
||||
id: "FN-002",
|
||||
description: "Created task",
|
||||
dependencies: [],
|
||||
column: "triage",
|
||||
}),
|
||||
logEntry: vi.fn().mockResolvedValue({}),
|
||||
...overrides,
|
||||
} as unknown as TaskStore;
|
||||
}
|
||||
|
||||
// Helper: create a mock store that returns a specific agent
|
||||
function createStoreWithAgentForExec(agentData: Partial<Agent> = {}): AgentStore {
|
||||
mockAgent = {
|
||||
id: "agent-001",
|
||||
name: "Test Agent",
|
||||
role: "executor",
|
||||
state: "active",
|
||||
taskId: "FN-001",
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
metadata: {},
|
||||
...agentData,
|
||||
} as Agent;
|
||||
|
||||
// Track saved runs so getRunDetail returns the most recent state
|
||||
const savedRuns: Map<string, AgentHeartbeatRun> = new Map();
|
||||
|
||||
return {
|
||||
recordHeartbeat: vi.fn().mockResolvedValue(undefined),
|
||||
updateAgentState: vi.fn().mockResolvedValue(undefined),
|
||||
updateAgent: vi.fn().mockResolvedValue(undefined),
|
||||
getAgent: vi.fn().mockResolvedValue(mockAgent),
|
||||
startHeartbeatRun: vi.fn().mockResolvedValue({
|
||||
id: "run-001",
|
||||
agentId: "agent-001",
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: null,
|
||||
status: "active",
|
||||
} as AgentHeartbeatRun),
|
||||
saveRun: vi.fn().mockImplementation(async (run: AgentHeartbeatRun) => {
|
||||
savedRuns.set(run.id, run);
|
||||
}),
|
||||
getRunDetail: vi.fn().mockImplementation(async (_agentId: string, runId: string) => {
|
||||
return savedRuns.get(runId) ?? {
|
||||
id: runId,
|
||||
agentId: "agent-001",
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
status: "completed" as const,
|
||||
};
|
||||
}),
|
||||
endHeartbeatRun: vi.fn().mockResolvedValue(undefined),
|
||||
getCachedAgent: vi.fn().mockReturnValue(null),
|
||||
} as unknown as AgentStore;
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
mockTaskStore = createMockTaskStore();
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
describe("dependency validation", () => {
|
||||
it("throws when taskStore is not configured", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
const monitor = new HeartbeatMonitor({ store, rootDir: "/tmp" });
|
||||
|
||||
await expect(
|
||||
monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" })
|
||||
).rejects.toThrow("HeartbeatMonitor not configured for execution (missing taskStore/rootDir)");
|
||||
});
|
||||
|
||||
it("throws when rootDir is not configured", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore });
|
||||
|
||||
await expect(
|
||||
monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" })
|
||||
).rejects.toThrow("HeartbeatMonitor not configured for execution (missing taskStore/rootDir)");
|
||||
});
|
||||
});
|
||||
|
||||
describe("graceful exit", () => {
|
||||
it("completes with no_assignment when agent has no taskId and no explicit taskId", async () => {
|
||||
const store = createStoreWithAgentForExec({ taskId: undefined });
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||
|
||||
expect(result).toBeDefined();
|
||||
expect(result.status).toBe("completed");
|
||||
expect(result.resultJson).toEqual({ reason: "no_assignment" });
|
||||
// Should NOT have created an agent session
|
||||
expect(mockedCreateKbAgent).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("completes with invalid_state when agent state is terminated", async () => {
|
||||
const store = createStoreWithAgentForExec({ state: "terminated" });
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||
|
||||
expect(result).toBeDefined();
|
||||
expect(result.status).toBe("completed");
|
||||
expect(result.resultJson).toEqual({ reason: "invalid_state", state: "terminated" });
|
||||
expect(mockedCreateKbAgent).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("completes as failed when agent not found in store", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
(store.getAgent as ReturnType<typeof vi.fn>).mockResolvedValue(null);
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||
|
||||
expect(result).toBeDefined();
|
||||
expect(result.status).toBe("failed");
|
||||
expect(result.stderrExcerpt).toContain("not found");
|
||||
});
|
||||
|
||||
it("completes with task_not_found when task does not exist", async () => {
|
||||
const store = createStoreWithAgentForExec({ taskId: "FN-MISSING" });
|
||||
mockTaskStore.getTask = vi.fn().mockRejectedValue(new Error("Task FN-MISSING not found"));
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||
|
||||
expect(result).toBeDefined();
|
||||
expect(result.status).toBe("completed");
|
||||
expect(result.resultJson).toEqual({ reason: "task_not_found", taskId: "FN-MISSING" });
|
||||
});
|
||||
});
|
||||
|
||||
describe("execution", () => {
|
||||
it("creates session with correct system prompt and tools", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
const mockSession = createMockAgentSession();
|
||||
mockedCreateKbAgent.mockResolvedValue({
|
||||
session: mockSession as any,
|
||||
});
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp/test" });
|
||||
|
||||
await monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" });
|
||||
|
||||
expect(mockedCreateKbAgent).toHaveBeenCalledOnce();
|
||||
const callArgs = mockedCreateKbAgent.mock.calls[0]![0];
|
||||
expect(callArgs.cwd).toBe("/tmp/test");
|
||||
expect(callArgs.systemPrompt).toBe(HEARTBEAT_SYSTEM_PROMPT);
|
||||
expect(callArgs.tools).toBe("readonly");
|
||||
expect(callArgs.customTools).toHaveLength(3);
|
||||
expect(callArgs.customTools![0]!.name).toBe("task_create");
|
||||
expect(callArgs.customTools![1]!.name).toBe("task_log");
|
||||
expect(callArgs.customTools![2]!.name).toBe("heartbeat_done");
|
||||
});
|
||||
|
||||
it("calls promptWithFallback with task context", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
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: "assignment", triggerDetail: "new task assigned" });
|
||||
|
||||
expect(mockSession.prompt).toHaveBeenCalledOnce();
|
||||
const promptArg = mockSession.prompt.mock.calls[0]![0] as string;
|
||||
expect(promptArg).toContain("agent-001");
|
||||
expect(promptArg).toContain("Test Task");
|
||||
expect(promptArg).toContain("assignment");
|
||||
expect(promptArg).toContain("new task assigned");
|
||||
expect(promptArg).toContain("PROMPT.md");
|
||||
});
|
||||
|
||||
it("completes run with status completed on successful execution", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
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).toBeDefined();
|
||||
expect(result.status).toBe("completed");
|
||||
// Agent state should be set back to active
|
||||
expect(store.updateAgentState).toHaveBeenCalledWith("agent-001", "active");
|
||||
// Session should be disposed
|
||||
expect(mockSession.dispose).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("uses explicit taskId override instead of agent.taskId", async () => {
|
||||
const store = createStoreWithAgentForExec({ taskId: "FN-DEFAULT" });
|
||||
const mockSession = createMockAgentSession();
|
||||
mockedCreateKbAgent.mockResolvedValue({
|
||||
session: mockSession as any,
|
||||
});
|
||||
|
||||
// Override getTask to return a different task
|
||||
mockTaskStore.getTask = vi.fn().mockResolvedValue({
|
||||
id: "FN-OVERRIDE",
|
||||
title: "Override Task",
|
||||
description: "Override description",
|
||||
prompt: "",
|
||||
steps: [],
|
||||
column: "todo",
|
||||
dependencies: [],
|
||||
log: [],
|
||||
attachments: [],
|
||||
createdAt: new Date().toISOString(),
|
||||
updatedAt: new Date().toISOString(),
|
||||
} as unknown as TaskDetail);
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
await monitor.executeHeartbeat({
|
||||
agentId: "agent-001",
|
||||
source: "on_demand",
|
||||
taskId: "FN-OVERRIDE",
|
||||
});
|
||||
|
||||
// Should have fetched the override task
|
||||
expect(mockTaskStore.getTask).toHaveBeenCalledWith("FN-OVERRIDE");
|
||||
// task_log tool should use the override task ID
|
||||
const callArgs = mockedCreateKbAgent.mock.calls[0]![0];
|
||||
const taskLogTool = callArgs.customTools![1]!;
|
||||
expect(taskLogTool.name).toBe("task_log");
|
||||
});
|
||||
|
||||
it("passes model config from agent runtimeConfig to createKbAgent", async () => {
|
||||
const store = createStoreWithAgentForExec({
|
||||
runtimeConfig: { modelProvider: "openai", modelId: "gpt-4o" },
|
||||
});
|
||||
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(mockedCreateKbAgent).toHaveBeenCalledOnce();
|
||||
const callArgs = mockedCreateKbAgent.mock.calls[0]![0];
|
||||
expect(callArgs.defaultProvider).toBe("openai");
|
||||
expect(callArgs.defaultModelId).toBe("gpt-4o");
|
||||
});
|
||||
|
||||
it("passes undefined model when runtimeConfig has no model", async () => {
|
||||
const store = createStoreWithAgentForExec({ runtimeConfig: {} });
|
||||
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" });
|
||||
|
||||
const callArgs = mockedCreateKbAgent.mock.calls[0]![0];
|
||||
expect(callArgs.defaultProvider).toBeUndefined();
|
||||
expect(callArgs.defaultModelId).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("heartbeat_done tool", () => {
|
||||
it("captures summary from heartbeat_done in resultJson", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
let capturedDoneTool: any;
|
||||
const mockSession = createMockAgentSession();
|
||||
mockedCreateKbAgent.mockImplementation(async (opts: any) => {
|
||||
capturedDoneTool = opts.customTools[2]; // heartbeat_done
|
||||
return { session: mockSession as any };
|
||||
});
|
||||
|
||||
// Simulate: when prompt is called, invoke the heartbeat_done tool
|
||||
mockSession.prompt = vi.fn().mockImplementation(async (prompt: string) => {
|
||||
// Simulate the agent calling heartbeat_done
|
||||
const result = await capturedDoneTool.execute("call-1", { summary: "Checked task, all good" });
|
||||
expect(result.content[0].text).toContain("Heartbeat complete");
|
||||
expect(result.content[0].text).toContain("Checked task, all good");
|
||||
});
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
const run = await monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" });
|
||||
|
||||
expect(run.resultJson).toBeDefined();
|
||||
expect((run.resultJson as any).summary).toBe("Checked task, all good");
|
||||
});
|
||||
|
||||
it("works without summary in heartbeat_done", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
let capturedDoneTool: any;
|
||||
const mockSession = createMockAgentSession();
|
||||
mockedCreateKbAgent.mockImplementation(async (opts: any) => {
|
||||
capturedDoneTool = opts.customTools[2];
|
||||
return { session: mockSession as any };
|
||||
});
|
||||
|
||||
mockSession.prompt = vi.fn().mockImplementation(async () => {
|
||||
await capturedDoneTool.execute("call-1", {});
|
||||
});
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
const run = await monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" });
|
||||
|
||||
expect(run.resultJson).toBeDefined();
|
||||
expect((run.resultJson as any).summary).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
describe("task_create tool", () => {
|
||||
it("creates a task in the store when task_create tool is called", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
let capturedCreateTool: any;
|
||||
const mockSession = createMockAgentSession();
|
||||
mockedCreateKbAgent.mockImplementation(async (opts: any) => {
|
||||
capturedCreateTool = opts.customTools[0]; // task_create
|
||||
return { session: mockSession as any };
|
||||
});
|
||||
|
||||
mockSession.prompt = vi.fn().mockImplementation(async () => {
|
||||
await capturedCreateTool.execute("call-1", { description: "Follow-up task" });
|
||||
});
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||
|
||||
expect(mockTaskStore.createTask).toHaveBeenCalledWith({
|
||||
description: "Follow-up task",
|
||||
dependencies: undefined,
|
||||
column: "triage",
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("error handling", () => {
|
||||
it("completes run as failed when createKbAgent throws", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
mockedCreateKbAgent.mockRejectedValue(new Error("Model unavailable"));
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||
|
||||
expect(result).toBeDefined();
|
||||
expect(result.status).toBe("failed");
|
||||
expect(result.stderrExcerpt).toContain("Model unavailable");
|
||||
// Agent state should be set to error
|
||||
expect(store.updateAgentState).toHaveBeenCalledWith("agent-001", "error");
|
||||
});
|
||||
|
||||
it("completes run as failed when promptWithFallback throws", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
const mockSession = createMockAgentSession();
|
||||
mockedCreateKbAgent.mockResolvedValue({
|
||||
session: mockSession as any,
|
||||
});
|
||||
mockSession.prompt = vi.fn().mockRejectedValue(new Error("Prompt failed"));
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||
|
||||
expect(result).toBeDefined();
|
||||
expect(result.status).toBe("failed");
|
||||
expect(result.stderrExcerpt).toContain("Prompt failed");
|
||||
// Session should still be disposed in finally block
|
||||
expect(mockSession.dispose).toHaveBeenCalled();
|
||||
// Agent should be untracked
|
||||
expect(monitor.getTrackedAgents()).not.toContain("agent-001");
|
||||
});
|
||||
});
|
||||
|
||||
describe("concurrency", () => {
|
||||
it("serializes concurrent executeHeartbeat calls for the same agent", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
const mockSession = createMockAgentSession();
|
||||
let promptCallCount = 0;
|
||||
|
||||
// Make prompt take some time to ensure overlap
|
||||
mockSession.prompt = vi.fn().mockImplementation(async () => {
|
||||
promptCallCount++;
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
});
|
||||
|
||||
mockedCreateKbAgent.mockResolvedValue({
|
||||
session: mockSession as any,
|
||||
});
|
||||
|
||||
// We need getRunDetail to return different runs for each call
|
||||
let runCount = 0;
|
||||
const concurrentSavedRuns: Map<string, AgentHeartbeatRun> = new Map();
|
||||
(store.startHeartbeatRun as ReturnType<typeof vi.fn>).mockImplementation(async () => {
|
||||
runCount++;
|
||||
return {
|
||||
id: `run-${runCount}`,
|
||||
agentId: "agent-001",
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: null,
|
||||
status: "active",
|
||||
} as AgentHeartbeatRun;
|
||||
});
|
||||
(store.saveRun as ReturnType<typeof vi.fn>).mockImplementation(async (run: AgentHeartbeatRun) => {
|
||||
concurrentSavedRuns.set(run.id, run);
|
||||
});
|
||||
(store.getRunDetail as ReturnType<typeof vi.fn>).mockImplementation(async (_agentId: string, runId: string) => {
|
||||
return concurrentSavedRuns.get(runId) ?? {
|
||||
id: runId,
|
||||
agentId: "agent-001",
|
||||
startedAt: new Date().toISOString(),
|
||||
endedAt: new Date().toISOString(),
|
||||
status: "completed" as const,
|
||||
};
|
||||
});
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
// Fire two concurrent executions
|
||||
const [result1, result2] = await Promise.all([
|
||||
monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" }),
|
||||
monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" }),
|
||||
]);
|
||||
|
||||
// Both should complete
|
||||
expect(result1).toBeDefined();
|
||||
expect(result2).toBeDefined();
|
||||
// Both should have called prompt (serialized, not concurrent)
|
||||
expect(promptCallCount).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("usage tracking", () => {
|
||||
it("records estimated output tokens in usageJson", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
const mockSession = createMockAgentSession();
|
||||
let onTextCallback: ((delta: string) => void) | undefined;
|
||||
|
||||
mockedCreateKbAgent.mockImplementation(async (opts: any) => {
|
||||
onTextCallback = opts.onText;
|
||||
return { session: mockSession as any };
|
||||
});
|
||||
|
||||
// Simulate text output
|
||||
mockSession.prompt = vi.fn().mockImplementation(async () => {
|
||||
// Simulate 100 chars of output (roughly 25 tokens at 4 chars/token)
|
||||
if (onTextCallback) {
|
||||
onTextCallback("A".repeat(100));
|
||||
}
|
||||
});
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||
|
||||
expect(result.usageJson).toBeDefined();
|
||||
expect(result.usageJson!.inputTokens).toBe(0);
|
||||
expect(result.usageJson!.outputTokens).toBe(25); // 100/4 = 25
|
||||
expect(result.usageJson!.cachedTokens).toBe(0);
|
||||
});
|
||||
|
||||
it("accumulates usage on agent record", async () => {
|
||||
const store = createStoreWithAgentForExec({
|
||||
totalInputTokens: 100,
|
||||
totalOutputTokens: 200,
|
||||
});
|
||||
const mockSession = createMockAgentSession();
|
||||
let onTextCallback: ((delta: string) => void) | undefined;
|
||||
|
||||
mockedCreateKbAgent.mockImplementation(async (opts: any) => {
|
||||
onTextCallback = opts.onText;
|
||||
return { session: mockSession as any };
|
||||
});
|
||||
|
||||
mockSession.prompt = vi.fn().mockImplementation(async () => {
|
||||
if (onTextCallback) {
|
||||
onTextCallback("A".repeat(100));
|
||||
}
|
||||
});
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||
|
||||
// Should update cumulative tokens: 200 + 25 = 225
|
||||
expect(store.updateAgent).toHaveBeenCalledWith("agent-001", {
|
||||
totalInputTokens: 100,
|
||||
totalOutputTokens: 225,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("cleanup", () => {
|
||||
it("disposes session and untracks agent even on error", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
const mockSession = createMockAgentSession();
|
||||
mockedCreateKbAgent.mockResolvedValue({
|
||||
session: mockSession as any,
|
||||
});
|
||||
mockSession.prompt = vi.fn().mockRejectedValue(new Error("Crash"));
|
||||
|
||||
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||
|
||||
await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||
|
||||
// Session disposed
|
||||
expect(mockSession.dispose).toHaveBeenCalled();
|
||||
// Agent untracked
|
||||
expect(monitor.getTrackedAgents()).not.toContain("agent-001");
|
||||
});
|
||||
|
||||
it("disposes session and untracks agent on success", async () => {
|
||||
const store = createStoreWithAgentForExec();
|
||||
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(mockSession.dispose).toHaveBeenCalled();
|
||||
expect(monitor.getTrackedAgents()).not.toContain("agent-001");
|
||||
});
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,8 +1,15 @@
|
||||
/**
|
||||
* HeartbeatMonitor - Runtime monitoring for agent health
|
||||
* HeartbeatMonitor - Runtime monitoring and execution for agents
|
||||
*
|
||||
* Monitors agents via periodic polling and detects missed heartbeats.
|
||||
* Follows the StuckTaskDetector pattern for consistency.
|
||||
* Monitors agents via periodic polling, detects missed heartbeats,
|
||||
* and provides the Paperclip-style heartbeat execution engine:
|
||||
*
|
||||
* wake → check inbox → work → exit
|
||||
*
|
||||
* When `executeHeartbeat()` is called (via API, timer, or assignment),
|
||||
* the system wakes the agent, checks its assigned task from AgentStore,
|
||||
* executes work in a lightweight agent session with `task_create` capability,
|
||||
* records results, and transitions the run to completed.
|
||||
*
|
||||
* Callback pattern (not EventEmitter):
|
||||
* - onMissed: Called when an agent misses its heartbeat
|
||||
@@ -10,7 +17,16 @@
|
||||
* - onTerminated: Called when an unresponsive agent is terminated
|
||||
*/
|
||||
|
||||
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig } from "@fusion/core";
|
||||
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, TaskStore, TaskDetail } from "@fusion/core";
|
||||
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
|
||||
import { Type, type Static } from "@mariozechner/pi-ai";
|
||||
import { createTaskCreateTool, createTaskLogTool } from "./agent-tools.js";
|
||||
import { heartbeatLog } from "./logger.js";
|
||||
|
||||
// Lazy import for pi — avoids pulling the pi SDK into the module graph
|
||||
// when heartbeat execution isn't needed.
|
||||
type CreateKbAgentFn = (options: import("./pi.js").AgentOptions) => Promise<import("./pi.js").AgentResult>;
|
||||
type PromptWithFallbackFn = (session: import("@mariozechner/pi-coding-agent").AgentSession, prompt: string) => Promise<void>;
|
||||
|
||||
/** Resolved per-agent heartbeat config after validation and fallback */
|
||||
interface ResolvedHeartbeatConfig {
|
||||
@@ -42,6 +58,12 @@ export interface HeartbeatMonitorOptions {
|
||||
onRunStarted?: (agentId: string, run: AgentHeartbeatRun) => void;
|
||||
/** Callback when a run completes */
|
||||
onRunCompleted?: (agentId: string, run: AgentHeartbeatRun) => void;
|
||||
/** TaskStore for task_create and task_log tools during heartbeat execution.
|
||||
* When not provided, executeHeartbeat() will throw. */
|
||||
taskStore?: TaskStore;
|
||||
/** Project root directory for agent session CWD.
|
||||
* When not provided, executeHeartbeat() will throw. */
|
||||
rootDir?: string;
|
||||
}
|
||||
|
||||
/** Options for waking up an agent */
|
||||
@@ -54,6 +76,18 @@ export interface WakeupOptions {
|
||||
contextSnapshot?: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/** Options for executing a heartbeat run */
|
||||
export interface HeartbeatExecutionOptions {
|
||||
/** Agent ID to execute heartbeat for */
|
||||
agentId: string;
|
||||
/** What triggered this heartbeat */
|
||||
source: HeartbeatInvocationSource;
|
||||
/** Human-readable trigger detail */
|
||||
triggerDetail?: string;
|
||||
/** Optional task ID override (uses agent.taskId if not set) */
|
||||
taskId?: string;
|
||||
}
|
||||
|
||||
/** Session interface for disposing agent resources */
|
||||
export interface AgentSession {
|
||||
/** Dispose the agent session (stop execution, cleanup resources) */
|
||||
@@ -71,9 +105,31 @@ interface TrackedAgent {
|
||||
sessionIdBefore?: string;
|
||||
}
|
||||
|
||||
/**
|
||||
* System prompt for heartbeat agent sessions.
|
||||
* Instructs the agent to perform a single-pass check on its assigned task
|
||||
* and use `task_create` / `task_log` to record findings or spawn follow-up work.
|
||||
*/
|
||||
export const HEARTBEAT_SYSTEM_PROMPT = `You are a heartbeat agent running in a short execution window.
|
||||
|
||||
Your job:
|
||||
1. Check your assigned task — read the description and PROMPT.md if present.
|
||||
2. Do ONE useful action: analyze, review, create follow-up tasks, or log findings.
|
||||
3. Use task_create to spawn follow-up work, task_log to record observations.
|
||||
4. Call heartbeat_done when finished with an optional summary of what was accomplished.
|
||||
|
||||
Keep work lightweight — this is a single-pass check, not a full implementation run.
|
||||
You have readonly file access plus task_create and task_log tools.`;
|
||||
|
||||
/** Parameter schema for the heartbeat_done tool */
|
||||
const heartbeatDoneParams = Type.Object({
|
||||
summary: Type.Optional(Type.String({ description: "Summary of what was accomplished this heartbeat" })),
|
||||
});
|
||||
|
||||
/**
|
||||
* HeartbeatMonitor monitors agents via periodic polling.
|
||||
* Detects missed heartbeats and auto-terminates unresponsive agents.
|
||||
* Detects missed heartbeats, auto-terminates unresponsive agents,
|
||||
* and provides the Paperclip-style execution engine via executeHeartbeat().
|
||||
*/
|
||||
export class HeartbeatMonitor {
|
||||
private store: AgentStore;
|
||||
@@ -86,6 +142,8 @@ export class HeartbeatMonitor {
|
||||
private onTerminated?: (agentId: string) => void;
|
||||
private onRunStarted?: (agentId: string, run: AgentHeartbeatRun) => void;
|
||||
private onRunCompleted?: (agentId: string, run: AgentHeartbeatRun) => void;
|
||||
private taskStore?: TaskStore;
|
||||
private rootDir?: string;
|
||||
|
||||
private trackedAgents: Map<string, TrackedAgent> = new Map();
|
||||
private agentStartLocks: Map<string, Promise<unknown>> = new Map();
|
||||
@@ -103,6 +161,8 @@ export class HeartbeatMonitor {
|
||||
this.onTerminated = options.onTerminated;
|
||||
this.onRunStarted = options.onRunStarted;
|
||||
this.onRunCompleted = options.onRunCompleted;
|
||||
this.taskStore = options.taskStore;
|
||||
this.rootDir = options.rootDir;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -344,6 +404,199 @@ export class HeartbeatMonitor {
|
||||
return this.trackedAgents.get(agentId)?.lastSeen;
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────
|
||||
// Heartbeat execution (Paperclip wake → check → work → exit)
|
||||
// ─────────────────────────────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Execute a heartbeat run for an agent.
|
||||
*
|
||||
* Implements the Paperclip-style execution model:
|
||||
* 1. Wake — start a heartbeat run record
|
||||
* 2. Check inbox — resolve the agent's assigned task
|
||||
* 3. Work — run a lightweight agent session with readonly tools + task_create/task_log
|
||||
* 4. Exit — record results and complete the run
|
||||
*
|
||||
* Per-agent execution is serialized via `withAgentStartLock` — concurrent calls
|
||||
* for the same agent wait for the previous run to complete.
|
||||
*
|
||||
* @param options - Execution options (agent ID, source, optional task override)
|
||||
* @returns The completed heartbeat run, or null if the monitor isn't configured for execution
|
||||
* @throws Error if taskStore or rootDir are not configured
|
||||
*/
|
||||
async executeHeartbeat(options: HeartbeatExecutionOptions): Promise<AgentHeartbeatRun> {
|
||||
const { agentId, source, triggerDetail, taskId: explicitTaskId } = options;
|
||||
|
||||
// Validate execution dependencies
|
||||
if (!this.taskStore || !this.rootDir) {
|
||||
throw new Error("HeartbeatMonitor not configured for execution (missing taskStore/rootDir)");
|
||||
}
|
||||
const taskStore = this.taskStore;
|
||||
const rootDir = this.rootDir;
|
||||
|
||||
// Serialize per-agent
|
||||
return this.withAgentStartLock(agentId, async () => {
|
||||
heartbeatLog.log(`Executing heartbeat for ${agentId} (source=${source})`);
|
||||
|
||||
// Start run
|
||||
const run = await this.startRun(agentId, { source, triggerDetail });
|
||||
|
||||
try {
|
||||
// Resolve agent
|
||||
const agent = await this.store.getAgent(agentId);
|
||||
if (!agent) {
|
||||
heartbeatLog.warn(`Agent ${agentId} not found — completing run as failed`);
|
||||
await this.completeRun(agentId, run.id, {
|
||||
status: "failed",
|
||||
stderrExcerpt: `Agent ${agentId} not found`,
|
||||
});
|
||||
return (await this.store.getRunDetail(agentId, run.id))!;
|
||||
}
|
||||
|
||||
// Resolve task assignment
|
||||
const taskId = explicitTaskId ?? agent.taskId;
|
||||
if (!taskId) {
|
||||
heartbeatLog.log(`Agent ${agentId} has no task assignment — graceful exit`);
|
||||
await this.completeRun(agentId, run.id, {
|
||||
status: "completed",
|
||||
resultJson: { reason: "no_assignment" },
|
||||
});
|
||||
return (await this.store.getRunDetail(agentId, run.id))!;
|
||||
}
|
||||
|
||||
// Validate agent state
|
||||
const validStates = ["active", "running", "idle"];
|
||||
if (!validStates.includes(agent.state)) {
|
||||
heartbeatLog.log(`Agent ${agentId} state is "${agent.state}" — graceful exit`);
|
||||
await this.completeRun(agentId, run.id, {
|
||||
status: "completed",
|
||||
resultJson: { reason: "invalid_state", state: agent.state },
|
||||
});
|
||||
return (await this.store.getRunDetail(agentId, run.id))!;
|
||||
}
|
||||
|
||||
// Fetch task context
|
||||
let taskDetail: TaskDetail;
|
||||
try {
|
||||
taskDetail = await taskStore.getTask(taskId);
|
||||
} catch {
|
||||
heartbeatLog.warn(`Task ${taskId} not found — graceful exit`);
|
||||
await this.completeRun(agentId, run.id, {
|
||||
status: "completed",
|
||||
resultJson: { reason: "task_not_found", taskId },
|
||||
});
|
||||
return (await this.store.getRunDetail(agentId, run.id))!;
|
||||
}
|
||||
|
||||
// Track usage via callbacks
|
||||
let outputLength = 0;
|
||||
let toolCallCount = 0;
|
||||
let heartbeatSummary: string | undefined;
|
||||
|
||||
// Create heartbeat_done tool
|
||||
const heartbeatDoneTool: ToolDefinition = {
|
||||
name: "heartbeat_done",
|
||||
label: "Heartbeat Done",
|
||||
description: "Signal that the heartbeat execution is complete. Call when finished.",
|
||||
parameters: heartbeatDoneParams,
|
||||
execute: async (_id: string, params: Static<typeof heartbeatDoneParams>) => {
|
||||
if (params.summary) {
|
||||
heartbeatSummary = params.summary;
|
||||
}
|
||||
return {
|
||||
content: [{
|
||||
type: "text" as const,
|
||||
text: `Heartbeat complete.${params.summary ? ` Summary: ${params.summary}` : ""}`,
|
||||
}],
|
||||
details: {},
|
||||
};
|
||||
},
|
||||
};
|
||||
|
||||
// Lazy-load createKbAgent and promptWithFallback
|
||||
const { createKbAgent, promptWithFallback } = await import("./pi.js");
|
||||
|
||||
// Create agent session
|
||||
const { session } = await createKbAgent({
|
||||
cwd: rootDir,
|
||||
systemPrompt: HEARTBEAT_SYSTEM_PROMPT,
|
||||
tools: "readonly",
|
||||
customTools: [
|
||||
createTaskCreateTool(taskStore),
|
||||
createTaskLogTool(taskStore, taskId),
|
||||
heartbeatDoneTool,
|
||||
],
|
||||
defaultProvider: agent.runtimeConfig?.modelProvider as string | undefined,
|
||||
defaultModelId: agent.runtimeConfig?.modelId as string | undefined,
|
||||
onText: (delta) => { outputLength += delta.length; },
|
||||
onToolEnd: () => { toolCallCount++; },
|
||||
});
|
||||
|
||||
// Track for monitoring
|
||||
this.trackAgent(agentId, { dispose: () => session.dispose() }, run.id);
|
||||
|
||||
try {
|
||||
// Build execution prompt
|
||||
const taskTitle = taskDetail.title ?? taskDetail.description.slice(0, 100);
|
||||
const executionPrompt = [
|
||||
`Heartbeat execution for agent "${agent.name}" (ID: ${agent.id})`,
|
||||
`Source: ${source}${triggerDetail ? ` (${triggerDetail})` : ""}`,
|
||||
`Assigned task: ${taskId} — ${taskTitle}`,
|
||||
"",
|
||||
"Task description:",
|
||||
taskDetail.description,
|
||||
"",
|
||||
taskDetail.prompt ? `PROMPT.md:\n${taskDetail.prompt}` : "No PROMPT.md available.",
|
||||
"",
|
||||
"Review the task status and take appropriate action. Call heartbeat_done when finished.",
|
||||
].join("\n");
|
||||
|
||||
// Execute
|
||||
await promptWithFallback(session, executionPrompt);
|
||||
|
||||
// Estimate output tokens (rough: ~4 chars per token)
|
||||
const estimatedOutputTokens = Math.ceil(outputLength / 4);
|
||||
|
||||
// Complete run successfully
|
||||
await this.completeRun(agentId, run.id, {
|
||||
status: "completed",
|
||||
usageJson: { inputTokens: 0, outputTokens: estimatedOutputTokens, cachedTokens: 0 },
|
||||
resultJson: { summary: heartbeatSummary, toolCallCount },
|
||||
});
|
||||
|
||||
heartbeatLog.log(`Heartbeat completed for ${agentId} (${toolCallCount} tool calls, ~${estimatedOutputTokens} output tokens)`);
|
||||
} catch (err) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
heartbeatLog.error(`Heartbeat execution failed for ${agentId}: ${errorMessage}`);
|
||||
await this.completeRun(agentId, run.id, {
|
||||
status: "failed",
|
||||
stderrExcerpt: errorMessage,
|
||||
});
|
||||
} finally {
|
||||
this.untrackAgent(agentId);
|
||||
try { session.dispose(); } catch { /* ignore */ }
|
||||
}
|
||||
|
||||
return (await this.store.getRunDetail(agentId, run.id))!;
|
||||
} catch (err) {
|
||||
const errorMessage = err instanceof Error ? err.message : String(err);
|
||||
heartbeatLog.error(`Heartbeat execution error for ${agentId}: ${errorMessage}`);
|
||||
|
||||
// Attempt to complete the run as failed if it's still active
|
||||
try {
|
||||
await this.completeRun(agentId, run.id, {
|
||||
status: "failed",
|
||||
stderrExcerpt: errorMessage,
|
||||
});
|
||||
} catch {
|
||||
// If completeRun also fails, the run remains active — nothing more we can do
|
||||
}
|
||||
|
||||
return (await this.store.getRunDetail(agentId, run.id))!;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────
|
||||
// Private methods
|
||||
// ─────────────────────────────────────────────────────────────────────────
|
||||
@@ -370,11 +623,6 @@ export class HeartbeatMonitor {
|
||||
};
|
||||
|
||||
try {
|
||||
// Synchronous read — AgentStore.getAgent is async, but we can't make this
|
||||
// method async without changing the call chain. Instead, we'll resolve
|
||||
// per-agent config on the checkMissedHeartbeats path (which is async).
|
||||
// For synchronous callers (isAgentHealthy), we use a cached approach.
|
||||
// For simplicity, we read from the store's underlying agent data.
|
||||
const agent = this.configStore.getCachedAgent?.(agentId);
|
||||
if (agent?.runtimeConfig) {
|
||||
const rc = agent.runtimeConfig;
|
||||
@@ -449,4 +697,4 @@ export class HeartbeatMonitor {
|
||||
// Notify callback
|
||||
this.onTerminated?.(tracked.agentId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
87
packages/engine/src/agent-tools.ts
Normal file
87
packages/engine/src/agent-tools.ts
Normal file
@@ -0,0 +1,87 @@
|
||||
/**
|
||||
* Shared agent tool factory functions.
|
||||
*
|
||||
* Extracted from TaskExecutor so they can be reused by other subsystems
|
||||
* (e.g., HeartbeatMonitor execution) without pulling in the full executor.
|
||||
*
|
||||
* The parameter schemas are canonical here — executor.ts imports and reuses them.
|
||||
*/
|
||||
|
||||
import type { TaskStore } from "@fusion/core";
|
||||
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
|
||||
import { Type, type Static } from "@mariozechner/pi-ai";
|
||||
|
||||
// ── Tool parameter schemas (canonical definitions) ────────────────────────
|
||||
|
||||
export const taskCreateParams = Type.Object({
|
||||
description: Type.String({ description: "What needs to be done" }),
|
||||
dependencies: Type.Optional(
|
||||
Type.Array(Type.String(), { description: "Task IDs this new task depends on (e.g. [\"KB-001\"])" }),
|
||||
),
|
||||
});
|
||||
|
||||
export const taskLogParams = Type.Object({
|
||||
message: Type.String({ description: "What happened" }),
|
||||
outcome: Type.Optional(Type.String({ description: "Result or consequence (optional)" })),
|
||||
});
|
||||
|
||||
// ── Tool factory functions ────────────────────────────────────────────────
|
||||
|
||||
/**
|
||||
* Create a `task_create` tool that creates a new task in triage.
|
||||
*
|
||||
* @param store - TaskStore for task persistence
|
||||
* @returns ToolDefinition for the `task_create` tool
|
||||
*/
|
||||
export function createTaskCreateTool(store: TaskStore): ToolDefinition {
|
||||
return {
|
||||
name: "task_create",
|
||||
label: "Create Task",
|
||||
description:
|
||||
"Create a new task for out-of-scope work discovered during execution. " +
|
||||
"The task goes into triage where it will be specified by the AI. " +
|
||||
"Optionally set dependencies (e.g., the new task depends on the current one, " +
|
||||
"or the current task should wait for the new one).",
|
||||
parameters: taskCreateParams,
|
||||
execute: async (_id: string, params: Static<typeof taskCreateParams>) => {
|
||||
const task = await store.createTask({
|
||||
description: params.description,
|
||||
dependencies: params.dependencies,
|
||||
column: "triage",
|
||||
});
|
||||
const deps = task.dependencies.length ? ` (depends on: ${task.dependencies.join(", ")})` : "";
|
||||
return {
|
||||
content: [{
|
||||
type: "text" as const,
|
||||
text: `Created ${task.id}: ${params.description}${deps}`,
|
||||
}],
|
||||
details: {},
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a `task_log` tool that logs an entry for a specific task.
|
||||
*
|
||||
* @param store - TaskStore for task persistence
|
||||
* @param taskId - The task ID to log entries against
|
||||
* @returns ToolDefinition for the `task_log` tool
|
||||
*/
|
||||
export function createTaskLogTool(store: TaskStore, taskId: string): ToolDefinition {
|
||||
return {
|
||||
name: "task_log",
|
||||
label: "Log Entry",
|
||||
description:
|
||||
"Log an important action, decision, or issue for this task. " +
|
||||
"Use for significant events — not every small step.",
|
||||
parameters: taskLogParams,
|
||||
execute: async (_id: string, params: Static<typeof taskLogParams>) => {
|
||||
await store.logEntry(taskId, params.message, params.outcome);
|
||||
return {
|
||||
content: [{ type: "text" as const, text: `Logged: ${params.message}` }],
|
||||
details: {},
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -22,9 +22,11 @@ import { computeRecoveryDecision, formatDelay, MAX_RECOVERY_RETRIES } from "./re
|
||||
import type { StuckTaskDetector, StuckTaskEvent } from "./stuck-task-detector.js";
|
||||
import { isContextLimitError } from "./context-limit-detector.js";
|
||||
import { StepSessionExecutor, type StepSessionExecutorOptions, type StepResult } from "./step-session-executor.js";
|
||||
import { createTaskCreateTool as sharedCreateTaskCreateTool, createTaskLogTool as sharedCreateTaskLogTool, taskCreateParams, taskLogParams } from "./agent-tools.js";
|
||||
|
||||
// Re-export for backward compatibility (tests import from executor.ts)
|
||||
export { summarizeToolArgs } from "./agent-logger.js";
|
||||
export { createTaskCreateTool, createTaskLogTool, taskCreateParams, taskLogParams } from "./agent-tools.js";
|
||||
|
||||
const STEP_STATUSES: StepStatus[] = ["pending", "in-progress", "done", "skipped"];
|
||||
|
||||
@@ -38,17 +40,7 @@ const taskUpdateParams = Type.Object({
|
||||
),
|
||||
});
|
||||
|
||||
const taskLogParams = Type.Object({
|
||||
message: Type.String({ description: "What happened" }),
|
||||
outcome: Type.Optional(Type.String({ description: "Result or consequence (optional)" })),
|
||||
});
|
||||
|
||||
const taskCreateParams = Type.Object({
|
||||
description: Type.String({ description: "What needs to be done" }),
|
||||
dependencies: Type.Optional(
|
||||
Type.Array(Type.String(), { description: "Task IDs this new task depends on (e.g. [\"KB-001\"])" }),
|
||||
),
|
||||
});
|
||||
// taskLogParams and taskCreateParams are imported from agent-tools.ts
|
||||
|
||||
const taskAddDepParams = Type.Object({
|
||||
task_id: Type.String({ description: "The ID of the task to depend on (e.g. \"KB-001\")" }),
|
||||
@@ -1514,51 +1506,11 @@ export class TaskExecutor {
|
||||
}
|
||||
|
||||
private createTaskLogTool(taskId: string): ToolDefinition {
|
||||
const store = this.store;
|
||||
return {
|
||||
name: "task_log",
|
||||
label: "Log Entry",
|
||||
description:
|
||||
"Log an important action, decision, or issue for this task. " +
|
||||
"Use for significant events — not every small step.",
|
||||
parameters: taskLogParams,
|
||||
execute: async (_id: string, params: Static<typeof taskLogParams>) => {
|
||||
await store.logEntry(taskId, params.message, params.outcome);
|
||||
return {
|
||||
content: [{ type: "text" as const, text: `Logged: ${params.message}` }],
|
||||
details: {},
|
||||
};
|
||||
},
|
||||
};
|
||||
return sharedCreateTaskLogTool(this.store, taskId);
|
||||
}
|
||||
|
||||
private createTaskCreateTool(): ToolDefinition {
|
||||
const store = this.store;
|
||||
return {
|
||||
name: "task_create",
|
||||
label: "Create Task",
|
||||
description:
|
||||
"Create a new task for out-of-scope work discovered during execution. " +
|
||||
"The task goes into triage where it will be specified by the AI. " +
|
||||
"Optionally set dependencies (e.g., the new task depends on the current one, " +
|
||||
"or the current task should wait for the new one).",
|
||||
parameters: taskCreateParams,
|
||||
execute: async (_id: string, params: Static<typeof taskCreateParams>) => {
|
||||
const task = await store.createTask({
|
||||
description: params.description,
|
||||
dependencies: params.dependencies,
|
||||
column: "triage",
|
||||
});
|
||||
const deps = task.dependencies.length ? ` (depends on: ${task.dependencies.join(", ")})` : "";
|
||||
return {
|
||||
content: [{
|
||||
type: "text" as const,
|
||||
text: `Created ${task.id}: ${params.description}${deps}`,
|
||||
}],
|
||||
details: {},
|
||||
};
|
||||
},
|
||||
};
|
||||
return sharedCreateTaskCreateTool(this.store);
|
||||
}
|
||||
|
||||
private createTaskAddDepTool(taskId: string): ToolDefinition {
|
||||
|
||||
@@ -79,3 +79,6 @@ export const hybridExecutorLog = createLogger("hybrid-executor");
|
||||
|
||||
/** Logger for the mission autopilot subsystem. */
|
||||
export const autopilotLog = createLogger("autopilot");
|
||||
|
||||
/** Logger for the heartbeat execution subsystem. */
|
||||
export const heartbeatLog = createLogger("heartbeat");
|
||||
|
||||
@@ -4,6 +4,8 @@ import type {
|
||||
Task,
|
||||
CentralCore,
|
||||
AgentStore,
|
||||
HeartbeatInvocationSource,
|
||||
AgentHeartbeatRun,
|
||||
} from "@fusion/core";
|
||||
import { Scheduler } from "../scheduler.js";
|
||||
import { TaskExecutor, type TaskExecutorOptions } from "../executor.js";
|
||||
@@ -214,6 +216,8 @@ export class InProcessRuntime
|
||||
this.heartbeatMonitor = new HeartbeatMonitor({
|
||||
store: this.agentStore,
|
||||
agentStore: this.agentStore, // enables per-agent config resolution
|
||||
taskStore: this.taskStore,
|
||||
rootDir: this.config.workingDirectory,
|
||||
onMissed: (agentId) => {
|
||||
runtimeLog.warn(`Agent ${agentId} missed heartbeat`);
|
||||
},
|
||||
@@ -389,6 +393,47 @@ export class InProcessRuntime
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the HeartbeatMonitor instance (if initialized).
|
||||
* Returns undefined when agent monitoring is not available.
|
||||
*/
|
||||
getHeartbeatMonitor(): HeartbeatMonitor | undefined {
|
||||
return this.heartbeatMonitor;
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute a heartbeat run for an agent.
|
||||
*
|
||||
* Delegates to HeartbeatMonitor.executeHeartbeat().
|
||||
* Throws if the runtime is not active or the heartbeat monitor is not initialized.
|
||||
*
|
||||
* @param agentId - The agent ID to execute a heartbeat for
|
||||
* @param source - What triggered this heartbeat
|
||||
* @param options - Optional task ID override and trigger detail
|
||||
* @returns The completed heartbeat run
|
||||
*/
|
||||
async executeHeartbeat(
|
||||
agentId: string,
|
||||
source: HeartbeatInvocationSource,
|
||||
options?: { taskId?: string; triggerDetail?: string }
|
||||
): Promise<AgentHeartbeatRun | null> {
|
||||
if (this.status !== "active") {
|
||||
throw new Error(`Cannot execute heartbeat: runtime status is ${this.status}`);
|
||||
}
|
||||
if (!this.heartbeatMonitor) {
|
||||
return null;
|
||||
}
|
||||
|
||||
runtimeLog.log(`Executing heartbeat for agent ${agentId} (source=${source})`);
|
||||
const result = await this.heartbeatMonitor.executeHeartbeat({
|
||||
agentId,
|
||||
source,
|
||||
...options,
|
||||
});
|
||||
runtimeLog.log(`Heartbeat completed for agent ${agentId}`);
|
||||
return result;
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the StuckTaskDetector for this runtime.
|
||||
*/
|
||||
|
||||
Reference in New Issue
Block a user