puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-07-29 12:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix chat fanout + unify system-injected agent prompts (#830)

Browse Source 
* Fix chat everyone fanout

* Unify system-injected agent prompts

Chat mentions, schedule fires, and notifyOnFinish now share one shape:
prompt wrapped in <paseo-system>...</paseo-system>, 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 \"<name>\" 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.
This commit is contained in:
Mohamed Boudra
2026-05-09 17:10:22 +08:00
committed by GitHub
parent 0229ef94f1
commit 34df55a10a
10 changed files with 608 additions and 308 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/server/src/server/agent/mcp-shared.test.ts → packages/server/src/server/agent/agent-prompt.test.ts
View File

@@ -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 () => {

243
packages/server/src/server/agent/agent-prompt.ts Normal file
View File

@@ -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<boolean> {
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 <paseo-system>…</paseo-system> 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 `<paseo-system>\n${reason}\n</paseo-system>`;
}
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<void> {
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");
}
}

4
packages/server/src/server/agent/mcp-server.ts
View File

@@ -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";

219
packages/server/src/server/agent/mcp-shared.ts
View File

@@ -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<boolean> {
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<void> {
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 = `<paseo-system>\nAgent ${childAgentId} (${title}) ${reason}.\n</paseo-system>`;
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 {

145
packages/server/src/server/chat/chat-mentions.test.ts
View File

@@ -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<StoredAgentRecord> & { 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<ManagedAgent> & { 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,
});

176
packages/server/src/server/chat/chat-mentions.ts
View File

@@ -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<string[]>;
limit?: number;
}
export interface NotifyChatMentionsInput extends ChatMentionNotificationInput {
logger: pino.Logger;
listStoredAgents: () => Promise<StoredAgentRecord[]>;
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<void>;
}
export async function notifyChatMentions(input: NotifyChatMentionsInput): Promise<void> {
const mentionAgentIds = await resolveChatMentionTargetAgentIds({
export async function prepareChatMentionFanout(
input: PrepareChatMentionFanoutInput,
): Promise<PrepareChatMentionFanoutResult> {
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<void> {
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<string>();
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<string>();
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;
}

30
packages/server/src/server/chat/chat-service.test.ts
View File

@@ -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({

14
packages/server/src/server/chat/chat-service.ts
View File

@@ -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<string[]> {
await this.load();
const room = this.resolveRoom(input.room);
const posters = new Set<string>();
for (const message of this.getRoomMessages(room.id)) {
posters.add(message.authorAgentId);
}
return Array.from(posters);
}
async waitForMessages(input: WaitForChatMessagesInput): Promise<ChatMessage[]> {
await this.load();
const room = this.resolveRoom(input.room);

40
packages/server/src/server/schedule/service.ts
View File

@@ -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<ScheduleExecutionResult>;
runner?: (schedule: StoredSchedule, runId: string) => Promise<ScheduleExecutionResult>;
}
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<ScheduleExecutionResult>;
private readonly runner: (
schedule: StoredSchedule,
runId: string,
) => Promise<ScheduleExecutionResult>;
private readonly runningScheduleIds = new Set<string>();
private tickTimer: ReturnType<typeof setInterval> | 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<void> {
@@ -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<ScheduleExecutionResult> {
private async executeSchedule(
schedule: StoredSchedule,
runId: string,
): Promise<ScheduleExecutionResult> {
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,

43
packages/server/src/server/session.ts
View File

@@ -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<void> {
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) {