diff --git a/apps/server/src/db/migrations/0030_agent-messages.sql b/apps/server/src/db/migrations/0030_agent-messages.sql new file mode 100644 index 00000000..5c17b67e --- /dev/null +++ b/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; diff --git a/apps/server/src/messages/store.ts b/apps/server/src/messages/store.ts new file mode 100644 index 00000000..b88c19c2 --- /dev/null +++ b/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 { + const result = await this.pool.query( + `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 { + // 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( + `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 { + 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 { + 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; + } +} diff --git a/apps/server/src/routes/activity.ts b/apps/server/src/routes/activity.ts index 5aad97fa..7ea00c9e 100644 --- a/apps/server/src/routes/activity.ts +++ b/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, }; }); diff --git a/apps/server/src/routes/messages.ts b/apps/server/src/routes/messages.ts new file mode 100644 index 00000000..7c3ed9ec --- /dev/null +++ b/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 { + 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 }; + }); +} diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 34f75103..0c919ab8 100644 --- a/apps/server/src/server.ts +++ b/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, diff --git a/apps/server/src/server/mcp-handlers.ts b/apps/server/src/server/mcp-handlers.ts index 7afcfc73..cf681b5b 100644 --- a/apps/server/src/server/mcp-handlers.ts +++ b/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( diff --git a/apps/server/src/server/ui-events.ts b/apps/server/src/server/ui-events.ts index f020a79f..3472875e 100644 --- a/apps/server/src/server/ui-events.ts +++ b/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 } | { diff --git a/apps/server/test/activity-routes.test.ts b/apps/server/test/activity-routes.test.ts index 407fe153..4760b516 100644 --- a/apps/server/test/activity-routes.test.ts +++ b/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'" + ); + }); }); diff --git a/apps/server/test/message-store.test.ts b/apps/server/test/message-store.test.ts new file mode 100644 index 00000000..4fbfc99d --- /dev/null +++ b/apps/server/test/message-store.test.ts @@ -0,0 +1,67 @@ +import { describe, it, expect, beforeAll, afterAll } from "vitest"; +import type { Pool } from "pg"; + +import { MessageStore } from "../src/messages/store.js"; +import { setupTestDb, teardownTestDb, runTestMigrations } from "./db/setup.js"; + +let pool: Pool; +let store: MessageStore; + +const REPO = "/repo/msg-test"; +const A = "agt_msg_a"; +const B = "agt_msg_b"; + +beforeAll(async () => { + pool = await setupTestDb(); + await runTestMigrations(); + store = new MessageStore(pool); +}); + +afterAll(async () => { + await teardownTestDb(); +}); + +describe("MessageStore", () => { + it("inserts and lists messages for both participants", async () => { + const inserted = await store.insertMessage({ + senderAgentId: A, + recipientAgentId: B, + senderName: "Alice", + recipientName: "Bob", + content: "hello bob", + delivered: true, + senderRepoRoot: REPO, + recipientRepoRoot: REPO, + }); + expect(inserted.id).toBeTruthy(); + expect(inserted.readAt).toBeNull(); + expect(inserted.delivered).toBe(true); + + const forSender = await store.listForAgent(A); + const forRecipient = await store.listForAgent(B); + expect(forSender.map((m) => m.content)).toContain("hello bob"); + expect(forRecipient.map((m) => m.content)).toContain("hello bob"); + }); + + it("counts and clears unread for the recipient only", async () => { + await store.insertMessage({ + senderAgentId: B, + recipientAgentId: A, + senderName: "Bob", + recipientName: "Alice", + content: "hi alice", + delivered: true, + senderRepoRoot: REPO, + recipientRepoRoot: REPO, + }); + + // A received one message ("hi alice") -> unread 1. A sent one -> not counted. + expect(await store.countUnreadForAgent(A)).toBe(1); + + const cleared = await store.markReadForAgent(A); + expect(cleared).toBe(1); + expect(await store.countUnreadForAgent(A)).toBe(0); + // Marking A read must not touch B's unread. + expect(await store.countUnreadForAgent(B)).toBe(1); + }); +}); diff --git a/apps/server/test/messages-routes.test.ts b/apps/server/test/messages-routes.test.ts new file mode 100644 index 00000000..694cf84b --- /dev/null +++ b/apps/server/test/messages-routes.test.ts @@ -0,0 +1,111 @@ +import { beforeEach, describe, expect, it } from "vitest"; + +import { useInjectApp } from "./helpers/inject-app.js"; +import { MessageStore } from "../src/messages/store.js"; + +const ctx = useInjectApp(); + +async function authedInject( + method: string, + url: string, + opts?: { payload?: unknown; headers?: Record } +): Promise> { + const cookie = await ctx.sessionCookie(); + const headers: Record = { cookie, ...opts?.headers }; + if (opts?.payload !== undefined && !headers["content-type"]) { + headers["content-type"] = "application/json"; + } + return ctx.app.inject({ + method: method as "GET" | "POST", + url, + headers, + ...(opts?.payload !== undefined ? { payload: opts.payload } : {}), + }); +} + +async function createAgent(name: string): Promise<{ id: string }> { + const res = await authedInject("POST", "/api/v1/agents", { + payload: { cwd: "/tmp", useWorktree: false, name }, + }); + expect(res.statusCode).toBe(201); + const agent = res.json().agent; + return { id: agent.id }; +} + +let agentA: string; +let agentB: string; + +beforeEach(async () => { + await ctx.pool.query("DELETE FROM agent_messages"); + await ctx.pool.query("DELETE FROM media_seen"); + await ctx.pool.query("DELETE FROM media"); + await ctx.pool.query("DELETE FROM job_runs"); + await ctx.pool.query("DELETE FROM jobs"); + await ctx.pool.query("DELETE FROM agents"); + + const a = await createAgent("Alice"); + const b = await createAgent("Bob"); + agentA = a.id; + agentB = b.id; + + const store = new MessageStore(ctx.pool); + await store.insertMessage({ + senderAgentId: agentA, + recipientAgentId: agentB, + senderName: "Alice", + recipientName: "Bob", + content: "hi", + delivered: true, + senderRepoRoot: "/repo", + recipientRepoRoot: "/repo", + }); +}); + +describe("GET /api/v1/agents/:id/messages (list)", () => { + it("returns 404 for nonexistent agent", async () => { + const res = await authedInject( + "GET", + "/api/v1/agents/agt_nonexistent/messages" + ); + expect(res.statusCode).toBe(404); + }); + + it("lists messages for an agent with a server-computed unread count", async () => { + const res = await authedInject("GET", `/api/v1/agents/${agentB}/messages`); + expect(res.statusCode).toBe(200); + const body = res.json() as { + messages: Array<{ content: string }>; + unreadCount: number; + }; + expect(body.messages.map((m) => m.content)).toContain("hi"); + // Bob is the recipient of the unread "hi" message. + expect(body.unreadCount).toBe(1); + }); +}); + +describe("POST /api/v1/agents/:id/messages/read (mark read)", () => { + it("returns 404 for nonexistent agent", async () => { + const res = await authedInject( + "POST", + "/api/v1/agents/agt_nonexistent/messages/read" + ); + expect(res.statusCode).toBe(404); + }); + + it("marks messages read and returns updated count", async () => { + const res = await authedInject( + "POST", + `/api/v1/agents/${agentB}/messages/read` + ); + expect(res.statusCode).toBe(200); + expect((res.json() as { ok: boolean; updated: number }).updated).toBe(1); + + // Second call should update 0 since the message is already read. + const res2 = await authedInject( + "POST", + `/api/v1/agents/${agentB}/messages/read` + ); + expect(res2.statusCode).toBe(200); + expect((res2.json() as { updated: number }).updated).toBe(0); + }); +}); diff --git a/apps/server/test/send-message-persistence.test.ts b/apps/server/test/send-message-persistence.test.ts new file mode 100644 index 00000000..aa812931 --- /dev/null +++ b/apps/server/test/send-message-persistence.test.ts @@ -0,0 +1,134 @@ +import { + afterAll, + beforeAll, + beforeEach, + describe, + expect, + it, + vi, +} from "vitest"; +import type { Pool } from "pg"; + +vi.mock("../src/shared/git/git-context.js", () => ({ + resolveRepoRoot: vi.fn(async () => "/repo"), + resolveWorktreeRoot: vi.fn(async () => "/repo"), +})); + +import { createMcpHandlers } from "../src/server/mcp-handlers.js"; +import { resolveRepoRoot } from "../src/shared/git/git-context.js"; +import { setupTestDb, teardownTestDb, runTestMigrations } from "./db/setup.js"; + +let pool: Pool; +let handlers: ReturnType; +let sendAgentPrompt: ReturnType; +let publishUiEvent: ReturnType; +let agentManager: { + getAgent: ReturnType; + listAgents: ReturnType; +}; + +const SENDER = { + id: "agt_sender", + name: "sender-agent", + cwd: "/repo", + status: "running", +}; +const RECEIVER = { + id: "agt_receiver", + name: "receiver-agent", + cwd: "/repo", + status: "running", +}; + +beforeAll(async () => { + pool = await setupTestDb(); + await runTestMigrations(); +}); + +afterAll(async () => { + await teardownTestDb(); +}); + +beforeEach(() => { + vi.mocked(resolveRepoRoot).mockResolvedValue("/repo"); + + sendAgentPrompt = vi.fn(async () => {}); + publishUiEvent = vi.fn(); + agentManager = { + getAgent: vi.fn(async (id: string) => + id === SENDER.id ? SENDER : id === RECEIVER.id ? RECEIVER : null + ), + listAgents: vi.fn(async () => [SENDER, RECEIVER]), + }; + + handlers = createMcpHandlers({ + pool, + mediaRoot: "/tmp/media", + agentManager: agentManager as any, + jobService: {} as any, + slackNotifier: {} as any, + publishUiEvent, + withStreamFlag: vi.fn((agent: any) => ({ ...agent, hasStream: false })), + sendAgentPrompt, + appLog: { + info: vi.fn(), + error: vi.fn(), + warn: vi.fn(), + debug: vi.fn(), + } as any, + } as unknown as Parameters[0]); +}); + +describe("sendMessage persistence", () => { + it("persists a row and emits message.created on send", async () => { + await handlers.sendMessage(SENDER.id, { + target: RECEIVER.id, + message: "ping", + senderRepoRoot: "/repo", + }); + + const rows = await pool.query( + "SELECT sender_agent_id, recipient_agent_id, content, delivered FROM agent_messages WHERE content = $1", + ["ping"] + ); + expect(rows.rows).toHaveLength(1); + expect(rows.rows[0].sender_agent_id).toBe(SENDER.id); + expect(rows.rows[0].recipient_agent_id).toBe(RECEIVER.id); + expect(rows.rows[0].delivered).toBe(true); + + expect(publishUiEvent).toHaveBeenCalledWith( + expect.objectContaining({ + type: "message.created", + senderAgentId: SENDER.id, + recipientAgentId: RECEIVER.id, + }) + ); + }); + + it("records a failed delivery as delivered=false and still throws", async () => { + sendAgentPrompt.mockRejectedValueOnce(new Error("tmux boom")); + + await expect( + handlers.sendMessage(SENDER.id, { + target: RECEIVER.id, + message: "will fail", + senderRepoRoot: "/repo", + }) + ).rejects.toThrow("tmux boom"); + + const rows = await pool.query( + "SELECT delivered FROM agent_messages WHERE content = $1", + ["will fail"] + ); + expect(rows.rows).toHaveLength(1); + expect(rows.rows[0].delivered).toBe(false); + + expect(publishUiEvent).toHaveBeenCalledWith( + expect.objectContaining({ + type: "message.created", + senderAgentId: SENDER.id, + recipientAgentId: RECEIVER.id, + }) + ); + }); +}); diff --git a/apps/web/src/components/app/agent-history-detail.tsx b/apps/web/src/components/app/agent-history-detail.tsx index 9b192393..ec169c73 100644 --- a/apps/web/src/components/app/agent-history-detail.tsx +++ b/apps/web/src/components/app/agent-history-detail.tsx @@ -1,5 +1,6 @@ import { useMemo, useState } from "react"; -import { ArrowLeft, Search } from "lucide-react"; +import { AnimatePresence, motion } from "framer-motion"; +import { ArrowLeft, ChevronDown, ChevronRight, Search } from "lucide-react"; import { Bar, BarChart, XAxis, YAxis } from "recharts"; import { @@ -34,6 +35,91 @@ import { EventTimeline, FeedbackTimeline, } from "@/components/app/agent-history-timeline"; +import { + MessageBubble, + groupByParticipant, + type Thread, +} from "@/components/app/messages-panel"; +import { type AgentMessage } from "@/hooks/use-agent-messages"; + +// ── History thread group ──────────────────────────────────────────── + +function HistoryThreadGroup({ + thread, + agentId, +}: { + thread: Thread; + agentId: string; +}): JSX.Element { + const [expanded, setExpanded] = useState(true); + + return ( +
+ + + {expanded && ( + +
+ {thread.messages.map((m) => ( + + ))} +
+ + )} + +
+ ); +} + +function HistoryMessages({ + messages, + agentId, +}: { + messages: AgentMessage[]; + agentId: string; +}): JSX.Element { + const threads = useMemo( + () => groupByParticipant(messages, agentId), + [messages, agentId] + ); + + return ( +
+ {threads.map((thread) => ( + + ))} +
+ ); +} // ── Helpers ───────────────────────────────────────────────────────── @@ -94,13 +180,14 @@ function DurationBar({ durations }: { durations: Record }) { // ── Detail tabs ───────────────────────────────────────────────────── -type DetailTab = "events" | "media" | "pins" | "feedback"; +type DetailTab = "events" | "media" | "pins" | "feedback" | "messages"; function DetailTabs({ events, media, pins, feedback, + messages, agentId, workspaceRoot, }: { @@ -108,6 +195,7 @@ function DetailTabs({ media: HistoryMedia[]; pins: AgentPin[]; feedback: HistoryFeedbackItem[]; + messages: AgentMessage[]; agentId: string; workspaceRoot: string | null; }) { @@ -137,18 +225,19 @@ function DetailTabs({ { key: "media", label: "Media", count: media.length }, { key: "pins", label: "Pins", count: pins.length }, { key: "feedback", label: "Feedback", count: feedback.length }, + { key: "messages", label: "Messages", count: messages.length }, ]; return ( <>
-
+
{tabs.map(({ key, label, count }) => (
@@ -307,7 +405,15 @@ export function AgentHistoryDetail({ ); } - const { agent, events, tokenUsage, media, feedback, stateDurations } = data; + const { + agent, + events, + tokenUsage, + media, + feedback, + messages, + stateDurations, + } = data; const durationMs = new Date(agent.updatedAt).getTime() - new Date(agent.createdAt).getTime(); const totalTokens = @@ -443,12 +549,13 @@ export function AgentHistoryDetail({
)} - {/* Tabbed: Events / Media / Pins / Feedback */} + {/* Tabbed: Events / Media / Pins / Feedback / Messages */} diff --git a/apps/web/src/components/app/agent-type-icon.tsx b/apps/web/src/components/app/agent-type-icon.tsx index 15095775..140ec108 100644 --- a/apps/web/src/components/app/agent-type-icon.tsx +++ b/apps/web/src/components/app/agent-type-icon.tsx @@ -1,4 +1,4 @@ -import { Terminal as TerminalIcon } from "lucide-react"; +import { Bot, Terminal as TerminalIcon } from "lucide-react"; import { siClaude, siCursor } from "simple-icons"; import { cn } from "@/lib/utils"; @@ -38,9 +38,12 @@ function normalizeAgentType( if (type === "terminal") { return "terminal"; } - if (type === "codex" || !type) { + if (type === "codex") { return "codex"; } + if (!type) { + return "unknown"; + } return "unknown"; } @@ -59,7 +62,9 @@ export function AgentTypeIcon({ ? "Cursor" : normalizedType === "terminal" ? "Terminal" - : "Codex"; + : normalizedType === "codex" + ? "Codex" + : "Agent"; const statusClass = eventType ? eventColorClass[eventType] : ""; const baseClass = statusClass ? "inline-flex h-5 w-5 shrink-0 items-center justify-center rounded border transition-colors duration-300" @@ -94,6 +99,18 @@ export function AgentTypeIcon({ ); } + if (normalizedType === "unknown") { + return ( + + + ); + } + const logoPath = normalizedType === "claude" ? siClaude.path diff --git a/apps/web/src/components/app/agents-view.tsx b/apps/web/src/components/app/agents-view.tsx index b5f9ba8d..95da7679 100644 --- a/apps/web/src/components/app/agents-view.tsx +++ b/apps/web/src/components/app/agents-view.tsx @@ -65,6 +65,10 @@ import { type IdeType } from "@/lib/ide-types"; import { type CenterTab } from "@/lib/store"; import { cn } from "@/lib/utils"; import { useAgentActions } from "@/hooks/use-agent-actions"; +import { + useAgentUnreadCount, + useMarkMessagesRead, +} from "@/hooks/use-agent-messages"; import { useAgents } from "@/hooks/use-agents"; import { useMedia } from "@/hooks/use-media"; import { useMediaSidebarState } from "@/hooks/use-media-sidebar-state"; @@ -329,6 +333,17 @@ export function AgentsView({ refreshMedia, } = useMedia(focusedAgentId, mediaPanelOpen); + const unreadMessageCount = useAgentUnreadCount(focusedAgentId); + const markMessagesRead = useMarkMessagesRead(focusedAgentId); + + // Only mark read when the sidebar is actually open on the Messages tab. + // MediaSidebarContent stays mounted while closed and the active tab is + // persisted per-agent, so gating on the tab alone would silently clear + // unread state for agents whose last-used tab was Messages. + useEffect(() => { + if (mediaPanelOpen && mediaActiveTab === "messages") markMessagesRead(); + }, [mediaPanelOpen, mediaActiveTab, markMessagesRead]); + const focusedAgentHasStream = focusedAgent?.hasStream ?? false; const focusedAgentStreamUrl = focusedAgentId ? `/api/v1/agents/${focusedAgentId}/stream` @@ -707,9 +722,9 @@ export function AgentsView({ data-testid="toggle-media-sidebar" > - {unseenMediaCount > 0 ? ( + {unseenMediaCount + unreadMessageCount > 0 ? ( - {unseenMediaCount} + {unseenMediaCount + unreadMessageCount} ) : null} @@ -893,12 +908,10 @@ export function AgentsView({ selectedAgentWorkspaceRoot={ focusedAgent?.worktreePath ?? focusedAgent?.cwd ?? null } - selectedAgentRepoRoot={ - focusedAgent?.gitContext?.repoRoot ?? focusedAgent?.cwd ?? null - } selectedAgentPins={focusedAgent?.pins ?? []} animatingMediaKeys={animatingMediaKeys} unseenMediaCount={unseenMediaCount} + unreadMessageCount={unreadMessageCount} mediaViewportRef={mediaViewportRef} setMediaOpen={setMediaOpen} activeTab={mediaActiveTab} @@ -944,12 +957,10 @@ export function AgentsView({ selectedAgentWorkspaceRoot={ focusedAgent?.worktreePath ?? focusedAgent?.cwd ?? null } - selectedAgentRepoRoot={ - focusedAgent?.gitContext?.repoRoot ?? focusedAgent?.cwd ?? null - } selectedAgentPins={focusedAgent?.pins ?? []} animatingMediaKeys={animatingMediaKeys} unseenMediaCount={unseenMediaCount} + unreadMessageCount={unreadMessageCount} mediaViewportRef={mediaViewportRef} activeTab={mediaActiveTab} setActiveTab={setMediaActiveTab} diff --git a/apps/web/src/components/app/media-sidebar.tsx b/apps/web/src/components/app/media-sidebar.tsx index 0a722407..4c84aaba 100644 --- a/apps/web/src/components/app/media-sidebar.tsx +++ b/apps/web/src/components/app/media-sidebar.tsx @@ -4,11 +4,11 @@ import { ChevronRight, ExternalLink, FileText, + Image, MonitorPlay, Pin, PinOff, X, - Image, File as FileIcon, Video, Upload, @@ -19,7 +19,7 @@ import { type AgentPin, type MediaFile } from "@/components/app/types"; import { type MediaSidebarTab } from "@/lib/store"; import { MediaActions } from "@/components/app/media-lightbox"; import { isTextFile, stripTimestamp } from "@/components/app/media-file-utils"; -import { BrainTabContent } from "@/components/app/brain-tab-content"; +import { MessagesPanel } from "@/components/app/messages-panel"; import { PinsPanel } from "@/components/app/pins-panel"; import { ReviewsSidebarContent } from "@/components/app/reviews-sidebar"; import { useAgentReviews } from "@/hooks/use-agent-reviews"; @@ -51,7 +51,6 @@ type MediaSidebarSharedProps = { selectedAgentId: string | null; selectedAgentName: string | null; selectedAgentWorkspaceRoot: string | null; - selectedAgentRepoRoot: string | null; selectedAgentPins: AgentPin[]; animatingMediaKeys: Set; mediaViewportRef: RefObject; @@ -59,6 +58,7 @@ type MediaSidebarSharedProps = { hasStream: boolean; streamUrl: string | null; unseenMediaCount: number; + unreadMessageCount: number; onUploadFile?: (agentId: string, file: File) => Promise; onNavigateToFile?: (filePath: string, lineStart: number | null) => void; }; @@ -393,7 +393,6 @@ export function MediaSidebarContent({ selectedAgentId, selectedAgentName, selectedAgentWorkspaceRoot, - selectedAgentRepoRoot, selectedAgentPins, animatingMediaKeys, mediaViewportRef, @@ -408,9 +407,13 @@ export function MediaSidebarContent({ onTogglePin, className, unseenMediaCount, + unreadMessageCount, onUploadFile, onNavigateToFile, -}: MediaSidebarContentProps & { unseenMediaCount: number }): JSX.Element { +}: MediaSidebarContentProps & { + unseenMediaCount: number; + unreadMessageCount: number; +}): JSX.Element { const { reviews } = useAgentReviews(selectedAgentId, !!selectedAgentId); const reviewUnresolvedCount = reviews.reduce( (sum, r) => sum + (r.itemCount - r.resolvedCount), @@ -426,11 +429,11 @@ export function MediaSidebarContent({ > {/* Tab header */}
-
+
@@ -570,24 +578,21 @@ export function MediaSidebarContent({
-
- +
); diff --git a/apps/web/src/components/app/messages-panel.tsx b/apps/web/src/components/app/messages-panel.tsx new file mode 100644 index 00000000..b2195d81 --- /dev/null +++ b/apps/web/src/components/app/messages-panel.tsx @@ -0,0 +1,265 @@ +import { useMemo, useCallback } from "react"; +import { useQuery } from "@tanstack/react-query"; +import { useAtom } from "jotai"; +import { AnimatePresence, motion } from "framer-motion"; +import { + MessageSquare, + ArrowUpRight, + ArrowDownLeft, + ChevronDown, + ChevronRight, + Copy, + Check, +} from "lucide-react"; + +import { AgentTypeIcon } from "@/components/app/agent-type-icon"; +import type { Agent } from "@/components/app/types"; +import { + useAgentMessages, + type AgentMessage, +} from "@/hooks/use-agent-messages"; +import { useCopyText } from "@/hooks/use-copy"; +import { messageGroupsCollapsedAtomFamily } from "@/lib/store"; +import { cn } from "@/lib/utils"; + +export type Thread = { + otherId: string; + otherName: string; + otherType?: string; + messages: AgentMessage[]; +}; + +export function groupByParticipant( + messages: AgentMessage[], + agentId: string, + agentMap?: Map +): Thread[] { + const threads = new Map(); + for (const m of messages) { + const isSent = m.senderAgentId === agentId; + const otherId = isSent ? m.recipientAgentId : m.senderAgentId; + const otherName = isSent ? m.recipientName : m.senderName; + const existing = threads.get(otherId); + if (existing) { + existing.messages.push(m); + } else { + const otherAgent = agentMap?.get(otherId); + threads.set(otherId, { + otherId, + otherName, + otherType: otherAgent?.type ?? undefined, + messages: [m], + }); + } + } + return Array.from(threads.values()); +} + +function relativeTime(iso: string): string { + const then = new Date(iso).getTime(); + const secs = Math.round((Date.now() - then) / 1000); + if (secs < 60) return `${secs}s ago`; + if (secs < 3600) return `${Math.round(secs / 60)}m ago`; + if (secs < 86400) return `${Math.round(secs / 3600)}h ago`; + return `${Math.round(secs / 86400)}d ago`; +} + +export function MessageBubble({ + message, + isSent, +}: { + message: AgentMessage; + isSent: boolean; +}): JSX.Element { + const [copied, copyText] = useCopyText(); + + return ( +
+
+
{message.content}
+
+ {isSent ? ( + + ) : ( + + )} + {relativeTime(message.createdAt)} + {!message.delivered && ( + + · not delivered + + )} + +
+
+
+ ); +} + +function ThreadGroup({ + thread, + agentId, + expanded, + onToggle, +}: { + thread: Thread; + agentId: string; + expanded: boolean; + onToggle: () => void; +}): JSX.Element { + return ( +
+ + + {expanded && ( + +
+ {thread.messages.map((m) => ( + + ))} +
+ + )} + +
+ ); +} + +export function MessagesPanel({ + agentId, +}: { + agentId: string | null; +}): JSX.Element { + const { messages, isLoading } = useAgentMessages(agentId); + const [collapsedIds, setCollapsedIds] = useAtom( + messageGroupsCollapsedAtomFamily(agentId ?? "") + ); + + const { data: agentMap } = useQuery>({ + queryKey: ["agents"], + select: (agents) => new Map(agents.map((a) => [a.id, a])), + }); + + const threads = useMemo( + () => (agentId ? groupByParticipant(messages, agentId, agentMap) : []), + [messages, agentId, agentMap] + ); + + const collapsedSet = useMemo(() => new Set(collapsedIds), [collapsedIds]); + + const toggleThread = useCallback( + (otherId: string) => { + setCollapsedIds((prev) => + prev.includes(otherId) + ? prev.filter((id) => id !== otherId) + : [...prev, otherId] + ); + }, + [setCollapsedIds] + ); + + if (!agentId) { + return ( +
+
+ +
No agent selected.
+
+
+ ); + } + + if (isLoading && messages.length === 0) { + return ( +
+
+
+ ); + } + + if (messages.length === 0) { + return ( +
+
+ +
This agent has no messages yet.
+
+
+ ); + } + + return ( +
+ {threads.map((thread) => ( + toggleThread(thread.otherId)} + /> + ))} +
+ ); +} diff --git a/apps/web/src/hooks/use-agent-history.ts b/apps/web/src/hooks/use-agent-history.ts index ea8a4a98..2f85829b 100644 --- a/apps/web/src/hooks/use-agent-history.ts +++ b/apps/web/src/hooks/use-agent-history.ts @@ -2,6 +2,7 @@ import { useQuery } from "@tanstack/react-query"; import { api } from "@/lib/api"; import { getRangeBounds, type ActivityRange } from "@/hooks/use-activity"; import { type AgentPin } from "@/components/app/types"; +import { type AgentMessage } from "@/hooks/use-agent-messages"; const HISTORY_QUERY_OPTIONS = { staleTime: 60_000, @@ -112,6 +113,7 @@ export type HistoryAgentDetail = { tokenUsage: HistoryTokenUsage; media: HistoryMedia[]; feedback: HistoryFeedbackItem[]; + messages: AgentMessage[]; stateDurations: Record; }; diff --git a/apps/web/src/hooks/use-agent-messages.ts b/apps/web/src/hooks/use-agent-messages.ts new file mode 100644 index 00000000..eefe984c --- /dev/null +++ b/apps/web/src/hooks/use-agent-messages.ts @@ -0,0 +1,94 @@ +import { useCallback } from "react"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; +import { api } from "@/lib/api"; + +export type AgentMessage = { + id: string; + senderAgentId: string; + recipientAgentId: string; + senderName: string; + recipientName: string; + content: string; + delivered: boolean; + readAt: string | null; + createdAt: string; +}; + +type MessagesPayload = { + messages: AgentMessage[]; + unreadCount: number; +}; + +function messagesQueryKey(agentId: string | null) { + return ["messages", agentId]; +} + +export function useAgentMessages(agentId: string | null): { + messages: AgentMessage[]; + isLoading: boolean; +} { + const { data, isLoading } = useQuery({ + queryKey: messagesQueryKey(agentId), + queryFn: async () => { + const payload = await api( + `/api/v1/agents/${agentId}/messages` + ); + return { + messages: payload.messages ?? [], + unreadCount: payload.unreadCount ?? 0, + }; + }, + enabled: !!agentId, + staleTime: 0, + refetchOnMount: true, + refetchOnWindowFocus: true, + refetchOnReconnect: true, + }); + + return { messages: data?.messages ?? [], isLoading }; +} + +export function useAgentUnreadCount(agentId: string | null): number { + const { data } = useQuery({ + queryKey: messagesQueryKey(agentId), + queryFn: async () => { + const payload = await api( + `/api/v1/agents/${agentId}/messages` + ); + return { + messages: payload.messages ?? [], + unreadCount: payload.unreadCount ?? 0, + }; + }, + enabled: !!agentId, + staleTime: 0, + refetchOnMount: true, + refetchOnWindowFocus: true, + refetchOnReconnect: true, + }); + + return data?.unreadCount ?? 0; +} + +export function useMarkMessagesRead(agentId: string | null): () => void { + const queryClient = useQueryClient(); + + const { mutate } = useMutation({ + mutationFn: async () => { + if (!agentId) return; + await api(`/api/v1/agents/${agentId}/messages/read`, { method: "POST" }); + }, + onSuccess: () => { + void queryClient.invalidateQueries({ + queryKey: messagesQueryKey(agentId), + exact: true, + }); + }, + }); + + const unreadCount = useAgentUnreadCount(agentId); + + return useCallback(() => { + if (agentId && unreadCount > 0) mutate(); + }, [agentId, unreadCount, mutate]); +} diff --git a/apps/web/src/hooks/use-media-sidebar-state.ts b/apps/web/src/hooks/use-media-sidebar-state.ts index 09682a54..ac1af592 100644 --- a/apps/web/src/hooks/use-media-sidebar-state.ts +++ b/apps/web/src/hooks/use-media-sidebar-state.ts @@ -9,6 +9,7 @@ import { reconcileDiffViewStateStorage, reconcileSplitPaneStateStorage, reconcileReviewDraftStorage, + reconcileMessageGroupsStateStorage, type MediaSidebarTab, } from "@/lib/store"; @@ -100,6 +101,7 @@ export function useMediaSidebarState({ reconcileDiffViewStateStorage(agentIds as string[]); reconcileSplitPaneStateStorage(agentIds as string[]); reconcileReviewDraftStorage(agentIds as string[]); + reconcileMessageGroupsStateStorage(agentIds as string[]); }, [agentIds]); useEffect(() => { diff --git a/apps/web/src/hooks/use-sse.ts b/apps/web/src/hooks/use-sse.ts index 1dce706a..d95c4137 100644 --- a/apps/web/src/hooks/use-sse.ts +++ b/apps/web/src/hooks/use-sse.ts @@ -43,6 +43,8 @@ type UiEvent = | { type: "job.changed" } | { type: "template.changed" } | { type: "brain.changed"; repoRoot: string } + | { type: "message.created"; senderAgentId: string; recipientAgentId: string } + | { type: "message.read"; agentId: string } | { type: "notification"; notificationId: string; @@ -213,6 +215,26 @@ export function useSSE(authState: AuthState): void { return; } + if (payload.type === "message.created") { + void queryClient.invalidateQueries({ + queryKey: ["messages", payload.senderAgentId], + exact: true, + }); + void queryClient.invalidateQueries({ + queryKey: ["messages", payload.recipientAgentId], + exact: true, + }); + return; + } + + if (payload.type === "message.read") { + void queryClient.invalidateQueries({ + queryKey: ["messages", payload.agentId], + exact: true, + }); + return; + } + if (payload.type === "notification") { const shown = showWebNotification(payload); if (shown) { diff --git a/apps/web/src/lib/store.ts b/apps/web/src/lib/store.ts index 55020051..2e3b22ec 100644 --- a/apps/web/src/lib/store.ts +++ b/apps/web/src/lib/store.ts @@ -137,7 +137,7 @@ export function reconcileAgentSidebarOrder( return nextOrder; } -export type MediaSidebarTab = "pins" | "media" | "brain" | "reviews"; +export type MediaSidebarTab = "pins" | "media" | "reviews" | "messages"; export type MediaSidebarState = { isOpen: boolean; @@ -327,3 +327,39 @@ export function reconcileSplitPaneStateStorage( keysToDelete.forEach((key) => window.localStorage.removeItem(key)); } + +// --------------------------------------------------------------------------- +// Message group collapsed state — per-agent set of collapsed thread IDs +// --------------------------------------------------------------------------- + +export const MESSAGE_GROUPS_STATE_STORAGE_PREFIX = + "dispatch:messageGroupsState:"; + +export const messageGroupsCollapsedAtomFamily = atomFamily((agentId: string) => + atomWithLocalStorage( + `${MESSAGE_GROUPS_STATE_STORAGE_PREFIX}${agentId}`, + [] + ) +); + +export function reconcileMessageGroupsStateStorage( + agentIds: Iterable +): void { + if (typeof window === "undefined") return; + + const liveAgentIds = new Set(agentIds); + const keysToDelete: string[] = []; + + for (let i = 0; i < window.localStorage.length; i += 1) { + const key = window.localStorage.key(i); + if (!key?.startsWith(MESSAGE_GROUPS_STATE_STORAGE_PREFIX)) continue; + + const agentId = key + .slice(MESSAGE_GROUPS_STATE_STORAGE_PREFIX.length) + .trim(); + if (!agentId || liveAgentIds.has(agentId)) continue; + keysToDelete.push(key); + } + + keysToDelete.forEach((key) => window.localStorage.removeItem(key)); +} diff --git a/docs/superpowers/specs/2026-06-30-agent-messages-design.md b/docs/superpowers/specs/2026-06-30-agent-messages-design.md new file mode 100644 index 00000000..141a81e0 --- /dev/null +++ b/docs/superpowers/specs/2026-06-30-agent-messages-design.md @@ -0,0 +1,199 @@ +# Design: Agent Messages — persistence, sidebar tab, and History + +**Date:** 2026-06-30 +**Status:** Approved (design phase) + +## Problem + +Agents can already message each other with the `dispatch_send_message` MCP tool, but +those messages are **ephemeral**: the send handler injects the message text straight +into the target agent's tmux session and keeps no record. There is no database row, no +SSE broadcast, and no UI. Unlike pins and brain artifacts — which persist and are surfaced +in the sidebar and History — messages leave no trace to view. + +We want agent-to-agent messages to be viewable with a UX similar to pins/brain artifacts: +a dedicated **Messages** tab in the agent sidebar, and a **Messages** view in the History +section. + +## Scope decisions (confirmed with the user) + +1. **Persist + view only.** Build persistence, broadcast, and the two read surfaces. Keep + the existing same-repo-only send rule. Design the schema so cross-repo becomes a later + config flip rather than a migration — but do **not** unblock cross-repo sending now. +2. **Sidebar tab is per selected agent** and shows _that agent's_ conversations (messages it + sent or received), grouped by the other participant. +3. **History placement is a new tab in the per-agent detail view**, alongside + Events / Media / Pins / Feedback. +4. **View-only from the UI.** Humans do not compose or reply to messages from the UI in this + feature; agents continue to send via the MCP tool. + +### Explicitly out of scope (YAGNI) + +- Unblocking cross-repo _send_ (schema is prepared for it; the send rule is unchanged). +- Composing / replying to messages from the UI. +- Message search / filtering within the sidebar tab. + +## Current-state facts (from codebase exploration) + +- **MCP tool:** `dispatch_send_message` — defined in + `apps/server/src/shared/mcp/messaging-tools.ts` (registration) and handled by the + `sendMessage` handler in `apps/server/src/server/mcp-handlers.ts` (~lines 847–947). + It resolves the target (agent id `agt_*` or fuzzy name match), enforces + same-repo-only via `senderRepoRoot` filtering, blocks self-send and non-running + targets, then injects a delimited envelope through `sendAgentPrompt` → + `injectAgentPrompt` (`apps/server/src/server/agent-prompts.ts`), which uses tmux + `sendCommand`. +- **No persistence today.** No message table exists. `agent_events` + (`apps/server/src/db/migrations/0001_baseline.sql`) logs state transitions only. +- **SSE:** `GET /api/v1/events` (`apps/server/src/routes/agents/events-routes.ts`) + broadcasts UI events (`agent.upsert`, `feedback.created`, `brain.changed`, …) via the + `publishUiEvent` broker. There is no message event type. +- **Sidebar tabs:** `apps/web/src/components/app/media-sidebar.tsx` renders the + Pins / Media / Brain tab bar and panels. Panels stay mounted and toggle via a `hidden` + class. Active tab is a per-agent Jotai atom family persisted to localStorage + (`MediaSidebarTab` union in `apps/web/src/lib/store.ts`; + `use-media-sidebar-state.ts`). Media shows an unseen-count badge. +- **History:** route `/activity/history[/:agentId]`. List in + `apps/web/src/components/app/agent-history-tab.tsx`; detail in + `agent-history-detail.tsx` with a `DetailTabs` component (Events / Media / Pins / + Feedback, each with a count badge). Backed by `GET /api/v1/history/agents/:id` + (`apps/server/src/routes/activity.ts`), which runs parallel queries for events, token + usage, media, and feedback. Hooks in `apps/web/src/hooks/use-agent-history.ts`. + +## Architecture + +Messages become first-class **persisted artifacts**, mirroring pins/brain/feedback. Three +layers, all additive — the existing ephemeral tmux delivery is unchanged: + +1. **Persistence** — a new `agent_messages` table. The `sendMessage` handler writes a row + right after attempting tmux injection (injection-first, so a DB error never blocks + delivery — see Error handling). +2. **Broadcast** — a new `message.created` SSE UI event, published through the same broker + as existing events, so live UIs update without polling. +3. **Surfacing** — a `Messages` sidebar tab (per selected agent) and a `Messages` tab in the + History detail view. Both read from the same API + query hooks. + +## Data model — `agent_messages` (new migration) + +New migration file under `apps/server/src/db/migrations/` following the existing numbering. + +| column | type | notes | +| --------------------- | --------------------- | ----------------------------------------------------------------------------------------------------------------- | +| `id` | uuid / text PK | consistent with existing tables | +| `sender_agent_id` | text FK → `agents.id` | | +| `recipient_agent_id` | text FK → `agents.id` | | +| `sender_name` | text | denormalized snapshot — names change, agents get archived | +| `recipient_name` | text | denormalized snapshot | +| `content` | text | message body | +| `delivered` | boolean | whether tmux injection succeeded | +| `read_at` | timestamptz null | powers unread badges; null = unread | +| `sender_repo_root` | text | stored now even though equal to recipient's today — the "cross-repo later is a config flip, not a migration" hook | +| `recipient_repo_root` | text | as above | +| `created_at` | timestamptz | default now | + +Indexes: `sender_agent_id`, `recipient_agent_id`, `created_at`. + +Rationale for a dedicated table (vs. reusing `agent_events`): messages have distinct +semantics (two participants, read state, delivery state) and distinct read patterns +(grouped by conversation). Denormalizing names + repo roots keeps History correct after +either agent is deleted. + +## Write path (server) + +In the `sendMessage` handler (`apps/server/src/server/mcp-handlers.ts`), after the target is +resolved and tmux injection is attempted: + +1. Insert one `agent_messages` row, capturing `delivered` from the injection result and the + resolved sender/recipient names and repo roots. +2. Publish a `message.created` UI event via the existing broker. + +Failed deliveries are still recorded (rendered as "not delivered"), so a message sent to a +stopped agent is not silently lost. The same-repo filter and all existing validation remain +in place ahead of this write. + +## Sidebar tab (per selected agent) + +- Add `"messages"` to the `MediaSidebarTab` union (`apps/web/src/lib/store.ts`). +- Add a fourth button to the tab bar in `media-sidebar.tsx`, with an **unread badge** reusing + the Media tab's badge pattern (count of messages with `read_at IS NULL` for the selected + agent). +- New `MessagesPanel` component: the selected agent's sent + received messages **grouped by + the other participant** into collapsible conversation threads (mirrors Brain's + Objects / Lists / Events sectioning). Each row shows direction (→ sent / ← received), the + other agent's name, content, and relative time. +- New hook `useAgentMessages(agentId)` → `GET /api/v1/agents/:id/messages`, query key + `["messages", agentId]`, invalidated on the `message.created` SSE event (wired in + `apps/web/src/hooks/use-sse.ts`). +- Panel stays mounted via the `hidden` class, like the other panels. +- Opening the tab marks the agent's visible messages read (clears the badge), consistent + with Media's "seen" behavior. This calls a small `POST`/`PATCH` mark-read endpoint that + sets `read_at`. + +### New server route + +`GET /api/v1/agents/:id/messages` (under `apps/server/src/routes/`, alongside the media +route) — returns messages where the agent is sender or recipient, ordered by `created_at`. +Plus a mark-read endpoint to set `read_at` for the agent's received messages. + +## History detail tab + +- Extend `DetailTab` to include `"messages"` in `agent-history-detail.tsx`; add a `Messages` + tab with a count badge next to Events / Media / Pins / Feedback. +- Extend `GET /api/v1/history/agents/:id` (`apps/server/src/routes/activity.ts`) to include a + `messages` array via a new parallel query — no new endpoint. +- Extend the `HistoryAgentDetail` type + `useHistoryAgentDetail` hook + (`apps/web/src/hooks/use-agent-history.ts`) to carry `messages`. +- New `MessageTimeline` component modeled on the existing `FeedbackTimeline` / + `EventTimeline` (`agent-history-timeline.tsx`), rendering messages chronologically with + sender → recipient labels and delivery state. + +## Data flow summary + +``` +agent A calls dispatch_send_message + -> sendMessage handler (same-repo filter, target resolution, self/running checks) + -> tmux injection (unchanged) ------------------> agent B's session + -> INSERT agent_messages row (delivered=?) + -> publishUiEvent("message.created") + -> SSE /api/v1/events + -> web: invalidate ["messages", senderAgentId] and ["messages", recipientAgentId] + -> MessagesPanel (sidebar) re-renders live +``` + +The **sidebar** Messages tab updates live via SSE. The **History** detail view is a +retrospective surface and is _not_ live-invalidated on `message.created` — consistent with +its sibling tabs (Events/Media/Pins/Feedback), it refetches on remount/refocus and per its +`staleTime`. This is intentional; History does not need to reflect in-flight messages. + +## Error handling + +- Tmux injection failure: still insert the row with `delivered = false`; the UI shows a + "not delivered" affordance. The MCP tool response keeps its existing `delivered` field. +- DB insert failure must not break message delivery to the agent — injection happens first; + a persistence error is logged and surfaced in the tool result but does not throw past + delivery. (Confirm ordering during implementation so a running agent still receives the + message even if the write fails.) +- Deleted/archived participants: History and sidebar render from denormalized names, so rows + remain readable. + +## Testing + +- **Unit (vitest):** + - `sendMessage` writes an `agent_messages` row and emits `message.created`. + - `delivered` reflects injection success vs. failure (target stopped → row with + `delivered = false`). + - Same-repo filter still enforced (cross-repo target rejected as today). + - `GET /api/v1/agents/:id/messages` returns sent + received, correct ordering. + - History detail query includes messages. + - Mark-read sets `read_at` only for the agent's received messages. +- **E2E (Playwright):** + - Launch two agents in one repo, send a message between them. + - Assert it appears in the sidebar Messages tab with an unread badge, and the badge clears + on open. + - Assert it appears in the History detail Messages tab. + - Screenshot published via `dispatch_share`. + +## Pre-completion checks + +`pnpm run check`; `pnpm run finalize:web` (web changed); `pnpm run test:e2e`; +`pnpm run test` (backend logic changed). diff --git a/e2e/agent-messages.spec.ts b/e2e/agent-messages.spec.ts new file mode 100644 index 00000000..bb26c521 --- /dev/null +++ b/e2e/agent-messages.spec.ts @@ -0,0 +1,127 @@ +import path from "node:path"; +import { expect, test } from "@playwright/test"; + +import { + cleanupE2EAgents, + clickAgentRow, + createAgentViaAPI, + loadApp, + seedAgentMessageViaDB, +} from "./helpers"; + +// Agents run inert (no tmux) in E2E, so the real dispatch_send_message path +// cannot run here. Messages are seeded directly into `agent_messages` to +// exercise the sidebar + history UI that reads them. + +test.describe("Agent messages", () => { + test.afterAll(async ({ request }) => { + await cleanupE2EAgents(request); + }); + + test("shows seeded messages in the sidebar and history detail", async ({ + page, + request, + }) => { + const agentA = await createAgentViaAPI(request, { + name: `e2e-agent-messages-a-${Date.now()}`, + }); + const otherAgentId = `e2e-agent-messages-other-${Date.now()}`; + + const receivedContent = `Incoming message ${Date.now()}`; + const sentContent = `Outgoing message ${Date.now()}`; + + // Unread message received by agent A — should trigger the sidebar's + // unread badge on the Messages tab. + await seedAgentMessageViaDB({ + senderAgentId: otherAgentId, + recipientAgentId: agentA.id, + senderName: "Other Agent", + recipientName: agentA.name, + content: receivedContent, + delivered: true, + read: false, + }); + + // A message sent by agent A, already read (no badge contribution). + await seedAgentMessageViaDB({ + senderAgentId: agentA.id, + recipientAgentId: otherAgentId, + senderName: agentA.name, + recipientName: "Other Agent", + content: sentContent, + delivered: true, + read: true, + }); + + await loadApp(page); + + await clickAgentRow(page, agentA.id); + const toggle = page.getByTestId("toggle-media-sidebar"); + await expect(toggle).toBeVisible(); + await toggle.click(); + + const mediaSidebar = page.getByTestId("media-sidebar"); + await expect(mediaSidebar).toBeVisible(); + + const messagesTabButton = mediaSidebar.getByRole("button", { + name: "Messages", + }); + const messagesUnreadBadge = messagesTabButton.locator( + "span.bg-destructive" + ); + + // Unread badge should be visible before opening the tab. + await expect(messagesUnreadBadge).toBeVisible(); + await expect(messagesUnreadBadge).toHaveText("1"); + + await messagesTabButton.click(); + + const messageItems = mediaSidebar.getByTestId("message-item"); + await expect( + messageItems.filter({ hasText: receivedContent }) + ).toBeVisible(); + await expect(messageItems.filter({ hasText: sentContent })).toBeVisible(); + + // Opening the tab marks the unread message as read, clearing the badge. + await expect(messagesUnreadBadge).toBeHidden({ timeout: 10_000 }); + // Confirm it stays cleared (guards against the badge re-appearing due to + // a stale re-render before the mark-read mutation settles). + await page.waitForTimeout(500); + await expect(messagesUnreadBadge).toBeHidden(); + + // Visual validation artifact for the sidebar Messages tab (published via + // dispatch_share since no browser MCP is available in this environment). + await page.screenshot({ + path: path.join( + process.env.E2E_SCREENSHOT_DIR ?? "/tmp", + "agent-messages-sidebar.png" + ), + }); + + await page.goto(`/activity/history/${agentA.id}`, { + waitUntil: "domcontentloaded", + }); + + const historyMessagesTabButton = page.getByRole("button", { + name: "Messages", + }); + await expect(historyMessagesTabButton).toBeVisible(); + await historyMessagesTabButton.click(); + + const historyMessages = page.getByTestId("message-item"); + await expect( + historyMessages.filter({ hasText: receivedContent }) + ).toBeVisible(); + await expect( + historyMessages.filter({ hasText: sentContent }) + ).toBeVisible(); + + // Visual validation artifact for the history detail Messages tab. + await page.screenshot({ + path: path.join( + process.env.E2E_SCREENSHOT_DIR ?? "/tmp", + "agent-messages-history.png" + ), + }); + }); +}); diff --git a/e2e/brain-sidebar.spec.ts b/e2e/brain-sidebar.spec.ts deleted file mode 100644 index 7011317f..00000000 --- a/e2e/brain-sidebar.spec.ts +++ /dev/null @@ -1,297 +0,0 @@ -import { expect, test, type Page } from "@playwright/test"; -import { Pool } from "pg"; - -import { - cleanupE2EAgents, - clickAgentRow, - createAgentViaAPI, - loadApp, -} from "./helpers"; - -const AUTH_HEADER = { - Authorization: `Bearer ${process.env.AUTH_TOKEN ?? "dev-token"}`, -}; - -const REPO_ROOT = "/tmp/e2e-brain-test"; - -async function seedBrainDataViaDB(agentId: string): Promise { - const connectionString = process.env.DATABASE_URL; - if (!connectionString) { - throw new Error("DATABASE_URL is required to seed brain data."); - } - - const pool = new Pool({ connectionString, max: 1 }); - try { - const client = await pool.connect(); - try { - await client.query("BEGIN"); - - await client.query( - `UPDATE agents SET git_context = $2::jsonb WHERE id = $1`, - [agentId, JSON.stringify({ repoRoot: REPO_ROOT, branch: "main" })] - ); - - await client.query( - `INSERT INTO brain_objects (repo_root, collection, name, value, revision, created_by_agent_id, updated_by_agent_id) - VALUES ($1, 'config', 'settings', '{"theme":"dark"}'::jsonb, 1, $2, $2), - ($1, 'state', 'cursor', '{"pos":42}'::jsonb, 1, $2, $2)`, - [REPO_ROOT, agentId] - ); - - await client.query( - `INSERT INTO brain_lists (repo_root, collection, name, revision, created_by_agent_id, updated_by_agent_id) - VALUES ($1, 'state', 'backlog', 1, $2, $2)`, - [REPO_ROOT, agentId] - ); - await client.query( - `INSERT INTO brain_list_items (repo_root, collection, name, item_index, value) - VALUES ($1, 'state', 'backlog', 0, '{"id":"a"}'::jsonb), - ($1, 'state', 'backlog', 1, '{"id":"b"}'::jsonb)`, - [REPO_ROOT] - ); - - await client.query( - `INSERT INTO brain_events (id, repo_root, collection, kind, subject, tags, value, agent_id) - VALUES (gen_random_uuid(), $1, 'reviews', 'assessment', 'ux-review', ARRAY['noise'], '{"score":0.8}'::jsonb, $2), - (gen_random_uuid(), $1, 'reviews', 'decision', NULL, ARRAY[]::text[], '{"action":"approve"}'::jsonb, $2)`, - [REPO_ROOT, agentId] - ); - - await client.query("COMMIT"); - } catch (error) { - await client.query("ROLLBACK"); - throw error; - } finally { - client.release(); - } - } finally { - await pool.end(); - } -} - -async function cleanupBrainDataViaDB(): Promise { - const connectionString = process.env.DATABASE_URL; - if (!connectionString) return; - - const pool = new Pool({ connectionString, max: 1 }); - try { - await pool.query("DELETE FROM brain_list_items WHERE repo_root = $1", [ - REPO_ROOT, - ]); - await pool.query("DELETE FROM brain_lists WHERE repo_root = $1", [ - REPO_ROOT, - ]); - await pool.query("DELETE FROM brain_objects WHERE repo_root = $1", [ - REPO_ROOT, - ]); - await pool.query("DELETE FROM brain_events WHERE repo_root = $1", [ - REPO_ROOT, - ]); - } finally { - await pool.end(); - } -} - -async function openMediaSidebarForAgent( - page: Page, - agent: { id: string; name: string } -) { - await clickAgentRow(page, agent.id); - const toggle = page.getByTestId("toggle-media-sidebar"); - await expect(toggle).toBeVisible({ timeout: 10_000 }); - await toggle.click(); -} - -test.describe("Brain sidebar tab", () => { - test.afterAll(async ({ request }) => { - await cleanupBrainDataViaDB(); - await cleanupE2EAgents(request); - }); - - test("shows brain data in sidebar tab", async ({ page, request }) => { - const agent = await createAgentViaAPI(request, { - name: `e2e-agent-brain-${Date.now()}`, - cwd: "/tmp", - }); - - await seedBrainDataViaDB(agent.id); - - await loadApp(page); - await openMediaSidebarForAgent(page, agent); - - const mediaSidebar = page.getByTestId("media-sidebar"); - await expect(mediaSidebar).toBeVisible(); - - await mediaSidebar.getByRole("button", { name: "Brain" }).click(); - - await expect(mediaSidebar.getByText("Objects")).toBeVisible({ - timeout: 10_000, - }); - await expect(mediaSidebar.getByText("Lists")).toBeVisible(); - await expect(mediaSidebar.getByText("Events")).toBeVisible(); - - await expect(mediaSidebar.getByText("settings")).toBeVisible(); - await expect(mediaSidebar.getByText("cursor")).toBeVisible(); - - await expect(mediaSidebar.getByText("backlog")).toBeVisible(); - await expect(mediaSidebar.getByText("2 items")).toBeVisible(); - - await expect(mediaSidebar.getByText("assessment")).toBeVisible(); - await expect(mediaSidebar.getByText("decision")).toBeVisible(); - await expect(mediaSidebar.getByText("ux-review")).toBeVisible(); - await expect(mediaSidebar.getByText("noise")).toBeVisible(); - }); - - test("shows empty state when agent has no brain data", async ({ - page, - request, - }) => { - const agent = await createAgentViaAPI(request, { - name: `e2e-agent-brain-empty-${Date.now()}`, - cwd: "/tmp", - }); - - await loadApp(page); - await openMediaSidebarForAgent(page, agent); - - const mediaSidebar = page.getByTestId("media-sidebar"); - await mediaSidebar.getByRole("button", { name: "Brain" }).click(); - - await expect( - mediaSidebar.getByText(/brain context available|brain data yet/i) - ).toBeVisible({ timeout: 10_000 }); - }); -}); - -test.describe("Brain API", () => { - let agentId: string; - - test.beforeAll(async ({ request }) => { - const agent = await createAgentViaAPI(request, { - name: `e2e-agent-brain-api-${Date.now()}`, - cwd: "/tmp", - }); - agentId = agent.id; - await seedBrainDataViaDB(agentId); - }); - - test.afterAll(async ({ request }) => { - await cleanupBrainDataViaDB(); - await cleanupE2EAgents(request); - }); - - test("GET /api/v1/brain/projects lists projects", async ({ request }) => { - const res = await request.get("/api/v1/brain/projects", { - headers: AUTH_HEADER, - }); - expect(res.ok()).toBeTruthy(); - const body = (await res.json()) as Array<{ - repoRoot: string; - objectCount: number; - }>; - const project = body.find((p) => p.repoRoot === REPO_ROOT); - expect(project).toBeDefined(); - expect(project!.objectCount).toBeGreaterThanOrEqual(2); - }); - - test("GET /api/v1/brain/collections returns summaries", async ({ - request, - }) => { - const res = await request.get( - `/api/v1/brain/collections?repoRoot=${encodeURIComponent(REPO_ROOT)}`, - { headers: AUTH_HEADER } - ); - expect(res.ok()).toBeTruthy(); - const body = (await res.json()) as Array<{ collection: string }>; - const collections = body.map((c) => c.collection); - expect(collections).toContain("config"); - expect(collections).toContain("state"); - expect(collections).toContain("reviews"); - }); - - test("GET /api/v1/brain/collections returns 400 without repoRoot", async ({ - request, - }) => { - const res = await request.get("/api/v1/brain/collections", { - headers: AUTH_HEADER, - }); - expect(res.status()).toBe(400); - }); - - test("GET /api/v1/brain/objects lists objects", async ({ request }) => { - const res = await request.get( - `/api/v1/brain/objects?repoRoot=${encodeURIComponent(REPO_ROOT)}`, - { headers: AUTH_HEADER } - ); - expect(res.ok()).toBeTruthy(); - const body = (await res.json()) as Array<{ name: string }>; - expect(body.length).toBeGreaterThanOrEqual(2); - }); - - test("GET /api/v1/brain/objects/:collection/:name gets a single object", async ({ - request, - }) => { - const res = await request.get( - `/api/v1/brain/objects/config/settings?repoRoot=${encodeURIComponent(REPO_ROOT)}`, - { headers: AUTH_HEADER } - ); - expect(res.ok()).toBeTruthy(); - const body = (await res.json()) as { name: string; value: unknown }; - expect(body.name).toBe("settings"); - expect(body.value).toEqual({ theme: "dark" }); - }); - - test("GET /api/v1/brain/lists lists with item counts", async ({ - request, - }) => { - const res = await request.get( - `/api/v1/brain/lists?repoRoot=${encodeURIComponent(REPO_ROOT)}`, - { headers: AUTH_HEADER } - ); - expect(res.ok()).toBeTruthy(); - const body = (await res.json()) as Array<{ - name: string; - itemCount: number; - }>; - const backlog = body.find((l) => l.name === "backlog"); - expect(backlog).toBeDefined(); - expect(backlog!.itemCount).toBe(2); - }); - - test("GET /api/v1/brain/events lists events", async ({ request }) => { - const res = await request.get( - `/api/v1/brain/events?repoRoot=${encodeURIComponent(REPO_ROOT)}`, - { headers: AUTH_HEADER } - ); - expect(res.ok()).toBeTruthy(); - const body = (await res.json()) as Array<{ kind: string }>; - expect(body.length).toBeGreaterThanOrEqual(2); - }); - - test("GET /api/v1/brain/agent-activity/:agentId returns activity", async ({ - request, - }) => { - const res = await request.get( - `/api/v1/brain/agent-activity/${agentId}?repoRoot=${encodeURIComponent(REPO_ROOT)}`, - { headers: AUTH_HEADER } - ); - expect(res.ok()).toBeTruthy(); - const body = (await res.json()) as { - objects: Array<{ name: string }>; - lists: Array<{ name: string }>; - events: Array<{ kind: string }>; - }; - expect(body.objects.length).toBeGreaterThanOrEqual(2); - expect(body.lists.length).toBeGreaterThanOrEqual(1); - expect(body.events.length).toBeGreaterThanOrEqual(2); - }); - - test("GET /api/v1/brain/agent-activity/:agentId returns 400 without repoRoot", async ({ - request, - }) => { - const res = await request.get(`/api/v1/brain/agent-activity/${agentId}`, { - headers: AUTH_HEADER, - }); - expect(res.status()).toBe(400); - }); -}); diff --git a/e2e/helpers.ts b/e2e/helpers.ts index bf5835e7..49bcd4ff 100644 --- a/e2e/helpers.ts +++ b/e2e/helpers.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import { Pool } from "pg"; import { type Page, type APIRequestContext } from "@playwright/test"; @@ -238,6 +239,55 @@ export async function setAgentRoleViaDB( } } +/** + * Seed a row directly into `agent_messages`, bypassing the real + * dispatch_send_message path (which requires two running tmux agents and + * cannot run in the inert-runtime E2E stack). `agent_messages` has no FK + * constraints, so the "other" participant id can be any string. + */ +export async function seedAgentMessageViaDB(message: { + senderAgentId: string; + recipientAgentId: string; + senderName: string; + recipientName: string; + content: string; + delivered?: boolean; + read?: boolean; + senderRepoRoot?: string | null; + recipientRepoRoot?: string | null; +}): Promise { + const connectionString = process.env.DATABASE_URL; + if (!connectionString) { + throw new Error("DATABASE_URL is required to seed agent messages."); + } + + const id = randomUUID(); + const pool = new Pool({ connectionString, max: 1 }); + try { + await pool.query( + `INSERT INTO agent_messages + (id, sender_agent_id, recipient_agent_id, sender_name, recipient_name, + content, delivered, read_at, sender_repo_root, recipient_repo_root) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)`, + [ + id, + message.senderAgentId, + message.recipientAgentId, + message.senderName, + message.recipientName, + message.content, + message.delivered ?? true, + message.read ? new Date() : null, + message.senderRepoRoot ?? null, + message.recipientRepoRoot ?? null, + ] + ); + } finally { + await pool.end(); + } + return id; +} + export async function seedPersonaRecheckFixtureViaDB( parentAgentId: string ): Promise<{ diff --git a/e2e/split-pane.spec.ts b/e2e/split-pane.spec.ts index cae1e20e..74472637 100644 --- a/e2e/split-pane.spec.ts +++ b/e2e/split-pane.spec.ts @@ -12,6 +12,9 @@ async function waitForAppShell( .getByText(agentName) .first() .waitFor({ state: "visible" }); + // Wait for the agent to actually be focused (URL-routed) so per-agent + // atoms like splitPaneState are active. + await page.getByTestId("current-session-name").waitFor({ state: "attached" }); } /** Seed split-pane state directly in localStorage — deterministic, avoids