From 68eb11ad50447082df44618d2687653d33e87921 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Mon, 2 Nov 2026 19:32:35 +0000 Subject: [PATCH 1/5] fix(agent-invocations): Keep one active agent invocation per child A child agent has at most one running or paused agent invocation. Before the worker marks an agent invocation running, it now ends every other active invocation of the same child. A saved finished Turn gives its result; otherwise the invocation fails and its paused Turn is abandoned. The started invocation reports for both. Move the named-child binding queries to bindings.ts so store.ts stays under the file-length limit. Co-Authored-By: David Cramer <23610+dcramer@users.noreply.github.com> --- .../src/chat/agent-invocations/AGENTS.md | 2 + .../src/chat/agent-invocations/bindings.ts | 61 ++++++++++++ .../src/chat/agent-invocations/message.ts | 2 +- .../src/chat/agent-invocations/store.ts | 82 ++++------------ .../junior/src/chat/agent-invocations/work.ts | 28 ++++++ .../integration/agent-invocation-work.test.ts | 95 ++++++++++++++++++- 6 files changed, 202 insertions(+), 68 deletions(-) create mode 100644 packages/junior/src/chat/agent-invocations/bindings.ts diff --git a/packages/junior/src/chat/agent-invocations/AGENTS.md b/packages/junior/src/chat/agent-invocations/AGENTS.md index 141ed80622..9b2ba4b3c0 100644 --- a/packages/junior/src/chat/agent-invocations/AGENTS.md +++ b/packages/junior/src/chat/agent-invocations/AGENTS.md @@ -9,5 +9,7 @@ id, so retries and repairs never add a second Turn, invocation, or row. - Pick the child worker from the stored `parentConversationId`, never from the Turn Source (a child watch-event Turn has an event Source). +- A child has at most one `running` or `paused` agent invocation. Before the + worker starts one, it ends every other active one of that child. - An Agent type (`agent-types.ts`) is data. Read its fields, never its name. Types do not limit tools. diff --git a/packages/junior/src/chat/agent-invocations/bindings.ts b/packages/junior/src/chat/agent-invocations/bindings.ts new file mode 100644 index 0000000000..77d59e50d6 --- /dev/null +++ b/packages/junior/src/chat/agent-invocations/bindings.ts @@ -0,0 +1,61 @@ +import { and, eq } from "drizzle-orm"; +import { getSqlExecutor } from "@/chat/db"; +import { juniorAgentBindings, juniorAgentInvocations } from "@/db/schema"; +import { agentBindingSchema, type AgentBinding } from "./types"; + +function bindingFromRow( + row: typeof juniorAgentBindings.$inferSelect, +): AgentBinding { + return agentBindingSchema.parse({ + childConversationId: row.childConversationId, + name: row.name, + parentConversationId: row.parentConversationId, + }); +} + +/** Read one named child binding in its parent-agent scope. */ +export async function getAgentBinding(args: { + name: string; + parentConversationId: string; +}): Promise { + const rows = await getSqlExecutor() + .db() + .select() + .from(juniorAgentBindings) + .where( + and( + eq(juniorAgentBindings.parentConversationId, args.parentConversationId), + eq(juniorAgentBindings.name, args.name), + ), + ); + return rows[0] ? bindingFromRow(rows[0]) : undefined; +} + +/** Find a direct child agent of one parent by name or child Conversation id. */ +export async function findChildAgentConversationId(args: { + parentConversationId: string; + target: string; +}): Promise { + const binding = await getAgentBinding({ + name: args.target, + parentConversationId: args.parentConversationId, + }); + if (binding) { + return binding.childConversationId; + } + const rows = await getSqlExecutor() + .db() + .select({ childConversationId: juniorAgentInvocations.childConversationId }) + .from(juniorAgentInvocations) + .where( + and( + eq( + juniorAgentInvocations.parentConversationId, + args.parentConversationId, + ), + eq(juniorAgentInvocations.childConversationId, args.target), + ), + ) + .limit(1); + return rows[0]?.childConversationId; +} diff --git a/packages/junior/src/chat/agent-invocations/message.ts b/packages/junior/src/chat/agent-invocations/message.ts index 5607d5f768..2176fa62b8 100644 --- a/packages/junior/src/chat/agent-invocations/message.ts +++ b/packages/junior/src/chat/agent-invocations/message.ts @@ -7,11 +7,11 @@ import type { } from "@/chat/agent/types"; import { actorFromRun, toolInvocationDestination } from "@/chat/agent/types"; import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; +import { findChildAgentConversationId } from "./bindings"; import { AgentMessageTargetError } from "./errors"; import { enqueueAgentMessageToParent } from "./result"; import { createAgentInvocationForMessage, - findChildAgentConversationId, getActiveAgentInvocationForConversation, } from "./store"; import { PARENT_AGENT_TARGET } from "./types"; diff --git a/packages/junior/src/chat/agent-invocations/store.ts b/packages/junior/src/chat/agent-invocations/store.ts index 9e0b41c005..1d97b59680 100644 --- a/packages/junior/src/chat/agent-invocations/store.ts +++ b/packages/junior/src/chat/agent-invocations/store.ts @@ -14,11 +14,10 @@ import { } from "drizzle-orm"; import { getConversationStore, getSqlExecutor } from "@/chat/db"; import { juniorAgentBindings, juniorAgentInvocations } from "@/db/schema"; +import { getAgentBinding } from "./bindings"; import { - agentBindingSchema, agentInvocationSchema, createAgentInvocationSchema, - type AgentBinding, type AgentInvocation, type AgentInvocationStatus, type CreateAgentInvocationInput, @@ -152,16 +151,6 @@ export function parseAgentInvocationResultMessageId( return invocationId || undefined; } -function bindingFromRow( - row: typeof juniorAgentBindings.$inferSelect, -): AgentBinding { - return agentBindingSchema.parse({ - childConversationId: row.childConversationId, - name: row.name, - parentConversationId: row.parentConversationId, - }); -} - /** Parse one agent invocation row. */ export function invocationFromRow( row: typeof juniorAgentInvocations.$inferSelect, @@ -227,24 +216,6 @@ function sameCreateInput( ); } -/** Read one named child binding in its parent-agent scope. */ -async function getAgentBinding(args: { - name: string; - parentConversationId: string; -}): Promise { - const rows = await getSqlExecutor() - .db() - .select() - .from(juniorAgentBindings) - .where( - and( - eq(juniorAgentBindings.parentConversationId, args.parentConversationId), - eq(juniorAgentBindings.name, args.name), - ), - ); - return rows[0] ? bindingFromRow(rows[0]) : undefined; -} - /** Read one durable agent invocation. */ export async function getAgentInvocation( invocationId: string, @@ -257,10 +228,10 @@ export async function getAgentInvocation( return rows[0] ? invocationFromRow(rows[0]) : undefined; } -/** Resolve the single invocation that may resume for one child conversation. */ -export async function getActiveAgentInvocationForConversation( +/** List the running and paused agent invocations of one child Conversation. */ +export async function listActiveAgentInvocationsForConversation( childConversationId: string, -): Promise { +): Promise { const rows = await getSqlExecutor() .db() .select() @@ -271,14 +242,22 @@ export async function getActiveAgentInvocationForConversation( inArray(juniorAgentInvocations.status, ["running", "paused"]), ), ) - .orderBy(asc(juniorAgentInvocations.createdAt)) - .limit(2); - if (rows.length > 1) { + .orderBy(asc(juniorAgentInvocations.createdAt)); + return rows.map(invocationFromRow); +} + +/** Resolve the single invocation that may resume for one child conversation. */ +export async function getActiveAgentInvocationForConversation( + childConversationId: string, +): Promise { + const active = + await listActiveAgentInvocationsForConversation(childConversationId); + if (active.length > 1) { throw new Error( `Child conversation ${childConversationId} has multiple active agent invocations`, ); } - return rows[0] ? invocationFromRow(rows[0]) : undefined; + return active[0]; } /** Read the newest agent invocation of one child Conversation. */ @@ -555,35 +534,6 @@ export async function createAgentInvocationFromLatest( return invocation; } -/** Find a direct child agent of one parent by name or child Conversation id. */ -export async function findChildAgentConversationId(args: { - parentConversationId: string; - target: string; -}): Promise { - const binding = await getAgentBinding({ - name: args.target, - parentConversationId: args.parentConversationId, - }); - if (binding) { - return binding.childConversationId; - } - const rows = await getSqlExecutor() - .db() - .select({ childConversationId: juniorAgentInvocations.childConversationId }) - .from(juniorAgentInvocations) - .where( - and( - eq( - juniorAgentInvocations.parentConversationId, - args.parentConversationId, - ), - eq(juniorAgentInvocations.childConversationId, args.target), - ), - ) - .limit(1); - return rows[0]?.childConversationId; -} - /** * Return whether a pending agent invocation still waits for older open work * on its child. diff --git a/packages/junior/src/chat/agent-invocations/work.ts b/packages/junior/src/chat/agent-invocations/work.ts index ba29fb96e9..769d569a8c 100644 --- a/packages/junior/src/chat/agent-invocations/work.ts +++ b/packages/junior/src/chat/agent-invocations/work.ts @@ -76,6 +76,7 @@ import { getAgentInvocationTurnId, getLatestAgentInvocationForConversation, isTerminalAgentInvocation, + listActiveAgentInvocationsForConversation, markAgentInvocationPaused, markAgentInvocationMailboxAppended, markAgentInvocationRunning, @@ -473,6 +474,24 @@ async function projectTerminalTurn( }); } +/** + * End an agent invocation that is still running or paused while other work + * of its child starts. A saved finished Turn gives its result. Otherwise it + * fails. + */ +async function endOtherActiveAgentInvocation( + invocation: AgentInvocation, +): Promise { + return ( + (await projectTerminalTurn(invocation)) ?? + (await completeAgentInvocation({ + invocationId: invocation.invocationId, + errorMessage: "Agent invocation was not resumed before newer work", + status: "failed", + })) + ); +} + /** Classify authorization failures that terminally block background work. */ function blockingInvocationError( error: unknown, @@ -690,6 +709,15 @@ export function createAgentInvocationWorker( await acknowledge(); return { status: "completed" }; } + // A child has at most one running or paused agent invocation. End any + // other active one first, so this one reports for both. + for (const active of await listActiveAgentInvocationsForConversation( + invocation.childConversationId, + )) { + if (active.invocationId !== invocation.invocationId) { + await finish(await endOtherActiveAgentInvocation(active)); + } + } invocation = (await markAgentInvocationRunning(invocation.invocationId)) ?? invocation; diff --git a/packages/junior/tests/integration/agent-invocation-work.test.ts b/packages/junior/tests/integration/agent-invocation-work.test.ts index 07d5638059..f9ccd33319 100644 --- a/packages/junior/tests/integration/agent-invocation-work.test.ts +++ b/packages/junior/tests/integration/agent-invocation-work.test.ts @@ -4,11 +4,15 @@ import { createApp, defineJuniorPlugins } from "@/app"; import type { PiMessage } from "@/chat/pi/messages"; import { completeAgentInvocation, + createAgentInvocation, + createAgentInvocationForMessage, getActiveAgentInvocationForConversation, getAgentInvocation, getAgentInvocationMessageId, getAgentInvocationResultMessageId, getAgentInvocationTurnId, + markAgentInvocationPaused, + markAgentInvocationRunning, } from "@/chat/agent-invocations/store"; import { withAgentInvocationResultActors } from "@/chat/agent-invocations/result"; import { EVENT_MESSAGE_AUTHOR, EVENT_SYSTEM_ACTOR } from "@/chat/events/actor"; @@ -20,7 +24,10 @@ import { CONVERSATION_WORK_MAX_RETRIES, getConversationWorkState, } from "@/chat/task-execution/store"; -import { createAndEnqueueAgentInvocation } from "@/chat/agent-invocations/work"; +import { + createAndEnqueueAgentInvocation, + enqueueAgentInvocation, +} from "@/chat/agent-invocations/work"; import { loadProjection } from "@/chat/conversations/projection"; import { getConversationEventStore } from "@/chat/db"; import { disconnectStateAdapter } from "@/chat/state/adapter"; @@ -639,6 +646,92 @@ describe("agent invocation conversation work", () => { }, ); + // A child with an active invocation left over from earlier work gets new + // mailbox input. Only the new invocation may be active when it runs. + it.each([{ state: "running" as const }, { state: "paused" as const }])( + "ends a leftover $state child invocation before new mailbox input runs", + async ({ state: orphanState }) => { + const { conversationStore, fixture } = await prepareParentConversation(); + const activeAtRun: unknown[] = []; + const { process, queue, state } = await createConversationWorkHarness({ + run: async (run) => { + activeAtRun.push( + await getActiveAgentInvocationForConversation(run.conversationId), + ); + throw new Error("stop after reading the run"); + }, + }); + try { + const orphan = await createAgentInvocation({ + ...invocationInput, + agentName: "researcher", + idempotencyKey: "orphan-1", + parentToolCallId: "call-spawn", + }); + const { childConversationId } = orphan; + await markAgentInvocationRunning(orphan.invocationId); + if (orphanState === "paused") { + await markAgentInvocationPaused(orphan.invocationId); + } + const orphanTurnId = getAgentInvocationTurnId(orphan.invocationId); + await upsertTurnRecord({ + actor: invocationInput.actor, + conversationId: childConversationId, + piMessages: [ + { + role: "user", + content: [{ type: "text", text: invocationInput.input }], + }, + ] as PiMessage[], + sliceId: 1, + source: { kind: "agent_invocation" }, + state: orphanState, + surface: "internal", + turnId: orphanTurnId, + turnStartMessageIndex: 0, + ...(orphanState === "paused" + ? { resumeReason: "yield" as const } + : undefined), + }); + const { invocation: queued } = await createAgentInvocationForMessage({ + actor: invocationInput.actor, + childConversationId, + destination, + idempotencyKey: "parent-turn:call-message", + input: "Also check the retry path.", + parentConversationId, + parentToolCallId: "call-message", + }); + await enqueueAgentInvocation(queued, { + conversationStore, + queue, + state, + }); + + await process(queue.takeMessageFor(childConversationId)); + + expect(activeAtRun).toEqual([ + expect.objectContaining({ + invocationId: queued.invocationId, + status: "running", + }), + ]); + // The queued work reports for both. + await expect( + getAgentInvocation(orphan.invocationId), + ).resolves.toMatchObject({ + parentMailboxStatus: "replaced", + status: "failed", + }); + await expect( + getTurnRecord(childConversationId, orphanTurnId), + ).resolves.not.toMatchObject({ state: orphanState }); + } finally { + await fixture.close(); + } + }, + ); + const missing = "Write access to getsentry/junior. Ask the user."; it.each([ { From 27c66a26a267dc6cbbf97b6b7686927e1c75581f Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Mon, 2 Nov 2026 19:59:01 +0000 Subject: [PATCH 2/5] fix(agent-invocations): Report each ended agent invocation to the parent Newer work of a child does not continue an earlier task, so the parent must get the earlier agent invocation's result on its own. Send it through the failed-result path instead of marking it replaced. Co-Authored-By: David Cramer <23610+dcramer@users.noreply.github.com> --- .../src/chat/agent-invocations/AGENTS.md | 3 +- .../junior/src/chat/agent-invocations/work.ts | 18 ++++-- .../integration/agent-invocation-work.test.ts | 56 +++++++++++++++++-- 3 files changed, 65 insertions(+), 12 deletions(-) diff --git a/packages/junior/src/chat/agent-invocations/AGENTS.md b/packages/junior/src/chat/agent-invocations/AGENTS.md index 9b2ba4b3c0..6d9cd996a0 100644 --- a/packages/junior/src/chat/agent-invocations/AGENTS.md +++ b/packages/junior/src/chat/agent-invocations/AGENTS.md @@ -10,6 +10,7 @@ - Pick the child worker from the stored `parentConversationId`, never from the Turn Source (a child watch-event Turn has an event Source). - A child has at most one `running` or `paused` agent invocation. Before the - worker starts one, it ends every other active one of that child. + worker starts one, it ends every other active one of that child. Each ended + one sends its own result to the parent. - An Agent type (`agent-types.ts`) is data. Read its fields, never its name. Types do not limit tools. diff --git a/packages/junior/src/chat/agent-invocations/work.ts b/packages/junior/src/chat/agent-invocations/work.ts index 769d569a8c..adce601bc9 100644 --- a/packages/junior/src/chat/agent-invocations/work.ts +++ b/packages/junior/src/chat/agent-invocations/work.ts @@ -477,19 +477,25 @@ async function projectTerminalTurn( /** * End an agent invocation that is still running or paused while other work * of its child starts. A saved finished Turn gives its result. Otherwise it - * fails. + * fails. The newer work does not continue this task, so the result goes to + * the parent on its own. */ async function endOtherActiveAgentInvocation( invocation: AgentInvocation, ): Promise { - return ( + const terminal = (await projectTerminalTurn(invocation)) ?? (await completeAgentInvocation({ invocationId: invocation.invocationId, - errorMessage: "Agent invocation was not resumed before newer work", + errorMessage: + "This task was not resumed. Newer work for this child agent started, so this task ended without a result.", status: "failed", - })) - ); + })); + return terminal && + isTerminalAgentInvocation(terminal) && + terminal.parentMailboxStatus === "replaced" + ? await reportReplacedAgentInvocation(terminal.invocationId) + : terminal; } /** Classify authorization failures that terminally block background work. */ @@ -710,7 +716,7 @@ export function createAgentInvocationWorker( return { status: "completed" }; } // A child has at most one running or paused agent invocation. End any - // other active one first, so this one reports for both. + // other active one first. Each one reports its own result. for (const active of await listActiveAgentInvocationsForConversation( invocation.childConversationId, )) { diff --git a/packages/junior/tests/integration/agent-invocation-work.test.ts b/packages/junior/tests/integration/agent-invocation-work.test.ts index f9ccd33319..de58995228 100644 --- a/packages/junior/tests/integration/agent-invocation-work.test.ts +++ b/packages/junior/tests/integration/agent-invocation-work.test.ts @@ -653,6 +653,8 @@ describe("agent invocation conversation work", () => { async ({ state: orphanState }) => { const { conversationStore, fixture } = await prepareParentConversation(); const activeAtRun: unknown[] = []; + // Integration tests do not run the agent. The runner reads the run and + // stops it, so its work retries until the test finishes the invocation. const { process, queue, state } = await createConversationWorkHarness({ run: async (run) => { activeAtRun.push( @@ -661,6 +663,13 @@ describe("agent invocation conversation work", () => { throw new Error("stop after reading the run"); }, }); + const parentMailbox = async () => + ( + await getConversationWorkState({ + conversationId: parentConversationId, + state, + }) + )?.messages ?? []; try { const orphan = await createAgentInvocation({ ...invocationInput, @@ -708,7 +717,9 @@ describe("agent invocation conversation work", () => { state, }); - await process(queue.takeMessageFor(childConversationId)); + await expect( + process(queue.takeMessageFor(childConversationId)), + ).resolves.toMatchObject({ status: "failed" }); expect(activeAtRun).toEqual([ expect.objectContaining({ @@ -716,16 +727,51 @@ describe("agent invocation conversation work", () => { status: "running", }), ]); - // The queued work reports for both. + await expect( + getTurnRecord(childConversationId, orphanTurnId), + ).resolves.not.toMatchObject({ state: orphanState }); + // The newer work does not continue the earlier task, so the earlier + // agent invocation sends its own result before the newer one ends. await expect( getAgentInvocation(orphan.invocationId), ).resolves.toMatchObject({ - parentMailboxStatus: "replaced", + parentMailboxStatus: "appended", status: "failed", }); + const orphanResult = expect.objectContaining({ + inboundMessageId: getAgentInvocationResultMessageId( + orphan.invocationId, + ), + input: expect.objectContaining({ + text: expect.stringMatching( + orphanState === "paused" + ? /failed\.[\s\S]*Error:\nThis task was not resumed\. Newer work for this child agent started/ + : /failed\.[\s\S]*Error:\nAgent invocation lost its worker/, + ), + }), + }); + await expect(parentMailbox()).resolves.toEqual([orphanResult]); + + await completeAgentInvocation({ + invocationId: queued.invocationId, + result: "Retry path checked", + status: "completed", + }); await expect( - getTurnRecord(childConversationId, orphanTurnId), - ).resolves.not.toMatchObject({ state: orphanState }); + process(queue.takeMessageFor(childConversationId)), + ).resolves.toMatchObject({ status: "completed" }); + + await expect(parentMailbox()).resolves.toEqual([ + orphanResult, + expect.objectContaining({ + inboundMessageId: getAgentInvocationResultMessageId( + queued.invocationId, + ), + input: expect.objectContaining({ + text: expect.stringMatching(/Result:\nRetry path checked/), + }), + }), + ]); } finally { await fixture.close(); } From 2c2b369be749e24a9c6385a39b3d5e221b4aed70 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Mon, 2 Nov 2026 20:11:13 +0000 Subject: [PATCH 3/5] docs(agent-invocations): Name both callers of reportReplacedAgentInvocation Co-Authored-By: David Cramer <23610+dcramer@users.noreply.github.com> --- packages/junior/src/chat/agent-invocations/store.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/junior/src/chat/agent-invocations/store.ts b/packages/junior/src/chat/agent-invocations/store.ts index 1d97b59680..162bdbb021 100644 --- a/packages/junior/src/chat/agent-invocations/store.ts +++ b/packages/junior/src/chat/agent-invocations/store.ts @@ -764,8 +764,9 @@ export async function listPendingAgentInvocationResults( } /** - * Send a `replaced` result to the parent now. A stop ended the open work of - * the child, so no later agent invocation sends the report. + * Send a `replaced` result to the parent now, when no later agent invocation + * sends the report. A stop ended the open work of the child, or newer work of + * the child ended this agent invocation and does not continue its task. */ export async function reportReplacedAgentInvocation( invocationId: string, From b35f67c0c2b97d8fa43abe31d5094fbd2131d870 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Mon, 2 Nov 2026 20:17:44 +0000 Subject: [PATCH 4/5] docs(agent-invocations): Keep the invocation invariant out of AGENTS.md The worker comment and JSDoc already state it. --- packages/junior/src/chat/agent-invocations/AGENTS.md | 3 --- 1 file changed, 3 deletions(-) diff --git a/packages/junior/src/chat/agent-invocations/AGENTS.md b/packages/junior/src/chat/agent-invocations/AGENTS.md index 6d9cd996a0..141ed80622 100644 --- a/packages/junior/src/chat/agent-invocations/AGENTS.md +++ b/packages/junior/src/chat/agent-invocations/AGENTS.md @@ -9,8 +9,5 @@ id, so retries and repairs never add a second Turn, invocation, or row. - Pick the child worker from the stored `parentConversationId`, never from the Turn Source (a child watch-event Turn has an event Source). -- A child has at most one `running` or `paused` agent invocation. Before the - worker starts one, it ends every other active one of that child. Each ended - one sends its own result to the parent. - An Agent type (`agent-types.ts`) is data. Read its fields, never its name. Types do not limit tools. From cbcb1aa1144b31e6b0d438fe57c373a616eaca69 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Mon, 2 Nov 2026 20:29:31 +0000 Subject: [PATCH 5/5] fix(agent-invocations): Mark an ended invocation's result pending in one write Ending another active invocation set its result `replaced` and then flipped it to `pending` in a second write. A crash between the writes left it `replaced`, which recovery does not send. `completeAgentInvocation` now takes `reportNow`, so the terminal write marks it `pending` directly. --- .../junior/src/chat/agent-invocations/store.ts | 18 ++++++++++-------- .../junior/src/chat/agent-invocations/work.ts | 15 +++++++-------- 2 files changed, 17 insertions(+), 16 deletions(-) diff --git a/packages/junior/src/chat/agent-invocations/store.ts b/packages/junior/src/chat/agent-invocations/store.ts index 162bdbb021..1d0566f6c0 100644 --- a/packages/junior/src/chat/agent-invocations/store.ts +++ b/packages/junior/src/chat/agent-invocations/store.ts @@ -764,9 +764,8 @@ export async function listPendingAgentInvocationResults( } /** - * Send a `replaced` result to the parent now, when no later agent invocation - * sends the report. A stop ended the open work of the child, or newer work of - * the child ended this agent invocation and does not continue its task. + * Send a `replaced` result to the parent now. A stop ended the open work of + * the child, so no later agent invocation sends the report. */ export async function reportReplacedAgentInvocation( invocationId: string, @@ -860,21 +859,22 @@ export async function markAgentInvocationPaused( * result. The same write marks the result for the parent mailbox. If the * child still has open work, the result is `replaced`: the parent gets one * report, from the agent invocation that finishes the child's work. + * `reportNow` sends the result even then, when that open work does not + * continue this task. */ export async function completeAgentInvocation( - args: + args: ( | { invocationId: string; result: string; status: "completed"; - nowMs?: number; } | { errorMessage: string; invocationId: string; status: "blocked" | "failed" | "stopped"; - nowMs?: number; - }, + } + ) & { nowMs?: number; reportNow?: boolean }, ): Promise { const nowMs = args.nowMs ?? Date.now(); const current = await getAgentInvocation(args.invocationId); @@ -885,7 +885,9 @@ export async function completeAgentInvocation( // each `replaced` result has open work that starts a later invocation. A // stop is always reported, because the stop also stops that open work. const parentMailboxStatus = - args.status !== "stopped" && (await hasOpenChildWork(current)) + !args.reportNow && + args.status !== "stopped" && + (await hasOpenChildWork(current)) ? "replaced" : "pending"; await getSqlExecutor() diff --git a/packages/junior/src/chat/agent-invocations/work.ts b/packages/junior/src/chat/agent-invocations/work.ts index adce601bc9..1cc72ed9e2 100644 --- a/packages/junior/src/chat/agent-invocations/work.ts +++ b/packages/junior/src/chat/agent-invocations/work.ts @@ -422,6 +422,7 @@ async function persistTerminalLifecycle( /** Stop an invocation whose worker died, or project its terminal turn result. */ async function projectTerminalTurn( invocation: AgentInvocation, + options: { reportNow?: boolean } = {}, ): Promise { let turn = await getTurnRecord( invocation.childConversationId, @@ -456,6 +457,7 @@ async function projectTerminalTurn( ? undefined : readBlockedReport(turn.piMessages.slice(turn.turnStartMessageIndex)); return await completeAgentInvocation({ + ...options, invocationId: invocation.invocationId, ...(failed ? { @@ -483,19 +485,16 @@ async function projectTerminalTurn( async function endOtherActiveAgentInvocation( invocation: AgentInvocation, ): Promise { - const terminal = - (await projectTerminalTurn(invocation)) ?? + return ( + (await projectTerminalTurn(invocation, { reportNow: true })) ?? (await completeAgentInvocation({ invocationId: invocation.invocationId, errorMessage: "This task was not resumed. Newer work for this child agent started, so this task ended without a result.", + reportNow: true, status: "failed", - })); - return terminal && - isTerminalAgentInvocation(terminal) && - terminal.parentMailboxStatus === "replaced" - ? await reportReplacedAgentInvocation(terminal.invocationId) - : terminal; + })) + ); } /** Classify authorization failures that terminally block background work. */