From 193b3188c6c0a1af7076ea959bf0a877048a5163 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Mon, 2 Nov 2026 21:33:20 +0000 Subject: [PATCH] feat(agents): Let top-level agents message each other messageAgent now accepts the conversation id of another root Conversation. The sender and target must be in the same tenant and scope (public or private), and a private target needs the person behind the sender's Actor to be a member. The target runs the message as an Event Turn with the event system Actor, so no sender authority or credentials cross over. The automated-turn limit gates delivery so two agents cannot loop. Co-Authored-By: David Cramer <23610+dcramer@users.noreply.github.com> --- TERMINOLOGY.md | 7 +- .../src/chat/agent-invocations/AGENTS.md | 3 + .../src/chat/agent-invocations/errors.ts | 15 +- .../src/chat/agent-invocations/message.ts | 19 +- .../agent-invocations/top-level-message.ts | 193 ++++++++++++++++++ packages/junior/src/chat/agent/types.ts | 17 +- .../src/chat/tools/runtime/message-agent.ts | 16 +- .../agent-invocation-message.test.ts | 187 ++++++++++++++++- 8 files changed, 440 insertions(+), 17 deletions(-) create mode 100644 packages/junior/src/chat/agent-invocations/top-level-message.ts diff --git a/TERMINOLOGY.md b/TERMINOLOGY.md index b0b6a16738..26910af937 100644 --- a/TERMINOLOGY.md +++ b/TERMINOLOGY.md @@ -82,9 +82,10 @@ Canonical words for Junior code and docs. One concept, one word. - **Agent binding**: a parent-scoped name for one child Conversation. - **Agent invocation**: one retry-safe task from a parent to a child, with its terminal result. Never shorten to "invocation". -- **Agent message**: `messageAgent` text to a parent or child. To a child it - starts an Agent invocation or steers a running one. To a parent it is an - Event Inbound message. It gives no user approval. +- **Agent message**: `messageAgent` text to a parent, a child, or another + top-level agent (the agent of a root Conversation). To a child it starts an + Agent invocation or steers a running one. To a parent or a top-level agent + it is an Event Inbound message. It gives no user approval. - **Agent type**: a `spawnAgent` preset: when to pick it, whether it forks history, and per-task instructions. It does not limit tools. - **Tool exposure**: who may call a tool: the model, the tool catalog diff --git a/packages/junior/src/chat/agent-invocations/AGENTS.md b/packages/junior/src/chat/agent-invocations/AGENTS.md index 141ed80622..e89f943dd1 100644 --- a/packages/junior/src/chat/agent-invocations/AGENTS.md +++ b/packages/junior/src/chat/agent-invocations/AGENTS.md @@ -11,3 +11,6 @@ Turn Source (a child watch-event Turn has an event Source). - An Agent type (`agent-types.ts`) is data. Read its fields, never its name. Types do not limit tools. +- A top-level agent message (`top-level-message.ts`) runs as the event system + Actor. It never carries the sender's Actor or credentials. Check its access + on each send: same tenant, same scope, and a member for a private target. diff --git a/packages/junior/src/chat/agent-invocations/errors.ts b/packages/junior/src/chat/agent-invocations/errors.ts index 7a6cad87d3..dc31d1903b 100644 --- a/packages/junior/src/chat/agent-invocations/errors.ts +++ b/packages/junior/src/chat/agent-invocations/errors.ts @@ -24,10 +24,23 @@ export class AgentMessageQueueFullError extends Error { } } -/** A messageAgent target is not the parent or a child of the sender. */ +/** + * A messageAgent target is not the parent, a child, or a top-level agent that + * the sender can message. + */ export class AgentMessageTargetError extends Error { constructor(message: string) { super(message); this.name = "AgentMessageTargetError"; } } + +/** A top-level agent paused automated Turns, so it takes no agent message. */ +export class AgentMessagePausedError extends Error { + constructor(conversationId: string) { + super( + `${conversationId} paused automated work until a person replies there`, + ); + this.name = "AgentMessagePausedError"; + } +} diff --git a/packages/junior/src/chat/agent-invocations/message.ts b/packages/junior/src/chat/agent-invocations/message.ts index 2176fa62b8..987f137731 100644 --- a/packages/junior/src/chat/agent-invocations/message.ts +++ b/packages/junior/src/chat/agent-invocations/message.ts @@ -14,6 +14,7 @@ import { createAgentInvocationForMessage, getActiveAgentInvocationForConversation, } from "./store"; +import { enqueueTopLevelAgentMessage } from "./top-level-message"; import { PARENT_AGENT_TARGET } from "./types"; import { enqueueAgentInvocation } from "./work"; @@ -25,8 +26,9 @@ type MessageOptions = { /** * Bind a Run's Actor and credentials to messages for its parent and child - * agents. A message to a child continues that child. A message to the parent - * wakes the parent like a result. + * agents, and for other top-level agents. A message to a child continues that + * child. A message to the parent wakes the parent like a result. A top-level + * agent can also message another top-level agent that it has access to. */ export function bindMessageAgent( run: AgentRun, @@ -69,9 +71,16 @@ export function bindMessageAgent( target: input.to, }); if (!childConversationId) { - throw new AgentMessageTargetError( - `"${input.to}" is not "${PARENT_AGENT_TARGET}" or a child agent of this conversation`, - ); + await enqueueTopLevelAgentMessage({ + actor, + messageKey: `${run.turnId}:${call.toolCallId}`, + queue, + senderConversationId: run.conversationId, + state: options.state, + text: input.text, + to: input.to, + }); + return { conversationId: input.to, delivery: "queued" }; } const { invocation, queued } = await createAgentInvocationForMessage({ actor, diff --git a/packages/junior/src/chat/agent-invocations/top-level-message.ts b/packages/junior/src/chat/agent-invocations/top-level-message.ts new file mode 100644 index 0000000000..9323a64f1a --- /dev/null +++ b/packages/junior/src/chat/agent-invocations/top-level-message.ts @@ -0,0 +1,193 @@ +/** + * Agent messages between top-level agents: root Conversations that are not + * in the same agent tree. A message wakes the target like any other Event. + * The target runs it as the event system Actor, so the sender's Actor and + * credentials never reach the target. + */ +import type { StateAdapter } from "chat"; +import type { Actor } from "@/chat/actor"; +import { readConversationAccessFromSql } from "@/api/conversations/access"; +import { botConfig } from "@/chat/config"; +import { getConversationStore, getDb } from "@/chat/db"; +import { + EVENT_WAIT_MS, + type EventMailboxMetadata, +} from "@/chat/events/notification"; +import { SYSTEM_IDENTITY_ID } from "@/chat/identities/actor"; +import { readActorIdentity } from "@/chat/plugins/viewer"; +import { admitAutomatedTurn } from "@/chat/services/automated-turn-limit"; +import { postAutomatedTurnLimitNoticeForConversation } from "@/chat/slack/automated-turn-limit-notice"; +import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; +import { appendAndEnqueueInboundMessage } from "@/chat/task-execution/store"; +import { AgentMessagePausedError, AgentMessageTargetError } from "./errors"; +import { PARENT_AGENT_TARGET } from "./types"; + +const AGENT_MESSAGE_EVENT_NAMESPACE = "junior"; +/** Event type of a message from another top-level agent. */ +const TOP_LEVEL_AGENT_MESSAGE_EVENT_TYPE = "agent.top_level_message"; + +type AgentScope = "private" | "public"; + +/** + * Check that a top-level agent can message another top-level agent. Both + * must be in the same provider tenant and have the same scope: public or + * private. To reach a private agent, the person behind the sender's Actor + * must be a member of the target Conversation. A system Actor has no + * person, so it reaches only public agents. Every target that the Actor + * cannot see gets the same error, so the error does not show that a private + * Conversation exists. + */ +async function requireTopLevelAgentAccess(args: { + actor: Actor; + senderConversationId: string; + to: string; +}): Promise { + const notReachable = new AgentMessageTargetError( + `"${args.to}" is not "${PARENT_AGENT_TARGET}", a child agent of this conversation, or a top-level agent that you can message`, + ); + if (args.to === args.senderConversationId) { + throw new AgentMessageTargetError(`"${args.to}" is this conversation`); + } + const conversationStore = getConversationStore(); + const [senderConversation, target] = await Promise.all([ + conversationStore.get({ conversationId: args.senderConversationId }), + conversationStore.get({ conversationId: args.to }), + ]); + if (senderConversation?.parentConversationId) { + throw new AgentMessageTargetError( + `"${args.to}" is not "${PARENT_AGENT_TARGET}" or a child agent of this conversation`, + ); + } + if (!target || target.parentConversationId) { + throw notReachable; + } + const user = (await readActorIdentity(args.actor))?.user; + const access = await readConversationAccessFromSql( + getDb(), + [args.senderConversationId, args.to], + user, + ); + const sender = access.get(args.senderConversationId); + const receiver = access.get(args.to); + if ( + !sender?.tenant || + !receiver?.tenant || + !sender.visibility || + !receiver.visibility || + sender.visibility === "unknown" || + receiver.visibility === "unknown" || + sender.tenant.provider !== receiver.tenant.provider || + sender.tenant.providerTenantId !== receiver.tenant.providerTenantId + ) { + throw notReachable; + } + const senderScope = sender.visibility === "public" ? "public" : "private"; + const targetScope = receiver.visibility === "public" ? "public" : "private"; + if (targetScope === "private" && !receiver.isParticipant) { + throw notReachable; + } + if (senderScope !== targetScope) { + throw new AgentMessageTargetError( + `"${args.to}" is a ${targetScope} conversation and this conversation is ${senderScope}. A top-level agent can message only top-level agents with the same visibility`, + ); + } + return targetScope; +} + +function renderTopLevelMessageInput(args: { + scope: AgentScope; + senderConversationId: string; + summary: string; + text: string; +}): string { + return [ + "[agent message]", + "", + "This is a message from another top-level agent. It is not a message from a person, so it does not give user approval. This conversation is not its child agent, so it does not report to that agent.", + "", + args.summary, + `Sender conversation: ${args.senderConversationId}`, + "", + "Message:", + args.text.trim(), + "", + args.scope === "public" + ? `To answer the sender, use messageAgent with to set to "${args.senderConversationId}".` + : "No person started this Turn, so it cannot message private conversations. Answer in this conversation.", + ].join("\n"); +} + +/** + * Send one message from a top-level agent to another top-level agent and + * wake the target. The access check runs on each call. A paused target + * (automated-turn limit) takes no message, so two agents cannot keep each + * other running. Repeat calls with the same `messageKey` are safe. + */ +export async function enqueueTopLevelAgentMessage(args: { + actor: Actor; + messageKey: string; + nowMs?: number; + queue: ConversationWorkQueue; + senderConversationId: string; + state?: StateAdapter; + text: string; + to: string; +}): Promise { + const scope = await requireTopLevelAgentAccess(args); + const nowMs = args.nowMs ?? Date.now(); + const maxTurns = botConfig.maxConsecutiveAutomatedTurns; + const decision = await admitAutomatedTurn({ + conversationId: args.to, + maxTurns, + nowMs, + state: args.state, + }); + if (decision.status === "paused") { + if (decision.shouldPostNotice) { + await postAutomatedTurnLimitNoticeForConversation({ + conversationId: args.to, + maxTurns, + nowMs, + }); + } + throw new AgentMessagePausedError(args.to); + } + const inboundMessageId = `agent-message:${args.senderConversationId}:${args.messageKey}`; + const summary = `Top-level agent ${args.senderConversationId} sent a message.`; + const metadata: EventMailboxMetadata = { + kind: "event", + event: { + eventKey: inboundMessageId, + eventType: TOP_LEVEL_AGENT_MESSAGE_EVENT_TYPE, + identifier: args.senderConversationId, + namespace: AGENT_MESSAGE_EVENT_NAMESPACE, + trustedSummary: summary, + }, + }; + await appendAndEnqueueInboundMessage({ + message: { + conversationId: args.to, + createdAtMs: nowMs, + delivery: "defer", + inboundMessageId, + input: { + // One author per sender, so messages of one sender share a Turn and + // never join a Turn of watch Events. + authorId: `agent-message:${args.senderConversationId}`, + authorIdentityId: SYSTEM_IDENTITY_ID, + metadata, + text: renderTopLevelMessageInput({ + scope, + senderConversationId: args.senderConversationId, + summary, + text: args.text, + }), + }, + receivedAtMs: nowMs, + source: "event", + }, + queue: args.queue, + queueDelayMs: EVENT_WAIT_MS, + state: args.state, + }); +} diff --git a/packages/junior/src/chat/agent/types.ts b/packages/junior/src/chat/agent/types.ts index ee4aae3bd4..6de08198f6 100644 --- a/packages/junior/src/chat/agent/types.ts +++ b/packages/junior/src/chat/agent/types.ts @@ -90,9 +90,15 @@ export type SpawnAgent = ( options: { signal?: AbortSignal; toolCallId: string }, ) => Promise; -/** Model-safe input for one message to the parent or a child agent. */ +/** + * Model-safe input for one message to the parent, a child agent, or another + * top-level agent. + */ export type MessageAgentInput = { - /** `parent`, or the name or Conversation id of a child agent. */ + /** + * `parent`, the name or Conversation id of a child agent, or the + * Conversation id of another top-level agent. + */ to: string; text: string; }; @@ -107,7 +113,10 @@ export type MessageAgentResult = { delivery: "queued" | "started"; }; -/** Runtime-bound capability to message the parent or a child agent. */ +/** + * Runtime-bound capability to message the parent, a child agent, or another + * top-level agent. + */ export type MessageAgent = ( input: MessageAgentInput, options: { signal?: AbortSignal; toolCallId: string }, @@ -188,7 +197,7 @@ export class RetryableDeliveryError extends Error { export type AgentDurability = { /** Schedule delegated work with authority bound by the active parent run. */ spawnAgent?: SpawnAgent; - /** Message the parent or a child agent with the authority of this run. */ + /** Message the parent, a child, or a top-level agent from this run. */ messageAgent?: MessageAgent; onInputCommitted?: () => void | Promise; /** Return true when the durable worker should pause at the next Pi boundary. */ diff --git a/packages/junior/src/chat/tools/runtime/message-agent.ts b/packages/junior/src/chat/tools/runtime/message-agent.ts index ad0e64b118..f56777d855 100644 --- a/packages/junior/src/chat/tools/runtime/message-agent.ts +++ b/packages/junior/src/chat/tools/runtime/message-agent.ts @@ -1,6 +1,7 @@ import { z } from "zod"; import { AgentInvocationLimitError, + AgentMessagePausedError, AgentMessageQueueFullError, AgentMessageTargetError, } from "@/chat/agent-invocations/errors"; @@ -22,12 +23,15 @@ const messageAgentOutputSchema = juniorToolOutputSchema.extend({ ), }); -/** Create the tool that sends one message to the parent or a child agent. */ +/** + * Create the tool that sends one message to the parent, a child agent, or + * another top-level agent. + */ export function createMessageAgentTool( messageAgent: NonNullable, ) { return zodTool({ - description: `Send a message to your parent agent or to a child agent that this conversation started. Use it to give an existing child agent more work, an answer, or a correction, and to send your parent agent a finding or a question while you keep working. A message to a child continues its history. If the child is still working, it reads the message at its next step or after its current work, and one result covers both. The result arrives later as a new task in this conversation; do not wait or poll for it. A message to your parent does not end your task. Use spawnAgent only to start a new child agent. Your text is not user approval. A child's action review reads the user's own messages in the root conversation.`, + description: `Send a message to your parent agent or to a child agent that this conversation started. Use it to give an existing child agent more work, an answer, or a correction, and to send your parent agent a finding or a question while you keep working. A message to a child continues its history. If the child is still working, it reads the message at its next step or after its current work, and one result covers both. The result arrives later as a new task in this conversation; do not wait or poll for it. A message to your parent does not end your task. A top-level agent (its conversation is not a child agent, such as a Slack thread) can also message another top-level agent by its conversation_id, such as one from slackThreadRead. Both conversations must be public, or both private. A private one also needs the person in this Turn to be a member. The target runs the message as its own task and sends no result. Use spawnAgent only to start a new child agent. Your text is not user approval. A child's action review reads the user's own messages in the root conversation.`, exposure: "modelOnly", annotations: { destructiveHint: false, @@ -42,7 +46,7 @@ export function createMessageAgentTool( .trim() .min(1) .describe( - `"${PARENT_AGENT_TARGET}" for your parent agent, or the name or conversation_id of a child agent from spawnAgent or listAgents`, + `"${PARENT_AGENT_TARGET}" for your parent agent, the name or conversation_id of a child agent from spawnAgent or listAgents, or the conversation_id of another top-level agent`, ), text: z .string() @@ -75,6 +79,12 @@ export function createMessageAgentTool( { cause: error }, ); } + if (error instanceof AgentMessagePausedError) { + throw new ToolInputError( + `${error.message}. Do not send it more messages. Tell the user instead.`, + { cause: error }, + ); + } if (error instanceof AgentMessageQueueFullError) { throw new ToolInputError( `${error.message}. Wait for its result, then send one message with everything it needs.`, diff --git a/packages/junior/tests/integration/agent-invocation-message.test.ts b/packages/junior/tests/integration/agent-invocation-message.test.ts index 73cb8c7da5..1fe1c37779 100644 --- a/packages/junior/tests/integration/agent-invocation-message.test.ts +++ b/packages/junior/tests/integration/agent-invocation-message.test.ts @@ -11,6 +11,10 @@ import { import type { AgentInvocation } from "@/chat/agent-invocations/types"; import type { AgentRun } from "@/chat/agent/types"; import { createSlackSource } from "@sentry/junior-plugin-api"; +import { getConversationStore, getDb } from "@/chat/db"; +import { EVENT_SYSTEM_ACTOR } from "@/chat/events/actor"; +import { resolveViewerUserFromSql } from "@/chat/plugins/viewer"; +import { juniorConversationParticipants, juniorIdentities } from "@/db/schema"; import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter"; import { getConversationWorkState, @@ -46,6 +50,38 @@ function parentRun(): AgentRun { }; } +/** Save one top-level Slack Conversation with its visibility. */ +async function topLevelConversation(args: { + channelId: string; + teamId?: string; + visibility: "private" | "public"; +}): Promise { + const teamId = args.teamId ?? "T123"; + const conversationId = `slack:${args.channelId}:1712345.0001`; + await getConversationStore().recordActivity({ + conversationId, + destination: { channelId: args.channelId, platform: "slack", teamId }, + nowMs: 1_000, + sessionSource: createSlackSource({ + channelId: args.channelId, + teamId, + threadTs: "1712345.0001", + visibility: args.visibility, + }), + source: "slack", + visibility: args.visibility, + }); + return conversationId; +} + +/** A Run of one top-level Conversation. */ +function topLevelRun( + conversationId: string, + actor: AgentRun["actor"] = parentActor, +): AgentRun { + return { ...parentRun(), actor, conversationId, turnId: "top-level-turn" }; +} + /** The Run of one child agent invocation, as the invocation worker builds it. */ function childRun(invocation: AgentInvocation): AgentRun { return { @@ -144,7 +180,7 @@ describe("messageAgent", () => { ), ).resolves.toMatchObject({ modelProfile: "handoff" }); await expect(sendFromParent("nobody", "call-unknown")).rejects.toThrow( - /"nobody" is not "parent" or a child agent/, + /"nobody" is not "parent", a child agent/, ); // Child to parent: the message wakes the parent as the child's Actor. @@ -324,4 +360,153 @@ describe("messageAgent", () => { await fixture.close(); } }); + + it("lets top-level agents message each other only with the same visibility", async () => { + const { fixture } = await prepareSlackParentConversation(); + const runner = readRunsAgentRunner(); + const harness = await createConversationWorkHarness(runner.agentRunner); + const { queue, state } = harness; + try { + const publicTarget = await topLevelConversation({ + channelId: "CPUB", + visibility: "public", + }); + const otherWorkspace = await topLevelConversation({ + channelId: "CFAR", + teamId: "T999", + visibility: "public", + }); + const privateSender = await topLevelConversation({ + channelId: "CPRIV1", + visibility: "private", + }); + const privateMember = await topLevelConversation({ + channelId: "CPRIV2", + visibility: "private", + }); + const privateOther = await topLevelConversation({ + channelId: "CPRIV3", + visibility: "private", + }); + // The person behind the Slack Actor is a member of one private target. + const ana = await resolveViewerUserFromSql(getDb(), "ana@example.com"); + await getDb() + .insert(juniorIdentities) + .values({ + createdAt: new Date(1), + id: "identity:T123:U123", + kind: "user", + provider: "slack", + providerSubjectId: "U123", + providerTenantId: "T123", + updatedAt: new Date(1), + userId: ana!.id, + }) + .onConflictDoUpdate({ + target: juniorIdentities.id, + set: { userId: ana!.id }, + }); + await getDb() + .insert(juniorConversationParticipants) + .values({ + lastMessageAt: new Date(1), + rootConversationId: privateMember, + userId: ana!.id, + }); + const send = async ( + run: AgentRun, + to: string, + toolCallId = "call-top-level", + ) => + await createMessageAgentTool(bindMessageAgent(run, { queue, state })!) + .execute!( + { text: "Can you take the deploy check?", to }, + { + toolCallId, + }, + ); + + // Public to public, in the same Slack workspace. A replay adds no + // second message. + const fromPublic = topLevelRun(slackParentConversationId); + await expect(send(fromPublic, publicTarget)).resolves.toEqual({ + conversation_id: publicTarget, + delivery: "queued", + }); + await send(fromPublic, publicTarget); + const messages = await eventMessages(publicTarget); + expect(messages).toHaveLength(1); + expect(messages[0]).toMatchObject({ + input: { + metadata: { + event: { + eventType: "agent.top_level_message", + identifier: slackParentConversationId, + namespace: "junior", + }, + }, + }, + }); + expect(messages[0]!.input.text).toContain( + "It is not a message from a person", + ); + expect(messages[0]!.input.text).toContain( + `to set to "${slackParentConversationId}"`, + ); + + // Not the same visibility, another workspace, or the sender itself. + await expect(send(fromPublic, privateMember)).rejects.toThrow( + "is a private conversation and this conversation is public", + ); + await expect(send(fromPublic, otherWorkspace)).rejects.toThrow( + "a top-level agent that you can message", + ); + await expect(send(fromPublic, slackParentConversationId)).rejects.toThrow( + "is this conversation", + ); + + // Private to private needs a member. A private target that the person + // is not in looks like a missing one. + const fromPrivate = topLevelRun(privateSender); + await expect(send(fromPrivate, privateMember)).resolves.toEqual({ + conversation_id: privateMember, + delivery: "queued", + }); + expect((await eventMessages(privateMember))[0]!.input.text).toContain( + "cannot message private conversations", + ); + await expect(send(fromPrivate, privateOther)).rejects.toThrow( + "a top-level agent that you can message", + ); + await expect(send(fromPrivate, publicTarget)).rejects.toThrow( + "is a public conversation and this conversation is private", + ); + // An event Turn has no person, so it reaches no private agent. + await expect( + send(topLevelRun(privateSender, EVENT_SYSTEM_ACTOR), privateMember), + ).rejects.toThrow("a top-level agent that you can message"); + + // A child agent reaches only its own tree. + const child = await createAgentInvocation({ + ...slackInvocationInput, + idempotencyKey: "top-level-child", + }); + await expect( + send(childRun(child), publicTarget, "call-from-child"), + ).rejects.toThrow(`"${publicTarget}" is not "parent" or a child agent`); + + // The target runs the message as an Event, not as the sender's Actor. + vi.spyOn(Date, "now").mockReturnValue(Date.now() + 60_000); + await harness.process(queue.takeMessageFor(publicTarget)); + vi.mocked(Date.now).mockRestore(); + expect(runner.runs).toEqual([ + expect.objectContaining({ + actor: EVENT_SYSTEM_ACTOR, + conversationId: publicTarget, + }), + ]); + } finally { + await fixture.close(); + } + }); });