fix(agents): preserve subagent wake delivery route

ooiuuii · ooiuuii · commit 3728ae01a850 · 2026-06-08T17:29:28.000+08:00
diff --git a/src/agents/embedded-agent-runner/run-state.ts b/src/agents/embedded-agent-runner/run-state.ts
@@ -6,6 +6,7 @@ import {
   getActiveReplyRunCount,
   listActiveReplyRunSessionKeys,
   listActiveReplyRunSessionIds,
+  type ReplyRunQueueMessageOptions,
 } from "../../auto-reply/reply/reply-run-registry.js";
 import { resolveGlobalSingleton } from "../../shared/global-singleton.js";
 
@@ -26,12 +27,9 @@ export type EmbeddedAgentQueueHandle = {
   sourceReplyDeliveryMode?: SourceReplyDeliveryMode;
 };
 
-export type EmbeddedAgentQueueMessageOptions = {
+export type EmbeddedAgentQueueMessageOptions = ReplyRunQueueMessageOptions & {
   steeringMode?: "all";
   debounceMs?: number;
-  deliveryTimeoutMs?: number;
-  waitForTranscriptCommit?: boolean;
-  sourceReplyDeliveryMode?: SourceReplyDeliveryMode;
 };
 
 export type ActiveEmbeddedRunSnapshot = {
diff --git a/src/agents/embedded-agent-runner/runs.test.ts b/src/agents/embedded-agent-runner/runs.test.ts
@@ -306,7 +306,14 @@ describe("embedded-agent runner run registry", () => {
     const outcome = await queueEmbeddedAgentMessageWithOutcomeAsync(
       "session-reply-run",
       "completion from child",
-      { waitForTranscriptCommit: true },
+      {
+        deliveryContext: {
+          channel: "bncr",
+          to: "Bncr:tgBot:chat:sender",
+          accountId: "Primary",
+        },
+        waitForTranscriptCommit: true,
+      },
     );
 
     expect(outcome.queued).toBe(true);
@@ -321,7 +328,14 @@ describe("embedded-agent runner run registry", () => {
     });
     expect(outcome.enqueuedAtMs).toEqual(expect.any(Number));
     expect(outcome.deliveredAtMs).toBeUndefined();
-    expect(queueMessage).toHaveBeenCalledWith("completion from child");
+    expect(queueMessage).toHaveBeenCalledWith("completion from child", {
+      deliveryContext: {
+        channel: "bncr",
+        to: "Bncr:tgBot:chat:sender",
+        accountId: "Primary",
+      },
+      waitForTranscriptCommit: true,
+    });
   });
 
   it("force-clears an aborted run that does not drain", async () => {
diff --git a/src/agents/embedded-agent-runner/runs.ts b/src/agents/embedded-agent-runner/runs.ts
@@ -372,7 +372,7 @@ function prepareEmbeddedAgentQueueMessage(
 ): PreparedEmbeddedAgentQueueMessage {
   const handle = ACTIVE_EMBEDDED_RUNS.get(sessionId);
   if (!handle) {
-    const queuedReplyRunMessage = queueReplyRunMessage(sessionId, text);
+    const queuedReplyRunMessage = queueReplyRunMessage(sessionId, text, options);
     if (queuedReplyRunMessage) {
       logMessageQueued({ sessionId, source: "embedded-agent-runner" });
       return {
diff --git a/src/agents/subagent-announce-delivery.test.ts b/src/agents/subagent-announce-delivery.test.ts
@@ -634,6 +634,18 @@ describe("deliverSubagentAnnouncement active requester steering", () => {
       accountId?: string;
       threadId?: string | number;
     };
+    completionDirectOrigin?: {
+      channel?: string;
+      to?: string;
+      accountId?: string;
+      threadId?: string | number;
+    };
+    directOrigin?: {
+      channel?: string;
+      to?: string;
+      accountId?: string;
+      threadId?: string | number;
+    };
   }) {
     const callGateway = createGatewayMock();
     let activityChecks = 0;
@@ -673,6 +685,8 @@ describe("deliverSubagentAnnouncement active requester steering", () => {
       triggerMessage: "child done",
       steerMessage: "child done",
       requesterOrigin: params.requesterOrigin,
+      completionDirectOrigin: params.completionDirectOrigin,
+      directOrigin: params.directOrigin,
       requesterIsSubagent: false,
       expectsCompletionMessage: false,
       directIdempotencyKey: "announce-no-external-route",
@@ -727,6 +741,32 @@ describe("deliverSubagentAnnouncement active requester steering", () => {
     expect(callGateway).not.toHaveBeenCalled();
   });
 
+  it("preserves the requester external route when the completion origin is internal", async () => {
+    const queueEmbeddedAgentMessageWithOutcome = createQueueOutcomeMock(true);
+    await deliverSteeredAnnouncement({
+      queueEmbeddedAgentMessageWithOutcome,
+      completionDirectOrigin: {
+        channel: "webchat",
+        to: "session:child-completion",
+        accountId: "webchat-account",
+      },
+      requesterOrigin: {
+        channel: "slack",
+        to: "channel:C123",
+        accountId: "acct-1",
+        threadId: "171.222",
+      },
+    });
+
+    const options = mockCallArg(queueEmbeddedAgentMessageWithOutcome, 0, 2);
+    expectRecordFields(options.deliveryContext, {
+      channel: "slack",
+      to: "channel:C123",
+      accountId: "acct-1",
+      threadId: "171.222",
+    });
+  });
+
   it.each(["followup", "collect", "interrupt"] as const)(
     "steers active requester announces even in %s mode",
     async (mode) => {
@@ -777,6 +817,11 @@ describe("deliverSubagentAnnouncement active requester steering", () => {
       "paperclip-session",
       "child done",
       {
+        deliveryContext: {
+          channel: "slack",
+          to: "channel:C123",
+          accountId: "acct-1",
+        },
         steeringMode: "all",
         debounceMs: 0,
         waitForTranscriptCommit: true,
@@ -788,6 +833,11 @@ describe("deliverSubagentAnnouncement active requester steering", () => {
       "paperclip-session",
       "child done",
       {
+        deliveryContext: {
+          channel: "slack",
+          to: "channel:C123",
+          accountId: "acct-1",
+        },
         steeringMode: "all",
         debounceMs: 0,
         deliveryTimeoutMs: 120_000,
diff --git a/src/agents/subagent-announce-delivery.ts b/src/agents/subagent-announce-delivery.ts
@@ -28,7 +28,11 @@ import {
 import { deriveSessionChatTypeFromKey } from "../sessions/session-chat-type-shared.js";
 import { isCronRunSessionKey, isCronSessionKey } from "../sessions/session-key-utils.js";
 import { isNonTerminalAgentRunStatus } from "../shared/agent-run-status.js";
-import { mergeDeliveryContext, normalizeDeliveryContext } from "../utils/delivery-context.js";
+import {
+  deliveryContextFromSession,
+  mergeDeliveryContext,
+  normalizeDeliveryContext,
+} from "../utils/delivery-context.js";
 import {
   INTERNAL_MESSAGE_CHANNEL,
   isDeliverableMessageChannel,
@@ -596,6 +600,7 @@ export function loadSessionEntryByKey(sessionKey: string) {
 }
 
 async function maybeSteerSubagentAnnounce(params: {
+  deliveryContext?: DeliveryContext;
   deliveryTimeoutMs?: number;
   requesterSessionKey: string;
   steerMessage: string;
@@ -624,7 +629,12 @@ async function maybeSteerSubagentAnnounce(params: {
 
   // Subagent announcements are internal handoffs into an active requester turn.
   // Queue modes such as followup/collect apply to user prompts, not this path.
+  const queueDeliveryContext = resolveActiveWakeDeliveryContext(
+    params.deliveryContext,
+    deliveryContextFromSession(entry),
+  );
   const queueOptions: EmbeddedAgentQueueMessageOptions = {
+    ...(queueDeliveryContext ? { deliveryContext: queueDeliveryContext } : {}),
     deliveryTimeoutMs: params.deliveryTimeoutMs,
     steeringMode: "all",
     ...(queueSettings.debounceMs !== undefined ? { debounceMs: queueSettings.debounceMs } : {}),
@@ -1176,6 +1186,41 @@ function collectMessagingToolDeliveredMediaUrlsForTarget(
   return Array.from(urls);
 }
 
+function isExternalQueueDeliveryContext(context?: DeliveryContext): boolean {
+  const normalized = normalizeDeliveryContext(context);
+  const channel = normalizeMessageChannel(normalized?.channel);
+  return Boolean(
+    normalized?.to &&
+    channel &&
+    isDeliverableMessageChannel(channel) &&
+    !isInternalMessageChannel(channel),
+  );
+}
+
+function resolveActiveWakeDeliveryContext(
+  ...contexts: Array<DeliveryContext | undefined>
+): DeliveryContext | undefined {
+  const normalizedContexts = contexts
+    .map((context) => normalizeDeliveryContext(context))
+    .filter((context): context is DeliveryContext => Boolean(context));
+  const externalContext = normalizedContexts.find((context) =>
+    isExternalQueueDeliveryContext(context),
+  );
+  if (externalContext) {
+    const sameChannelFallback = normalizedContexts.find(
+      (context) =>
+        context !== externalContext &&
+        normalizeMessageChannel(context.channel) ===
+          normalizeMessageChannel(externalContext.channel),
+    );
+    return mergeDeliveryContext(externalContext, sameChannelFallback);
+  }
+  return normalizedContexts.reduce<DeliveryContext | undefined>(
+    (merged, context) => mergeDeliveryContext(merged, context),
+    undefined,
+  );
+}
+
 function stripNonDeliverableChannelForCompletionOrigin(
   context?: DeliveryContext,
 ): DeliveryContext | undefined {
@@ -1680,6 +1725,12 @@ export async function deliverSubagentAnnouncement(params: {
     signal: params.signal,
     steer: async () =>
       await maybeSteerSubagentAnnounce({
+        deliveryContext: resolveActiveWakeDeliveryContext(
+          params.completionDirectOrigin,
+          params.directOrigin,
+          params.requesterOrigin,
+          params.requesterSessionOrigin,
+        ),
         deliveryTimeoutMs: resolveSubagentAnnounceTimeoutMs(
           subagentAnnounceDeliveryDeps.getRuntimeConfig(),
         ),
diff --git a/src/auto-reply/reply/effective-reply-route.test.ts b/src/auto-reply/reply/effective-reply-route.test.ts
@@ -227,6 +227,40 @@ describe("resolveEffectiveReplyRoute", () => {
     });
   });
 
+  it("uses established external route for subagent completion wake handoffs", () => {
+    expect(
+      resolveEffectiveReplyRoute({
+        ctx: ctx({
+          Provider: "webchat",
+          Surface: "webchat",
+          OriginatingChannel: "webchat",
+          OriginatingTo: "session:dashboard",
+          AccountId: "webchat-account",
+          InputProvenance: {
+            kind: "inter_session",
+            sourceTool: "subagent_announce",
+            sourceChannel: "webchat",
+          },
+        }),
+        entry: entry({
+          deliveryContext: {
+            channel: "bncr",
+            to: "Bncr:tgBot:chat:sender",
+            accountId: "Primary",
+          },
+          lastChannel: "webchat",
+          lastTo: "session:dashboard",
+          lastAccountId: "webchat-account",
+        }),
+      }),
+    ).toEqual({
+      channel: "bncr",
+      to: "Bncr:tgBot:chat:sender",
+      accountId: "Primary",
+      inheritedExternalRoute: true,
+    });
+  });
+
   it("keeps normal webchat turns on their live route", () => {
     expect(
       resolveEffectiveReplyRoute({
diff --git a/src/auto-reply/reply/effective-reply-route.ts b/src/auto-reply/reply/effective-reply-route.ts
@@ -1,7 +1,10 @@
 /** Resolves the effective reply route from current context and persisted session route. */
 import type { SessionEntry } from "../../config/sessions/types.js";
 import { stringifyRouteThreadId } from "../../plugin-sdk/channel-route.js";
-import type { InputProvenance } from "../../sessions/input-provenance.js";
+import {
+  shouldPreserveUserFacingSessionStateForInputProvenance,
+  type InputProvenance,
+} from "../../sessions/input-provenance.js";
 import { INTERNAL_MESSAGE_CHANNEL, normalizeMessageChannel } from "../../utils/message-channel.js";
 import type { FinalizedMsgContext } from "../templating.js";
 
@@ -31,11 +34,16 @@ export function isSystemEventProvider(provider?: string): boolean {
   return provider === "heartbeat" || provider === "cron-event" || provider === "exec-event";
 }
 
-function isSessionsSendInterSessionHandoff(inputProvenance: InputProvenance | undefined): boolean {
-  return (
-    inputProvenance?.kind === "inter_session" &&
-    inputProvenance.sourceTool?.toLowerCase() === "sessions_send"
-  );
+function shouldInheritExternalRouteForInterSessionHandoff(
+  inputProvenance: InputProvenance | undefined,
+): boolean {
+  if (inputProvenance?.kind !== "inter_session") {
+    return false;
+  }
+  if (inputProvenance.sourceTool?.toLowerCase() === "sessions_send") {
+    return true;
+  }
+  return shouldPreserveUserFacingSessionStateForInputProvenance(inputProvenance);
 }
 
 function resolveTrustedInheritedThreadId(
@@ -70,7 +78,7 @@ export function resolveEffectiveReplyRoute(params: {
   const persistedDeliveryContext = params.entry?.deliveryContext;
   const persistedDeliveryChannel = normalizeMessageChannel(persistedDeliveryContext?.channel);
   if (
-    isSessionsSendInterSessionHandoff(params.ctx.InputProvenance) &&
+    shouldInheritExternalRouteForInterSessionHandoff(params.ctx.InputProvenance) &&
     currentSurface === INTERNAL_MESSAGE_CHANNEL &&
     persistedDeliveryChannel &&
     persistedDeliveryChannel !== INTERNAL_MESSAGE_CHANNEL &&
diff --git a/src/auto-reply/reply/reply-run-registry.ts b/src/auto-reply/reply/reply-run-registry.ts
@@ -6,18 +6,27 @@ import {
 } from "../../logging/diagnostic-run-activity.js";
 import { resolveGlobalSingleton } from "../../shared/global-singleton.js";
 import { resolveTimerTimeoutMs } from "../../shared/number-coercion.js";
+import type { DeliveryContext } from "../../utils/delivery-context.shared.js";
+import type { SourceReplyDeliveryMode } from "../get-reply-options.types.js";
 
 export type ReplyRunKey = string;
 
 export type ReplyBackendKind = "embedded" | "cli";
 
 export type ReplyBackendCancelReason = "user_abort" | "restart" | "superseded";
 
+export type ReplyRunQueueMessageOptions = {
+  deliveryContext?: DeliveryContext;
+  deliveryTimeoutMs?: number;
+  sourceReplyDeliveryMode?: SourceReplyDeliveryMode;
+  waitForTranscriptCommit?: boolean;
+};
+
 export type ReplyBackendHandle = {
   readonly kind: ReplyBackendKind;
   cancel(reason?: ReplyBackendCancelReason): void;
   isStreaming(): boolean;
-  queueMessage?: (text: string) => Promise<void>;
+  queueMessage?: (text: string, options?: ReplyRunQueueMessageOptions) => Promise<void>;
   /**
    * Compatibility-only hook so legacy "abort compacting runs" paths can still
    * find embedded runs that are compacting during the main run phase.
@@ -526,7 +535,11 @@ export function isReplyRunStreamingForSessionId(sessionId: string): boolean {
   return getAttachedBackend(operation)?.isStreaming() ?? false;
 }
 
-export function queueReplyRunMessage(sessionId: string, text: string): boolean {
+export function queueReplyRunMessage(
+  sessionId: string,
+  text: string,
+  options?: ReplyRunQueueMessageOptions,
+): boolean {
   const operation = resolveReplyRunForCurrentSessionId(sessionId);
   const backend = operation ? getAttachedBackend(operation) : undefined;
   if (!operation || operation.phase !== "running" || !backend?.queueMessage) {
@@ -535,7 +548,11 @@ export function queueReplyRunMessage(sessionId: string, text: string): boolean {
   if (!backend.isStreaming()) {
     return false;
   }
-  void backend.queueMessage(text);
+  if (options === undefined) {
+    void backend.queueMessage(text);
+  } else {
+    void backend.queueMessage(text, options);
+  }
   return true;
 }
 
