From 34df55a10ad90c09722d9344207535875d4e4cfe Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 9 May 2026 17:10:22 +0800 Subject: [PATCH] Fix chat fanout + unify system-injected agent prompts (#830) * Fix chat everyone fanout * Unify system-injected agent prompts Chat mentions, schedule fires, and notifyOnFinish now share one shape: prompt wrapped in ..., no user-message turn recorded, no auto-unarchive. Agents can recognize system-injected context vs a real user turn. - Move agent prompt orchestration out of mcp-shared.ts into a neutral agent-prompt.ts module (mcp is an interface, not the home). - Add `unarchive` and `recordUserMessage` flags to sendPromptToAgent. - Add formatSystemNotificationPrompt helper. - Schedule fires now identify themselves: "Schedule \"\" fired (id=..., run=...)." Plumb runId through runner so the paseo.schedule-run label matches the actual run record. - Replace chat-mention validateChatMentionFanout + duplicate resolveChatMentionTargetAgentIds with single prepareChatMentionFanout that owns expansion + cap + poster snapshot. --- ...cp-shared.test.ts => agent-prompt.test.ts} | 2 +- .../server/src/server/agent/agent-prompt.ts | 243 ++++++++++++++++++ .../server/src/server/agent/mcp-server.ts | 4 +- .../server/src/server/agent/mcp-shared.ts | 219 +--------------- .../src/server/chat/chat-mentions.test.ts | 145 +++++++++-- .../server/src/server/chat/chat-mentions.ts | 176 +++++++++---- .../src/server/chat/chat-service.test.ts | 30 +++ .../server/src/server/chat/chat-service.ts | 14 + .../server/src/server/schedule/service.ts | 40 +-- packages/server/src/server/session.ts | 43 +++- 10 files changed, 608 insertions(+), 308 deletions(-) rename packages/server/src/server/agent/{mcp-shared.test.ts => agent-prompt.test.ts} (97%) create mode 100644 packages/server/src/server/agent/agent-prompt.ts diff --git a/packages/server/src/server/agent/mcp-shared.test.ts b/packages/server/src/server/agent/agent-prompt.test.ts similarity index 97% rename from packages/server/src/server/agent/mcp-shared.test.ts rename to packages/server/src/server/agent/agent-prompt.test.ts index ff0d12cc7..9223ed36d 100644 --- a/packages/server/src/server/agent/mcp-shared.test.ts +++ b/packages/server/src/server/agent/agent-prompt.test.ts @@ -3,7 +3,7 @@ import { expect, it, vi } from "vitest"; import { createTestLogger } from "../../test-utils/test-logger.js"; import { AgentManager } from "./agent-manager.js"; import { AgentStorage } from "./agent-storage.js"; -import { setupFinishNotification } from "./mcp-shared.js"; +import { setupFinishNotification } from "./agent-prompt.js"; import type { AgentManagerEvent, ManagedAgent } from "./agent-manager.js"; it("does not notify archived callers", async () => { diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts new file mode 100644 index 000000000..5135c77e1 --- /dev/null +++ b/packages/server/src/server/agent/agent-prompt.ts @@ -0,0 +1,243 @@ +import type { Logger } from "pino"; + +import type { AgentPromptInput, AgentRunOptions } from "./agent-sdk-types.js"; +import type { AgentManager } from "./agent-manager.js"; +import type { AgentStorage } from "./agent-storage.js"; +import { ensureAgentLoaded } from "./agent-loading.js"; + +export interface StartAgentRunOptions { + replaceRunning?: boolean; + runOptions?: AgentRunOptions; +} + +export function startAgentRun( + agentManager: AgentManager, + agentId: string, + prompt: AgentPromptInput, + logger: Logger, + options?: StartAgentRunOptions, +): { outOfBand: boolean } { + // Out-of-band commands (e.g. /goal pause) must run WITHOUT canceling an + // in-flight turn — replaceAgentRun would interrupt the running turn. The + // intercept lives at this layer so it covers every prompt entrypoint. + if (agentManager.tryRunOutOfBand(agentId, prompt)) { + return { outOfBand: true }; + } + const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId)); + const runOptions = options?.runOptions; + const iterator = shouldReplace + ? agentManager.replaceAgentRun(agentId, prompt, runOptions) + : agentManager.streamAgent(agentId, prompt, runOptions); + void (async () => { + try { + for await (const _ of iterator) { + // Events are broadcast via AgentManager subscribers. + } + } catch (error) { + logger.error({ err: error, agentId }, "Agent stream failed"); + } + })(); + return { outOfBand: false }; +} + +/** + * Clear the archived flag from a stored agent record. + * Shared across Session (app/WS), MCP, and CLI so every surface that acts on + * an archived agent unarchives it the same way. + */ +export async function unarchiveAgentState( + agentStorage: AgentStorage, + agentManager: AgentManager, + agentId: string, +): Promise { + const record = await agentStorage.get(agentId); + if (!record || !record.archivedAt) { + return false; + } + const updatedAt = new Date().toISOString(); + await agentStorage.upsert({ + ...record, + archivedAt: null, + updatedAt, + }); + agentManager.notifyAgentState(agentId); + return true; +} + +/** + * Wrap a body in … so the receiving agent + * recognizes the prompt as system-injected context — not a user turn. + * Used by chat mentions, schedule fires, and notify-on-finish. + */ +export function formatSystemNotificationPrompt(reason: string): string { + return `\n${reason}\n`; +} + +export interface SendPromptToAgentParams { + agentManager: AgentManager; + agentStorage: AgentStorage; + agentId: string; + /** Prompt to dispatch to the provider (may include image blocks or wrapped text). */ + prompt: AgentPromptInput; + /** Raw user text to record in the timeline. Required when recordUserMessage is true. */ + userMessageText?: string; + messageId?: string; + runOptions?: AgentRunOptions; + /** Optional mode to set on the agent before the run starts. */ + sessionMode?: string; + /** + * Default true. When false, archived agents are skipped instead of being + * unarchived. Use false for system-injected prompts (chat mentions, + * schedule fires, notify-on-finish). + */ + unarchive?: boolean; + /** + * Default true. When false, the prompt is dispatched to the provider but + * no user-message turn is added to the timeline — the app shows nothing. + * Use false for system-injected prompts. + */ + recordUserMessage?: boolean; + logger: Logger; +} + +/** + * Full send-prompt orchestration: (optional unarchive) → load → (optional + * mode change) → (optional record user message) → start run. + * + * Every surface that sends a prompt to an agent (Session/WS, MCP, CLI-through-MCP, + * chat mentions, notify-on-finish) MUST go through this so behavior can never + * drift between them. + * + * When `unarchive` is false and the agent is archived, the call is a silent + * no-op (returns `{ outOfBand: false }`) — the agent is not run. + */ +export async function sendPromptToAgent( + params: SendPromptToAgentParams, +): Promise<{ outOfBand: boolean }> { + const unarchive = params.unarchive ?? true; + const recordUserMessage = params.recordUserMessage ?? true; + + if (recordUserMessage && params.userMessageText === undefined) { + throw new Error("userMessageText is required when recordUserMessage is true"); + } + + const record = await params.agentStorage.get(params.agentId); + if (record?.archivedAt) { + if (!unarchive) { + return { outOfBand: false }; + } + await unarchiveAgentState(params.agentStorage, params.agentManager, params.agentId); + } + + await ensureAgentLoaded(params.agentId, { + agentManager: params.agentManager, + agentStorage: params.agentStorage, + logger: params.logger, + }); + + if (params.sessionMode) { + await params.agentManager.setAgentMode(params.agentId, params.sessionMode); + } + + if (recordUserMessage && params.userMessageText !== undefined) { + try { + params.agentManager.recordUserMessage(params.agentId, params.userMessageText, { + messageId: params.messageId, + emitState: false, + }); + } catch (error) { + params.logger.error({ err: error, agentId: params.agentId }, "Failed to record user message"); + } + } + + return startAgentRun(params.agentManager, params.agentId, params.prompt, params.logger, { + replaceRunning: true, + runOptions: params.runOptions, + }); +} + +export interface SetupFinishNotificationParams { + agentManager: AgentManager; + agentStorage: AgentStorage; + childAgentId: string; + callerAgentId: string; + logger: Logger; +} + +export function setupFinishNotification(params: SetupFinishNotificationParams): void { + const { agentManager, agentStorage, childAgentId, callerAgentId, logger } = params; + let hasSeenRunning = false; + let fired = false; + let unsubscribe: (() => void) | null = null; + + async function notify(reason: "finished" | "errored" | "needs permission"): Promise { + if (fired) { + return; + } + fired = true; + unsubscribe?.(); + + const title = agentManager.getAgent(childAgentId)?.config?.title ?? childAgentId; + const body = `Agent ${childAgentId} (${title}) ${reason}.`; + + await sendPromptToAgent({ + agentManager, + agentStorage, + agentId: callerAgentId, + prompt: formatSystemNotificationPrompt(body), + unarchive: false, + recordUserMessage: false, + logger, + }); + } + + unsubscribe = agentManager.subscribe( + (event) => { + if (fired) { + return; + } + + if (event.type === "agent_state") { + if (event.agent.lifecycle === "running") { + hasSeenRunning = true; + return; + } + if (event.agent.lifecycle === "error") { + void notify("errored"); + return; + } + if (event.agent.lifecycle === "idle" && hasSeenRunning) { + void notify("finished"); + return; + } + if (event.agent.lifecycle === "closed") { + fired = true; + unsubscribe?.(); + return; + } + return; + } + + if (event.event.type === "permission_requested") { + void notify("needs permission"); + } + }, + { agentId: childAgentId, replayState: false }, + ); + + // Check if the child is already running (catches the case where + // the lifecycle flipped before our subscribe call was processed). + // Do NOT treat an immediate "idle" as "finished" — the agent may + // not have started yet (streamAgent sets a pending run before + // transitioning to "running"). + const childSnapshot = agentManager.getAgent(childAgentId); + if (!childSnapshot || childSnapshot.lifecycle === "closed") { + unsubscribe(); + return; + } + if (childSnapshot.lifecycle === "running") { + hasSeenRunning = true; + } else if (childSnapshot.lifecycle === "error") { + void notify("errored"); + } +} diff --git a/packages/server/src/server/agent/mcp-server.ts b/packages/server/src/server/agent/mcp-server.ts index 39c4f13fc..b1337f1ca 100644 --- a/packages/server/src/server/agent/mcp-server.ts +++ b/packages/server/src/server/agent/mcp-server.ts @@ -60,13 +60,11 @@ import { parseDurationString, resolveRequiredProviderModel, sanitizePermissionRequest, - sendPromptToAgent, - setupFinishNotification, serializeSnapshotWithMetadata, - startAgentRun, toScheduleSummary, waitForAgentWithTimeout, } from "./mcp-shared.js"; +import { sendPromptToAgent, setupFinishNotification, startAgentRun } from "./agent-prompt.js"; import type { GitHubService } from "../../services/github-service.js"; import type { WorkspaceGitService } from "../workspace-git-service.js"; import type { CreatePaseoWorktreeInput } from "../paseo-worktree-service.js"; diff --git a/packages/server/src/server/agent/mcp-shared.ts b/packages/server/src/server/agent/mcp-shared.ts index 4a41867cb..acbe84c10 100644 --- a/packages/server/src/server/agent/mcp-shared.ts +++ b/packages/server/src/server/agent/mcp-shared.ts @@ -1,16 +1,11 @@ import { z } from "zod"; import type { Logger } from "pino"; -import type { - AgentPromptInput, - AgentPermissionRequest, - AgentRunOptions, -} from "./agent-sdk-types.js"; +import type { AgentPermissionRequest } from "./agent-sdk-types.js"; import type { AgentManager, ManagedAgent, WaitForAgentResult } from "./agent-manager.js"; import { curateAgentActivity } from "./activity-curator.js"; import { selectItemsByProjectedLimit } from "./timeline-projection.js"; import type { AgentStorage } from "./agent-storage.js"; -import { ensureAgentLoaded } from "./agent-loading.js"; import { serializeAgentSnapshot } from "../messages.js"; import { StoredScheduleSchema } from "../schedule/types.js"; import type { AgentProvider } from "./agent-sdk-types.js"; @@ -91,11 +86,6 @@ export function resolveRequiredProviderModel( }; } -export interface StartAgentRunOptions { - replaceRunning?: boolean; - runOptions?: AgentRunOptions; -} - /** * Wraps agentManager.waitForAgentEvent with a self-imposed timeout. * Returns a friendly message when timeout occurs, rather than letting @@ -168,213 +158,6 @@ export async function waitForAgentWithTimeout( } } -export function startAgentRun( - agentManager: AgentManager, - agentId: string, - prompt: AgentPromptInput, - logger: Logger, - options?: StartAgentRunOptions, -): { outOfBand: boolean } { - // Out-of-band commands (e.g. /goal pause) must run WITHOUT canceling an - // in-flight turn — replaceAgentRun would interrupt the running turn. The - // intercept lives at this layer so it covers every prompt entrypoint. - if (agentManager.tryRunOutOfBand(agentId, prompt)) { - return { outOfBand: true }; - } - const shouldReplace = Boolean(options?.replaceRunning && agentManager.hasInFlightRun(agentId)); - const runOptions = options?.runOptions; - const iterator = shouldReplace - ? agentManager.replaceAgentRun(agentId, prompt, runOptions) - : agentManager.streamAgent(agentId, prompt, runOptions); - void (async () => { - try { - for await (const _ of iterator) { - // Events are broadcast via AgentManager subscribers. - } - } catch (error) { - logger.error({ err: error, agentId }, "Agent stream failed"); - } - })(); - return { outOfBand: false }; -} - -/** - * Clear the archived flag from a stored agent record. - * Shared across Session (app/WS), MCP, and CLI so every surface that acts on - * an archived agent unarchives it the same way. - */ -export async function unarchiveAgentState( - agentStorage: AgentStorage, - agentManager: AgentManager, - agentId: string, -): Promise { - const record = await agentStorage.get(agentId); - if (!record || !record.archivedAt) { - return false; - } - const updatedAt = new Date().toISOString(); - await agentStorage.upsert({ - ...record, - archivedAt: null, - updatedAt, - }); - agentManager.notifyAgentState(agentId); - return true; -} - -export interface SendPromptToAgentParams { - agentManager: AgentManager; - agentStorage: AgentStorage; - agentId: string; - /** Raw user text to record in the timeline. */ - userMessageText: string; - /** Prompt to dispatch to the provider (may include image blocks or wrapped text). */ - prompt: AgentPromptInput; - messageId?: string; - runOptions?: AgentRunOptions; - /** Optional mode to set on the agent before the run starts. */ - sessionMode?: string; - logger: Logger; -} - -/** - * Full send-prompt orchestration: unarchive → load → (optional mode change) → - * record user message → start run. - * - * Every surface that sends a prompt to an agent (Session/WS, MCP, CLI-through-MCP) - * MUST go through this so behavior can never drift between them. - */ -export async function sendPromptToAgent( - params: SendPromptToAgentParams, -): Promise<{ outOfBand: boolean }> { - const { - agentManager, - agentStorage, - agentId, - userMessageText, - prompt, - messageId, - runOptions, - sessionMode, - logger, - } = params; - - await unarchiveAgentState(agentStorage, agentManager, agentId); - - await ensureAgentLoaded(agentId, { - agentManager, - agentStorage, - logger, - }); - - if (sessionMode) { - await agentManager.setAgentMode(agentId, sessionMode); - } - - try { - agentManager.recordUserMessage(agentId, userMessageText, { - messageId, - emitState: false, - }); - } catch (error) { - logger.error({ err: error, agentId }, "Failed to record user message"); - } - - return startAgentRun(agentManager, agentId, prompt, logger, { - replaceRunning: true, - runOptions, - }); -} - -interface SetupFinishNotificationParams { - agentManager: AgentManager; - agentStorage: AgentStorage; - childAgentId: string; - callerAgentId: string; - logger: Logger; -} - -export function setupFinishNotification(params: SetupFinishNotificationParams): void { - const { agentManager, agentStorage, childAgentId, callerAgentId, logger } = params; - let hasSeenRunning = false; - let fired = false; - let unsubscribe: (() => void) | null = null; - - async function notify(reason: "finished" | "errored" | "needs permission"): Promise { - if (fired) { - return; - } - fired = true; - unsubscribe?.(); - - if (!agentManager.getAgent(callerAgentId)) { - return; - } - - const callerRecord = await agentStorage.get(callerAgentId); - if (callerRecord?.archivedAt) { - return; - } - - const title = agentManager.getAgent(childAgentId)?.config?.title ?? childAgentId; - const prompt = `\nAgent ${childAgentId} (${title}) ${reason}.\n`; - - startAgentRun(agentManager, callerAgentId, prompt, logger, { - replaceRunning: true, - }); - } - - unsubscribe = agentManager.subscribe( - (event) => { - if (fired) { - return; - } - - if (event.type === "agent_state") { - if (event.agent.lifecycle === "running") { - hasSeenRunning = true; - return; - } - if (event.agent.lifecycle === "error") { - void notify("errored"); - return; - } - if (event.agent.lifecycle === "idle" && hasSeenRunning) { - void notify("finished"); - return; - } - if (event.agent.lifecycle === "closed") { - fired = true; - unsubscribe?.(); - return; - } - return; - } - - if (event.event.type === "permission_requested") { - void notify("needs permission"); - } - }, - { agentId: childAgentId, replayState: false }, - ); - - // Check if the child is already running (catches the case where - // the lifecycle flipped before our subscribe call was processed). - // Do NOT treat an immediate "idle" as "finished" — the agent may - // not have started yet (streamAgent sets a pending run before - // transitioning to "running"). - const childSnapshot = agentManager.getAgent(childAgentId); - if (!childSnapshot || childSnapshot.lifecycle === "closed") { - unsubscribe(); - return; - } - if (childSnapshot.lifecycle === "running") { - hasSeenRunning = true; - } else if (childSnapshot.lifecycle === "error") { - void notify("errored"); - } -} - export function sanitizePermissionRequest( permission: AgentPermissionRequest | null | undefined, ): AgentPermissionRequest | null { diff --git a/packages/server/src/server/chat/chat-mentions.test.ts b/packages/server/src/server/chat/chat-mentions.test.ts index dbc02842a..b209abb0d 100644 --- a/packages/server/src/server/chat/chat-mentions.test.ts +++ b/packages/server/src/server/chat/chat-mentions.test.ts @@ -5,46 +5,150 @@ import type { ManagedAgent } from "../agent/agent-manager.js"; import { buildChatMentionNotification, notifyChatMentions, - resolveChatMentionTargetAgentIds, + prepareChatMentionFanout, } from "./chat-mentions.js"; function storedAgent(overrides: Partial & { id: string }): StoredAgentRecord { - return { internal: false, archivedAt: null, ...overrides } as StoredAgentRecord; + return { + internal: false, + archivedAt: null, + lastStatus: "idle", + ...overrides, + } as StoredAgentRecord; } function liveAgent(overrides: Partial & { id: string }): ManagedAgent { - return { internal: false, ...overrides } as ManagedAgent; + return { internal: false, lifecycle: "idle", ...overrides } as ManagedAgent; +} + +async function prepare(input: { + authorAgentId: string; + mentionAgentIds: string[]; + storedAgents?: StoredAgentRecord[]; + liveAgents?: ManagedAgent[]; + roomPosterAgentIds?: string[]; + limit?: number; +}) { + const result = await prepareChatMentionFanout({ + authorAgentId: input.authorAgentId, + mentionAgentIds: input.mentionAgentIds, + storedAgents: input.storedAgents ?? [], + liveAgents: input.liveAgents ?? [], + listRoomPosterAgentIds: async () => input.roomPosterAgentIds ?? [], + limit: input.limit, + }); + if (!result.ok) { + throw new Error(`expected ok prepare, got error: ${result.error}`); + } + return result.prepared; } describe("chat mentions", () => { - test("@everyone expands to active non-archived agents", () => { - const targets = resolveChatMentionTargetAgentIds({ + test("@everyone in an empty room resolves to no targets", async () => { + const prepared = await prepare({ + authorAgentId: "author-agent", + mentionAgentIds: ["everyone"], + storedAgents: [storedAgent({ id: "unrelated-agent" })], + liveAgents: [liveAgent({ id: "live-unrelated-agent" })], + roomPosterAgentIds: [], + }); + + expect(prepared.targetMentionAgentIds).toEqual([]); + }); + + test("@everyone in a single-poster room excludes the author", async () => { + const prepared = await prepare({ + authorAgentId: "author-agent", + mentionAgentIds: ["everyone"], + storedAgents: [storedAgent({ id: "author-agent" })], + roomPosterAgentIds: ["author-agent"], + }); + + expect(prepared.targetMentionAgentIds).toEqual([]); + }); + + test("@everyone only expands to active posters in the room", async () => { + const prepared = await prepare({ authorAgentId: "author-agent", mentionAgentIds: ["everyone"], storedAgents: [ storedAgent({ id: "agent-a" }), - storedAgent({ id: "agent-b", archivedAt: "2026-03-28T00:00:00.000Z" }), - storedAgent({ id: "author-agent" }), - storedAgent({ id: "internal-agent", internal: true }), + storedAgent({ id: "agent-b" }), + storedAgent({ id: "unrelated-agent" }), ], - liveAgents: [ - liveAgent({ id: "agent-c" }), - liveAgent({ id: "internal-live", internal: true }), + liveAgents: [liveAgent({ id: "agent-c" }), liveAgent({ id: "live-unrelated-agent" })], + roomPosterAgentIds: ["agent-a", "agent-b", "agent-c"], + }); + + expect([...prepared.targetMentionAgentIds].sort()).toEqual(["agent-a", "agent-b", "agent-c"]); + }); + + test("@everyone excludes archived and error-state agents", async () => { + const prepared = await prepare({ + authorAgentId: "author-agent", + mentionAgentIds: ["everyone"], + storedAgents: [ + storedAgent({ id: "active-agent" }), + storedAgent({ id: "archived-agent", archivedAt: "2026-03-28T00:00:00.000Z" }), + storedAgent({ id: "stored-error-agent", lastStatus: "error" }), + ], + liveAgents: [liveAgent({ id: "live-error-agent", lifecycle: "error" })], + roomPosterAgentIds: [ + "active-agent", + "archived-agent", + "stored-error-agent", + "live-error-agent", ], }); - expect(targets.sort()).toEqual(["agent-a", "agent-c"]); + expect(prepared.targetMentionAgentIds).toEqual(["active-agent"]); }); - test("@everyone deduplicates with explicit mentions and keeps explicit non-everyone mentions", () => { - const targets = resolveChatMentionTargetAgentIds({ + test("@everyone deduplicates with explicit mentions and keeps explicit non-everyone mentions", async () => { + const prepared = await prepare({ authorAgentId: "author-agent", mentionAgentIds: ["everyone", "agent-a", "custom-title"], storedAgents: [storedAgent({ id: "agent-a" })], liveAgents: [liveAgent({ id: "agent-b" })], + roomPosterAgentIds: ["agent-a", "agent-b"], }); - expect(targets.sort()).toEqual(["agent-a", "agent-b", "custom-title"]); + expect([...prepared.targetMentionAgentIds].sort()).toEqual([ + "agent-a", + "agent-b", + "custom-title", + ]); + }); + + test("does not list room posters when @everyone is not mentioned", async () => { + const listRoomPosterAgentIds = vi.fn(async () => []); + const result = await prepareChatMentionFanout({ + authorAgentId: "author-agent", + mentionAgentIds: ["agent-a"], + storedAgents: [storedAgent({ id: "agent-a" })], + liveAgents: [], + listRoomPosterAgentIds, + }); + + expect(result.ok).toBe(true); + expect(listRoomPosterAgentIds).not.toHaveBeenCalled(); + }); + + test("rejects @everyone fan-out above the hard cap", async () => { + const posters = Array.from({ length: 26 }, (_, index) => `agent-${index}`); + const result = await prepareChatMentionFanout({ + authorAgentId: "author-agent", + mentionAgentIds: ["everyone"], + storedAgents: posters.map((id) => storedAgent({ id })), + liveAgents: [], + listRoomPosterAgentIds: async () => posters, + }); + + expect(result).toEqual({ + ok: false, + error: + "@everyone would notify 26 agents, which exceeds the limit of 25. Narrow the room or mention specific agents.", + }); }); test("notification body strips inline mentions but keeps the room context", () => { @@ -68,14 +172,21 @@ describe("chat mentions", () => { warn: vi.fn(), } as unknown as pino.Logger; + const storedAgents = [storedAgent({ id: "agent-a" })]; + const liveAgents = [liveAgent({ id: "agent-b" })]; + await notifyChatMentions({ room: "coord-room", authorAgentId: "author-agent", body: "@everyone Check status", mentionAgentIds: ["everyone"], logger, - listStoredAgents: async () => [storedAgent({ id: "agent-a" })], - listLiveAgents: () => [liveAgent({ id: "agent-b" })], + storedAgents, + liveAgents, + prepared: { + targetMentionAgentIds: ["agent-a", "agent-b"], + roomPosterAgentIds: ["agent-a", "agent-b"], + }, resolveAgentIdentifier, sendAgentMessage, }); diff --git a/packages/server/src/server/chat/chat-mentions.ts b/packages/server/src/server/chat/chat-mentions.ts index 9740c1b43..f07237dc4 100644 --- a/packages/server/src/server/chat/chat-mentions.ts +++ b/packages/server/src/server/chat/chat-mentions.ts @@ -2,6 +2,8 @@ import type pino from "pino"; import type { StoredAgentRecord } from "../agent/agent-storage.js"; import type { ManagedAgent } from "../agent/agent-manager.js"; +export const CHAT_MENTION_FANOUT_LIMIT = 25; + export interface ChatMentionNotificationInput { room: string; authorAgentId: string; @@ -9,24 +11,64 @@ export interface ChatMentionNotificationInput { mentionAgentIds: string[]; } +export interface PreparedChatMentionFanout { + targetMentionAgentIds: string[]; + roomPosterAgentIds: string[]; +} + +export type PrepareChatMentionFanoutResult = + | { ok: true; prepared: PreparedChatMentionFanout } + | { ok: false; error: string }; + +export interface PrepareChatMentionFanoutInput { + authorAgentId: string; + mentionAgentIds: string[]; + storedAgents: StoredAgentRecord[]; + liveAgents: ManagedAgent[]; + listRoomPosterAgentIds: () => Promise; + limit?: number; +} + export interface NotifyChatMentionsInput extends ChatMentionNotificationInput { logger: pino.Logger; - listStoredAgents: () => Promise; - listLiveAgents: () => ManagedAgent[]; + storedAgents: StoredAgentRecord[]; + liveAgents: ManagedAgent[]; + prepared: PreparedChatMentionFanout; resolveAgentIdentifier: ( identifier: string, ) => Promise<{ ok: true; agentId: string } | { ok: false; error: string }>; sendAgentMessage: (agentId: string, text: string) => Promise; } -export async function notifyChatMentions(input: NotifyChatMentionsInput): Promise { - const mentionAgentIds = await resolveChatMentionTargetAgentIds({ +export async function prepareChatMentionFanout( + input: PrepareChatMentionFanoutInput, +): Promise { + const mentionsEveryone = input.mentionAgentIds.includes("everyone"); + const roomPosterAgentIds = mentionsEveryone ? await input.listRoomPosterAgentIds() : []; + const targetMentionAgentIds = expandChatMentionTargets({ authorAgentId: input.authorAgentId, mentionAgentIds: input.mentionAgentIds, - storedAgents: await input.listStoredAgents(), - liveAgents: input.listLiveAgents(), + storedAgents: input.storedAgents, + liveAgents: input.liveAgents, + roomPosterAgentIds, }); - if (mentionAgentIds.length === 0) { + + if (mentionsEveryone) { + const limit = input.limit ?? CHAT_MENTION_FANOUT_LIMIT; + if (targetMentionAgentIds.length > limit) { + return { + ok: false, + error: `@everyone would notify ${targetMentionAgentIds.length} agents, which exceeds the limit of ${limit}. Narrow the room or mention specific agents.`, + }; + } + } + + return { ok: true, prepared: { targetMentionAgentIds, roomPosterAgentIds } }; +} + +export async function notifyChatMentions(input: NotifyChatMentionsInput): Promise { + const { targetMentionAgentIds } = input.prepared; + if (targetMentionAgentIds.length === 0) { return; } @@ -34,11 +76,11 @@ export async function notifyChatMentions(input: NotifyChatMentionsInput): Promis room: input.room, authorAgentId: input.authorAgentId, body: input.body, - mentionAgentIds, + mentionAgentIds: targetMentionAgentIds, }); await Promise.all( - mentionAgentIds.map(async (mentionedAgentId) => { + targetMentionAgentIds.map(async (mentionedAgentId) => { const resolved = await input.resolveAgentIdentifier(mentionedAgentId); if (!resolved.ok) { input.logger.warn( @@ -48,6 +90,19 @@ export async function notifyChatMentions(input: NotifyChatMentionsInput): Promis return; } + // Re-check eligibility on the resolved canonical id: explicit mentions may + // be custom titles that resolve to an archived or error-state agent. + if ( + !isChatMentionTargetEligible({ + agentId: resolved.agentId, + authorAgentId: input.authorAgentId, + storedAgents: input.storedAgents, + liveAgents: input.liveAgents, + }) + ) { + return; + } + try { await input.sendAgentMessage(resolved.agentId, notification); } catch (error) { @@ -60,45 +115,6 @@ export async function notifyChatMentions(input: NotifyChatMentionsInput): Promis ); } -export function resolveChatMentionTargetAgentIds(input: { - authorAgentId: string; - mentionAgentIds: string[]; - storedAgents: StoredAgentRecord[]; - liveAgents: ManagedAgent[]; -}): string[] { - const targets = new Set(); - const mentionsEveryone = input.mentionAgentIds.includes("everyone"); - - for (const mentionAgentId of input.mentionAgentIds) { - if (mentionAgentId === "everyone") { - continue; - } - if (mentionAgentId !== input.authorAgentId) { - targets.add(mentionAgentId); - } - } - - if (!mentionsEveryone) { - return Array.from(targets); - } - - for (const record of input.storedAgents) { - if (record.internal || record.archivedAt || record.id === input.authorAgentId) { - continue; - } - targets.add(record.id); - } - - for (const agent of input.liveAgents) { - if (agent.internal || agent.id === input.authorAgentId) { - continue; - } - targets.add(agent.id); - } - - return Array.from(targets); -} - export function buildChatMentionNotification(input: ChatMentionNotificationInput): string { const mentioned = input.mentionAgentIds.map((agentId) => `@${agentId}`).join(", "); const bodyWithoutMentions = input.body.replace(/(^|\s)@[A-Za-z0-9][A-Za-z0-9._-]*/g, "$1").trim(); @@ -112,3 +128,67 @@ export function buildChatMentionNotification(input: ChatMentionNotificationInput `Read the room with: paseo chat read ${input.room} --limit 20`, ].join("\n"); } + +function expandChatMentionTargets(input: { + authorAgentId: string; + mentionAgentIds: string[]; + storedAgents: StoredAgentRecord[]; + liveAgents: ManagedAgent[]; + roomPosterAgentIds: string[]; +}): string[] { + const candidates = new Set(); + const mentionsEveryone = input.mentionAgentIds.includes("everyone"); + + for (const mentionAgentId of input.mentionAgentIds) { + if (mentionAgentId === "everyone" || mentionAgentId === input.authorAgentId) { + continue; + } + candidates.add(mentionAgentId); + } + + if (mentionsEveryone) { + for (const posterAgentId of input.roomPosterAgentIds) { + if (posterAgentId !== input.authorAgentId) { + candidates.add(posterAgentId); + } + } + } + + const targets: string[] = []; + for (const candidate of candidates) { + if ( + isChatMentionTargetEligible({ + agentId: candidate, + authorAgentId: input.authorAgentId, + storedAgents: input.storedAgents, + liveAgents: input.liveAgents, + }) + ) { + targets.push(candidate); + } + } + return targets; +} + +function isChatMentionTargetEligible(input: { + agentId: string; + authorAgentId: string; + storedAgents: StoredAgentRecord[]; + liveAgents: ManagedAgent[]; +}): boolean { + if (input.agentId === input.authorAgentId) { + return false; + } + + const stored = input.storedAgents.find((record) => record.id === input.agentId); + if (stored?.internal || stored?.archivedAt || stored?.lastStatus === "error") { + return false; + } + + const live = input.liveAgents.find((agent) => agent.id === input.agentId); + if (live) { + return !live.internal && live.lifecycle !== "error"; + } + + return true; +} diff --git a/packages/server/src/server/chat/chat-service.test.ts b/packages/server/src/server/chat/chat-service.test.ts index d4702f064..0b9ed6a1f 100644 --- a/packages/server/src/server/chat/chat-service.test.ts +++ b/packages/server/src/server/chat/chat-service.test.ts @@ -94,6 +94,36 @@ describe("FileBackedChatService", () => { expect(detail.room.lastMessageAt).toBeTruthy(); }); + test("lists unique agents who have posted to a room", async () => { + const room = await service.createRoom({ name: "incident-room" }); + const otherRoom = await service.createRoom({ name: "other-room" }); + await sendChatMessage({ + room: room.name, + authorAgentId: "agent-a", + body: "first", + }); + await sendChatMessage({ + room: room.name, + authorAgentId: "agent-b", + body: "second", + }); + await sendChatMessage({ + room: room.name, + authorAgentId: "agent-a", + body: "third", + }); + await sendChatMessage({ + room: otherRoom.name, + authorAgentId: "unrelated-agent", + body: "different room", + }); + + await expect(service.listRoomPosterAgentIds({ room: room.name })).resolves.toEqual([ + "agent-a", + "agent-b", + ]); + }); + test("waits for new messages after a cursor and times out with an empty result", async () => { const room = await service.createRoom({ name: "loop-status" }); const first = await sendChatMessage({ diff --git a/packages/server/src/server/chat/chat-service.ts b/packages/server/src/server/chat/chat-service.ts index 622a174cf..e3554ec8a 100644 --- a/packages/server/src/server/chat/chat-service.ts +++ b/packages/server/src/server/chat/chat-service.ts @@ -89,6 +89,10 @@ export interface ReadChatMessagesInput { authorAgentId?: string; } +export interface ListChatRoomPosterAgentIdsInput { + room: string; +} + export interface WaitForChatMessagesInput { room: string; afterMessageId?: string | null; @@ -249,6 +253,16 @@ export class FileBackedChatService { return filtered.slice(filtered.length - limit); } + async listRoomPosterAgentIds(input: ListChatRoomPosterAgentIdsInput): Promise { + await this.load(); + const room = this.resolveRoom(input.room); + const posters = new Set(); + for (const message of this.getRoomMessages(room.id)) { + posters.add(message.authorAgentId); + } + return Array.from(posters); + } + async waitForMessages(input: WaitForChatMessagesInput): Promise { await this.load(); const room = this.resolveRoom(input.room); diff --git a/packages/server/src/server/schedule/service.ts b/packages/server/src/server/schedule/service.ts index d961ab91f..2326a73c5 100644 --- a/packages/server/src/server/schedule/service.ts +++ b/packages/server/src/server/schedule/service.ts @@ -3,9 +3,10 @@ import { join } from "node:path"; import type { Logger } from "pino"; import { AgentManager } from "../agent/agent-manager.js"; import type { AgentStorage } from "../agent/agent-storage.js"; -import type { AgentPromptInput, AgentSessionConfig } from "../agent/agent-sdk-types.js"; +import type { AgentSessionConfig } from "../agent/agent-sdk-types.js"; import { curateAgentActivity } from "../agent/activity-curator.js"; import { ensureAgentLoaded } from "../agent/agent-loading.js"; +import { formatSystemNotificationPrompt } from "../agent/agent-prompt.js"; import { getUnattendedModeId } from "../agent/provider-manifest.js"; import { ScheduleStore } from "./store.js"; import { computeNextRunAt, validateScheduleCadence } from "./cron.js"; @@ -29,6 +30,13 @@ function trimOptionalName(value: string | null | undefined): string | null { return trimmed.length > 0 ? trimmed : null; } +function buildScheduleFireBody(schedule: StoredSchedule, runId: string): string { + const heading = schedule.name + ? `Schedule "${schedule.name}" fired (id=${schedule.id}, run=${runId}).` + : `Schedule fired (id=${schedule.id}, run=${runId}).`; + return `${heading}\n${schedule.prompt}`; +} + function normalizePrompt(prompt: string): string { const trimmed = prompt.trim(); if (!trimmed) { @@ -126,17 +134,13 @@ function buildRunOutput(params: { return null; } -function buildAgentPrompt(text: string): AgentPromptInput { - return text; -} - export interface ScheduleServiceOptions { paseoHome: string; logger: Logger; agentManager: AgentManager; agentStorage: AgentStorage; now?: () => Date; - runner?: (schedule: StoredSchedule) => Promise; + runner?: (schedule: StoredSchedule, runId: string) => Promise; } export class ScheduleService { @@ -145,7 +149,10 @@ export class ScheduleService { private readonly agentManager: AgentManager; private readonly agentStorage: AgentStorage; private readonly now: () => Date; - private readonly runner: (schedule: StoredSchedule) => Promise; + private readonly runner: ( + schedule: StoredSchedule, + runId: string, + ) => Promise; private readonly runningScheduleIds = new Set(); private tickTimer: ReturnType | null = null; @@ -155,7 +162,7 @@ export class ScheduleService { this.agentManager = options.agentManager; this.agentStorage = options.agentStorage; this.now = options.now ?? (() => new Date()); - this.runner = options.runner ?? ((schedule) => this.executeSchedule(schedule)); + this.runner = options.runner ?? ((schedule, runId) => this.executeSchedule(schedule, runId)); } async start(): Promise { @@ -407,7 +414,7 @@ export class ScheduleService { await this.store.put(scheduleWithRun); try { - const result = await this.runner(scheduleWithRun); + const result = await this.runner(scheduleWithRun, runId); await this.finishRun({ scheduleId: schedule.id, runId, @@ -486,7 +493,12 @@ export class ScheduleService { await this.store.put(updated); } - private async executeSchedule(schedule: StoredSchedule): Promise { + private async executeSchedule( + schedule: StoredSchedule, + runId: string, + ): Promise { + const wrappedPrompt = formatSystemNotificationPrompt(buildScheduleFireBody(schedule, runId)); + if (schedule.target.type === "agent") { const record = await this.agentStorage.get(schedule.target.agentId); if (record?.archivedAt) { @@ -501,8 +513,7 @@ export class ScheduleService { if (this.agentManager.hasInFlightRun(agent.id)) { throw new Error(`Agent ${agent.id} already has an active run`); } - this.agentManager.recordUserMessage(agent.id, schedule.prompt, { emitState: false }); - const result = await this.agentManager.runAgent(agent.id, buildAgentPrompt(schedule.prompt)); + const result = await this.agentManager.runAgent(agent.id, wrappedPrompt); const timelineText = curateAgentActivity(result.timeline); return { agentId: agent.id, @@ -531,11 +542,10 @@ export class ScheduleService { }; const labels = { "paseo.schedule-id": schedule.id, - "paseo.schedule-run": randomUUID(), + "paseo.schedule-run": runId, }; const agent = await this.agentManager.createAgent(config, undefined, { labels }); - this.agentManager.recordUserMessage(agent.id, schedule.prompt, { emitState: false }); - const result = await this.agentManager.runAgent(agent.id, buildAgentPrompt(schedule.prompt)); + const result = await this.agentManager.runAgent(agent.id, wrappedPrompt); const timelineText = curateAgentActivity(result.timeline); return { agentId: agent.id, diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 0dedd2cff..d8db06f5c 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -64,7 +64,11 @@ import { toAgentPersistenceHandle, } from "./persistence-hooks.js"; import { ensureAgentLoaded } from "./agent/agent-loading.js"; -import { sendPromptToAgent, unarchiveAgentState } from "./agent/mcp-shared.js"; +import { + formatSystemNotificationPrompt, + sendPromptToAgent, + unarchiveAgentState, +} from "./agent/agent-prompt.js"; import { experimental_createMCPClient } from "ai"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import type { VoiceCallerContext, VoiceSpeakHandler } from "./voice-types.js"; @@ -192,8 +196,12 @@ import { toResolver, type Resolvable } from "./speech/provider-resolver.js"; import type { SpeechReadinessSnapshot, SpeechReadinessState } from "./speech/speech-runtime.js"; import type pino from "pino"; import { resolveClientMessageId } from "./client-message-id.js"; -import { ChatServiceError, FileBackedChatService } from "./chat/chat-service.js"; -import { notifyChatMentions } from "./chat/chat-mentions.js"; +import { + ChatServiceError, + FileBackedChatService, + parseMentionAgentIds, +} from "./chat/chat-service.js"; +import { notifyChatMentions, prepareChatMentionFanout } from "./chat/chat-mentions.js"; import { LoopService } from "./loop-service.js"; import { ScheduleService } from "./schedule/service.js"; import { execCommand } from "../utils/spawn.js"; @@ -8414,6 +8422,20 @@ export class Session { ): Promise { try { const authorAgentId = request.authorAgentId?.trim() || this.clientId; + const mentionAgentIds = parseMentionAgentIds(request.body); + const storedAgents = await this.agentStorage.list(); + const liveAgents = this.agentManager.listAgents(); + const fanout = await prepareChatMentionFanout({ + authorAgentId, + mentionAgentIds, + storedAgents, + liveAgents, + listRoomPosterAgentIds: () => + this.chatService.listRoomPosterAgentIds({ room: request.room }), + }); + if (!fanout.ok) { + throw new ChatServiceError("chat_mention_fanout_limit_exceeded", fanout.error); + } const message = await this.chatService.dispatchMessage({ room: request.room, authorAgentId, @@ -8434,11 +8456,20 @@ export class Session { body: request.body, mentionAgentIds: message.mentionAgentIds, logger: this.sessionLogger, - listStoredAgents: () => this.agentStorage.list(), - listLiveAgents: () => this.agentManager.listAgents(), + storedAgents, + liveAgents, + prepared: fanout.prepared, resolveAgentIdentifier: (identifier) => this.resolveAgentIdentifier(identifier), sendAgentMessage: async (agentId, text) => { - await this.handleSendAgentMessage(agentId, text); + await sendPromptToAgent({ + agentManager: this.agentManager, + agentStorage: this.agentStorage, + agentId, + prompt: formatSystemNotificationPrompt(text), + unarchive: false, + recordUserMessage: false, + logger: this.sessionLogger, + }); }, }); } catch (error) {