feat(FN-1065): add immediate message response mode for heartbeats
- Add MessageResponseMode and messageResponseMode runtime config to core agent heartbeat types and exports - Extend MessageStore with an onMessageToAgent hook plus runtime hook updates for agent-directed messages - Wire HeartbeatMonitor to wake agents on incoming messages when messageResponseMode is immediate and state is eligible - Update AgentDetailView with a Message Response Mode setting, validation, and runtimeConfig persistence - Expand core and engine tests to cover hook behavior, wake-on-message triggering, and start/stop hook lifecycle
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||
import { HeartbeatMonitor, HeartbeatTriggerScheduler, type AgentSession, type HeartbeatExecutionOptions, HEARTBEAT_SYSTEM_PROMPT } from "./agent-heartbeat.js";
|
||||
import type { AgentStore, AgentHeartbeatRun, TaskStore, TaskDetail, Agent } from "@fusion/core";
|
||||
import type { AgentStore, AgentHeartbeatRun, TaskStore, TaskDetail, Agent, MessageStore, Message } from "@fusion/core";
|
||||
|
||||
// Mock logger to suppress noise in test output
|
||||
vi.mock("./logger.js", () => {
|
||||
@@ -43,6 +43,31 @@ function createMockSession(): AgentSession {
|
||||
};
|
||||
}
|
||||
|
||||
function createMockMessageStore(onSetHook?: (hook: (message: Message) => void) => void): MessageStore {
|
||||
return {
|
||||
setMessageToAgentHook: vi.fn((hook: (message: Message) => void) => {
|
||||
onSetHook?.(hook);
|
||||
}),
|
||||
} as unknown as MessageStore;
|
||||
}
|
||||
|
||||
function createMessage(overrides: Partial<Message> = {}): Message {
|
||||
const now = new Date().toISOString();
|
||||
return {
|
||||
id: "msg-001",
|
||||
fromId: "user-1",
|
||||
fromType: "user",
|
||||
toId: "agent-1",
|
||||
toType: "agent",
|
||||
content: "hello",
|
||||
type: "user-to-agent",
|
||||
read: false,
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
describe("HeartbeatMonitor", () => {
|
||||
let store: AgentStore;
|
||||
let monitor: HeartbeatMonitor;
|
||||
@@ -139,6 +164,164 @@ describe("HeartbeatMonitor", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("wake-on-message", () => {
|
||||
it("executes heartbeat when messageResponseMode is immediate", () => {
|
||||
let messageHook: ((message: Message) => void) | undefined;
|
||||
const messageStore = createMockMessageStore((hook) => {
|
||||
messageHook = hook;
|
||||
});
|
||||
const configStore = createMockStore({
|
||||
getCachedAgent: vi.fn().mockReturnValue({
|
||||
id: "agent-1",
|
||||
state: "active",
|
||||
runtimeConfig: { messageResponseMode: "immediate" },
|
||||
}),
|
||||
});
|
||||
|
||||
const customMonitor = new HeartbeatMonitor({
|
||||
store,
|
||||
agentStore: configStore,
|
||||
messageStore,
|
||||
});
|
||||
const executeHeartbeatSpy = vi
|
||||
.spyOn(customMonitor, "executeHeartbeat")
|
||||
.mockResolvedValue({ id: "run-1" } as AgentHeartbeatRun);
|
||||
|
||||
customMonitor.start();
|
||||
messageHook?.(createMessage({ toId: "agent-1", toType: "agent" }));
|
||||
|
||||
expect(executeHeartbeatSpy).toHaveBeenCalledWith({
|
||||
agentId: "agent-1",
|
||||
source: "on_demand",
|
||||
triggerDetail: "wake-on-message",
|
||||
});
|
||||
|
||||
customMonitor.stop();
|
||||
});
|
||||
|
||||
it("does not execute heartbeat when messageResponseMode is on-heartbeat or unset", () => {
|
||||
let messageHook: ((message: Message) => void) | undefined;
|
||||
const messageStore = createMockMessageStore((hook) => {
|
||||
messageHook = hook;
|
||||
});
|
||||
const getCachedAgent = vi
|
||||
.fn()
|
||||
.mockReturnValueOnce({
|
||||
id: "agent-1",
|
||||
state: "active",
|
||||
runtimeConfig: { messageResponseMode: "on-heartbeat" },
|
||||
})
|
||||
.mockReturnValueOnce({
|
||||
id: "agent-1",
|
||||
state: "active",
|
||||
runtimeConfig: {},
|
||||
});
|
||||
const configStore = createMockStore({ getCachedAgent });
|
||||
|
||||
const customMonitor = new HeartbeatMonitor({
|
||||
store,
|
||||
agentStore: configStore,
|
||||
messageStore,
|
||||
});
|
||||
const executeHeartbeatSpy = vi
|
||||
.spyOn(customMonitor, "executeHeartbeat")
|
||||
.mockResolvedValue({ id: "run-1" } as AgentHeartbeatRun);
|
||||
|
||||
customMonitor.start();
|
||||
messageHook?.(createMessage({ toId: "agent-1", toType: "agent", id: "msg-1" }));
|
||||
messageHook?.(createMessage({ toId: "agent-1", toType: "agent", id: "msg-2" }));
|
||||
|
||||
expect(executeHeartbeatSpy).not.toHaveBeenCalled();
|
||||
|
||||
customMonitor.stop();
|
||||
});
|
||||
|
||||
it("does not execute heartbeat when agent is paused or error", () => {
|
||||
let messageHook: ((message: Message) => void) | undefined;
|
||||
const messageStore = createMockMessageStore((hook) => {
|
||||
messageHook = hook;
|
||||
});
|
||||
const getCachedAgent = vi
|
||||
.fn()
|
||||
.mockReturnValueOnce({
|
||||
id: "agent-1",
|
||||
state: "paused",
|
||||
runtimeConfig: { messageResponseMode: "immediate" },
|
||||
})
|
||||
.mockReturnValueOnce({
|
||||
id: "agent-1",
|
||||
state: "error",
|
||||
runtimeConfig: { messageResponseMode: "immediate" },
|
||||
});
|
||||
const configStore = createMockStore({ getCachedAgent });
|
||||
|
||||
const customMonitor = new HeartbeatMonitor({
|
||||
store,
|
||||
agentStore: configStore,
|
||||
messageStore,
|
||||
});
|
||||
const executeHeartbeatSpy = vi
|
||||
.spyOn(customMonitor, "executeHeartbeat")
|
||||
.mockResolvedValue({ id: "run-1" } as AgentHeartbeatRun);
|
||||
|
||||
customMonitor.start();
|
||||
messageHook?.(createMessage({ toId: "agent-1", toType: "agent", id: "msg-paused" }));
|
||||
messageHook?.(createMessage({ toId: "agent-1", toType: "agent", id: "msg-error" }));
|
||||
|
||||
expect(executeHeartbeatSpy).not.toHaveBeenCalled();
|
||||
|
||||
customMonitor.stop();
|
||||
});
|
||||
|
||||
it("registers the message hook on start and clears it on stop", () => {
|
||||
const hooks: Array<(message: Message) => void> = [];
|
||||
const messageStore = createMockMessageStore((hook) => {
|
||||
hooks.push(hook);
|
||||
});
|
||||
const customMonitor = new HeartbeatMonitor({ store, messageStore });
|
||||
|
||||
customMonitor.start();
|
||||
|
||||
expect(messageStore.setMessageToAgentHook).toHaveBeenCalledTimes(1);
|
||||
expect(hooks).toHaveLength(1);
|
||||
|
||||
customMonitor.stop();
|
||||
|
||||
expect(messageStore.setMessageToAgentHook).toHaveBeenCalledTimes(2);
|
||||
expect(hooks).toHaveLength(2);
|
||||
expect(hooks[0]).not.toBe(hooks[1]);
|
||||
});
|
||||
|
||||
it("ignores non-agent messages", () => {
|
||||
let messageHook: ((message: Message) => void) | undefined;
|
||||
const messageStore = createMockMessageStore((hook) => {
|
||||
messageHook = hook;
|
||||
});
|
||||
const configStore = createMockStore({
|
||||
getCachedAgent: vi.fn().mockReturnValue({
|
||||
id: "agent-1",
|
||||
state: "active",
|
||||
runtimeConfig: { messageResponseMode: "immediate" },
|
||||
}),
|
||||
});
|
||||
const customMonitor = new HeartbeatMonitor({
|
||||
store,
|
||||
agentStore: configStore,
|
||||
messageStore,
|
||||
});
|
||||
const executeHeartbeatSpy = vi
|
||||
.spyOn(customMonitor, "executeHeartbeat")
|
||||
.mockResolvedValue({ id: "run-1" } as AgentHeartbeatRun);
|
||||
|
||||
customMonitor.start();
|
||||
messageHook?.(createMessage({ toType: "user", toId: "user-1" }));
|
||||
|
||||
expect(executeHeartbeatSpy).not.toHaveBeenCalled();
|
||||
|
||||
customMonitor.stop();
|
||||
});
|
||||
});
|
||||
|
||||
describe("isActive", () => {
|
||||
it("reflects monitor state (false when not started)", () => {
|
||||
expect(monitor.isActive()).toBe(false);
|
||||
|
||||
@@ -17,7 +17,7 @@
|
||||
* - onTerminated: Called when an unresponsive agent is terminated
|
||||
*/
|
||||
|
||||
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, TaskStore, TaskDetail } from "@fusion/core";
|
||||
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, Message, MessageStore, TaskStore, TaskDetail } from "@fusion/core";
|
||||
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
|
||||
import { Type, type Static } from "@mariozechner/pi-ai";
|
||||
import { createTaskCreateTool, createTaskLogTool, taskCreateParams } from "./agent-tools.js";
|
||||
@@ -42,6 +42,8 @@ export interface HeartbeatMonitorOptions {
|
||||
/** Optional separate AgentStore reference for reading per-agent runtimeConfig.
|
||||
* If not provided, falls back to `store`. */
|
||||
agentStore?: AgentStore;
|
||||
/** Optional MessageStore for wake-on-message behavior */
|
||||
messageStore?: MessageStore;
|
||||
/** Polling interval in milliseconds (default: 30000) */
|
||||
pollIntervalMs?: number;
|
||||
/** Heartbeat timeout in milliseconds (default: 60000) */
|
||||
@@ -144,6 +146,7 @@ export class HeartbeatMonitor {
|
||||
private onRunCompleted?: (agentId: string, run: AgentHeartbeatRun) => void;
|
||||
private taskStore?: TaskStore;
|
||||
private rootDir?: string;
|
||||
private messageStore?: MessageStore;
|
||||
|
||||
private trackedAgents: Map<string, TrackedAgent> = new Map();
|
||||
private agentStartLocks: Map<string, Promise<unknown>> = new Map();
|
||||
@@ -166,6 +169,7 @@ export class HeartbeatMonitor {
|
||||
this.onRunCompleted = options.onRunCompleted;
|
||||
this.taskStore = options.taskStore;
|
||||
this.rootDir = options.rootDir;
|
||||
this.messageStore = options.messageStore;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -176,6 +180,9 @@ export class HeartbeatMonitor {
|
||||
if (this.isRunning) return;
|
||||
|
||||
this.isRunning = true;
|
||||
if (this.messageStore) {
|
||||
this.messageStore.setMessageToAgentHook(this.handleMessageToAgent.bind(this));
|
||||
}
|
||||
this.pollInterval = setInterval(() => {
|
||||
void this.checkMissedHeartbeats();
|
||||
}, this.pollIntervalMs);
|
||||
@@ -186,6 +193,9 @@ export class HeartbeatMonitor {
|
||||
* Does not untrack agents - they remain in memory.
|
||||
*/
|
||||
stop(): void {
|
||||
if (this.messageStore) {
|
||||
this.messageStore.setMessageToAgentHook(() => {});
|
||||
}
|
||||
if (!this.isRunning) return;
|
||||
|
||||
this.isRunning = false;
|
||||
@@ -417,6 +427,36 @@ export class HeartbeatMonitor {
|
||||
return this.trackedAgents.get(agentId)?.lastSeen;
|
||||
}
|
||||
|
||||
private handleMessageToAgent(message: Message): void {
|
||||
if (message.toType !== "agent") {
|
||||
return;
|
||||
}
|
||||
|
||||
const agent = this.configStore.getCachedAgent(message.toId);
|
||||
if (!agent) {
|
||||
return;
|
||||
}
|
||||
|
||||
const runtimeConfig = agent.runtimeConfig as AgentHeartbeatConfig | undefined;
|
||||
if (runtimeConfig?.messageResponseMode !== "immediate") {
|
||||
return;
|
||||
}
|
||||
|
||||
const validStates = new Set(["active", "idle", "running"]);
|
||||
if (!validStates.has(agent.state)) {
|
||||
return;
|
||||
}
|
||||
|
||||
void this.executeHeartbeat({
|
||||
agentId: message.toId,
|
||||
source: "on_demand",
|
||||
triggerDetail: "wake-on-message",
|
||||
}).catch((error) => {
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
heartbeatLog.warn(`Wake-on-message heartbeat failed for ${message.toId}: ${errorMessage}`);
|
||||
});
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────
|
||||
// Heartbeat execution (Paperclip wake → check → work → exit)
|
||||
// ─────────────────────────────────────────────────────────────────────────
|
||||
|
||||
Reference in New Issue
Block a user