From 0193f9ebffced2d67c16c5b0a2182d8af997bd4e Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Sat, 17 Oct 2026 01:47:15 +0000 Subject: [PATCH] fix(agents): Record whose work a child agent does in its session log A watch, timer, or result Turn of a child agent runs as the event system Actor. Action review used that Run Actor to read the root Conversation, so the root user's own requests were labeled as another person, and routine steps on the child's own work were blocked. spawnAgent and messageAgent now put the calling Turn's person in the meta of the task message that they write into the child, as onBehalfOf. Each child Turn saves it in the provenance of its instruction, and a Turn without a stamped task keeps the value of the child's latest instruction. Action review reads the root Conversation as onBehalfOf. The Run Actor, credentials, and child user intent do not change, and agent text still has no author Actor. Nested results and messages now take the Actor and credentials of the child's latest task, not of its newest agent invocation, so the grants of a timer Turn never pass to later work. Supersedes #2175. Co-Authored-By: David Cramer --- .../guardian/child-agent-approval.eval.ts | 105 ++++---- .../src/chat/agent-invocations/README.md | 48 +++- .../src/chat/agent-invocations/child-input.ts | 104 ++++++++ .../src/chat/agent-invocations/markers.ts | 6 +- .../src/chat/agent-invocations/message.ts | 7 +- .../chat/agent-invocations/on-behalf-of.ts | 77 ++++++ .../src/chat/agent-invocations/result.ts | 6 +- .../agent-invocations/root-conversation.ts | 25 +- .../src/chat/agent-invocations/spawn.ts | 3 +- .../src/chat/agent-invocations/store.ts | 51 ---- .../junior/src/chat/agent-invocations/work.ts | 35 ++- packages/junior/src/chat/agent/index.ts | 7 +- packages/junior/src/chat/agent/tools.ts | 6 +- packages/junior/src/chat/agent/types.ts | 6 + .../junior/src/chat/conversations/README.md | 3 + .../src/chat/conversations/provenance.ts | 16 +- .../component/agent-invocation-worker.test.ts | 44 +++- .../agent-invocation-concurrency.test.ts | 232 +++++++++++++++++- .../agent-invocation-nested.test.ts | 52 ++++ .../agent-invocation-user-approval.test.ts | 12 +- 20 files changed, 688 insertions(+), 157 deletions(-) create mode 100644 packages/junior/src/chat/agent-invocations/child-input.ts create mode 100644 packages/junior/src/chat/agent-invocations/on-behalf-of.ts diff --git a/packages/junior-evals/evals/guardian/child-agent-approval.eval.ts b/packages/junior-evals/evals/guardian/child-agent-approval.eval.ts index 1f2ad1ae4d..1c8ef41707 100644 --- a/packages/junior-evals/evals/guardian/child-agent-approval.eval.ts +++ b/packages/junior-evals/evals/guardian/child-agent-approval.eval.ts @@ -3,9 +3,11 @@ * * A child agent's task comes from its parent agent, so the task text is not * user intent. Core gives Guardian a fixed child `userIntent` and reads the - * root Conversation at review time. Only `user:` lines from the child's own - * Actor, or the `automation:` instruction of an Automation run, can approve. - * They approve what they name and the conventional steps of that work. + * root Conversation at review time. Only `user:` lines from the person whose + * work the child does, or the `automation:` instruction of an Automation run, + * can approve. A watch Turn of the child runs as the event Actor, but still + * reads the root for that person. They approve what they name and the + * conventional steps of that work. */ import { describeEval } from "vitest-evals"; import { @@ -533,48 +535,65 @@ describeEval("Guardian Child Agent Approval Snapshots", guardianEvals, (it) => { }); } - it("when a Watch on the child's own PR delivers review feedback, allow marking it addressed", async ({ - run, - }) => { - await run({ - expectedDecision: "allow", - proposal: childProposal({ - context: childWatchContext({ - eventType: "pull_request.review_comment.created", - identifier: `${repo}#${prNumber}`, - namespace: "github", + // A Watch Turn of the child runs as the event Actor. Core reads the root + // for the person whose work the child does, so the user's request is still + // a `user:` line. + for (const { name, input, tool } of [ + { + name: "when a Watch on the child's own PR delivers review feedback, allow marking it addressed", + input: feedbackInput, + tool: { + ...updateFeedbackTool, + proposalDescription: `Mark review comment ${reviewCommentId} on ${repo} as addressed (+1).`, + }, + }, + { + name: "when a Watch on the child's own PR delivers review feedback, allow resolving the handled thread", + input: { repo, threadId: reviewThreadId }, + tool: { + ...resolveThreadTool, + proposalDescription: `Resolve review thread ${reviewThreadId} on ${repo}.`, + }, + }, + ]) { + it(name, async ({ run }) => { + await run({ + expectedDecision: "allow", + proposal: childProposal({ + context: childWatchContext({ + eventType: "pull_request.review_comment.created", + identifier: `${repo}#${prNumber}`, + namespace: "github", + }), + entries: [ + { role: "parent agent", text: parentTask(fixRequest) }, + pullRequestCreated, + { + role: "parent agent", + text: renderTaskInput({ + about: `${repo}#${prNumber}`, + guidance: + 'After acting on actionable feedback, call github_updatePullRequestFeedback with the exact repo, commentId, and commentKind from Verified details, and status "addressed" or "declined". Then resolve the thread with github_resolvePullRequestReviewThread.', + instructions: + "Address review feedback on the pull request that I opened.", + trustedSummary: `New review comment on ${repo}#${prNumber}.`, + verifiedDetails: { + commentId: reviewCommentId, + commentKind: "review", + repo, + }, + externalText: reviewComment, + }), + }, + ...fixPushed, + ], + input, + root: fixRoot, + tool, }), - entries: [ - { role: "parent agent", text: parentTask(fixRequest) }, - pullRequestCreated, - { - role: "parent agent", - text: renderTaskInput({ - about: `${repo}#${prNumber}`, - guidance: - 'After acting on actionable feedback, call github_updatePullRequestFeedback with the exact repo, commentId, and commentKind from Verified details, and status "addressed" or "declined". Then resolve the thread with github_resolvePullRequestReviewThread.', - instructions: - "Address review feedback on the pull request that I opened.", - trustedSummary: `New review comment on ${repo}#${prNumber}.`, - verifiedDetails: { - commentId: reviewCommentId, - commentKind: "review", - repo, - }, - externalText: reviewComment, - }), - }, - ...fixPushed, - ], - input: feedbackInput, - root: fixRoot, - tool: { - ...updateFeedbackTool, - proposalDescription: `Mark review comment ${reviewCommentId} on ${repo} as addressed (+1).`, - }, - }), + }); }); - }); + } it("when the assistant said each agent would open a PR for the requested fix, allow the child's PR", async ({ run, diff --git a/packages/junior/src/chat/agent-invocations/README.md b/packages/junior/src/chat/agent-invocations/README.md index 75a870cbfd..dbede59c39 100644 --- a/packages/junior/src/chat/agent-invocations/README.md +++ b/packages/junior/src/chat/agent-invocations/README.md @@ -151,8 +151,10 @@ valid targets. agent invocation ends as `completed` with `parentMailboxStatus: "steered"`. The running agent invocation reports for both. A message joins only with the Actor and credentials of the running Turn. Otherwise it and every later - parent message wait. If the child finishes without another model call, the - message runs after that work, and the earlier work ends as `replaced`. + parent message wait. It also joins only for the same `onBehalfOf` (see + User Approval). A message without one keeps the running one. If the child + finishes without another model call, the message runs after that work, and + the earlier work ends as `replaced`. - **Child to parent.** The message wakes the parent as an Event with namespace `junior` and type `agent.message`. It uses the same mailbox path and Actor rules as a child result: the Actor comes from the sender's agent @@ -171,9 +173,11 @@ A child Conversation has no Location, so it cannot run an Event Turn. When results starts a new agent invocation on that child instead. Its id comes from the result's agent invocation id, so retries do not start a second one. An agent message from one of those children starts an agent invocation the same -way; its id comes from the message's Inbound message id. It uses the parent, -Actor, and credentials of the child's latest agent invocation, and it skips the -queue and active-count checks. +way; its id comes from the message's Inbound message id. It uses the parent of +the child's latest agent invocation, and the Actor and credentials of the +child's latest task from `spawnAgent` or `messageAgent`. A watch event is not a +task, so the grants of a timer never pass to later work. It skips the queue and +active-count checks. The parent gets one report for the child's work. A finished agent invocation is `replaced` instead of sent when the child still has open work: another open @@ -237,9 +241,9 @@ When Guardian reviews an action of a child, `root-conversation.ts` walks rewrite these events. Nothing is copied into the child, and nothing new is stored. -- Only a root message whose author has the same Actor identity as the child - is labeled `user`. It is compared by platform, team, and user id. When - either Actor is not known, the message does not count. +- Only a root message whose author has the same Actor identity as the + child's `onBehalfOf` is labeled `user`. It is compared by platform, team, + and user id. When either Actor is not known, the message does not count. - When the root is an Automation run, the dispatch input message is that run's own instruction. It is labeled `automation` for the run's system Actor. This is the authority that the run already has. Guardian reviews the @@ -266,6 +270,34 @@ stored. its text. An edit of a message older than the read is left out, so an edit never moves a message after a newer one, such as a revocation. +### Whose Work a Child Does + +`onBehalfOf` is the Actor whose work a child does. It is not the Run Actor. A +watch, timer, or result Turn of a child runs as the event system Actor, but it +still does the work of the person who gave the task. `on-behalf-of.ts` owns +this rule. The value is in the child's session log, so no SQL column holds it: + +- `spawnAgent` and `messageAgent` put it in the `meta` of the task message + that they write into the child. A root Turn gives its Actor. A child Turn + gives its own `onBehalfOf`. An Event Turn on a root is not from a person, + so its task has none. +- Each child Turn uses the value of its own task message. Without one, it + keeps the value of the child's latest instruction. So a watch Turn, or a + message from a root Event Turn, keeps the person of the earlier task. A + child that a root Event Turn starts has none. +- The Turn saves the value in the `onBehalfOf` field of its instruction + provenance. The instruction still has no `actor`, so agent text never + becomes user intent. The latest instruction is read across history + versions, so compaction and handoff do not lose it. +- The model and tool input cannot set it. It never changes the Run Actor, + credentials, or a timer's own grants. +- A child that started before this field existed has no value. It gets no + `user` lines until a new task gives it one. There is no fallback, such as + the Run Actor or the thread owner. +- A heartbeat repair of a task message cannot know the value. The child then + keeps the value of its latest instruction, the same as for a message from a + root Event Turn. + Guardian renders the projection outside the untrusted proposal JSON, in a `` section with a new random key for each request. Each line has its real author as a prefix. The rules for a child diff --git a/packages/junior/src/chat/agent-invocations/child-input.ts b/packages/junior/src/chat/agent-invocations/child-input.ts new file mode 100644 index 0000000000..3e699f250e --- /dev/null +++ b/packages/junior/src/chat/agent-invocations/child-input.ts @@ -0,0 +1,104 @@ +/** + * Creates the agent invocation that gives one input to a child Conversation: + * a result or message from its own child agent, or its own watch events. + * + * The parent did not start this work, so it is not a task. It runs with the + * Actor and credentials of the child's latest task. A watch event or a result + * is never a task, so the credentials of one timer Turn never pass to later + * work. + */ +import { and, asc, desc, eq, isNotNull } from "drizzle-orm"; +import { getSqlExecutor } from "@/chat/db"; +import { juniorAgentInvocations } from "@/db/schema"; +import { getAgentInvocation, invocationFromRow } from "./store"; +import type { AgentInvocation } from "./types"; + +/** + * Read the latest task of one child Conversation: the newest agent invocation + * that its parent started with `spawnAgent` or `messageAgent`. The first agent + * invocation of a child always comes from `spawnAgent`, also when it has no + * tool call id. + */ +async function getLatestAgentTask( + childConversationId: string, +): Promise { + const db = getSqlExecutor().db(); + const [latest] = await db + .select() + .from(juniorAgentInvocations) + .where( + and( + eq(juniorAgentInvocations.childConversationId, childConversationId), + isNotNull(juniorAgentInvocations.parentToolCallId), + ), + ) + .orderBy(desc(juniorAgentInvocations.createdAt)) + .limit(1); + if (latest) { + return invocationFromRow(latest); + } + const [first] = await db + .select() + .from(juniorAgentInvocations) + .where(eq(juniorAgentInvocations.childConversationId, childConversationId)) + .orderBy(asc(juniorAgentInvocations.createdAt)) + .limit(1); + return first ? invocationFromRow(first) : undefined; +} + +/** + * Create or replay the agent invocation that gives one input to the child + * Conversation of `latest`. It continues that Conversation for the same + * parent. It runs as the Actor and with the credentials of the child's latest + * task, unless `runAs` sets them. It skips the queue and active-count checks + * because it continues work that already passed them. + */ +export async function createAgentInvocationForChildInput( + args: { + input: string; + invocationId: string; + latest: AgentInvocation; + /** Actor, credentials, and Source for watch events. */ + runAs?: Pick; + }, + nowMs = Date.now(), +): Promise { + const { invocationId, latest } = args; + const task = await getLatestAgentTask(latest.childConversationId); + if (!task) { + throw new Error( + `Child conversation ${latest.childConversationId} has no agent task`, + ); + } + const runAs = args.runAs ?? task; + await getSqlExecutor() + .db() + .insert(juniorAgentInvocations) + .values({ + invocationId, + parentConversationId: latest.parentConversationId, + childConversationId: latest.childConversationId, + agentName: latest.agentName ?? null, + input: args.input, + actor: runAs.actor, + credentialContext: runAs.credentialContext ?? null, + source: args.runAs?.source ?? { kind: "agent_invocation" }, + destination: latest.destination, + modelProfile: latest.modelProfile ?? null, + reasoningLevel: latest.reasoningLevel ?? null, + status: "pending", + mailboxStatus: "pending", + createdAt: new Date(nowMs), + updatedAt: new Date(nowMs), + terminalAt: null, + }) + .onConflictDoNothing(); + const invocation = await getAgentInvocation(invocationId); + if ( + !invocation || + invocation.childConversationId !== latest.childConversationId + ) { + throw new Error(`Agent invocation creation raced for ${invocationId}`); + } + return invocation; +} diff --git a/packages/junior/src/chat/agent-invocations/markers.ts b/packages/junior/src/chat/agent-invocations/markers.ts index a4ca951c0a..8f9b7be9dc 100644 --- a/packages/junior/src/chat/agent-invocations/markers.ts +++ b/packages/junior/src/chat/agent-invocations/markers.ts @@ -4,6 +4,7 @@ * The child Conversation gets `message` events for its input and its result, * because a child has no Location that records visible messages. */ +import type { Actor } from "@/chat/actor"; import { getConversationEventStore } from "@/chat/db"; import { getAgentInvocationMessageId } from "./store"; import type { AgentInvocation, TerminalAgentInvocation } from "./types"; @@ -16,10 +17,12 @@ const SUBAGENT_KIND = "agent"; * record its input as a user message in the child Conversation. The dashboard * pairs the parent event with `subagent_ended` to show the child and its * status. It must be written before the child mailbox append, so the child - * cannot end first. + * cannot end first. `onBehalfOf` of a parent task goes on the child message + * (see `on-behalf-of.ts`). */ export async function recordSubagentStarted( invocation: AgentInvocation, + onBehalfOf?: Actor, ): Promise { const eventStore = getConversationEventStore(); // The mailbox input has the same id, so the child Turn links to it. @@ -32,6 +35,7 @@ export async function recordSubagentStarted( messageId, role: "user", text: invocation.input, + ...(onBehalfOf ? { meta: { onBehalfOf } } : undefined), }, createdAtMs: invocation.createdAtMs, }, diff --git a/packages/junior/src/chat/agent-invocations/message.ts b/packages/junior/src/chat/agent-invocations/message.ts index 5607d5f768..4da967541c 100644 --- a/packages/junior/src/chat/agent-invocations/message.ts +++ b/packages/junior/src/chat/agent-invocations/message.ts @@ -14,6 +14,7 @@ import { findChildAgentConversationId, getActiveAgentInvocationForConversation, } from "./store"; +import { taskOnBehalfOf } from "./on-behalf-of"; import { PARENT_AGENT_TARGET } from "./types"; import { enqueueAgentInvocation } from "./work"; @@ -85,7 +86,11 @@ export function bindMessageAgent( parentConversationId: run.conversationId, parentToolCallId: call.toolCallId, }); - await enqueueAgentInvocation(invocation, { ...options, queue }); + await enqueueAgentInvocation(invocation, { + ...options, + onBehalfOf: taskOnBehalfOf(run, actor), + queue, + }); return { conversationId: childConversationId, delivery: queued ? "queued" : "started", diff --git a/packages/junior/src/chat/agent-invocations/on-behalf-of.ts b/packages/junior/src/chat/agent-invocations/on-behalf-of.ts new file mode 100644 index 0000000000..19381c6663 --- /dev/null +++ b/packages/junior/src/chat/agent-invocations/on-behalf-of.ts @@ -0,0 +1,77 @@ +/** + * Records and reads the Actor whose work a child agent does. + * + * A child agent never has a message from a person, and a watch, timer, or + * result Turn of a child runs as the event system Actor. Action review still + * reads the root Conversation for the person whose work it is. The runtime + * records that Actor as `onBehalfOf` in the child's session log: + * + * - `spawnAgent` and `messageAgent` put it in the `meta` of the task message + * that they write into the child. It is the Actor of a root Turn, or the + * `onBehalfOf` of a child Turn. An Event Turn on a root is not from a + * person, so its task has none. + * - Each child Turn uses the value of its own task message. Without one, it + * keeps the value of the child's latest instruction. The Turn saves the + * value in the provenance of its own instruction, so the next Turn keeps + * it. + * + * Only the runtime sets it. It never changes the Run Actor or credentials. + */ +import { actorSchema } from "@sentry/junior-plugin-api"; +import type { Actor } from "@/chat/actor"; +import type { AgentRun } from "@/chat/agent/types"; +import { sameActorIdentity } from "@/chat/conversations/provenance"; +import { getConversationEventStore } from "@/chat/db"; +import { EVENT_SYSTEM_ACTOR } from "@/chat/events/actor"; +import { getAgentInvocationMessageId } from "./store"; +import type { AgentInvocation } from "./types"; + +/** + * Return the Actor whose work a task from this Run serves. `actor` is the + * Run Actor. Returns nothing for an Event Turn on a root. + */ +export function taskOnBehalfOf( + run: Pick, + actor: Actor, +): Actor | undefined { + if (run.parentConversationId) { + return run.onBehalfOf; + } + return sameActorIdentity(actor, EVENT_SYSTEM_ACTOR) ? undefined : actor; +} + +/** Read the `onBehalfOf` of one parent task from its task message. */ +export async function readTaskOnBehalfOf( + invocation: AgentInvocation, +): Promise { + if (!invocation.parentToolCallId) { + return undefined; + } + const messageId = getAgentInvocationMessageId(invocation.invocationId); + const event = await getConversationEventStore().loadByIdempotencyKey( + invocation.childConversationId, + `message:${messageId}`, + ); + const value = + event?.data.type === "message" ? event.data.meta?.onBehalfOf : undefined; + return value === undefined ? undefined : actorSchema.parse(value); +} + +/** + * Return the Actor whose work the Turn of one agent invocation does: the + * value of its own task, else the value of the child's latest instruction. + */ +export async function readAgentInvocationOnBehalfOf( + invocation: AgentInvocation, +): Promise { + const task = await readTaskOnBehalfOf(invocation); + if (task) { + return task; + } + const latest = await getConversationEventStore().loadLatestInstruction( + invocation.childConversationId, + ); + return latest?.data.type === "user_message" + ? latest.data.provenance.onBehalfOf + : undefined; +} diff --git a/packages/junior/src/chat/agent-invocations/result.ts b/packages/junior/src/chat/agent-invocations/result.ts index 36bb7df904..46690df9dc 100644 --- a/packages/junior/src/chat/agent-invocations/result.ts +++ b/packages/junior/src/chat/agent-invocations/result.ts @@ -22,8 +22,8 @@ import type { ConversationWorkerContext, ConversationWorkerResult, } from "@/chat/task-execution/worker"; +import { createAgentInvocationForChildInput } from "./child-input"; import { - createAgentInvocationFromLatest, getAgentInvocation, getAgentInvocationIdForEvents, getAgentInvocationIdForMessage, @@ -368,7 +368,7 @@ export async function startAgentInvocationsForChildInput( // The events run as the event system Actor, as on a root. A timer may // use the grants of the person who set it. const { actor, source } = events[0]!; - const invocation = await createAgentInvocationFromLatest({ + const invocation = await createAgentInvocationForChildInput({ input: joinMailboxText(messages), invocationId: getAgentInvocationIdForEvents( messages[0]!.inboundMessageId, @@ -400,7 +400,7 @@ export async function startAgentInvocationsForChildInput( ) { continue; } - const invocation = await createAgentInvocationFromLatest({ + const invocation = await createAgentInvocationForChildInput({ input: message.input.text.trim(), invocationId: input.kind === "result" diff --git a/packages/junior/src/chat/agent-invocations/root-conversation.ts b/packages/junior/src/chat/agent-invocations/root-conversation.ts index 5589a9a9f7..d33ee747e0 100644 --- a/packages/junior/src/chat/agent-invocations/root-conversation.ts +++ b/packages/junior/src/chat/agent-invocations/root-conversation.ts @@ -8,9 +8,11 @@ * handoff never rewrite these events, so an early request or a later * revocation is still there after the root's model history changes. * - * Only a message whose author has the same Actor identity as the child counts - * as authority. Each other message keeps its real author as a label and is - * only context. Nothing is copied into the child. + * Only a message whose author has the same Actor identity as the Actor whose + * work the child does counts as authority. That Actor is the `onBehalfOf` of + * the child's latest task, not the Run Actor. Each other message keeps its + * real author as a label and is only context. Nothing is copied into the + * child. */ import type { Actor } from "@/chat/actor"; import { getConversationEventStore, getConversationStore } from "@/chat/db"; @@ -160,17 +162,18 @@ async function readAutomationInstruction( /** * Read a bounded projection of the root Conversation for Guardian review of - * one child agent action. `actor` is the child's Actor. Only messages that - * the same Actor wrote, or the Automation run's own instruction for a system - * Actor, are labeled as authority. When `conversationId` is a root, it - * returns nothing. + * one child agent action. `onBehalfOf` is the Actor whose work the child + * does. Only messages that this Actor wrote, or the Automation run's own + * instruction for its system Actor, are labeled as authority. Without it, + * nothing is. When `conversationId` is a root, it returns nothing. */ export async function readRootConversation(args: { - actor: Actor | undefined; conversationId: string; + onBehalfOf: Actor | undefined; }): Promise { const root = await findRootConversation(args.conversationId); - if (!root || root.conversationId === args.conversationId || !args.actor) { + const actor = args.onBehalfOf; + if (!root || root.conversationId === args.conversationId || !actor) { return { messages: [], truncated: false }; } const page = await getConversationEventStore().query(root.conversationId, { @@ -208,7 +211,7 @@ export async function readRootConversation(args: { continue; } all.push({ - author: messageLabel(root, data, args.actor), + author: messageLabel(root, data, actor), messageId: data.messageId, text: truncate(text, MAX_ROOT_MESSAGE_CHARS), }); @@ -220,7 +223,7 @@ export async function readRootConversation(args: { instruction && instructionText ? [ { - author: messageLabel(root, instruction, args.actor), + author: messageLabel(root, instruction, actor), text: truncate(instructionText, MAX_ROOT_MESSAGE_CHARS), }, ] diff --git a/packages/junior/src/chat/agent-invocations/spawn.ts b/packages/junior/src/chat/agent-invocations/spawn.ts index 03cb773c2b..b22eddd94e 100644 --- a/packages/junior/src/chat/agent-invocations/spawn.ts +++ b/packages/junior/src/chat/agent-invocations/spawn.ts @@ -3,6 +3,7 @@ import type { ConversationStore } from "@/chat/conversations/store"; import type { AgentRun, SpawnAgent, SpawnAgentInput } from "@/chat/agent/types"; import { actorFromRun, toolInvocationDestination } from "@/chat/agent/types"; import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; +import { taskOnBehalfOf } from "./on-behalf-of"; import { createAndEnqueueAgentInvocation } from "./work"; type SpawnOptions = { @@ -45,7 +46,7 @@ export function bindSpawnAgent( : undefined), ...(input.title ? { title: input.title } : undefined), }, - { ...options, queue }, + { ...options, onBehalfOf: taskOnBehalfOf(run, actor), queue }, ); return { childConversationId: invocation.childConversationId, diff --git a/packages/junior/src/chat/agent-invocations/store.ts b/packages/junior/src/chat/agent-invocations/store.ts index 6699b04048..83b0a6b79b 100644 --- a/packages/junior/src/chat/agent-invocations/store.ts +++ b/packages/junior/src/chat/agent-invocations/store.ts @@ -480,57 +480,6 @@ export async function createAgentInvocation( }); } -/** - * Create or replay the agent invocation that gives one input to a child - * Conversation: a result or message from its own child agent, or its own - * watch events. It continues that Conversation for the same parent as - * `latest`, and for the same Actor and credentials unless `runAs` sets them. - * It skips the queue and active-count checks because it continues work that - * already passed them. - */ -export async function createAgentInvocationFromLatest( - args: { - input: string; - invocationId: string; - latest: AgentInvocation; - /** Actor, credentials, and Source for watch events. */ - runAs?: Pick; - }, - nowMs = Date.now(), -): Promise { - const { invocationId } = args; - await getSqlExecutor() - .db() - .insert(juniorAgentInvocations) - .values({ - invocationId, - parentConversationId: args.latest.parentConversationId, - childConversationId: args.latest.childConversationId, - agentName: args.latest.agentName ?? null, - input: args.input, - actor: (args.runAs ?? args.latest).actor, - credentialContext: (args.runAs ?? args.latest).credentialContext ?? null, - source: args.runAs?.source ?? { kind: "agent_invocation" }, - destination: args.latest.destination, - modelProfile: args.latest.modelProfile ?? null, - reasoningLevel: args.latest.reasoningLevel ?? null, - status: "pending", - mailboxStatus: "pending", - createdAt: new Date(nowMs), - updatedAt: new Date(nowMs), - terminalAt: null, - }) - .onConflictDoNothing(); - const invocation = await getAgentInvocation(invocationId); - if ( - !invocation || - invocation.childConversationId !== args.latest.childConversationId - ) { - throw new Error(`Agent invocation creation raced for ${invocationId}`); - } - return invocation; -} - /** Find a direct child agent of one parent by name or child Conversation id. */ export async function findChildAgentConversationId(args: { parentConversationId: string; diff --git a/packages/junior/src/chat/agent-invocations/work.ts b/packages/junior/src/chat/agent-invocations/work.ts index bc6238171e..a0e911a0ce 100644 --- a/packages/junior/src/chat/agent-invocations/work.ts +++ b/packages/junior/src/chat/agent-invocations/work.ts @@ -7,8 +7,9 @@ import { } from "@/chat/agent/prompt"; import type { AgentRun, AgentSteeringMessage } from "@/chat/agent/types"; import type { Location } from "@/chat/conversations/location"; +import type { Actor } from "@/chat/actor"; import { - agentInstructionProvenance, + agentInstructionProvenanceFor, sameActorIdentity, } from "@/chat/conversations/provenance"; import type { ConversationStore } from "@/chat/conversations/store"; @@ -79,6 +80,10 @@ import type { import { recordSubagentEnded, recordSubagentStarted } from "./markers"; import { stopChildAgentsOf } from "./stop"; import { readBlockedReport } from "./blocked-report"; +import { + readAgentInvocationOnBehalfOf, + readTaskOnBehalfOf, +} from "./on-behalf-of"; const agentInvocationMailboxMetadataSchema = z .object({ @@ -90,6 +95,8 @@ const agentInvocationMailboxMetadataSchema = z type EnqueueOptions = { conversationStore?: ConversationStore; nowMs?: number; + /** Whose work a parent task serves. Only spawnAgent and messageAgent set it. */ + onBehalfOf?: Actor; queue: ConversationWorkQueue; state?: StateAdapter; }; @@ -147,7 +154,7 @@ export async function enqueueAgentInvocation( ) { return; } - await recordSubagentStarted(invocation); + await recordSubagentStarted(invocation, options.onBehalfOf); const nowMs = options.nowMs ?? Date.now(); await appendAndEnqueueInboundMessage({ message: buildAgentInvocationInboundMessage(invocation, nowMs), @@ -252,14 +259,16 @@ export async function resolveAgentInvocationId( /** * Give queued parent messages to the running agent invocation of a child as - * steering messages. A message joins only with the Actor and credentials of - * the running Turn. The first message that does not match stops the scan, so - * it and every later message wait for the current work. Each message that - * joins ends as `steered`, and the running agent invocation reports for it. + * steering messages. A message joins only with the Actor, credentials, and + * `onBehalfOf` of the running Turn. A message without `onBehalfOf` keeps the + * running one. The first message that does not match stops the scan, so it + * and every later message wait for the current work. Each message that joins + * ends as `steered`, and the running agent invocation reports for it. */ function createAgentMessageSteeringDrain( context: ConversationWorkerContext, running: AgentInvocation, + onBehalfOf: Actor | undefined, ) { const turnId = getAgentInvocationTurnId(running.invocationId); return async ( @@ -295,12 +304,15 @@ function createAgentMessageSteeringDrain( } continue; } + const candidateOnBehalfOf = await readTaskOnBehalfOf(candidate); if ( !sameActorIdentity(candidate.actor, running.actor) || !isDeepStrictEqual( candidate.credentialContext, running.credentialContext, - ) + ) || + (candidateOnBehalfOf && + !sameActorIdentity(candidateOnBehalfOf, onBehalfOf)) ) { break; } @@ -320,7 +332,7 @@ function createAgentMessageSteeringDrain( saved, joining.map( (candidate): AgentSteeringMessage => ({ - provenance: agentInstructionProvenance, + provenance: agentInstructionProvenanceFor(onBehalfOf), text: agentInvocationInputText(candidate), timestampMs: candidate.createdAtMs, }), @@ -656,6 +668,7 @@ export function createAgentInvocationWorker( let sandboxRef: SandboxRef | undefined; let history: PiMessage[]; let location: Location | undefined; + let onBehalfOf: Actor | undefined; let canSpawn = false; try { if (invocation.childConversationId !== context.conversationId) { @@ -695,13 +708,15 @@ export function createAgentInvocationWorker( surface: "internal", turnId, }); - const [persisted, projection, tree] = await Promise.all([ + const [persisted, projection, tree, turnOnBehalfOf] = await Promise.all([ getPersistedThreadState(invocation.childConversationId), openConversationProjection({ conversationId: invocation.childConversationId, }), readChildDepthAndRootLocation(invocation.childConversationId), + readAgentInvocationOnBehalfOf(invocation), ]); + onBehalfOf = turnOnBehalfOf; sandboxRef = getPersistedSandboxState(persisted); history = projection.messages; location = tree.location; @@ -739,6 +754,7 @@ export function createAgentInvocationWorker( history, actor: invocation.actor, credentialContext: invocation.credentialContext, + ...(onBehalfOf ? { onBehalfOf } : undefined), destination: invocation.destination, source: invocation.source, ...(location ? { location } : undefined), @@ -756,6 +772,7 @@ export function createAgentInvocationWorker( drainSteeringMessages: createAgentMessageSteeringDrain( context, invocation, + onBehalfOf, ), onInputCommitted: acknowledge, shouldYield: context.shouldYield, diff --git a/packages/junior/src/chat/agent/index.ts b/packages/junior/src/chat/agent/index.ts index 0ccf882752..78029c8ca9 100644 --- a/packages/junior/src/chat/agent/index.ts +++ b/packages/junior/src/chat/agent/index.ts @@ -50,7 +50,7 @@ import { recordTurnRoute, } from "@/chat/conversations/projection"; import { - agentInstructionProvenance, + agentInstructionProvenanceFor, contextProvenance, instructionActors, instructionProvenanceFor, @@ -491,10 +491,11 @@ async function executeAgentRunInPrivacyContext( // Steering appends to this array as it drains, so `run.actors` stays a // pure, live projection of committed instruction provenance. // A child agent's input is a parent agent task or another agent's - // output. An agent wrote it, so it has no Actor. + // output. An agent wrote it, so it has no Actor. It records whose work it + // is, so the child's next Turn keeps that Actor. const childAgent = Boolean(run.parentConversationId); const turnStartProvenance = childAgent - ? agentInstructionProvenance + ? agentInstructionProvenanceFor(run.onBehalfOf) : instructionProvenanceFor(actor); const committedInstructionProvenance: ConversationMessageProvenance[] = [ ...(existingSessionRecord?.piMessageProvenance ?? []), diff --git a/packages/junior/src/chat/agent/tools.ts b/packages/junior/src/chat/agent/tools.ts index 2623744032..f215aa4bf4 100644 --- a/packages/junior/src/chat/agent/tools.ts +++ b/packages/junior/src/chat/agent/tools.ts @@ -445,13 +445,15 @@ export async function wireAgentTools( evidence: () => buildToolActionEvidence(args.currentAgentMessages(), { childAgent }), // A child has no message from a person. Its authority is in the root - // Conversation, read again for each action. + // Conversation, read again for each action, for the Actor whose work + // the child does. A watch event Turn runs as the event system Actor, so + // the Run Actor is not that Actor. ...(childAgent ? { rootConversation: async () => await readRootConversation({ - actor: args.currentActor, conversationId: args.run.conversationId, + onBehalfOf: args.run.onBehalfOf, }), } : undefined), diff --git a/packages/junior/src/chat/agent/types.ts b/packages/junior/src/chat/agent/types.ts index 2fd4d2056e..d101091abe 100644 --- a/packages/junior/src/chat/agent/types.ts +++ b/packages/junior/src/chat/agent/types.ts @@ -257,6 +257,12 @@ export type AgentRun = { actor?: Actor; /** Credential authority projected from actor (plus optional subject). */ credentialContext?: CredentialContext; + /** + * The Actor whose work a child agent Run serves, from the child's latest + * task. Only the runtime sets it. Action review reads the root Conversation + * as this Actor. It never changes `actor` or credentials. Unset for a root. + */ + onBehalfOf?: Actor; source: Source; /** Conversation Location available to this Run. */ location?: Location; diff --git a/packages/junior/src/chat/conversations/README.md b/packages/junior/src/chat/conversations/README.md index bf26a7d37e..64b774a57e 100644 --- a/packages/junior/src/chat/conversations/README.md +++ b/packages/junior/src/chat/conversations/README.md @@ -44,6 +44,9 @@ facts: results that were never platform chat messages. `user_message` provenance distinguishes user instructions from ambient context. +Its `actor` is who wrote the message. An instruction that an agent wrote has no +`actor`. It can have `onBehalfOf`, the Actor whose work it serves. Only the +runtime sets it (see `../agent-invocations/README.md`). Tool calls remain ordered content inside the `assistant_message` that produced them; the corresponding results are separate `tool_result` events. diff --git a/packages/junior/src/chat/conversations/provenance.ts b/packages/junior/src/chat/conversations/provenance.ts index fffaec20be..9555a993f3 100644 --- a/packages/junior/src/chat/conversations/provenance.ts +++ b/packages/junior/src/chat/conversations/provenance.ts @@ -14,7 +14,10 @@ const conversationMessageAuthoritySchema = z.union([ export const conversationMessageProvenanceSchema = z .object({ authority: conversationMessageAuthoritySchema, + /** Who wrote the message. Agent-written text has none. */ actor: actorSchema.optional(), + /** Whose work an agent-written instruction serves. Runtime-only. */ + onBehalfOf: actorSchema.optional(), }) .strict(); @@ -40,11 +43,16 @@ export function instructionProvenanceFor( /** * Provenance of an instruction that an agent wrote, such as a parent agent * task in a child Conversation. It has no Actor, so it never counts as user - * intent. + * intent. `onBehalfOf` is the Actor whose work the instruction serves. Only + * the runtime sets it, and it never makes that Actor the writer. */ -export const agentInstructionProvenance: ConversationMessageProvenance = { - authority: "instruction", -}; +export function agentInstructionProvenanceFor( + onBehalfOf: Actor | undefined, +): ConversationMessageProvenance { + return onBehalfOf + ? { authority: "instruction", onBehalfOf } + : { authority: "instruction" }; +} /** Unattributed ambient-context provenance for non-instruction messages. */ export const contextProvenance: ConversationMessageProvenance = { diff --git a/packages/junior/tests/component/agent-invocation-worker.test.ts b/packages/junior/tests/component/agent-invocation-worker.test.ts index 4c4709e0c2..8db2a87766 100644 --- a/packages/junior/tests/component/agent-invocation-worker.test.ts +++ b/packages/junior/tests/component/agent-invocation-worker.test.ts @@ -2,7 +2,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import type { PiMessage } from "@/chat/pi/messages"; import { buildSteeringPiMessage } from "@/chat/agent/prompt"; import type { AgentSteeringMessage } from "@/chat/agent/types"; -import { agentInstructionProvenance } from "@/chat/conversations/provenance"; +import { agentInstructionProvenanceFor } from "@/chat/conversations/provenance"; import { createAgentInvocation, createAgentInvocationForMessage, @@ -11,6 +11,7 @@ import { getAgentInvocationTurnId, markAgentInvocationRunning, } from "@/chat/agent-invocations/store"; +import { recordSubagentStarted } from "@/chat/agent-invocations/markers"; import { createListAgentsTool } from "@/chat/tools/runtime/list-agents"; import { buildAgentInvocationInboundMessage, @@ -238,11 +239,16 @@ describe("agent invocation worker", () => { it("lets a running child read a parent message as a steering message", async () => { const fixture = await prepareParentConversation(); try { + // The task and each message record whose work they are, as spawnAgent + // and messageAgent do. + const person = { platform: "local", userId: "local-user" } as const; const created = await createAgentInvocation({ ...INVOCATION_INPUT, agentName: "researcher", idempotencyKey: "steer-task", + parentToolCallId: "call-steer-task", }); + await recordSubagentStarted(created, person); await markAgentInvocationRunning(created.invocationId); const { invocation: message, queued } = await createAgentInvocationForMessage({ @@ -253,6 +259,7 @@ describe("agent invocation worker", () => { parentToolCallId: "call-steer-message", }); expect(queued).toBe(true); + await recordSubagentStarted(message, person); // An earlier drain saved this message in the Turn, then failed before // it marked the agent invocation read. const { invocation: savedMessage } = @@ -270,7 +277,7 @@ describe("agent invocation worker", () => { destination: DESTINATION, piMessages: [ buildSteeringPiMessage({ - provenance: agentInstructionProvenance, + provenance: agentInstructionProvenanceFor(person), text: agentInvocationInputText(savedMessage), timestampMs: savedMessage.createdAtMs, }), @@ -282,6 +289,20 @@ describe("agent invocation worker", () => { surface: "internal", }); + // A message for another person's work does not join, so it waits. + const { invocation: otherPersonMessage } = + await createAgentInvocationForMessage({ + ...INVOCATION_INPUT, + childConversationId: created.childConversationId, + idempotencyKey: "steer-other-person", + input: "Do this for someone else.", + parentToolCallId: "call-steer-other-person", + }); + await recordSubagentStarted(otherPersonMessage, { + ...person, + userId: "other-user", + }); + // The runner drains steering messages, as the agent does before its // next model call. const steered: AgentSteeringMessage[][] = []; @@ -319,6 +340,7 @@ describe("agent invocation worker", () => { ownMessage, buildAgentInvocationInboundMessage(message), buildAgentInvocationInboundMessage(savedMessage), + buildAgentInvocationInboundMessage(otherPersonMessage), ]), ); return []; @@ -338,11 +360,12 @@ describe("agent invocation worker", () => { expect(context.attempt.drain).toHaveBeenCalledWith(expect.any(Function), { deferred: true, }); - // The parent's text is labeled as agent-written and has no Actor. + // The parent's text is labeled as agent-written and has no Actor. It + // keeps whose work it is. expect(steered).toEqual([ [ expect.objectContaining({ - provenance: agentInstructionProvenance, + provenance: agentInstructionProvenanceFor(person), text: expect.stringMatching( /^\[parent agent task\][\s\S]*Also check the retry path\.$/, ), @@ -366,11 +389,15 @@ describe("agent invocation worker", () => { status: "completed", }); } - // The running agent invocation reports for all of them. + await expect( + getAgentInvocation(otherPersonMessage.invocationId), + ).resolves.toMatchObject({ status: "pending" }); + // The running agent invocation answers the messages that joined. The + // waiting message runs next and reports for the child. await expect( getAgentInvocation(created.invocationId), ).resolves.toMatchObject({ - parentMailboxStatus: "pending", + parentMailboxStatus: "replaced", result: "Checked both paths.", status: "completed", }); @@ -380,10 +407,9 @@ describe("agent invocation worker", () => { ).resolves.toMatchObject({ agents: [ { - last_result: "completed", name: "researcher", - status: "idle", - task: `Skip the "cache" path & .`, + status: "pending", + task: "Do this for someone else.", }, ], }); diff --git a/packages/junior/tests/integration/agent-invocation-concurrency.test.ts b/packages/junior/tests/integration/agent-invocation-concurrency.test.ts index 6e963e78d1..7c36fd02e7 100644 --- a/packages/junior/tests/integration/agent-invocation-concurrency.test.ts +++ b/packages/junior/tests/integration/agent-invocation-concurrency.test.ts @@ -1,10 +1,18 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import type { StreamFn } from "@earendil-works/pi-agent-core"; -import type { AgentRun } from "@/chat/agent/types"; +import { createLocalSource, type Source } from "@sentry/junior-plugin-api"; +import type { Actor } from "@/chat/actor"; +import type { AgentRun, SpawnAgentInput } from "@/chat/agent/types"; +import { bindMessageAgent } from "@/chat/agent-invocations/message"; +import { readRootConversation } from "@/chat/agent-invocations/root-conversation"; +import { bindSpawnAgent } from "@/chat/agent-invocations/spawn"; +import { credentialContextForActor } from "@/chat/credentials/context"; +import { getConversationEventStore } from "@/chat/db"; import { AgentInvocationLimitError } from "@/chat/agent-invocations/errors"; import { createAgentInvocationForMessage, getAgentInvocation, + getLatestAgentInvocationForConversation, MAX_ACTIVE_AGENT_INVOCATIONS_PER_PARENT, } from "@/chat/agent-invocations/store"; import { @@ -13,8 +21,14 @@ import { routeAgentInvocationWork, createAndEnqueueAgentInvocation, } from "@/chat/agent-invocations/work"; -import type { CreateAgentInvocationInput } from "@/chat/agent-invocations/types"; -import { startAgentInvocationsForChildInput } from "@/chat/agent-invocations/result"; +import type { + CreateAgentInvocationInput, + TerminalAgentInvocation, +} from "@/chat/agent-invocations/types"; +import { + enqueueAgentInvocationResult, + startAgentInvocationsForChildInput, +} from "@/chat/agent-invocations/result"; import { EVENT_SYSTEM_ACTOR } from "@/chat/events/actor"; import { getEventCatalog } from "@/chat/events/runtime-catalog"; import { runTimerWatchHeartbeat } from "@/chat/events/timers"; @@ -53,9 +67,43 @@ async function listAgents() { .execute!({}, {}); } -function taskModel(request: AgentRun): StreamFn { +function taskModel(request: AgentRun, waitFor?: Promise): StreamFn { return createModelStream([ - { type: "text", text: `result:${request.instruction.text}` }, + { + type: "text", + text: `result:${request.instruction.text}`, + ...(waitFor ? { waitFor } : undefined), + }, + ]); +} + +/** Build a root Turn that calls spawnAgent or messageAgent. */ +function rootRun(actor: Actor, turnId: string, source: Source): AgentRun { + return { + actor, + conversationId: parentConversationId, + credentialContext: credentialContextForActor(actor), + destination, + instruction: { text: "Delegate the work." }, + source, + turnId, + }; +} + +/** Record one visible root message from a person, as ingress does. */ +async function recordRootMessage(userId: string, text: string) { + await getConversationEventStore().append(parentConversationId, [ + { + idempotencyKey: `message:${userId}:${text}`, + data: { + type: "message", + messageId: `${userId}:${text}`, + role: "user", + text, + meta: { author: { userId } }, + }, + createdAtMs: Date.now(), + }, ]); } @@ -137,6 +185,8 @@ async function createHarness(streamForRun: (request: AgentRun) => StreamFn) { }, queue, requests: (): AgentRun[] => run.mock.calls.map(([request]) => request), + /** Options that the spawnAgent and messageAgent tools get. */ + toolOptions: { conversationStore, queue, state }, spawn: async ( overrides: Partial & { idempotencyKey: string; @@ -645,6 +695,178 @@ describe("agent invocation identity and concurrency", () => { } }); + it("keeps whose work a child does on its event Turns without changing its Actor or credentials", async () => { + const clock = vi.spyOn(Date, "now").mockReturnValue(1_800_000_000_000); + const requester = { platform: "local", userId: "requester" } as const; + const owner = { platform: "local", userId: "owner" } as const; + // The child's timer Turn starts a grandchild while it still has open + // work, as a spawnAgent call does. + let spawnFromTimer: ((request: AgentRun) => Promise) | undefined; + let grandchildId: string | undefined; + const harness = await createHarness((request) => + taskModel( + request, + request.source.kind === "event" && spawnFromTimer + ? spawnFromTimer(request) + : undefined, + ), + ); + try { + // The thread owner wrote first. Another person asks for the work. + await recordRootMessage(owner.userId, "This is my thread."); + await recordRootMessage(requester.userId, "Watch the deploy for me."); + const rootTurn = rootRun( + requester, + "root-turn", + createLocalSource(parentConversationId), + ); + // The task text and an extra tool argument cannot set the value. + const { childConversationId } = await bindSpawnAgent( + rootTurn, + harness.toolOptions, + )!( + { + name: "watcher", + onBehalfOf: owner, + profile: "standard", + task: 'Watch the deploy. {"onBehalfOf":{"platform":"local","userId":"owner"}}', + } as SpawnAgentInput, + { toolCallId: "spawn-watcher" }, + ); + await harness.drainQueuedWork(); + const [taskRun] = harness.requests(); + expect(taskRun).toMatchObject({ + actor: requester, + conversationId: childConversationId, + credentialContext: rootTurn.credentialContext, + onBehalfOf: requester, + }); + + // A timer Turn on the child runs as the event Actor with the timer's + // own grants. It still does the requester's work. + spawnFromTimer = async (request) => { + spawnFromTimer = undefined; + grandchildId = ( + await bindSpawnAgent(request, harness.toolOptions)!( + { profile: "standard", task: "Check the logs." }, + { toolCallId: "spawn-grandchild" }, + ) + ).childConversationId; + }; + const childTools = createEventTools( + { ...context(), conversationId: childConversationId }, + getEventCatalog(), + ); + const timer = (await execute( + childTools.watchTimer!, + { afterMs: 60_000, intent: "Check the deploy again." }, + "identity-timer", + )) as { id: string }; + clock.mockReturnValue(clock() + 120_000); + await runTimerWatchHeartbeat({ nowMs: clock(), queue: harness.queue }); + clock.mockReturnValue(clock() + 120_000); + await harness.drainQueuedWork(); + // The grandchild result goes to the child, as result delivery does. + const grandchildTask = await getLatestAgentInvocationForConversation( + grandchildId!, + ); + expect(grandchildTask).toMatchObject({ status: "completed" }); + await enqueueAgentInvocationResult( + grandchildTask as TerminalAgentInvocation, + harness.toolOptions, + ); + await harness.drainQueuedWork(); + const runsIn = (conversationId: string | undefined) => + harness + .requests() + .filter((request) => request.conversationId === conversationId); + const [, timerRun, resultRun] = runsIn(childConversationId); + expect(timerRun).toMatchObject({ + actor: EVENT_SYSTEM_ACTOR, + credentialContext: { + actor: EVENT_SYSTEM_ACTOR, + subject: expect.objectContaining({ + binding: expect.objectContaining({ watchId: timer.id }), + }), + }, + onBehalfOf: requester, + source: { kind: "event", eventType: "timer.fired" }, + }); + // Action review in the watch Turn reads the requester's message as + // `user`. The thread owner is another person. + await expect( + readRootConversation({ + conversationId: childConversationId, + onBehalfOf: timerRun!.onBehalfOf, + }), + ).resolves.toMatchObject({ + messages: [ + { author: "other person", text: "This is my thread." }, + { author: "user", text: "Watch the deploy for me." }, + ], + }); + + // The grandchild gets the same person from the child's timer Turn. Its + // Actor and credentials are those of the Turn that started it. + expect(runsIn(grandchildId)).toEqual([ + expect.objectContaining({ + actor: EVENT_SYSTEM_ACTOR, + credentialContext: timerRun!.credentialContext, + onBehalfOf: requester, + }), + ]); + // The grandchild result continues the child's task, not the timer. + expect(resultRun).toMatchObject({ + actor: requester, + credentialContext: rootTurn.credentialContext, + onBehalfOf: requester, + }); + + // A root watch Turn has no person. Its message keeps the child's + // person, and the child runs with the Actor of that Turn. + const rootWatchTurn = rootRun(EVENT_SYSTEM_ACTOR, "root-watch-turn", { + eventKey: "deploy-1", + eventType: "deployment.succeeded", + identifier: "deploy", + kind: "event", + namespace: "vercel", + }); + await bindMessageAgent(rootWatchTurn, harness.toolOptions)!( + { text: "The deploy finished.", to: "watcher" }, + { toolCallId: "message-watcher" }, + ); + await harness.drainQueuedWork(); + expect(runsIn(childConversationId).at(-1)).toMatchObject({ + actor: EVENT_SYSTEM_ACTOR, + credentialContext: rootWatchTurn.credentialContext, + onBehalfOf: requester, + }); + + // A child that a root watch Turn starts has no person, so no root + // message counts as `user`. There is no fallback to the owner. + const { childConversationId: eventChild } = await bindSpawnAgent( + rootWatchTurn, + harness.toolOptions, + )!( + { profile: "standard", task: "Summarize the deploy." }, + { toolCallId: "spawn-from-watch" }, + ); + await harness.drainQueuedWork(); + const [eventChildRun] = runsIn(eventChild); + expect(eventChildRun).toBeDefined(); + expect(eventChildRun!.onBehalfOf).toBeUndefined(); + await expect( + readRootConversation({ + conversationId: eventChild, + onBehalfOf: eventChildRun!.onBehalfOf, + }), + ).resolves.toEqual({ messages: [], truncated: false }); + } finally { + clock.mockRestore(); + await harness.close(); + } + }); + it("lets a parent list and stop the watches of its child agents", async () => { const clock = vi.spyOn(Date, "now").mockReturnValue(1_800_000_000_000); const harness = await createHarness(taskModel); diff --git a/packages/junior/tests/integration/agent-invocation-nested.test.ts b/packages/junior/tests/integration/agent-invocation-nested.test.ts index 963fdabb2a..2de4c1a7bf 100644 --- a/packages/junior/tests/integration/agent-invocation-nested.test.ts +++ b/packages/junior/tests/integration/agent-invocation-nested.test.ts @@ -3,6 +3,7 @@ import { completeAgentInvocation, createAgentInvocation, getAgentInvocation, + getAgentInvocationIdForEvents, getAgentInvocationIdForResult, getAgentInvocationResultMessageId, } from "@/chat/agent-invocations/store"; @@ -13,6 +14,10 @@ import { import type { TerminalAgentInvocation } from "@/chat/agent-invocations/types"; import { createAgentInvocationWorker } from "@/chat/agent-invocations/work"; import type { AgentRun } from "@/chat/agent/types"; +import { credentialContextForActor } from "@/chat/credentials/context"; +import { bindTimerWatchCredentialSubject } from "@/chat/credentials/subject"; +import { EVENT_SYSTEM_ACTOR } from "@/chat/events/actor"; +import { createEventInboundMessage } from "@/chat/events/notification"; import { migrateSchema } from "@/chat/conversations/sql/migrations"; import { createSqlStore } from "@/chat/conversations/sql/store"; import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter"; @@ -132,8 +137,12 @@ describe("nested agent invocations", () => { ...slackInvocationInput, idempotencyKey: "nested-child", }); + const userCredentials = credentialContextForActor( + slackInvocationInput.actor, + ); const helper = await createAgentInvocation({ ...slackInvocationInput, + credentialContext: userCredentials, idempotencyKey: "nested-helper", parentConversationId: child.childConversationId, }); @@ -184,10 +193,53 @@ describe("nested agent invocations", () => { "Grand helper fact: 42.", ); expect(grandResult).toMatchObject({ parentMailboxStatus: "pending" }); + // A timer fires on the helper first. Its agent invocation runs as the + // event system Actor with the grants of that one timer. + const timerMessage = createEventInboundMessage({ + event: { + eventKey: "helper-timer", + eventType: "timer.fired", + identifier: "helper-timer", + namespace: "junior", + occurredAtMs: 1, + trustedSummary: "Timer fired", + }, + // The burst window has passed. + receivedAtMs: 0, + subscription: { + conversationId: helper.childConversationId, + credentialSubject: bindTimerWatchCredentialSubject({ + conversationId: helper.childConversationId, + userId: slackInvocationInput.actor.userId, + watchId: "helper-timer", + }), + id: "helper-timer", + }, + text: "Check the grand helper.", + }); + await startAgentInvocationsForChildInput( + context(helper.childConversationId, [timerMessage]), + { queue, state }, + ); + const timerInvocation = (await getAgentInvocation( + getAgentInvocationIdForEvents(timerMessage.inboundMessageId), + ))!; + expect(timerInvocation).toMatchObject({ + actor: EVENT_SYSTEM_ACTOR, + credentialContext: { + subject: expect.objectContaining({ allowedWhen: "timer-watch" }), + }, + }); + await expect( + complete(timerInvocation.invocationId, "Still waiting."), + ).resolves.toMatchObject({ parentMailboxStatus: "replaced" }); + // The result that arrives after the timer continues the helper's task. + // It does not take the event Actor or the timer's grants. const helperContinuation = await sendToChild(grandResult); expect(helperContinuation).toMatchObject({ actor: slackInvocationInput.actor, childConversationId: helper.childConversationId, + credentialContext: userCredentials, input: expect.stringContaining("Grand helper fact: 42."), parentConversationId: child.childConversationId, status: "pending", diff --git a/packages/junior/tests/integration/agent-invocation-user-approval.test.ts b/packages/junior/tests/integration/agent-invocation-user-approval.test.ts index 1816906216..37d2bc8f83 100644 --- a/packages/junior/tests/integration/agent-invocation-user-approval.test.ts +++ b/packages/junior/tests/integration/agent-invocation-user-approval.test.ts @@ -109,8 +109,8 @@ describe("user approval for child agents", () => { // Action review reads the root, not the task. Only the user's own // message is labeled `user`. - const read = async (conversationId: string, actor?: Actor) => - await readRootConversation({ actor, conversationId }); + const read = async (conversationId: string, onBehalfOf?: Actor) => + await readRootConversation({ conversationId, onBehalfOf }); await expect(read(child.childConversationId, user)).resolves.toEqual({ messages: [ { author: "user", text: "Which images did we publish here?" }, @@ -217,8 +217,8 @@ describe("user approval for child agents", () => { ); const read = await readRootConversation({ - actor: user, conversationId: child.childConversationId, + onBehalfOf: user, }); expect(read.messages).toHaveLength(16); expect(read.messages.at(-1)).toEqual({ @@ -272,8 +272,8 @@ describe("user approval for child agents", () => { await expect( readRootConversation({ - actor: eventActor, conversationId: child.childConversationId, + onBehalfOf: eventActor, }), ).resolves.toEqual({ messages: [{ author: "automation", text: instruction }], @@ -282,8 +282,8 @@ describe("user approval for child agents", () => { // Another system Actor does not get this run's authority. await expect( readRootConversation({ - actor: { platform: "system", name: "scheduler" }, conversationId: child.childConversationId, + onBehalfOf: { platform: "system", name: "scheduler" }, }), ).resolves.toMatchObject({ messages: [{ author: "event", text: instruction }], @@ -298,8 +298,8 @@ describe("user approval for child agents", () => { }); } const long = await readRootConversation({ - actor: eventActor, conversationId: child.childConversationId, + onBehalfOf: eventActor, }); expect(long.messages[0]).toEqual({ author: "automation",