feat(messages): persist agent-to-agent messages, surface in sidebar + History

apps/server/src/db/migrations/0030_agent-messages.sql

@@ -0,0 +1,28 @@
+-- Agent-to-agent messages: durable record of dispatch_send_message deliveries.
+-- Delivery itself stays ephemeral (tmux injection); this table is for viewing.
+
+CREATE TABLE IF NOT EXISTS agent_messages (
+  id uuid PRIMARY KEY,
+  sender_agent_id text NOT NULL,
+  recipient_agent_id text NOT NULL,
+  sender_name text NOT NULL,
+  recipient_name text NOT NULL,
+  content text NOT NULL,
+  delivered boolean NOT NULL DEFAULT false,
+  read_at timestamptz,
+  -- Stored even though sender/recipient repo roots are equal today (same-repo
+  -- send rule). Present so cross-repo messaging becomes a config flip, not a
+  -- migration.
+  sender_repo_root text,
+  recipient_repo_root text,
+  created_at timestamptz NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS agent_messages_sender_created_idx
+  ON agent_messages (sender_agent_id, created_at DESC);
+
+CREATE INDEX IF NOT EXISTS agent_messages_recipient_created_idx
+  ON agent_messages (recipient_agent_id, created_at DESC);
+
+CREATE INDEX IF NOT EXISTS agent_messages_recipient_unread_idx
+  ON agent_messages (recipient_agent_id) WHERE read_at IS NULL;

apps/server/src/messages/store.ts

@@ -0,0 +1,112 @@
+import { randomUUID } from "node:crypto";
+import type { Pool } from "pg";
+
+export type StoredMessage = {
+  id: string;
+  senderAgentId: string;
+  recipientAgentId: string;
+  senderName: string;
+  recipientName: string;
+  content: string;
+  delivered: boolean;
+  readAt: string | null;
+  senderRepoRoot: string | null;
+  recipientRepoRoot: string | null;
+  createdAt: string;
+};
+
+export type InsertMessageInput = Omit<
+  StoredMessage,
+  "id" | "readAt" | "createdAt"
+>;
+
+type Row = {
+  id: string;
+  sender_agent_id: string;
+  recipient_agent_id: string;
+  sender_name: string;
+  recipient_name: string;
+  content: string;
+  delivered: boolean;
+  read_at: Date | null;
+  sender_repo_root: string | null;
+  recipient_repo_root: string | null;
+  created_at: Date;
+};
+
+function toStoredMessage(row: Row): StoredMessage {
+  return {
+    id: row.id,
+    senderAgentId: row.sender_agent_id,
+    recipientAgentId: row.recipient_agent_id,
+    senderName: row.sender_name,
+    recipientName: row.recipient_name,
+    content: row.content,
+    delivered: row.delivered,
+    readAt: row.read_at ? row.read_at.toISOString() : null,
+    senderRepoRoot: row.sender_repo_root,
+    recipientRepoRoot: row.recipient_repo_root,
+    createdAt: row.created_at.toISOString(),
+  };
+}
+
+export class MessageStore {
+  constructor(private readonly pool: Pool) {}
+
+  async insertMessage(input: InsertMessageInput): Promise<StoredMessage> {
+    const result = await this.pool.query<Row>(
+      `INSERT INTO agent_messages
+         (id, sender_agent_id, recipient_agent_id, sender_name, recipient_name,
+          content, delivered, sender_repo_root, recipient_repo_root)
+       VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+       RETURNING *`,
+      [
+        randomUUID(),
+        input.senderAgentId,
+        input.recipientAgentId,
+        input.senderName,
+        input.recipientName,
+        input.content,
+        input.delivered,
+        input.senderRepoRoot,
+        input.recipientRepoRoot,
+      ]
+    );
+    return toStoredMessage(result.rows[0]);
+  }
+
+  async listForAgent(agentId: string): Promise<StoredMessage[]> {
+    // Bound the result set (mirrors the LIMIT 500 cap on the history-detail
+    // query in activity.ts). Take the most recent 500, then return them in
+    // ascending order for chronological rendering.
+    const result = await this.pool.query<Row>(
+      `SELECT * FROM (
+         SELECT * FROM agent_messages
+          WHERE sender_agent_id = $1 OR recipient_agent_id = $1
+          ORDER BY created_at DESC
+          LIMIT 500
+       ) recent
+       ORDER BY created_at ASC`,
+      [agentId]
+    );
+    return result.rows.map(toStoredMessage);
+  }
+
+  async countUnreadForAgent(agentId: string): Promise<number> {
+    const result = await this.pool.query<{ count: string }>(
+      `SELECT COUNT(*)::text AS count FROM agent_messages
+        WHERE recipient_agent_id = $1 AND read_at IS NULL`,
+      [agentId]
+    );
+    return Number(result.rows[0].count);
+  }
+
+  async markReadForAgent(agentId: string): Promise<number> {
+    const result = await this.pool.query(
+      `UPDATE agent_messages SET read_at = now()
+        WHERE recipient_agent_id = $1 AND read_at IS NULL`,
+      [agentId]
+    );
+    return result.rowCount ?? 0;
+  }
+}

apps/server/src/routes/activity.ts

@@ -585,6 +585,7 @@ export async function registerActivityRoutes(
       tokenByModelResult,
       mediaResult,
       feedbackResult,
+      messagesResult,
     ] = await Promise.all([
       deps.pool.query<{
         id: number;
@@ -659,6 +660,34 @@ export async function registerActivityRoutes(
            LIMIT 500`,
         [id]
       ),
+      deps.pool.query<{
+        id: string;
+        senderAgentId: string;
+        recipientAgentId: string;
+        senderName: string;
+        recipientName: string;
+        content: string;
+        delivered: boolean;
+        readAt: string | null;
+        createdAt: string;
+      }>(
+        `SELECT id,
+                sender_agent_id AS "senderAgentId",
+                recipient_agent_id AS "recipientAgentId",
+                sender_name AS "senderName",
+                recipient_name AS "recipientName",
+                content, delivered,
+                read_at AS "readAt",
+                created_at AS "createdAt"
+           FROM (
+             SELECT * FROM agent_messages
+              WHERE sender_agent_id = $1 OR recipient_agent_id = $1
+              ORDER BY created_at DESC
+              LIMIT 500
+           ) recent
+           ORDER BY created_at ASC`,
+        [id]
+      ),
     ]);
 
     const eventRows: ActivityEventRow[] = eventsResult.rows.map((row) => ({
@@ -674,6 +703,7 @@ export async function registerActivityRoutes(
       tokenUsage: { ...tokenResult.rows[0], by_model: tokenByModelResult.rows },
       media: mediaResult.rows,
       feedback: feedbackResult.rows,
+      messages: messagesResult.rows,
       stateDurations: stats.stateDurations,
     };
   });

apps/server/src/routes/messages.ts

@@ -0,0 +1,48 @@
+import type { FastifyInstance } from "fastify";
+import type { Pool } from "pg";
+
+import { MessageStore } from "../messages/store.js";
+
+type MessagesRouteDeps = {
+  pool: Pool;
+  publishUiEvent: (event: unknown) => void;
+};
+
+export async function registerMessagesRoutes(
+  app: FastifyInstance,
+  deps: MessagesRouteDeps
+): Promise<void> {
+  const store = new MessageStore(deps.pool);
+
+  app.get("/api/v1/agents/:id/messages", async (request, reply) => {
+    const id = (request.params as { id?: string }).id ?? "";
+    const exists = await deps.pool.query("SELECT 1 FROM agents WHERE id = $1", [
+      id,
+    ]);
+    if (exists.rows.length === 0) {
+      return reply.code(404).send({ error: "Agent not found." });
+    }
+    // unreadCount is computed server-side (uses the partial unread index) so
+    // the badge stays accurate even though listForAgent is capped.
+    const [messages, unreadCount] = await Promise.all([
+      store.listForAgent(id),
+      store.countUnreadForAgent(id),
+    ]);
+    return { messages, unreadCount };
+  });
+
+  app.post("/api/v1/agents/:id/messages/read", async (request, reply) => {
+    const id = (request.params as { id?: string }).id ?? "";
+    const exists = await deps.pool.query("SELECT 1 FROM agents WHERE id = $1", [
+      id,
+    ]);
+    if (exists.rows.length === 0) {
+      return reply.code(404).send({ error: "Agent not found." });
+    }
+    const updated = await store.markReadForAgent(id);
+    if (updated > 0) {
+      deps.publishUiEvent({ type: "message.read", agentId: id });
+    }
+    return { ok: true, updated };
+  });
+}

apps/server/src/server.ts

@@ -105,6 +105,7 @@ import { registerFeedbackRoutes } from "./routes/feedback.js";
 import { registerJobRoutes } from "./routes/jobs.js";
 import { registerTemplateRoutes } from "./routes/templates.js";
 import { registerMediaRoutes } from "./routes/media.js";
+import { registerMessagesRoutes } from "./routes/messages.js";
 import { registerMcpRoutes } from "./routes/mcp.js";
 import { registerPersonaReviewRoutes } from "./routes/persona-reviews.js";
 import { registerPersonalityRoutes } from "./routes/personalities.js";
@@ -581,6 +582,11 @@ async function registerRoutes() {
     publishUiEvent: (event) => uiEventBroker.publish(event as UiEvent),
   });
 
+  await registerMessagesRoutes(app, {
+    pool,
+    publishUiEvent: (event) => uiEventBroker.publish(event as UiEvent),
+  });
+
   await registerAgentRoutes(app, {
     pool,
     appLog: app.log,

apps/server/src/server/mcp-handlers.ts

@@ -28,6 +28,7 @@ import { resolveHeadSha } from "../shared/git/worktree.js";
 import { isMediaFile, isTextFile, resolveMediaDir } from "../shared/media.js";
 import type { PublishUiEvent, SendAgentPrompt } from "./mcp-handler-types.js";
 import { createReviewHandlers } from "./mcp-review-handlers.js";
+import { MessageStore } from "../messages/store.js";
 
 const AGENT_LATEST_EVENT_TYPES = [
   "working",
@@ -538,14 +539,60 @@ export function createMcpHandlers(deps: CreateMcpHandlersDeps) {
       });
       const prompt = `--- DISPATCH MESSAGE ---\n${envelope}\n--- END MESSAGE ---\nReply with dispatch_send_message using the replyTarget above.`;
 
+      // Deliver first: a persistence failure must never block delivery.
+      let delivered = false;
+      let deliveryError: unknown = null;
       try {
         await sendAgentPrompt(target.id, prompt, { swallowFailure: false });
+        delivered = true;
       } catch (err) {
+        deliveryError = err;
         appLog.error(
           { err, senderId: agentId, targetId: target.id },
           "dispatch_send_message: tmux delivery failed"
         );
-        throw err;
+      }
+
+      // Record the message (including failed deliveries) so it is viewable.
+      // Persistence must never block delivery, so a failed insert is swallowed
+      // and logged. Only announce message.created when the row actually landed,
+      // otherwise the UI would refetch and find nothing.
+      const recipientRepoRoot = await resolveRepoRoot(target.cwd).catch(
+        () => null
+      );
+      const messageStore = new MessageStore(pool);
+      const persisted = await messageStore
+        .insertMessage({
+          senderAgentId: agentId,
+          recipientAgentId: target.id,
+          senderName: sender.name,
+          recipientName: target.name,
+          content: input.message,
+          delivered,
+          senderRepoRoot,
+          recipientRepoRoot,
+        })
+        .then(() => true)
+        .catch((err) => {
+          appLog.error(
+            { err, senderId: agentId, targetId: target.id },
+            "dispatch_send_message: failed to persist message"
+          );
+          return false;
+        });
+
+      if (persisted) {
+        publishUiEvent({
+          type: "message.created",
+          senderAgentId: agentId,
+          recipientAgentId: target.id,
+        });
+      }
+
+      if (!delivered) {
+        throw deliveryError instanceof Error
+          ? deliveryError
+          : new Error(`Failed to deliver message to "${target.name}".`);
       }
 
       appLog.info(

apps/server/src/server/ui-events.ts

@@ -20,6 +20,12 @@ export type UiEvent =
   | { type: "agent.deleted"; agentId: string }
   | { type: "media.changed"; agentId: string }
   | { type: "media.seen"; agentId: string; keys: string[] }
+  | {
+      type: "message.created";
+      senderAgentId: string;
+      recipientAgentId: string;
+    }
+  | { type: "message.read"; agentId: string }
   | { type: "stream.started"; agentId: string }
   | { type: "stream.stopped"; agentId: string }
   | {

apps/server/test/activity-routes.test.ts

@@ -782,4 +782,24 @@ describe("GET /api/v1/history/agents/:id", () => {
     expect(feedback[0].description).toBe("potential XSS");
     expect(feedback[0].persona).toBe("security-review");
   });
+
+  it("includes agent messages in the history detail", async () => {
+    const agentId = await createAgent({ name: "history-agent" });
+    // Seed a message where the history agent is the sender.
+    await ctx.pool.query(
+      `INSERT INTO agent_messages
+         (id, sender_agent_id, recipient_agent_id, sender_name, recipient_name, content, delivered)
+       VALUES ('11111111-1111-1111-1111-111111111111', $1, 'agt_other', 'Hist', 'Other', 'history msg', true)`,
+      [agentId]
+    );
+
+    const res = await authedInject("GET", `/api/v1/history/agents/${agentId}`);
+    expect(res.statusCode).toBe(200);
+    const body = res.json() as { messages: Array<{ content: string }> };
+    expect(body.messages.map((m) => m.content)).toContain("history msg");
+
+    await ctx.pool.query(
+      "DELETE FROM agent_messages WHERE id = '11111111-1111-1111-1111-111111111111'"
+    );
+  });
 });