fix(agent-invocations): Keep one active agent invocation per child

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<AgentBinding | undefined> {
+  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<string | undefined> {
+  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;
+}

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

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<AgentBinding | undefined> {
-  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<AgentInvocation | undefined> {
+): Promise<AgentInvocation[]> {
   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<AgentInvocation | undefined> {
+  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<string | undefined> {
-  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.
@@ -909,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<AgentInvocation | undefined> {
   const nowMs = args.nowMs ?? Date.now();
   const current = await getAgentInvocation(args.invocationId);
@@ -934,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()

packages/junior/src/chat/agent-invocations/work.ts

@@ -76,6 +76,7 @@ import {
   getAgentInvocationTurnId,
   getLatestAgentInvocationForConversation,
   isTerminalAgentInvocation,
+  listActiveAgentInvocationsForConversation,
   markAgentInvocationPaused,
   markAgentInvocationMailboxAppended,
   markAgentInvocationRunning,
@@ -421,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<AgentInvocation | undefined> {
   let turn = await getTurnRecord(
     invocation.childConversationId,
@@ -455,6 +457,7 @@ async function projectTerminalTurn(
       ? undefined
       : readBlockedReport(turn.piMessages.slice(turn.turnStartMessageIndex));
   return await completeAgentInvocation({
+    ...options,
     invocationId: invocation.invocationId,
     ...(failed
       ? {
@@ -473,6 +476,27 @@ 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. The newer work does not continue this task, so the result goes to
+ * the parent on its own.
+ */
+async function endOtherActiveAgentInvocation(
+  invocation: AgentInvocation,
+): Promise<AgentInvocation | undefined> {
+  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",
+    }))
+  );
+}
+
 /** Classify authorization failures that terminally block background work. */
 function blockingInvocationError(
   error: unknown,
@@ -690,6 +714,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. Each one reports its own result.
+      for (const active of await listActiveAgentInvocationsForConversation(
+        invocation.childConversationId,
+      )) {
+        if (active.invocationId !== invocation.invocationId) {
+          await finish(await endOtherActiveAgentInvocation(active));
+        }
+      }
       invocation =
         (await markAgentInvocationRunning(invocation.invocationId)) ??
         invocation;