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

feat(FN-1949): add send_message and read_messages agent tools

Browse Source 
- Add send_message tool factory for executor agents to send messages
- Add read_messages tool factory to retrieve agent inbox messages
- Update heartbeat to process pending messages during agent wake cycles
- Include message tools in executor agent tool context
- Add comprehensive unit tests for message tools
- Add heartbeat message processing tests
- Add end-to-end agent messaging test
- Create changeset for @gsxdsm/fusion package
This commit is contained in:
Fusion
2026-04-16 13:50:10 -07:00
committed by gsxdsm
parent 4e66baf204
commit 9e1c18e0c8
6 changed files with 857 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

352
packages/engine/src/agent-heartbeat.test.ts
View File

@@ -362,6 +362,78 @@ describe("HeartbeatMonitor", () => {
customMonitor.stop();
});
describe("createHeartbeatTools - message tools", () => {
let mockTaskStore: TaskStore;
let mockSession: ReturnType<typeof createMockSession>;
let capturedTools: any[] = [];
beforeEach(() => {
mockTaskStore = {
createTask: vi.fn().mockResolvedValue({ id: "FN-002", description: "test", dependencies: [], column: "triage" }),
logEntry: vi.fn().mockResolvedValue(undefined),
upsertTaskDocument: vi.fn().mockResolvedValue({
id: "doc-1", taskId: "FN-001", key: "test", content: "test", revision: 1, author: "agent",
createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(),
}),
getTaskDocument: vi.fn().mockResolvedValue(null),
getTaskDocuments: vi.fn().mockResolvedValue([]),
} as unknown as TaskStore;
mockSession = createMockSession();
capturedTools = [];
});
it("includes send_message and read_messages tools when messageStore is available", () => {
const messageStore = createMockMessageStore();
const customMonitor = new HeartbeatMonitor({
store,
messageStore,
taskStore: mockTaskStore,
rootDir: "/tmp",
});
const tools = customMonitor.createHeartbeatTools("agent-001", mockTaskStore, "FN-001", undefined, undefined, messageStore);
const toolNames = tools.map((t) => t.name);
expect(toolNames).toContain("send_message");
expect(toolNames).toContain("read_messages");
});
it("does not include message tools when messageStore is not provided", () => {
const customMonitor = new HeartbeatMonitor({
store,
taskStore: mockTaskStore,
rootDir: "/tmp",
});
const tools = customMonitor.createHeartbeatTools("agent-001", mockTaskStore, "FN-001");
const toolNames = tools.map((t) => t.name);
expect(toolNames).not.toContain("send_message");
expect(toolNames).not.toContain("read_messages");
});
it("does not include message tools when messageStore is undefined even if other params are passed", () => {
const customMonitor = new HeartbeatMonitor({
store,
taskStore: mockTaskStore,
rootDir: "/tmp",
});
const tools = customMonitor.createHeartbeatTools(
"agent-001",
mockTaskStore,
"FN-001",
undefined,
undefined,
undefined
);
const toolNames = tools.map((t) => t.name);
expect(toolNames).not.toContain("send_message");
expect(toolNames).not.toContain("read_messages");
});
});
});
describe("isActive", () => {
@@ -1433,6 +1505,286 @@ describe("HeartbeatMonitor", () => {
});
});
describe("executeHeartbeat - message processing", () => {
it("includes unread messages in prompt when woken by wake-on-message", async () => {
const store = createStoreWithAgentForExec();
const mockSession = createMockAgentSession();
mockedCreateKbAgent.mockResolvedValue({ session: mockSession as any });
const messages = [
createMessage({
id: "msg-1",
fromId: "agent-2",
content: "Hello from agent-2",
createdAt: "2024-01-15T10:30:00.000Z",
}),
createMessage({
id: "msg-2",
fromId: "user-1",
content: "Hello from user",
createdAt: "2024-01-15T11:00:00.000Z",
}),
];
const messageStore = {
setMessageToAgentHook: vi.fn(),
getInbox: vi.fn().mockReturnValue(messages),
markAllAsRead: vi.fn(),
} as unknown as MessageStore;
const monitor = new HeartbeatMonitor({
store,
messageStore,
taskStore: mockTaskStore,
rootDir: "/tmp",
});
const result = await monitor.executeHeartbeat({
agentId: "agent-001",
source: "on_demand",
triggerDetail: "wake-on-message",
});
expect(result.status).toBe("completed");
expect(mockedCreateKbAgent).toHaveBeenCalled();
expect(messageStore.getInbox).toHaveBeenCalledWith("agent-001", "agent", { read: false, limit: 10 });
// Verify execution prompt (passed to promptWithFallback) included the messages
// The execution prompt is passed to session.prompt by promptWithFallback mock
const promptCalls = mockSession.prompt.mock.calls;
expect(promptCalls.length).toBeGreaterThan(0);
const executionPrompt = promptCalls[promptCalls.length - 1][0];
expect(executionPrompt).toContain("Pending Messages:");
expect(executionPrompt).toContain("Hello from agent-2");
expect(executionPrompt).toContain("Hello from user");
});
it("does not include message section when no unread messages", async () => {
const store = createStoreWithAgentForExec();
const mockSession = createMockAgentSession();
mockedCreateKbAgent.mockResolvedValue({ session: mockSession as any });
const messageStore = {
setMessageToAgentHook: vi.fn(),
getInbox: vi.fn().mockReturnValue([]),
markAllAsRead: vi.fn(),
} as unknown as MessageStore;
const monitor = new HeartbeatMonitor({
store,
messageStore,
taskStore: mockTaskStore,
rootDir: "/tmp",
});
const result = await monitor.executeHeartbeat({
agentId: "agent-001",
source: "on_demand",
triggerDetail: "wake-on-message",
});
expect(result.status).toBe("completed");
// Verify prompt did NOT include pending messages section
// Note: without wake-on-message trigger, no messages are fetched
// so the prompt won't have the Pending Messages section at all
});
it("marks messages as read after successful heartbeat execution", async () => {
const store = createStoreWithAgentForExec();
const mockSession = createMockAgentSession();
mockedCreateKbAgent.mockResolvedValue({ session: mockSession as any });
const messages = [
createMessage({
id: "msg-1",
fromId: "agent-2",
content: "Hello from agent-2",
}),
];
const messageStore = {
setMessageToAgentHook: vi.fn(),
getInbox: vi.fn().mockReturnValue(messages),
markAllAsRead: vi.fn(),
} as unknown as MessageStore;
const monitor = new HeartbeatMonitor({
store,
messageStore,
taskStore: mockTaskStore,
rootDir: "/tmp",
});
const result = await monitor.executeHeartbeat({
agentId: "agent-001",
source: "on_demand",
triggerDetail: "wake-on-message",
});
expect(result.status).toBe("completed");
expect(messageStore.markAllAsRead).toHaveBeenCalledWith("agent-001", "agent");
});
it("does not mark messages as read on failed heartbeat execution", async () => {
const store = createStoreWithAgentForExec();
const mockSession = createMockAgentSession();
mockSession.prompt = vi.fn().mockRejectedValue(new Error("Execution failed"));
mockedCreateKbAgent.mockResolvedValue({ session: mockSession as any });
const messages = [
createMessage({
id: "msg-1",
fromId: "agent-2",
content: "Hello from agent-2",
}),
];
const messageStore = {
setMessageToAgentHook: vi.fn(),
getInbox: vi.fn().mockReturnValue(messages),
markAllAsRead: vi.fn(),
} as unknown as MessageStore;
const monitor = new HeartbeatMonitor({
store,
messageStore,
taskStore: mockTaskStore,
rootDir: "/tmp",
});
const result = await monitor.executeHeartbeat({
agentId: "agent-001",
source: "on_demand",
triggerDetail: "wake-on-message",
});
expect(result.status).toBe("failed");
expect(messageStore.markAllAsRead).not.toHaveBeenCalled();
});
it("does not fetch messages when not wake-on-message trigger", async () => {
const store = createStoreWithAgentForExec();
const mockSession = createMockAgentSession();
mockedCreateKbAgent.mockResolvedValue({ session: mockSession as any });
const messageStore = {
setMessageToAgentHook: vi.fn(),
getInbox: vi.fn(),
markAllAsRead: vi.fn(),
} as unknown as MessageStore;
const monitor = new HeartbeatMonitor({
store,
messageStore,
taskStore: mockTaskStore,
rootDir: "/tmp",
});
// Use a regular trigger (not wake-on-message)
await monitor.executeHeartbeat({
agentId: "agent-001",
source: "timer",
triggerDetail: "scheduled",
});
expect(messageStore.getInbox).not.toHaveBeenCalled();
expect(messageStore.markAllAsRead).not.toHaveBeenCalled();
});
describe("end-to-end agent-to-agent message flow", () => {
it("proves full message flow from send to wake to processing to reply", async () => {
// Import real MessageStore for this test
const core = await import("@fusion/core");
const Database = core.Database;
const RealMessageStore = core.MessageStore;
// Setup: create real temp database and MessageStore
const tmpDir = await import("node:fs/promises").then((fs) =>
fs.mkdtemp(require("node:path").join(require("node:os").tmpdir(), "fn-e2e-message-"))
);
const kbDir = require("node:path").join(tmpDir, ".fusion");
await require("node:fs").promises.mkdir(kbDir, { recursive: true });
let messageStore: InstanceType<typeof RealMessageStore>;
let db: InstanceType<typeof Database> | undefined;
try {
db = new Database(kbDir);
db.init();
messageStore = new RealMessageStore(db);
// Agent A sends a message to Agent B
const sentMessage = messageStore.sendMessage({
fromId: "agent-alpha",
fromType: "agent",
toId: "agent-beta",
toType: "agent",
content: "Hello Agent Beta, please process task FN-001.",
type: "agent-to-agent",
metadata: { taskId: "FN-001" },
});
expect(sentMessage.id).toBeDefined();
expect(sentMessage.content).toContain("Hello Agent Beta");
// Verify message is stored in Agent B's inbox
const inbox = messageStore.getInbox("agent-beta", "agent", { read: false });
expect(inbox).toHaveLength(1);
expect(inbox[0].id).toBe(sentMessage.id);
expect(inbox[0].content).toBe(sentMessage.content);
// Create a monitor for Agent B with real MessageStore
const store = createStoreWithAgentForExec({
id: "agent-beta",
name: "Agent Beta",
state: "active",
taskId: "FN-001",
});
const mockSession = createMockAgentSession();
mockedCreateKbAgent.mockResolvedValue({ session: mockSession as any });
const monitor = new HeartbeatMonitor({
store,
messageStore,
taskStore: mockTaskStore,
rootDir: "/tmp",
});
// Execute heartbeat to process the message (simulating wake-on-message trigger)
const result = await monitor.executeHeartbeat({
agentId: "agent-beta",
source: "on_demand",
triggerDetail: "wake-on-message",
});
// Verify heartbeat completed
expect(result.status).toBe("completed");
// Verify the execution prompt included the message
const promptCalls = mockSession.prompt.mock.calls;
expect(promptCalls.length).toBeGreaterThan(0);
const executionPrompt = promptCalls[promptCalls.length - 1][0];
expect(executionPrompt).toContain("Hello Agent Beta");
expect(executionPrompt).toContain("Pending Messages:");
// Verify messages were marked as read after successful processing
expect(messageStore.getMailbox("agent-beta", "agent").unreadCount).toBe(0);
// Cleanup
await monitor.stop();
} finally {
// Cleanup temp directory
try {
db?.close();
await require("node:fs/promises").rm(tmpDir, { recursive: true, force: true });
} catch {
// ignore cleanup errors
}
}
});
});
});
describe("executeHeartbeat - inbox selection", () => {
const makeInboxSelection = (taskId: string, priority: "in_progress" | "todo" | "blocked" = "todo") => {
const now = new Date().toISOString();

65
packages/engine/src/agent-heartbeat.ts
View File

@@ -20,7 +20,7 @@
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask, BlockedStateSnapshot, RunMutationContext } from "@fusion/core";
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
import { Type, type Static } from "@mariozechner/pi-ai";
import { createTaskCreateTool, createTaskLogToolWithContext, createTaskDocumentWriteTool, createTaskDocumentReadTool, createListAgentsTool, createDelegateTaskTool, taskCreateParams } from "./agent-tools.js";
import { createTaskCreateTool, createTaskLogToolWithContext, createTaskDocumentWriteTool, createTaskDocumentReadTool, createListAgentsTool, createDelegateTaskTool, createSendMessageTool, createReadMessagesTool, taskCreateParams } from "./agent-tools.js";
import { AgentLogger } from "./agent-logger.js";
import { heartbeatLog } from "./logger.js";
import { createRunAuditor, type EngineRunContext } from "./run-audit.js";
@@ -136,7 +136,22 @@ Keep work lightweight — this is a single-pass check, not a full implementation
You have readonly file access plus task_create, task_log, and task_document tools.
**Task Documents:** Save important findings with task_document_write(key="...", content="...").
Documents persist across sessions and are visible in the dashboard's Documents tab.`;
Documents persist across sessions and are visible in the dashboard's Documents tab.
## Processing Messages
When you are woken by an incoming message (source includes "wake-on-message"), you should:
1. Use read_messages to check your inbox for unread messages.
2. Review each message and determine the appropriate action:
- If the message requires a response, use send_message to reply.
- If the message is informational, acknowledge it by logging with task_log.
- If the message requests work, create a follow-up task with task_create or handle it directly.
3. After processing messages, continue with your normal heartbeat duties.
When sending messages:
- Be concise and clear about what you need or what you've done.
- Include relevant context (task IDs, file paths) in metadata when applicable.
- Use agent-to-agent for inter-agent communication.`;
/** Parameter schema for the heartbeat_done tool */
const heartbeatDoneParams = Type.Object({
@@ -908,7 +923,8 @@ export class HeartbeatMonitor {
const { buildSessionSkillContextSync } = await import("./session-skill-context.js");
// Build tools with task creation tracking and run context for mutation correlation
const heartbeatTools = this.createHeartbeatTools(agentId, taskStore, taskId, runContext, audit);
// Pass messageStore for messaging tools (send_message, read_messages)
const heartbeatTools = this.createHeartbeatTools(agentId, taskStore, taskId, runContext, audit, this.messageStore);
heartbeatTools.push(heartbeatDoneTool);
agentLogger = new AgentLogger({
@@ -954,6 +970,17 @@ export class HeartbeatMonitor {
// Build execution prompt
const taskTitle = taskDetail.title ?? taskDetail.description.slice(0, 100);
// Fetch unread messages when woken by message trigger
let pendingMessages: Message[] = [];
if (triggerDetail === "wake-on-message" && this.messageStore) {
try {
pendingMessages = this.messageStore.getInbox(agentId, "agent", { read: false, limit: 10 });
} catch {
// Non-critical — if message fetch fails, proceed without messages
heartbeatLog.warn(`Failed to fetch inbox messages for ${agentId} during wake-on-message`);
}
}
const triggeringCommentLines: string[] = [];
if (effectiveTriggeringCommentIds && effectiveTriggeringCommentIds.length > 0) {
const commentLookup = new Map<string, { author: string; text: string }>();
@@ -983,6 +1010,19 @@ export class HeartbeatMonitor {
}
}
// Build pending messages section
const pendingMessagesLines: string[] = [];
if (pendingMessages.length > 0) {
pendingMessagesLines.push(
"",
"Pending Messages:",
...pendingMessages.map((msg) => {
const timestamp = new Date(msg.createdAt).toLocaleString();
return `- [from: ${msg.fromId}] ${msg.content} (${timestamp})`;
}),
);
}
const executionPrompt = [
`Heartbeat execution for agent "${agent.name}" (ID: ${agent.id})`,
`Source: ${source}${triggerDetail ? ` (${triggerDetail})` : ""}`,
@@ -993,6 +1033,7 @@ export class HeartbeatMonitor {
"",
taskDetail.prompt ? `PROMPT.md:\n${taskDetail.prompt}` : "No PROMPT.md available.",
...triggeringCommentLines,
...pendingMessagesLines,
"",
"Review the task status and take appropriate action. Call heartbeat_done when finished.",
].join("\n");
@@ -1004,6 +1045,16 @@ export class HeartbeatMonitor {
const estimatedOutputTokens = Math.ceil(outputLength / 4);
await flushAgentLogger();
// Mark messages as read after successful processing (only if messages were included in prompt)
if (pendingMessages.length > 0 && this.messageStore) {
try {
this.messageStore.markAllAsRead(agentId, "agent");
} catch {
// Non-critical — mark as read failed, messages remain unread
heartbeatLog.warn(`Failed to mark messages as read for ${agentId}`);
}
}
// Complete run successfully
const completionResultJson: Record<string, unknown> = {
summary: heartbeatSummary,
@@ -1074,6 +1125,7 @@ export class HeartbeatMonitor {
* @param taskId - The assigned task ID (for task_log context)
* @param runContext - Optional run context for mutation correlation
* @param audit - Optional run auditor for audit trail (FN-1404)
* @param messageStore - Optional MessageStore for messaging tools
* @returns Array of ToolDefinitions for the heartbeat session
*/
createHeartbeatTools(
@@ -1082,6 +1134,7 @@ export class HeartbeatMonitor {
taskId: string,
runContext?: RunMutationContext,
audit?: ReturnType<typeof createRunAuditor>,
messageStore?: MessageStore,
): ToolDefinition[] {
const tools: ToolDefinition[] = [];
@@ -1133,6 +1186,12 @@ export class HeartbeatMonitor {
tools.push(createListAgentsTool(this.store));
tools.push(createDelegateTaskTool(this.store, taskStore));
// Messaging tools — when MessageStore is available, agents can send and receive messages
if (messageStore) {
tools.push(createSendMessageTool(messageStore, agentId));
tools.push(createReadMessagesTool(messageStore, agentId));
}
return tools;
}

297
packages/engine/src/agent-tools.test.ts Normal file
View File

@@ -0,0 +1,297 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { createSendMessageTool, createReadMessagesTool, sendMessageParams, readMessagesParams } from "./agent-tools.js";
import type { MessageStore, Message } from "@fusion/core";
// Mock logger
vi.mock("./logger.js", () => {
const createMockLogger = () => ({
log: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
});
return {
createLogger: vi.fn(() => createMockLogger()),
heartbeatLog: createMockLogger(),
};
});
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: "Test message",
type: "agent-to-agent",
read: false,
createdAt: now,
updatedAt: now,
...overrides,
};
}
function createMockMessageStore(overrides: Partial<MessageStore> = {}): MessageStore {
return {
sendMessage: vi.fn(),
getInbox: vi.fn().mockReturnValue([]),
markAsRead: vi.fn(),
markAllAsRead: vi.fn(),
...overrides,
} as unknown as MessageStore;
}
describe("createSendMessageTool", () => {
let messageStore: ReturnType<typeof createMockMessageStore>;
let tool: ReturnType<typeof createSendMessageTool>;
beforeEach(() => {
messageStore = createMockMessageStore();
tool = createSendMessageTool(messageStore, "agent-sender");
});
// Helper to call tool execute with correct signature
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const executeTool = async (tool: any, params: unknown) => {
return tool.execute("call-1", params, undefined, undefined, undefined);
};
it("creates a tool with name 'send_message'", () => {
expect(tool.name).toBe("send_message");
});
it("creates a tool with correct label", () => {
expect(tool.label).toBe("Send Message");
});
it("creates a tool with a description mentioning recipient waking", () => {
expect(tool.description).toContain("messageResponseMode");
});
it("calls messageStore.sendMessage with correct parameters", async () => {
const mockMessage = createMessage({ id: "msg-123" });
vi.mocked(messageStore.sendMessage).mockReturnValue(mockMessage);
const result = await executeTool(tool, {
to_id: "agent-recipient",
content: "Hello, world!",
});
expect(messageStore.sendMessage).toHaveBeenCalledWith({
fromId: "agent-sender",
fromType: "agent",
toId: "agent-recipient",
toType: "agent",
content: "Hello, world!",
type: "agent-to-agent",
});
expect(result.content[0]).toEqual({ type: "text", text: "Message sent to agent-recipient (ID: msg-123)" });
expect(result.details).toEqual({ messageId: "msg-123" });
});
it("defaults type to 'agent-to-agent' when not specified", async () => {
const mockMessage = createMessage();
vi.mocked(messageStore.sendMessage).mockReturnValue(mockMessage);
await executeTool(tool, {
to_id: "agent-2",
content: "Test",
});
expect(messageStore.sendMessage).toHaveBeenCalledWith(
expect.objectContaining({ type: "agent-to-agent" })
);
});
it("uses provided type when specified", async () => {
const mockMessage = createMessage();
vi.mocked(messageStore.sendMessage).mockReturnValue(mockMessage);
await executeTool(tool, {
to_id: "user-1",
content: "Test",
type: "agent-to-user",
});
expect(messageStore.sendMessage).toHaveBeenCalledWith(
expect.objectContaining({ type: "agent-to-user" })
);
});
it("returns error for empty content", async () => {
const result = await executeTool(tool, {
to_id: "agent-2",
content: " ",
});
expect(result.content[0]).toEqual({ type: "text", text: "ERROR: Message content cannot be empty" });
expect(messageStore.sendMessage).not.toHaveBeenCalled();
});
it("returns error for content exceeding 2000 characters", async () => {
const longContent = "a".repeat(2001);
const result = await executeTool(tool, {
to_id: "agent-2",
content: longContent,
});
expect(result.content[0]).toEqual({ type: "text", text: "ERROR: Message content exceeds 2000 character limit" });
expect(messageStore.sendMessage).not.toHaveBeenCalled();
});
it("returns error when messageStore.sendMessage throws", async () => {
vi.mocked(messageStore.sendMessage).mockImplementation(() => {
throw new Error("Database error");
});
const result = await executeTool(tool, {
to_id: "agent-2",
content: "Test",
});
expect(result.content[0]).toEqual({ type: "text", text: "ERROR: Failed to send message: Database error" });
});
it("trims content before validation", async () => {
const mockMessage = createMessage();
vi.mocked(messageStore.sendMessage).mockReturnValue(mockMessage);
const result = await executeTool(tool, {
to_id: "agent-2",
content: " test ",
});
expect(result.content[0]).toEqual({ type: "text", text: expect.stringContaining("Message sent") });
expect(messageStore.sendMessage).toHaveBeenCalledWith(
expect.objectContaining({ content: "test" })
);
});
});
describe("createReadMessagesTool", () => {
let messageStore: ReturnType<typeof createMockMessageStore>;
let tool: ReturnType<typeof createReadMessagesTool>;
beforeEach(() => {
messageStore = createMockMessageStore();
tool = createReadMessagesTool(messageStore, "agent-1");
});
// Helper to call tool execute with correct signature
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const executeTool = async (tool: any, params: unknown) => {
return tool.execute("call-1", params, undefined, undefined, undefined);
};
it("creates a tool with name 'read_messages'", () => {
expect(tool.name).toBe("read_messages");
});
it("creates a tool with correct label", () => {
expect(tool.label).toBe("Read Messages");
});
it("creates a tool with description mentioning unread messages", () => {
expect(tool.description).toContain("unread messages");
});
it("calls messageStore.getInbox with correct agent ID", async () => {
await executeTool(tool, {});
expect(messageStore.getInbox).toHaveBeenCalledWith("agent-1", "agent", expect.any(Object));
});
it("defaults to unread_only: true", async () => {
await executeTool(tool, {});
expect(messageStore.getInbox).toHaveBeenCalledWith("agent-1", "agent", {
read: false,
limit: 20,
});
});
it("uses provided unread_only value", async () => {
await executeTool(tool, { unread_only: false });
expect(messageStore.getInbox).toHaveBeenCalledWith("agent-1", "agent", {
limit: 20,
});
});
it("uses provided limit value", async () => {
await executeTool(tool, { limit: 5 });
expect(messageStore.getInbox).toHaveBeenCalledWith("agent-1", "agent", {
read: false,
limit: 5,
});
});
it("returns 'No messages' when inbox is empty", async () => {
vi.mocked(messageStore.getInbox).mockReturnValue([]);
const result = await executeTool(tool, {});
expect(result.content[0]).toEqual({ type: "text", text: "No messages" });
});
it("returns formatted message list with sender, content, and timestamp", async () => {
const messages = [
createMessage({
id: "msg-1",
fromId: "agent-2",
content: "Hello there",
createdAt: "2024-01-15T10:30:00.000Z",
read: false,
}),
createMessage({
id: "msg-2",
fromId: "user-1",
content: "Another message",
createdAt: "2024-01-15T11:00:00.000Z",
read: true,
}),
];
vi.mocked(messageStore.getInbox).mockReturnValue(messages);
const result = await executeTool(tool, {});
const text = result.content[0];
expect(text).toMatchObject({ type: "text" });
expect((text as { text: string }).text).toContain("Messages (2)");
expect((text as { text: string }).text).toContain("[unread] [from: agent-2] Hello there");
expect((text as { text: string }).text).toContain("[read] [from: user-1] Another message");
expect(result.details).toEqual({ messages });
});
it("returns error when messageStore.getInbox throws", async () => {
vi.mocked(messageStore.getInbox).mockImplementation(() => {
throw new Error("Database error");
});
const result = await executeTool(tool, {});
expect(result.content[0]).toEqual({ type: "text", text: "ERROR: Failed to read messages: Database error" });
});
it("uses default limit of 20", async () => {
vi.mocked(messageStore.getInbox).mockReturnValue([]);
await executeTool(tool, { unread_only: false });
expect(messageStore.getInbox).toHaveBeenCalledWith("agent-1", "agent", { limit: 20 });
});
});
describe("sendMessageParams schema", () => {
it("is defined and exported", () => {
expect(sendMessageParams).toBeDefined();
});
});
describe("readMessagesParams schema", () => {
it("is defined and exported", () => {
expect(readMessagesParams).toBeDefined();
});
});

133
packages/engine/src/agent-tools.ts
View File

@@ -7,7 +7,7 @@
* The parameter schemas are canonical here — executor.ts imports and reuses them.
*/
import type { AgentStore, AgentState, AgentCapability, TaskDocument, TaskDocumentCreateInput, TaskStore, RunMutationContext } from "@fusion/core";
import type { AgentStore, AgentState, AgentCapability, TaskDocument, TaskDocumentCreateInput, TaskStore, RunMutationContext, MessageStore, Message } from "@fusion/core";
import { isEphemeralAgent } from "@fusion/core";
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
import { Type, type Static } from "@mariozechner/pi-ai";
@@ -67,6 +67,20 @@ export const delegateTaskParams = Type.Object({
),
});
export const sendMessageParams = Type.Object({
to_id: Type.String({ description: "Recipient agent ID (e.g. 'agent-abc123')" }),
content: Type.String({ description: "Message body (1-2000 characters)" }),
type: Type.Optional(Type.Union([
Type.Literal("agent-to-agent"),
Type.Literal("agent-to-user"),
], { description: "Message type (defaults to 'agent-to-agent')" })),
});
export const readMessagesParams = Type.Object({
unread_only: Type.Optional(Type.Boolean({ description: "Only return unread messages (default: true)" })),
limit: Type.Optional(Type.Number({ description: "Max messages to return (default: 20)" })),
});
// ── Tool factory functions ────────────────────────────────────────────────
/**
@@ -426,3 +440,120 @@ export function createDelegateTaskTool(agentStore: AgentStore, taskStore: TaskSt
},
};
}
/**
* Create a `send_message` tool that sends a message to another agent or user.
*
* @param messageStore - MessageStore for message persistence
* @param fromAgentId - The agent ID sending the message
* @returns ToolDefinition for the `send_message` tool
*/
export function createSendMessageTool(messageStore: MessageStore, fromAgentId: string): ToolDefinition {
return {
name: "send_message",
label: "Send Message",
description:
"Send a message to another agent or user. The recipient will be woken if they have " +
"`messageResponseMode: 'immediate'` configured.",
parameters: sendMessageParams,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
execute: async (_id: string, params: Static<typeof sendMessageParams>, _signal?: any, _onUpdate?: any, _ctx?: any) => {
// Validate content length
const content = params.content.trim();
if (content.length === 0) {
return {
content: [{ type: "text" as const, text: "ERROR: Message content cannot be empty" }],
details: {},
};
}
if (content.length > 2000) {
return {
content: [{ type: "text" as const, text: "ERROR: Message content exceeds 2000 character limit" }],
details: {},
};
}
try {
const message = messageStore.sendMessage({
fromId: fromAgentId,
fromType: "agent",
toId: params.to_id,
toType: "agent",
content,
type: params.type ?? "agent-to-agent",
});
return {
content: [{
type: "text" as const,
text: `Message sent to ${params.to_id} (ID: ${message.id})`,
}],
details: { messageId: message.id },
};
} catch (err) {
const errorMessage = err instanceof Error ? err.message : String(err);
return {
content: [{ type: "text" as const, text: `ERROR: Failed to send message: ${errorMessage}` }],
details: {},
};
}
},
};
}
/**
* Create a `read_messages` tool that reads inbox messages for an agent.
*
* @param messageStore - MessageStore for message retrieval
* @param agentId - The agent ID whose inbox to read
* @returns ToolDefinition for the `read_messages` tool
*/
export function createReadMessagesTool(messageStore: MessageStore, agentId: string): ToolDefinition {
return {
name: "read_messages",
label: "Read Messages",
description: "Read your inbox messages. Returns unread messages by default.",
parameters: readMessagesParams,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
execute: async (_id: string, params: Static<typeof readMessagesParams>, _signal?: any, _onUpdate?: any, _ctx?: any) => {
const unreadOnly = params.unread_only ?? true;
const limit = params.limit ?? 20;
try {
const filter = {
...(unreadOnly ? { read: false as const } : {}),
limit,
};
const messages = messageStore.getInbox(agentId, "agent", filter);
if (messages.length === 0) {
return {
content: [{ type: "text" as const, text: "No messages" }],
details: {},
};
}
const lines = messages.map((msg: Message) => {
const timestamp = new Date(msg.createdAt).toLocaleString();
const readStatus = msg.read ? "[read] " : "[unread] ";
return `${readStatus}[from: ${msg.fromId}] ${msg.content} (${timestamp})`;
});
return {
content: [{
type: "text" as const,
text: `Messages (${messages.length}):\n${lines.join("\n")}`,
}],
details: { messages },
};
} catch (err) {
const errorMessage = err instanceof Error ? err.message : String(err);
return {
content: [{ type: "text" as const, text: `ERROR: Failed to read messages: ${errorMessage}` }],
details: {},
};
}
},
};
}

9
packages/engine/src/executor.ts
View File

@@ -35,6 +35,7 @@ import {
createDelegateTaskTool,
createListAgentsTool,
createReflectOnPerformanceTool,
createSendMessageTool,
createTaskCreateTool as sharedCreateTaskCreateTool,
createTaskDocumentReadTool as sharedCreateTaskDocumentReadTool,
createTaskDocumentWriteTool as sharedCreateTaskDocumentWriteTool,
@@ -47,12 +48,14 @@ export { summarizeToolArgs } from "./agent-logger.js";
export {
createDelegateTaskTool,
createListAgentsTool,
createSendMessageTool,
createTaskCreateTool,
createTaskDocumentReadTool,
createTaskDocumentWriteTool,
createTaskLogTool,
delegateTaskParams,
listAgentsParams,
sendMessageParams,
taskCreateParams,
taskLogParams,
} from "./agent-tools.js";
@@ -314,6 +317,8 @@ export interface TaskExecutorOptions {
reflectionService?: AgentReflectionService;
/** Plugin runner for invoking plugin hooks and providing plugin tools. */
pluginRunner?: PluginRunner;
/** MessageStore for sending messages to other agents. When provided, executor agents gain send_message capability. */
messageStore?: import("@fusion/core").MessageStore;
missionStore?: MissionStore;
onSliceComplete?: (slice: Slice) => void;
onStart?: (task: Task, worktreePath: string) => void;
@@ -1415,6 +1420,10 @@ export class TaskExecutor {
createListAgentsTool(this.options.agentStore),
createDelegateTaskTool(this.options.agentStore, this.store),
] : []),
// Messaging tool — allows executor agents to send messages to other agents.
...(this.options.messageStore && assignedAgentId ? [
createSendMessageTool(this.options.messageStore, assignedAgentId),
] : []),
// Add plugin tools from PluginRunner
...(this.options.pluginRunner?.getPluginTools() ?? []),
];