feat(FN-4177): add heartbeat messaging support for chat rooms

The merge delivers room heartbeat messaging for agents, allowing heartbeat-triggered agents to send messages into chat rooms and subscribe to room activity. Core changes include new chat-store room APIs, engine heartbeat room message tools, and corresponding tests, with a small dashboard ChatView up

Fusion-Task-Id: FN-4177
This commit is contained in:
Fusion
2026-05-12 18:16:49 -07:00
committed by gsxdsm
parent df52cc2e84
commit e7fc3a9d17
13 changed files with 632 additions and 16 deletions

View File

@@ -0,0 +1,5 @@
---
"@runfusion/fusion": patch
---
Surface pending chat-room messages during agent heartbeats and let permanent agents reply with a new `fn_post_room_message` tool.

View File

@@ -825,11 +825,14 @@ This normalization applies on send and mailbox reads, so replies from agents sti
### How It Works ### How It Works
Heartbeat runs now surface both direct-message inbox traffic and recent room activity for rooms the agent belongs to. Room traffic is lookback-based (bounded to the prior completed heartbeat / `lastHeartbeatAt`, capped at 24 hours) and is only shown when there are unread/recent messages worth surfacing.
1. **Message Prefetch**: When `messageStore` is available, heartbeat runs fetch up to 10 unread inbox messages for the agent. 1. **Message Prefetch**: When `messageStore` is available, heartbeat runs fetch up to 10 unread inbox messages for the agent.
2. **Prompt Injection**: Pending messages are injected into the execution prompt with message ID, sender, and timestamp information. 2. **Room Prefetch**: When `chatStore` is available, heartbeat runs fetch up to 10 recent room messages per active room (30 total max, self-authored room messages excluded).
3. **Reply Guidance**: System instructions remind agents to reply with `reply_to_message_id` for linked threads. 3. **Prompt Injection**: Pending messages are injected into the execution prompt with message ID, sender, and timestamp information, followed by a **Pending Room Messages** section grouped by room.
4. **Mark as Read**: After successful heartbeat completion, messages are marked as read. 4. **Reply Guidance**: System instructions remind agents to reply with `reply_to_message_id` for direct messages and use `fn_post_room_message` only when room content is relevant to the agents role/identity.
5. **Failed Runs**: If the heartbeat execution fails, messages remain unread for retry on the next run. 5. **Mark as Read**: After successful heartbeat completion, direct inbox messages are marked as read.
6. **Failed Runs**: If the heartbeat execution fails, inbox messages remain unread for retry on the next run.
### Message Response Modes ### Message Response Modes

View File

@@ -34,6 +34,7 @@ These tools are **not** part of the user-invokable extension surface. They are i
| `fn_agent_delete` | executor, heartbeat | Delete a non-ephemeral direct-report agent | `agent_id` (string), optional: `force` (boolean), `reassign_to` (string) | | `fn_agent_delete` | executor, heartbeat | Delete a non-ephemeral direct-report agent | `agent_id` (string), optional: `force` (boolean), `reassign_to` (string) |
| `fn_send_message` | executor, step-session, heartbeat | Send inbox messages to agents/users | `to_id` (string), `content` (string), `type?` (`agent-to-agent` \| `agent-to-user`), `reply_to_message_id?` (string) | | `fn_send_message` | executor, step-session, heartbeat | Send inbox messages to agents/users | `to_id` (string), `content` (string), `type?` (`agent-to-agent` \| `agent-to-user`), `reply_to_message_id?` (string) |
| `fn_read_messages` | executor, step-session, heartbeat | Read inbox messages | `unread_only?` (boolean), `limit?` (number) | | `fn_read_messages` | executor, step-session, heartbeat | Read inbox messages | `unread_only?` (boolean), `limit?` (number) |
| `fn_post_room_message` | heartbeat | Post a message to a chat room the agent is a member of | `roomId` (string), `content` (string), `replyToMessageId?` (string), `mentions?` (string[]) |
## Triage-only runtime tools (`triage.ts`) ## Triage-only runtime tools (`triage.ts`)

View File

@@ -104,6 +104,48 @@ describe("ChatStore — rooms (FN-3805..FN-3811 contract)", () => {
expect(updated.attachments?.[0]?.id).toBe("att-room"); expect(updated.attachments?.[0]?.id).toBe("att-room");
}); });
it("returns only messages after sinceIso", async () => {
const room = store.createRoom({ name: "since-test" });
store.addRoomMessage(room.id, { role: "user", content: "before" });
await new Promise((r) => setTimeout(r, 5));
const sinceIso = new Date().toISOString();
await new Promise((r) => setTimeout(r, 5));
const after = store.addRoomMessage(room.id, { role: "user", content: "after" });
expect(store.listRoomMessagesSince(room.id, sinceIso).map((message) => message.id)).toEqual([after.id]);
});
it("excludes authored agent messages when excludeSenderAgentId is set", async () => {
const room = store.createRoom({ name: "exclude-self" });
store.addRoomMessage(room.id, { role: "assistant", content: "own", senderAgentId: "agent-1" });
await new Promise((r) => setTimeout(r, 5));
const other = store.addRoomMessage(room.id, { role: "assistant", content: "other", senderAgentId: "agent-2" });
const user = store.addRoomMessage(room.id, { role: "user", content: "user" });
expect(
store.listRoomMessagesSince(room.id, "1970-01-01T00:00:00.000Z", { excludeSenderAgentId: "agent-1" }).map((message) => message.id),
).toEqual([other.id, user.id]);
});
it("respects the limit cap", async () => {
const room = store.createRoom({ name: "limit-test" });
store.addRoomMessage(room.id, { role: "user", content: "one" });
store.addRoomMessage(room.id, { role: "user", content: "two" });
store.addRoomMessage(room.id, { role: "user", content: "three" });
expect(store.listRoomMessagesSince(room.id, "1970-01-01T00:00:00.000Z", { limit: 2 }).map((message) => message.content)).toEqual([
"one",
"two",
]);
});
it("returns empty when there are no new room messages", () => {
const room = store.createRoom({ name: "empty-test" });
store.addRoomMessage(room.id, { role: "user", content: "old" });
expect(store.listRoomMessagesSince(room.id, new Date().toISOString())).toEqual([]);
});
it("keeps cross-room and direct-vs-room histories isolated", () => { it("keeps cross-room and direct-vs-room histories isolated", () => {
const session = store.createSession({ agentId: "agent-1" }); const session = store.createSession({ agentId: "agent-1" });
store.addMessage(session.id, { role: "user", content: "direct" }); store.addMessage(session.id, { role: "user", content: "direct" });

View File

@@ -936,6 +936,29 @@ export class ChatStore extends EventEmitter<ChatStoreEvents> {
return rows.map((row) => this.rowToRoomMessage(row)); return rows.map((row) => this.rowToRoomMessage(row));
} }
listRoomMessagesSince(
roomId: string,
sinceIso: string,
options?: { excludeSenderAgentId?: string; limit?: number },
): ChatRoomMessage[] {
const whereClauses: string[] = ["roomId = ?", "createdAt > ?"];
const params: Array<string | number | null> = [roomId, sinceIso];
if (options?.excludeSenderAgentId) {
whereClauses.push("(senderAgentId IS NULL OR senderAgentId != ?)");
params.push(options.excludeSenderAgentId);
}
const rows = this.db.prepare(`
SELECT * FROM chat_room_messages
WHERE ${whereClauses.join(" AND ")}
ORDER BY createdAt ASC
LIMIT ?
`).all(...params, options?.limit ?? 50) as ChatRoomMessageRow[];
return rows.map((row) => this.rowToRoomMessage(row));
}
getRoomMessage(id: string): ChatRoomMessage | undefined { getRoomMessage(id: string): ChatRoomMessage | undefined {
const row = this.db.prepare("SELECT * FROM chat_room_messages WHERE id = ?").get(id) as ChatRoomMessageRow | undefined; const row = this.db.prepare("SELECT * FROM chat_room_messages WHERE id = ?").get(id) as ChatRoomMessageRow | undefined;
return row ? this.rowToRoomMessage(row) : undefined; return row ? this.rowToRoomMessage(row) : undefined;

View File

@@ -2203,7 +2203,7 @@ export function ChatView({ projectId, addToast, experimentalFeatures }: ChatView
<div className="chat-session-list chat-sidebar-list"> <div className="chat-session-list chat-sidebar-list">
{rooms.rooms.map((room) => { {rooms.rooms.map((room) => {
const isActive = rooms.activeRoom?.id === room.id; const isActive = rooms.activeRoom?.id === room.id;
const memberCount = isActive ? rooms.activeRoomMembers.length : "—"; const memberCount = isActive ? rooms.activeRoomMembers.length : null;
return ( return (
<div <div
key={room.id} key={room.id}
@@ -2229,7 +2229,9 @@ export function ChatView({ projectId, addToast, experimentalFeatures }: ChatView
> >
<span className="chat-room-item-details"> <span className="chat-room-item-details">
<span className="chat-room-item-name">#{room.name}</span> <span className="chat-room-item-name">#{room.name}</span>
<span className="chat-room-item-meta">{memberCount} {memberCount === 1 ? "member" : "members"}</span> {memberCount !== null ? (
<span className="chat-room-item-meta">{memberCount} {memberCount === 1 ? "member" : "members"}</span>
) : null}
</span> </span>
<button <button
type="button" type="button"

View File

@@ -12,6 +12,7 @@ import {
createTaskLogToolWithContext, createTaskLogToolWithContext,
createSendMessageTool, createSendMessageTool,
createReadMessagesTool, createReadMessagesTool,
createPostRoomMessageTool,
createResearchTools, createResearchTools,
qmdAgentMemoryCollectionName, qmdAgentMemoryCollectionName,
readAgentMemoryWorkspaceLongTerm, readAgentMemoryWorkspaceLongTerm,
@@ -19,6 +20,7 @@ import {
readMessagesParams, readMessagesParams,
} from "../agent-tools.js"; } from "../agent-tools.js";
import * as core from "@fusion/core"; import * as core from "@fusion/core";
import { ChatStore, Database } from "@fusion/core";
import type { MessageStore, Message } from "@fusion/core"; import type { MessageStore, Message } from "@fusion/core";
import { getEnabledPluginTools, getResearchToolSurfaceStatus } from "../tool-availability.js"; import { getEnabledPluginTools, getResearchToolSurfaceStatus } from "../tool-availability.js";
@@ -956,6 +958,75 @@ describe("createSendMessageTool", () => {
}); });
}); });
describe("createPostRoomMessageTool", () => {
let roomDir: string;
let fusionDir: string;
let db: Database;
let chatStore: ChatStore;
beforeEach(() => {
roomDir = join(tmpdir(), `fusion-room-tool-${Date.now()}-${Math.random().toString(16).slice(2)}`);
fusionDir = join(roomDir, ".fusion");
db = new Database(fusionDir, { inMemory: true });
db.init();
chatStore = new ChatStore(fusionDir, db);
});
afterEach(() => {
db.close();
});
it("posts to a room when the agent is a member", async () => {
const room = chatStore.createRoom({ name: "ops", memberAgentIds: ["agent-sender"] });
const tool = createPostRoomMessageTool(chatStore, "agent-sender");
const result = await tool.execute("call-1", {
roomId: room.id,
content: " Ready to help ",
mentions: ["agent-2"],
} as any, undefined, undefined, {} as any);
const messages = chatStore.getRoomMessages(room.id);
expect(messages).toHaveLength(1);
expect(messages[0]).toMatchObject({
role: "assistant",
senderAgentId: "agent-sender",
content: "Ready to help",
mentions: ["agent-2"],
});
expect(result.details).toEqual({ messageId: messages[0]!.id });
});
it("blocks non-members from posting", async () => {
const room = chatStore.createRoom({ name: "ops", memberAgentIds: ["agent-other"] });
const tool = createPostRoomMessageTool(chatStore, "agent-sender");
const result = await tool.execute("call-1", {
roomId: room.id,
content: "Hello",
} as any, undefined, undefined, {} as any);
expect(result.content[0]).toEqual({
type: "text",
text: `ERROR: Agent agent-sender is not a member of room ${room.id}`,
});
expect(chatStore.getRoomMessages(room.id)).toHaveLength(0);
});
it("stores reply metadata when replyToMessageId is provided", async () => {
const room = chatStore.createRoom({ name: "ops", memberAgentIds: ["agent-sender"] });
const tool = createPostRoomMessageTool(chatStore, "agent-sender");
await tool.execute("call-1", {
roomId: room.id,
content: "Replying",
replyToMessageId: "rmsg-parent",
} as any, undefined, undefined, {} as any);
expect(chatStore.getRoomMessages(room.id)[0]?.metadata).toEqual({ replyToMessageId: "rmsg-parent" });
});
});
describe("createResearchTools", () => { describe("createResearchTools", () => {
const baseSettings = { const baseSettings = {
researchGlobalEnabled: true, researchGlobalEnabled: true,

View File

@@ -0,0 +1,220 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { AgentStore, ChatStore, TaskStore } from "@fusion/core";
import { HeartbeatMonitor } from "../agent-heartbeat.js";
const sessionCapture = vi.hoisted(() => ({
prompt: "",
customTools: [] as Array<{ name: string; execute: (...args: any[]) => Promise<any> }>,
}));
vi.mock("../logger.js", async () => {
const { createMockLogger, formatMockError } = await import("./heartbeat-test-helpers.js");
return {
createLogger: vi.fn(() => createMockLogger()),
heartbeatLog: createMockLogger(),
formatError: formatMockError,
runtimeLog: createMockLogger(),
};
});
vi.mock("../pi.js", () => ({
promptWithFallback: vi.fn(async (session: any, prompt: string) => {
await session.prompt(prompt);
}),
}));
vi.mock("../agent-session-helpers.js", async () => {
const actual = await vi.importActual<typeof import("../agent-session-helpers.js")>("../agent-session-helpers.js");
return {
...actual,
createResolvedAgentSession: vi.fn(async (options: any) => {
sessionCapture.customTools = options.customTools ?? [];
return {
session: {
prompt: async (prompt: string) => {
sessionCapture.prompt = prompt;
},
dispose: vi.fn(),
getSessionStats: () => ({ tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 } }),
},
};
}),
};
});
type Harness = {
rootDir: string;
globalDir: string;
taskStore: TaskStore;
agentStore: AgentStore;
chatStore: ChatStore;
agentId: string;
};
async function createHarness(): Promise<Harness> {
const rootDir = mkdtempSync(join(tmpdir(), "hb-room-root-"));
const globalDir = mkdtempSync(join(tmpdir(), "hb-room-global-"));
const taskStore = new TaskStore(rootDir, globalDir, { inMemoryDb: true });
await taskStore.init();
const agentStore = new AgentStore({ rootDir: taskStore.getFusionDir(), taskStore, inMemoryDb: true });
const chatStore = new ChatStore(taskStore.getFusionDir(), taskStore.getDatabase());
const agent = await agentStore.createAgent({
name: "Room Heartbeat Agent",
role: "engineer",
soul: "Surfaces relevant room updates.",
runtimeConfig: { enabled: true },
});
return { rootDir, globalDir, taskStore, agentStore, chatStore, agentId: agent.id };
}
describe("heartbeat room messages", () => {
let harness: Harness | null = null;
beforeEach(() => {
sessionCapture.prompt = "";
sessionCapture.customTools = [];
});
afterEach(() => {
if (harness) {
rmSync(harness.rootDir, { recursive: true, force: true });
rmSync(harness.globalDir, { recursive: true, force: true });
harness = null;
}
});
it("omits room section and tool when no chatStore is configured", async () => {
harness = await createHarness();
const monitor = new HeartbeatMonitor({
store: harness.agentStore,
taskStore: harness.taskStore,
rootDir: harness.rootDir,
});
await monitor.executeHeartbeat({ agentId: harness.agentId, source: "timer" as any });
expect(sessionCapture.prompt).not.toContain("Pending Room Messages:");
expect(sessionCapture.customTools.map((tool) => tool.name)).not.toContain("fn_post_room_message");
});
it("shows only rooms with new messages", async () => {
harness = await createHarness();
const staleRoom = harness.chatStore.createRoom({ name: "stale-room", memberAgentIds: [harness.agentId] });
const freshRoom = harness.chatStore.createRoom({ name: "fresh-room", memberAgentIds: [harness.agentId] });
harness.chatStore.addRoomMessage(staleRoom.id, { role: "user", content: "too old" });
await new Promise((resolve) => setTimeout(resolve, 5));
const sinceIso = new Date().toISOString();
await harness.agentStore.saveRun({
id: "run-prev-fresh",
agentId: harness.agentId,
startedAt: new Date(Date.now() - 1_000).toISOString(),
endedAt: sinceIso,
status: "completed",
});
await new Promise((resolve) => setTimeout(resolve, 5));
const freshMessage = harness.chatStore.addRoomMessage(freshRoom.id, { role: "user", content: "needs review" });
const monitor = new HeartbeatMonitor({
store: harness.agentStore,
taskStore: harness.taskStore,
rootDir: harness.rootDir,
chatStore: harness.chatStore,
});
await monitor.executeHeartbeat({ agentId: harness.agentId, source: "timer" as any });
expect(sessionCapture.prompt).toContain("Pending Room Messages:");
expect(sessionCapture.prompt).toContain(`fresh-room (${freshRoom.id})`);
expect(sessionCapture.prompt).toContain(freshMessage.id);
expect(sessionCapture.prompt).not.toContain(`stale-room (${staleRoom.id})`);
});
it("excludes messages older than the lookback cutoff", async () => {
harness = await createHarness();
const room = harness.chatStore.createRoom({ name: "lookback", memberAgentIds: [harness.agentId] });
harness.chatStore.addRoomMessage(room.id, { role: "user", content: "old room note" });
await new Promise((resolve) => setTimeout(resolve, 5));
const cutoff = new Date().toISOString();
await harness.agentStore.saveRun({
id: "run-prev-lookback",
agentId: harness.agentId,
startedAt: new Date(Date.now() - 1_000).toISOString(),
endedAt: cutoff,
status: "completed",
});
await new Promise((resolve) => setTimeout(resolve, 5));
harness.chatStore.addRoomMessage(room.id, { role: "user", content: "fresh room note" });
const monitor = new HeartbeatMonitor({
store: harness.agentStore,
taskStore: harness.taskStore,
rootDir: harness.rootDir,
chatStore: harness.chatStore,
});
await monitor.executeHeartbeat({ agentId: harness.agentId, source: "timer" as any });
expect(sessionCapture.prompt).toContain("fresh room note");
expect(sessionCapture.prompt).not.toContain("old room note");
});
it("shows a truncated marker when total surfaced room messages overflow the cap", async () => {
harness = await createHarness();
for (let roomIndex = 0; roomIndex < 4; roomIndex += 1) {
const room = harness.chatStore.createRoom({ name: `overflow-${roomIndex}`, memberAgentIds: [harness.agentId] });
for (let messageIndex = 0; messageIndex < 10; messageIndex += 1) {
harness.chatStore.addRoomMessage(room.id, {
role: "user",
content: `message ${roomIndex}-${messageIndex}`,
});
}
}
const monitor = new HeartbeatMonitor({
store: harness.agentStore,
taskStore: harness.taskStore,
rootDir: harness.rootDir,
chatStore: harness.chatStore,
});
await monitor.executeHeartbeat({ agentId: harness.agentId, source: "timer" as any });
expect(sessionCapture.prompt).toContain("(10 more truncated)");
});
it("registers fn_post_room_message and posts through the real ChatStore", async () => {
harness = await createHarness();
const room = harness.chatStore.createRoom({ name: "reply-room", memberAgentIds: [harness.agentId] });
harness.chatStore.addRoomMessage(room.id, { role: "user", content: "can you confirm?" });
const monitor = new HeartbeatMonitor({
store: harness.agentStore,
taskStore: harness.taskStore,
rootDir: harness.rootDir,
chatStore: harness.chatStore,
});
await monitor.executeHeartbeat({ agentId: harness.agentId, source: "timer" as any });
const postTool = sessionCapture.customTools.find((tool) => tool.name === "fn_post_room_message");
expect(postTool).toBeDefined();
const result = await postTool!.execute("call-1", {
roomId: room.id,
content: "Confirmed.",
replyToMessageId: "rmsg-parent",
});
const posted = harness.chatStore.getRoomMessages(room.id).find((message) => message.id === result.details.messageId);
expect(posted).toMatchObject({
senderAgentId: harness.agentId,
content: "Confirmed.",
metadata: { replyToMessageId: "rmsg-parent" },
});
});
});

View File

@@ -16,6 +16,8 @@ import {
} from "../agent-heartbeat.js"; } from "../agent-heartbeat.js";
import { AgentLogger } from "../agent-logger.js"; import { AgentLogger } from "../agent-logger.js";
import * as agentTools from "../agent-tools.js"; import * as agentTools from "../agent-tools.js";
import * as sessionHelpers from "../agent-session-helpers.js";
import { AgentStore as RealAgentStore, TaskStore as RealTaskStore, ChatStore } from "@fusion/core";
import type { AgentStore, AgentHeartbeatRun, TaskStore, TaskDetail, Agent, MessageStore, Message, AgentBudgetStatus } from "@fusion/core"; import type { AgentStore, AgentHeartbeatRun, TaskStore, TaskDetail, Agent, MessageStore, Message, AgentBudgetStatus } from "@fusion/core";
import { createMockStore, createMockSession, createMockMessageStore, createMessage, createBudgetStatus } from "./heartbeat-test-helpers.js"; import { createMockStore, createMockSession, createMockMessageStore, createMessage, createBudgetStatus } from "./heartbeat-test-helpers.js";
vi.mock("../logger.js", async () => { vi.mock("../logger.js", async () => {
@@ -666,6 +668,59 @@ describe("clearRunState", () => {
}); });
}); });
describe("room-message prompt injection", () => {
it("includes pending room messages and excludes self-authored room traffic", async () => {
const rootDir = mkdtempSync(join(tmpdir(), "hb-room-prompt-"));
const globalDir = mkdtempSync(join(tmpdir(), "hb-room-global-"));
const taskStore = new RealTaskStore(rootDir, globalDir, { inMemoryDb: true });
await taskStore.init();
const agentStore = new RealAgentStore({ rootDir: taskStore.getFusionDir(), taskStore, inMemoryDb: true });
const chatStore = new ChatStore(taskStore.getFusionDir(), taskStore.getDatabase());
const agent = await agentStore.createAgent({
name: "Room Prompt Agent",
role: "engineer",
soul: "Responds to relevant room updates.",
runtimeConfig: { enabled: true },
});
const room = chatStore.createRoom({ name: "engineering", memberAgentIds: [agent.id] });
chatStore.addRoomMessage(room.id, { role: "assistant", senderAgentId: agent.id, content: "self message" });
const otherMessage = chatStore.addRoomMessage(room.id, { role: "user", content: "please investigate the queue" });
let capturedPrompt = "";
const createSessionSpy = vi.spyOn(sessionHelpers, "createResolvedAgentSession").mockImplementation(async (options: any) => ({
session: {
prompt: async (prompt: string) => {
capturedPrompt = prompt;
},
dispose: vi.fn(),
getSessionStats: () => ({ tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 } }),
},
options,
}) as any);
try {
const monitor = new HeartbeatMonitor({
store: agentStore as unknown as AgentStore,
taskStore: taskStore as unknown as TaskStore,
rootDir,
chatStore,
});
await monitor.executeHeartbeat({ agentId: agent.id, source: "timer" as any });
expect(capturedPrompt).toContain("Pending Room Messages:");
expect(capturedPrompt).toContain(room.name);
expect(capturedPrompt).toContain(otherMessage.id);
expect(capturedPrompt).not.toContain("self message");
} finally {
createSessionSpy.mockRestore();
rmSync(rootDir, { recursive: true, force: true });
rmSync(globalDir, { recursive: true, force: true });
}
});
});
// ───────────────────────────────────────────────────────────────────────── // ─────────────────────────────────────────────────────────────────────────
// HeartbeatTriggerScheduler tests // HeartbeatTriggerScheduler tests
// ───────────────────────────────────────────────────────────────────────── // ─────────────────────────────────────────────────────────────────────────

View File

@@ -17,12 +17,12 @@
* - onTerminated: Called when a heartbeat run is terminated * - onTerminated: Called when a heartbeat run is terminated
*/ */
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask, RunMutationContext, Settings, AgentConfigRevision, ReflectionStore } from "@fusion/core"; import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask, RunMutationContext, Settings, AgentConfigRevision, ReflectionStore, ChatStore, ChatRoom, ChatRoomMessage } from "@fusion/core";
import { ApprovalRequestStore, buildExecutionMemoryInstructions, isEphemeralAgent, hasAgentIdentity, resolveEffectiveAgentPermissionPolicy, canAgentTakeImplementationTask, canAgentTakeImplementationTaskForExplicitRouting } from "@fusion/core"; import { ApprovalRequestStore, buildExecutionMemoryInstructions, isEphemeralAgent, hasAgentIdentity, resolveEffectiveAgentPermissionPolicy, canAgentTakeImplementationTask, canAgentTakeImplementationTaskForExplicitRouting } from "@fusion/core";
import type { ToolDefinition } from "@mariozechner/pi-coding-agent"; import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
import { Type, type Static } from "@mariozechner/pi-ai"; import { Type, type Static } from "@mariozechner/pi-ai";
import { createHash } from "node:crypto"; import { createHash } from "node:crypto";
import { createTaskCreateTool, createTaskLogToolWithContext, createTaskDocumentWriteTool, createTaskDocumentReadTool, createListAgentsTool, createDelegateTaskTool, createGetAgentConfigTool, createUpdateAgentConfigTool, createAgentCreateTool, createAgentDeleteTool, createSendMessageTool, createReadMessagesTool, createMemoryTools, createReadEvaluationsTool, createUpdateIdentityTool, createReflectOnPerformanceTool, createWebFetchTool, readAgentMemoryWorkspaceLongTerm, taskCreateParams } from "./agent-tools.js"; import { createTaskCreateTool, createTaskLogToolWithContext, createTaskDocumentWriteTool, createTaskDocumentReadTool, createListAgentsTool, createDelegateTaskTool, createGetAgentConfigTool, createUpdateAgentConfigTool, createAgentCreateTool, createAgentDeleteTool, createSendMessageTool, createReadMessagesTool, createPostRoomMessageTool, createMemoryTools, createReadEvaluationsTool, createUpdateIdentityTool, createReflectOnPerformanceTool, createWebFetchTool, readAgentMemoryWorkspaceLongTerm, taskCreateParams } from "./agent-tools.js";
import { AgentLogger } from "./agent-logger.js"; import { AgentLogger } from "./agent-logger.js";
import { import {
resolveAgentInstructionsWithRatings, resolveAgentInstructionsWithRatings,
@@ -60,6 +60,8 @@ export interface HeartbeatMonitorOptions {
agentStore?: AgentStore; agentStore?: AgentStore;
/** Optional MessageStore for wake-on-message behavior */ /** Optional MessageStore for wake-on-message behavior */
messageStore?: MessageStore; messageStore?: MessageStore;
/** Optional ChatStore for room-message visibility during heartbeats */
chatStore?: ChatStore;
/** Polling interval in milliseconds (default: 3600000) */ /** Polling interval in milliseconds (default: 3600000) */
pollIntervalMs?: number; pollIntervalMs?: number;
/** Heartbeat timeout in milliseconds (default: 60000) */ /** Heartbeat timeout in milliseconds (default: 60000) */
@@ -274,7 +276,7 @@ Examples of ONE useful coordination action:
Keep work lightweight — this is a single-pass coordination check, not an implementation run. Keep work lightweight — this is a single-pass coordination check, not an implementation run.
You have workspace read tools (for context gathering) plus fn_task_create, fn_task_log, fn_task_document tools, You have workspace read tools (for context gathering) plus fn_task_create, fn_task_log, fn_task_document tools,
fn_send_message, fn_read_messages, fn_list_agents, fn_delegate_task, and memory tools. fn_send_message, fn_read_messages, fn_post_room_message, fn_list_agents, fn_delegate_task, and memory tools.
**Task Documents:** Save important findings with fn_task_document_write(key="...", content="..."). **Task Documents:** Save important findings with fn_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.
@@ -319,7 +321,10 @@ When you are woken by an incoming message (source includes "wake-on-message"), y
- If the message is informational, acknowledge it by logging with fn_task_log. - If the message is informational, acknowledge it by logging with fn_task_log.
- If the message requests net-new work, create a follow-up task with fn_task_create. - If the message requests net-new work, create a follow-up task with fn_task_create.
- If ownership is clear and an agent is available, delegate using fn_delegate_task. - If ownership is clear and an agent is available, delegate using fn_delegate_task.
4. After processing messages, continue with your normal heartbeat duties. 4. If a Pending Room Messages section is present, review it too:
- Use fn_post_room_message only when the room content is relevant to your role, soul, or identity.
- Reference room message IDs when replying so humans can trace context.
5. After processing messages, continue with your normal heartbeat duties.
Example flow: Example flow:
- Read unread messages → identify "needs action" item → reply with intent (reply_to_message_id) → create/delegate task if execution is needed → log key decision. - Read unread messages → identify "needs action" item → reply with intent (reply_to_message_id) → create/delegate task if execution is needed → log key decision.
@@ -363,7 +368,7 @@ You have coding-capable workspace tools (read/write/edit/bash within worktree bo
- fn_get_agent_config and fn_update_agent_config (for direct reports only) - fn_get_agent_config and fn_update_agent_config (for direct reports only)
- fn_memory_search, fn_memory_get, and fn_memory_append - fn_memory_search, fn_memory_get, and fn_memory_append
- fn_heartbeat_done - fn_heartbeat_done
- fn_send_message and fn_read_messages when messaging is enabled for this run (they may not always be available) - fn_send_message, fn_read_messages, and fn_post_room_message when messaging/room tools are enabled for this run (they may not always be available)
## Triage and Routing Decisions ## Triage and Routing Decisions
@@ -400,7 +405,8 @@ When you are woken by an incoming message (source includes "wake-on-message"), y
- If the message is informational, acknowledge it and respond via fn_send_message when appropriate. - If the message is informational, acknowledge it and respond via fn_send_message when appropriate.
- If the message requests work, create a follow-up task with fn_task_create. - If the message requests work, create a follow-up task with fn_task_create.
- If the request has a clear owner and fn_delegate_task is available, delegate it directly. - If the request has a clear owner and fn_delegate_task is available, delegate it directly.
3. After processing messages, continue with your ambient work. 3. If a Pending Room Messages section is present, review it too and use fn_post_room_message only when the room content is relevant to your role or identity.
4. After processing messages, continue with your ambient work.
Example flow: Example flow:
- Read inbox → classify message → reply with reply_to_message_id → create/delegate follow-up if needed → finish with fn_heartbeat_done. - Read inbox → classify message → reply with reply_to_message_id → create/delegate follow-up if needed → finish with fn_heartbeat_done.
@@ -428,7 +434,8 @@ export const HEARTBEAT_PROCEDURE = `## Heartbeat Procedure (run every tick, in o
section of your system prompt. section of your system prompt.
2. **Inbox** — when fn_read_messages is available, call it immediately and 2. **Inbox** — when fn_read_messages is available, call it immediately and
process unread/pending messages before any other action; reply with process unread/pending messages before any other action; reply with
reply_to_message_id when answering. reply_to_message_id when answering. If Pending Room Messages are present,
review them in the prompt and use fn_post_room_message only when relevant.
3. **Wake delta** — read the Wake Delta block above. The wake reason is the 3. **Wake delta** — read the Wake Delta block above. The wake reason is the
highest-priority change for this heartbeat. If you were woken by a comment highest-priority change for this heartbeat. If you were woken by a comment
or a message, acknowledge it before doing anything else. or a message, acknowledge it before doing anything else.
@@ -477,7 +484,8 @@ export const HEARTBEAT_NO_TASK_PROCEDURE = `## Heartbeat Procedure (run every ti
section of your system prompt. section of your system prompt.
2. **Inbox** — when fn_read_messages is available, call it immediately and 2. **Inbox** — when fn_read_messages is available, call it immediately and
process unread/pending messages before any other action; reply with process unread/pending messages before any other action; reply with
reply_to_message_id when answering. reply_to_message_id when answering. If Pending Room Messages are present,
review them in the prompt and use fn_post_room_message only when relevant.
3. **Wake delta** — read the Wake Delta block above. The wake reason is the 3. **Wake delta** — read the Wake Delta block above. The wake reason is the
highest-priority change for this heartbeat. If you were woken by a comment highest-priority change for this heartbeat. If you were woken by a comment
or a message, acknowledge it before doing anything else. or a message, acknowledge it before doing anything else.
@@ -594,6 +602,7 @@ export class HeartbeatMonitor {
private taskStore?: TaskStore; private taskStore?: TaskStore;
private rootDir?: string; private rootDir?: string;
private messageStore?: MessageStore; private messageStore?: MessageStore;
private chatStore?: ChatStore;
private pluginRunner?: import("./plugin-runner.js").PluginRunner; private pluginRunner?: import("./plugin-runner.js").PluginRunner;
private reflectionStore?: ReflectionStore; private reflectionStore?: ReflectionStore;
private reflectionService?: AgentReflectionService; private reflectionService?: AgentReflectionService;
@@ -622,12 +631,104 @@ export class HeartbeatMonitor {
this.taskStore = options.taskStore; this.taskStore = options.taskStore;
this.rootDir = options.rootDir; this.rootDir = options.rootDir;
this.messageStore = options.messageStore; this.messageStore = options.messageStore;
this.chatStore = options.chatStore;
this.pluginRunner = options.pluginRunner; this.pluginRunner = options.pluginRunner;
this.reflectionStore = options.reflectionStore; this.reflectionStore = options.reflectionStore;
this.reflectionService = options.reflectionService; this.reflectionService = options.reflectionService;
this.selfImproveService = options.selfImproveService; this.selfImproveService = options.selfImproveService;
} }
getChatStore(): ChatStore | undefined {
return this.chatStore;
}
private async resolveRoomMessageSinceIso(agent: Agent, activeRunId: string): Promise<string> {
const twentyFourHoursAgo = new Date(Date.now() - 24 * 60 * 60 * 1000).toISOString();
try {
const recentRuns = await this.store.getRecentRuns(agent.id, 10);
const previousCompletedRun = recentRuns.find((candidate) => candidate.id !== activeRunId && candidate.endedAt);
const candidateIso = previousCompletedRun?.endedAt ?? agent.lastHeartbeatAt ?? twentyFourHoursAgo;
const candidateTime = Date.parse(candidateIso);
if (!Number.isFinite(candidateTime)) {
return twentyFourHoursAgo;
}
return new Date(Math.max(candidateTime, Date.now() - 24 * 60 * 60 * 1000)).toISOString();
} catch (error) {
heartbeatLog.warn(`Failed to resolve room-message lookback for ${agent.id}: ${error instanceof Error ? error.message : String(error)}`);
const fallbackTime = Date.parse(agent.lastHeartbeatAt ?? twentyFourHoursAgo);
if (!Number.isFinite(fallbackTime)) {
return twentyFourHoursAgo;
}
return new Date(Math.max(fallbackTime, Date.now() - 24 * 60 * 60 * 1000)).toISOString();
}
}
private getPendingRoomMessagesSection(entries: Array<{ room: ChatRoom; messages: ChatRoomMessage[] }>, truncatedCount: number): string[] {
if (entries.length === 0) {
return [];
}
const lines = ["", "Pending Room Messages:"];
for (const entry of entries) {
lines.push(`- [room: ${entry.room.name} (${entry.room.id})]`);
for (const message of entry.messages) {
const normalized = message.content.replace(/\s+/g, " ").trim();
const truncatedContent = normalized.length > 180 ? `${normalized.slice(0, 179)}` : normalized;
lines.push(` - [from: ${message.senderAgentId ?? "user"}] [${message.id}] ${truncatedContent}`);
}
}
if (truncatedCount > 0) {
lines.push(` - (${truncatedCount} more truncated)`);
}
return lines;
}
private async getPendingRoomMessages(agent: Agent, sinceIso: string): Promise<{
entries: Array<{ room: ChatRoom; messages: ChatRoomMessage[] }>;
total: number;
truncatedCount: number;
}> {
if (!this.chatStore) {
return { entries: [], total: 0, truncatedCount: 0 };
}
try {
const rooms = this.chatStore.listRoomsForAgent(agent.id, { status: "active" });
const entries: Array<{ room: ChatRoom; messages: ChatRoomMessage[] }> = [];
let total = 0;
let surfaced = 0;
let truncatedCount = 0;
for (const room of rooms) {
const messages = this.chatStore.listRoomMessagesSince(room.id, sinceIso, {
excludeSenderAgentId: agent.id,
limit: 10,
});
if (messages.length === 0) {
continue;
}
total += messages.length;
const remaining = 30 - surfaced;
if (remaining <= 0) {
truncatedCount += messages.length;
continue;
}
const surfacedMessages = messages.slice(0, remaining);
truncatedCount += messages.length - surfacedMessages.length;
entries.push({ room, messages: surfacedMessages });
surfaced += surfacedMessages.length;
}
return { entries, total, truncatedCount };
} catch (error) {
heartbeatLog.warn(`Failed to fetch room messages for ${agent.id}: ${error instanceof Error ? error.message : String(error)}`);
return { entries: [], total: 0, truncatedCount: 0 };
}
}
private getApprovalRequestStore(): ApprovalRequestStore { private getApprovalRequestStore(): ApprovalRequestStore {
if (!this.approvalRequestStore) { if (!this.approvalRequestStore) {
if (!this.taskStore) { if (!this.taskStore) {
@@ -1808,6 +1909,9 @@ export class HeartbeatMonitor {
heartbeatTools.push(createSendMessageTool(this.messageStore, agentId)); heartbeatTools.push(createSendMessageTool(this.messageStore, agentId));
heartbeatTools.push(createReadMessagesTool(this.messageStore, agentId)); heartbeatTools.push(createReadMessagesTool(this.messageStore, agentId));
} }
if (this.chatStore) {
heartbeatTools.push(createPostRoomMessageTool(this.chatStore, agentId));
}
heartbeatTools.push(createReadEvaluationsTool(this.store, this.reflectionStore, agentId)); heartbeatTools.push(createReadEvaluationsTool(this.store, this.reflectionStore, agentId));
heartbeatTools.push(createUpdateIdentityTool(this.store, agentId)); heartbeatTools.push(createUpdateIdentityTool(this.store, agentId));
@@ -2029,6 +2133,13 @@ export class HeartbeatMonitor {
let pendingMessages: Message[] = []; let pendingMessages: Message[] = [];
let executionPrompt: string; let executionPrompt: string;
const sinceIso = await this.resolveRoomMessageSinceIso(agent, run.id);
const pendingRoomMessages = await this.getPendingRoomMessages(agent, sinceIso);
const pendingRoomMessagesLines = this.getPendingRoomMessagesSection(
pendingRoomMessages.entries,
pendingRoomMessages.truncatedCount,
);
// Derive a stable wake reason from source, triggerDetail, and trigger // Derive a stable wake reason from source, triggerDetail, and trigger
// type so the agent can change its strategy based on *why* it woke up. // type so the agent can change its strategy based on *why* it woke up.
// Mirrors paperclip's PAPERCLIP_WAKE_REASON (see plan: wake delta). // Mirrors paperclip's PAPERCLIP_WAKE_REASON (see plan: wake delta).
@@ -2103,6 +2214,7 @@ export class HeartbeatMonitor {
`- wake reason: ${wakeReason}`, `- wake reason: ${wakeReason}`,
`- assigned task: none`, `- assigned task: none`,
`- pending messages: ${pendingMessages.length}`, `- pending messages: ${pendingMessages.length}`,
`- pending room messages: ${pendingRoomMessages.total}`,
`- auto-claim relevant tasks: ${autoClaimEnabled ? "enabled" : "disabled"}`, `- auto-claim relevant tasks: ${autoClaimEnabled ? "enabled" : "disabled"}`,
"", "",
"Treat this wake delta as the highest-priority change for this heartbeat.", "Treat this wake delta as the highest-priority change for this heartbeat.",
@@ -2137,6 +2249,7 @@ export class HeartbeatMonitor {
"prioritize tasks that align with your role and soul before creating net-new tasks.", "prioritize tasks that align with your role and soul before creating net-new tasks.",
...candidateLines, ...candidateLines,
...pendingMessagesLines, ...pendingMessagesLines,
...pendingRoomMessagesLines,
"", "",
"Your soul, instructions, and memory are already loaded in the system prompt.", "Your soul, instructions, and memory are already loaded in the system prompt.",
"Focus on work that benefits the project without requiring a specific task context.", "Focus on work that benefits the project without requiring a specific task context.",
@@ -2215,6 +2328,7 @@ export class HeartbeatMonitor {
`- wake reason: ${wakeReason}`, `- wake reason: ${wakeReason}`,
`- assigned task: ${taskId}`, `- assigned task: ${taskId}`,
`- pending messages: ${pendingMessages.length}`, `- pending messages: ${pendingMessages.length}`,
`- pending room messages: ${pendingRoomMessages.total}`,
`- triggering comments: ${effectiveTriggeringCommentIds?.length ?? 0}`, `- triggering comments: ${effectiveTriggeringCommentIds?.length ?? 0}`,
"", "",
"Treat this wake delta as the highest-priority change for this heartbeat.", "Treat this wake delta as the highest-priority change for this heartbeat.",
@@ -2232,6 +2346,7 @@ export class HeartbeatMonitor {
taskDetail!.prompt ? `PROMPT.md:\n${taskDetail!.prompt}` : "No PROMPT.md available.", taskDetail!.prompt ? `PROMPT.md:\n${taskDetail!.prompt}` : "No PROMPT.md available.",
...triggeringCommentLines, ...triggeringCommentLines,
...pendingMessagesLines, ...pendingMessagesLines,
...pendingRoomMessagesLines,
...(reportsHealthSection ? ["", reportsHealthSection] : []), ...(reportsHealthSection ? ["", reportsHealthSection] : []),
"", "",
"Run the Heartbeat Procedure above. Call fn_heartbeat_done when finished.", "Run the Heartbeat Procedure above. Call fn_heartbeat_done when finished.",
@@ -2595,6 +2710,9 @@ export class HeartbeatMonitor {
tools.push(createSendMessageTool(messageStore, agentId)); tools.push(createSendMessageTool(messageStore, agentId));
tools.push(createReadMessagesTool(messageStore, agentId)); tools.push(createReadMessagesTool(messageStore, agentId));
} }
if (this.chatStore) {
tools.push(createPostRoomMessageTool(this.chatStore, agentId));
}
tools.push(createReadEvaluationsTool(this.store, this.reflectionStore, agentId)); tools.push(createReadEvaluationsTool(this.store, this.reflectionStore, agentId));
tools.push(createUpdateIdentityTool(this.store, agentId)); tools.push(createUpdateIdentityTool(this.store, agentId));

View File

@@ -11,7 +11,7 @@ import { appendFile, mkdir, readFile, readdir, stat, writeFile } from "node:fs/p
import { existsSync } from "node:fs"; import { existsSync } from "node:fs";
import { createHash } from "node:crypto"; import { createHash } from "node:crypto";
import { join, relative, resolve } from "node:path"; import { join, relative, resolve } from "node:path";
import type { AgentStore, AgentState, AgentCapability, AgentUpdateInput, TaskDocument, TaskDocumentCreateInput, TaskStore, RunMutationContext, MessageStore, Message, SourceType, Settings, ResearchRun, ResearchRunStatus, TaskCreateInput, ReflectionStore, ApprovalRequestStore, ProjectSettings } from "@fusion/core"; import type { AgentStore, AgentState, AgentCapability, AgentUpdateInput, TaskDocument, TaskDocumentCreateInput, TaskStore, RunMutationContext, MessageStore, Message, SourceType, Settings, ResearchRun, ResearchRunStatus, TaskCreateInput, ReflectionStore, ApprovalRequestStore, ProjectSettings, ChatStore } from "@fusion/core";
import { DASHBOARD_USER_ID, canAgentTakeImplementationTaskForExplicitRouting, dailyMemoryPath, ensureOpenClawMemoryFiles, extractAgentProvisioningRequest, formatRoleMismatchReason, getMemoryBackendCapabilities, getProjectMemory, isEphemeralAgent, memoryLongTermPath, normalizeMessageParticipant, resolveAgentProvisioningPolicy, resolveMemoryBackend, resolveResearchSettings, resolveTitleSummarizerSettingsModel, scheduleQmdProjectMemoryRefresh, searchProjectMemory, shouldSkipBackgroundQmdRefresh, summarizeTitle } from "@fusion/core"; import { DASHBOARD_USER_ID, canAgentTakeImplementationTaskForExplicitRouting, dailyMemoryPath, ensureOpenClawMemoryFiles, extractAgentProvisioningRequest, formatRoleMismatchReason, getMemoryBackendCapabilities, getProjectMemory, isEphemeralAgent, memoryLongTermPath, normalizeMessageParticipant, resolveAgentProvisioningPolicy, resolveMemoryBackend, resolveResearchSettings, resolveTitleSummarizerSettingsModel, scheduleQmdProjectMemoryRefresh, searchProjectMemory, shouldSkipBackgroundQmdRefresh, summarizeTitle } from "@fusion/core";
import { ResearchOrchestrator } from "./research-orchestrator.js"; import { ResearchOrchestrator } from "./research-orchestrator.js";
import { ResearchProviderRegistry } from "./research/provider-registry.js"; import { ResearchProviderRegistry } from "./research/provider-registry.js";
@@ -159,6 +159,13 @@ export const readMessagesParams = Type.Object({
limit: Type.Optional(Type.Number({ description: "Max messages to return (default: 20)" })), limit: Type.Optional(Type.Number({ description: "Max messages to return (default: 20)" })),
}); });
export const postRoomMessageParams = Type.Object({
roomId: Type.String({ description: "Room ID to post into" }),
content: Type.String({ description: "Room message body (1-2000 characters)" }),
replyToMessageId: Type.Optional(Type.String({ description: "Optional ID of the room message you are replying to" })),
mentions: Type.Optional(Type.Array(Type.String(), { description: "Optional agent IDs to mention in the room message" })),
});
export const memorySearchParams = Type.Object({ export const memorySearchParams = Type.Object({
query: Type.String({ description: "Search terms for durable project memory. Use focused keywords, not a full prompt." }), query: Type.String({ description: "Search terms for durable project memory. Use focused keywords, not a full prompt." }),
limit: Type.Optional(Type.Number({ description: "Maximum snippets to return (default: 5, max: 20)" })), limit: Type.Optional(Type.Number({ description: "Maximum snippets to return (default: 5, max: 20)" })),
@@ -2125,6 +2132,71 @@ export function createResearchTools(options: ResearchToolsOptions): ToolDefiniti
return [runTool, listTool, getTool, cancelTool]; return [runTool, listTool, getTool, cancelTool];
} }
export function createPostRoomMessageTool(chatStore: ChatStore, fromAgentId: string): ToolDefinition {
return {
name: "fn_post_room_message",
label: "Post Room Message",
description:
"Post a message to a room you are a member of. Room membership is enforced before posting, " +
"so only reply when the room content is relevant to your role or identity.",
parameters: postRoomMessageParams,
execute: async (_id: string, params: Static<typeof postRoomMessageParams>) => {
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: {},
};
}
const replyToMessageId = params.replyToMessageId?.trim();
if (params.replyToMessageId !== undefined && !replyToMessageId) {
return {
content: [{ type: "text" as const, text: "ERROR: replyToMessageId must be a non-empty string" }],
details: {},
};
}
try {
const isMember = chatStore.listRoomMembers(params.roomId).some((member) => member.agentId === fromAgentId);
if (!isMember) {
return {
content: [{ type: "text" as const, text: `ERROR: Agent ${fromAgentId} is not a member of room ${params.roomId}` }],
details: {},
isError: true,
};
}
const message = chatStore.addRoomMessage(params.roomId, {
role: "assistant",
senderAgentId: fromAgentId,
content,
mentions: params.mentions ?? [],
...(replyToMessageId ? { metadata: { replyToMessageId } } : {}),
});
return {
content: [{ type: "text" as const, text: `Room message posted to ${params.roomId} (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 post room message: ${errorMessage}` }],
details: {},
isError: true,
};
}
},
};
}
export function createReadMessagesTool(messageStore: MessageStore, agentId: string): ToolDefinition { export function createReadMessagesTool(messageStore: MessageStore, agentId: string): ToolDefinition {
const REPLY_CONTEXT_CONTENT_MAX_CHARS = 400; const REPLY_CONTEXT_CONTENT_MAX_CHARS = 400;

View File

@@ -542,6 +542,7 @@ describe("InProcessRuntime", () => {
await runtime.start(); await runtime.start();
const monitor = runtime.getHeartbeatMonitor(); const monitor = runtime.getHeartbeatMonitor();
expect(monitor).toBeDefined(); expect(monitor).toBeDefined();
expect(monitor?.getChatStore()).toBeDefined();
}, 30000); }, 30000);
// Regression: heartbeat auto-claim path was warning // Regression: heartbeat auto-claim path was warning

View File

@@ -11,7 +11,7 @@ import type {
MessageStore, MessageStore,
RoutineStore, RoutineStore,
} from "@fusion/core"; } from "@fusion/core";
import { isEphemeralAgent } from "@fusion/core"; import { ChatStore, isEphemeralAgent } from "@fusion/core";
import { Scheduler } from "../scheduler.js"; import { Scheduler } from "../scheduler.js";
import type { PrMonitor, PrComment } from "../pr-monitor.js"; import type { PrMonitor, PrComment } from "../pr-monitor.js";
import type { PrInfo } from "@fusion/core"; import type { PrInfo } from "@fusion/core";
@@ -114,6 +114,7 @@ export class InProcessRuntime
private missionAutopilot?: MissionAutopilot; private missionAutopilot?: MissionAutopilot;
private triageProcessor?: TriageProcessor; private triageProcessor?: TriageProcessor;
private messageStore?: MessageStore; private messageStore?: MessageStore;
private chatStore?: ChatStore;
private concurrencyChangedListener?: (state: { globalMaxConcurrent: number }) => void; private concurrencyChangedListener?: (state: { globalMaxConcurrent: number }) => void;
/** /**
* Optional callback the runtime forwards to SelfHealingManager so that * Optional callback the runtime forwards to SelfHealingManager so that
@@ -450,12 +451,14 @@ export class InProcessRuntime
// Already started — nothing to do // Already started — nothing to do
} }
if (!this.heartbeatMonitor && this.agentStore) { if (!this.heartbeatMonitor && this.agentStore) {
this.chatStore ??= new ChatStore(this.taskStore.getFusionDir(), this.taskStore.getDatabase());
this.heartbeatMonitor = new HeartbeatMonitor({ this.heartbeatMonitor = new HeartbeatMonitor({
store: this.agentStore, store: this.agentStore,
agentStore: this.agentStore, // enables per-agent config resolution agentStore: this.agentStore, // enables per-agent config resolution
taskStore: this.taskStore, taskStore: this.taskStore,
rootDir: this.config.workingDirectory, rootDir: this.config.workingDirectory,
messageStore: this.messageStore, messageStore: this.messageStore,
chatStore: this.chatStore,
pluginRunner: this.pluginRunner, pluginRunner: this.pluginRunner,
reflectionStore: reflectionStoreForService, reflectionStore: reflectionStoreForService,
reflectionService, reflectionService,