FN-8208: validate agent message recipients before delivery
Prevent false-success agent messages by validating recipients before delivery. - Wire AgentStore into every send-message tool registration. - Reject missing or unvalidated agent recipients before persistence or wake-up. - Add recipient validation coverage and a patch changeset. Files changed: .../fn-8208-send-message-recipient-validation.md | 7 ++ .../chat-send-message-agentstore-wiring.test.ts | 11 ++ packages/dashboard/src/chat.ts | 2 +- ...tools-send-message-recipient-validation.test.ts | 116 +++++++++++++++++++++ packages/engine/src/agent-heartbeat.ts | 4 +- packages/engine/src/agent-tools.ts | 24 ++++- packages/engine/src/executor.ts | 2 +- packages/engine/src/step-session-executor.ts | 2 +- 8 files changed, 162 insertions(+), 6 deletions(-) Fusion-Task-Id: FN-8208 Fusion-Task-Lineage: 84752bb8-c9f7-42bd-87c1-9681ac442789 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-8208-send-message-recipient-validation.md
Normal file
7
.changeset/fn-8208-send-message-recipient-validation.md
Normal file
@@ -0,0 +1,7 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
summary: Reject messages addressed to nonexistent agent recipients.
|
||||
category: fix
|
||||
dev: fn_send_message now validates agent recipients through the async AgentStore lookup before delivery.
|
||||
@@ -0,0 +1,11 @@
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { describe, expect, it } from "vitest";
|
||||
|
||||
describe("dashboard chat send-message registration", () => {
|
||||
it("forwards ChatManager's AgentStore to the messaging tool factory", async () => {
|
||||
const source = await readFile(fileURLToPath(new URL("../chat.ts", import.meta.url)), "utf8");
|
||||
|
||||
expect(source).toContain("createSendMessageTool(this.messageStore, agent.id, { agentStore: this.agentStore })");
|
||||
});
|
||||
});
|
||||
@@ -2375,7 +2375,7 @@ export class ChatManager {
|
||||
|
||||
const messagingTools = agent?.id && this.messageStore
|
||||
? [
|
||||
createSendMessageTool(this.messageStore, agent.id),
|
||||
createSendMessageTool(this.messageStore, agent.id, { agentStore: this.agentStore }),
|
||||
createReadMessagesTool(this.messageStore, agent.id),
|
||||
]
|
||||
: [];
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
import { readFile } from "node:fs/promises";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { createSendMessageTool } from "../agent-tools.js";
|
||||
|
||||
function firstText(result: { content: Array<{ type: string; text?: string }> }): string {
|
||||
const first = result.content[0];
|
||||
return first?.type === "text" ? (first.text ?? "") : "";
|
||||
}
|
||||
|
||||
function createMessageStoreHarness() {
|
||||
const wakeSpy = vi.fn();
|
||||
const sendMessage = vi.fn(async (input: Record<string, unknown>) => {
|
||||
if (input.toType === "agent") {
|
||||
await wakeSpy(input);
|
||||
}
|
||||
return { id: "msg-1" };
|
||||
});
|
||||
return { messageStore: { sendMessage }, sendMessage, wakeSpy };
|
||||
}
|
||||
|
||||
async function executeSend(
|
||||
tool: ReturnType<typeof createSendMessageTool>,
|
||||
params: Record<string, unknown>,
|
||||
) {
|
||||
return tool.execute("1", params as never, undefined, undefined, {});
|
||||
}
|
||||
|
||||
describe("createSendMessageTool recipient validation", () => {
|
||||
it("rejects a missing agent before persistence or wake", async () => {
|
||||
const { messageStore, sendMessage, wakeSpy } = createMessageStoreHarness();
|
||||
const agentStore = { getAgent: vi.fn().mockResolvedValue(null) };
|
||||
const tool = createSendMessageTool(messageStore as never, "agent-a", { agentStore: agentStore as never });
|
||||
|
||||
const result = await executeSend(tool, { to_id: "agent-does-not-exist", content: "hello", type: "agent-to-agent" });
|
||||
|
||||
expect(firstText(result as never)).toMatch(/^ERROR: Recipient agent 'agent-does-not-exist' does not exist/);
|
||||
expect(agentStore.getAgent).toHaveBeenCalledWith("agent-does-not-exist");
|
||||
expect(sendMessage).not.toHaveBeenCalled();
|
||||
expect(wakeSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("uses async getAgent and still sends and wakes an existing recipient", async () => {
|
||||
const { messageStore, sendMessage, wakeSpy } = createMessageStoreHarness();
|
||||
// Mimics PostgreSQL mode: only async getAgent is available; there is no sync cache.
|
||||
const agentStore = { getAgent: vi.fn().mockResolvedValue({ id: "agent-b" }) };
|
||||
const tool = createSendMessageTool(messageStore as never, "agent-a", { agentStore: agentStore as never });
|
||||
|
||||
const result = await executeSend(tool, { to_id: "agent-b", content: "hello" });
|
||||
|
||||
expect(firstText(result as never)).toContain("Message sent to agent-b");
|
||||
expect(agentStore.getAgent).toHaveBeenCalledWith("agent-b");
|
||||
expect(sendMessage).toHaveBeenCalledTimes(1);
|
||||
expect(wakeSpy).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("does not resolve user recipients and preserves no-resolver behavior", async () => {
|
||||
const userHarness = createMessageStoreHarness();
|
||||
const agentStore = { getAgent: vi.fn() };
|
||||
const userTool = createSendMessageTool(userHarness.messageStore as never, "agent-a", { agentStore: agentStore as never });
|
||||
|
||||
const userResult = await executeSend(userTool, { to_id: "dashboard-user", content: "hello", type: "agent-to-user" });
|
||||
|
||||
expect(firstText(userResult as never)).toContain("Message sent to dashboard-user");
|
||||
expect(agentStore.getAgent).not.toHaveBeenCalled();
|
||||
expect(userHarness.sendMessage).toHaveBeenCalledTimes(1);
|
||||
|
||||
const legacyHarness = createMessageStoreHarness();
|
||||
const legacyTool = createSendMessageTool(legacyHarness.messageStore as never, "agent-a");
|
||||
const legacyResult = await executeSend(legacyTool, { to_id: "unknown-agent", content: "hello" });
|
||||
expect(firstText(legacyResult as never)).toContain("Message sent to unknown-agent");
|
||||
expect(legacyHarness.sendMessage).toHaveBeenCalledTimes(1);
|
||||
|
||||
expect(firstText(await executeSend(legacyTool, { to_id: "agent-b", content: " " }) as never)).toBe("ERROR: Message content cannot be empty");
|
||||
expect(firstText(await executeSend(legacyTool, { to_id: "agent-b", content: "hello", reply_to_message_id: " " }) as never)).toBe("ERROR: reply_to_message_id must be a non-empty string");
|
||||
});
|
||||
|
||||
it("does not report delivery when recipient validation is unavailable", async () => {
|
||||
const failedLookupHarness = createMessageStoreHarness();
|
||||
const failedLookupTool = createSendMessageTool(failedLookupHarness.messageStore as never, "agent-a", {
|
||||
agentStore: { getAgent: vi.fn().mockRejectedValue(new Error("database unavailable")) } as never,
|
||||
});
|
||||
|
||||
const result = await executeSend(failedLookupTool, { to_id: "agent-b", content: "hello" });
|
||||
expect(firstText(result as never)).toMatch(/^ERROR: Recipient agent 'agent-b' could not be validated/);
|
||||
expect(failedLookupHarness.sendMessage).not.toHaveBeenCalled();
|
||||
expect(failedLookupHarness.wakeSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rejects an undefined recipient lookup result", async () => {
|
||||
const { messageStore, sendMessage, wakeSpy } = createMessageStoreHarness();
|
||||
const tool = createSendMessageTool(messageStore as never, "agent-a", {
|
||||
agentStore: { getAgent: vi.fn().mockResolvedValue(undefined) } as never,
|
||||
});
|
||||
|
||||
const result = await executeSend(tool, { to_id: "agent-does-not-exist", content: "hello" });
|
||||
expect(firstText(result as never)).toMatch(/^ERROR: Recipient agent 'agent-does-not-exist' does not exist/);
|
||||
expect(sendMessage).not.toHaveBeenCalled();
|
||||
expect(wakeSpy).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("send-message registration wiring", () => {
|
||||
it("forwards the in-scope AgentStore at every engine registration", async () => {
|
||||
const [executor, stepSessionExecutor, heartbeat] = await Promise.all([
|
||||
readFile(fileURLToPath(new URL("../executor.ts", import.meta.url)), "utf8"),
|
||||
readFile(fileURLToPath(new URL("../step-session-executor.ts", import.meta.url)), "utf8"),
|
||||
readFile(fileURLToPath(new URL("../agent-heartbeat.ts", import.meta.url)), "utf8"),
|
||||
]);
|
||||
|
||||
expect(executor).toContain("createSendMessageTool(this.options.messageStore, assignedAgentId, { autoRecovery: settings.autoRecovery, runAudit: audit, taskStore: this.store, settings, agentStore: this.options.agentStore })");
|
||||
expect(stepSessionExecutor).toContain("createSendMessageTool(this.options.messageStore, taskDetail.assignedAgentId, { autoRecovery: settings.autoRecovery, taskStore: this.options.store!, settings, agentStore: this.options.agentStore })");
|
||||
expect(heartbeat.match(/createSendMessageTool\(this\.messageStore, agentId, \{ agentStore: this\.store \}\)/g)).toHaveLength(1);
|
||||
expect(heartbeat.match(/createSendMessageTool\(messageStore, agentId, \{ agentStore: this\.store \}\)/g)).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
@@ -2503,7 +2503,7 @@ export class HeartbeatMonitor {
|
||||
|
||||
// Messaging tools — when MessageStore is available
|
||||
if (this.messageStore) {
|
||||
heartbeatTools.push(createSendMessageTool(this.messageStore, agentId));
|
||||
heartbeatTools.push(createSendMessageTool(this.messageStore, agentId, { agentStore: this.store }));
|
||||
heartbeatTools.push(createReadMessagesTool(this.messageStore, agentId));
|
||||
}
|
||||
if (this.chatStore) {
|
||||
@@ -3764,7 +3764,7 @@ export class HeartbeatMonitor {
|
||||
|
||||
// Messaging tools — when MessageStore is available, agents can send and receive messages
|
||||
if (messageStore) {
|
||||
tools.push(createSendMessageTool(messageStore, agentId));
|
||||
tools.push(createSendMessageTool(messageStore, agentId, { agentStore: this.store }));
|
||||
tools.push(createReadMessagesTool(messageStore, agentId));
|
||||
}
|
||||
if (this.chatStore) {
|
||||
|
||||
@@ -4214,7 +4214,7 @@ export function createAskQuestionTool(): ToolDefinition {
|
||||
export function createSendMessageTool(
|
||||
messageStore: MessageStore,
|
||||
fromAgentId: string,
|
||||
options?: { autoRecovery?: ProjectSettings["autoRecovery"]; runAudit?: RunAuditor; taskStore?: TaskStore; settings?: Settings },
|
||||
options?: { autoRecovery?: ProjectSettings["autoRecovery"]; runAudit?: RunAuditor; taskStore?: TaskStore; settings?: Settings; agentStore?: AgentStore },
|
||||
): ToolDefinition {
|
||||
const deliveryHandler = new MessageDeliveryAutoRecoveryHandler({
|
||||
runAudit: options?.runAudit ?? { database: async () => {}, git: async () => {}, filesystem: async () => {}, sandbox: async () => {} },
|
||||
@@ -4261,6 +4261,28 @@ export function createSendMessageTool(
|
||||
};
|
||||
}
|
||||
|
||||
/*
|
||||
FNXC:AgentMessaging 2026-07-28-12:10:
|
||||
Agent-to-agent sends must reject missing recipients rather than store an unread, undeliverable message and report false delivery success. Use async getAgent instead of getCachedAgent because the synchronous cache always returns null in PostgreSQL mode. A lookup failure is validation-unavailable and must block the send; only a successful lookup may establish delivery confidence.
|
||||
*/
|
||||
if (recipient.type === "agent" && options?.agentStore) {
|
||||
let resolvedRecipient: Awaited<ReturnType<AgentStore["getAgent"]>> | undefined;
|
||||
try {
|
||||
resolvedRecipient = await options.agentStore.getAgent(recipient.id);
|
||||
} catch {
|
||||
return {
|
||||
content: [{ type: "text" as const, text: `ERROR: Recipient agent '${params.to_id}' could not be validated — message not sent` }],
|
||||
details: {},
|
||||
};
|
||||
}
|
||||
if (resolvedRecipient == null) {
|
||||
return {
|
||||
content: [{ type: "text" as const, text: `ERROR: Recipient agent '${params.to_id}' does not exist — message not sent` }],
|
||||
details: {},
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
const result = await deliveryHandler.runWithBoundedRetry({
|
||||
run: async () => messageStore.sendMessage({
|
||||
fromId: fromAgentId,
|
||||
|
||||
@@ -11674,7 +11674,7 @@ export class TaskExecutor {
|
||||
] : []),
|
||||
// Messaging tools — allows executor agents to send and receive messages.
|
||||
...(this.options.messageStore && assignedAgentId ? [
|
||||
createSendMessageTool(this.options.messageStore, assignedAgentId, { autoRecovery: settings.autoRecovery, runAudit: audit, taskStore: this.store, settings }),
|
||||
createSendMessageTool(this.options.messageStore, assignedAgentId, { autoRecovery: settings.autoRecovery, runAudit: audit, taskStore: this.store, settings, agentStore: this.options.agentStore }),
|
||||
createReadMessagesTool(this.options.messageStore, assignedAgentId),
|
||||
] : []),
|
||||
// Add plugin tools from PluginRunner
|
||||
|
||||
@@ -1321,7 +1321,7 @@ export class StepSessionExecutor {
|
||||
const messagingTools =
|
||||
this.options.messageStore && taskDetail.assignedAgentId
|
||||
? [
|
||||
createSendMessageTool(this.options.messageStore, taskDetail.assignedAgentId, { autoRecovery: settings.autoRecovery, taskStore: this.options.store!, settings }),
|
||||
createSendMessageTool(this.options.messageStore, taskDetail.assignedAgentId, { autoRecovery: settings.autoRecovery, taskStore: this.options.store!, settings, agentStore: this.options.agentStore }),
|
||||
createReadMessagesTool(this.options.messageStore, taskDetail.assignedAgentId),
|
||||
]
|
||||
: [];
|
||||
|
||||
Reference in New Issue
Block a user