From 6cb06f5fbcf5cfaf25ca9a90ef3921b0fb730744 Mon Sep 17 00:00:00 2001 From: Ted Li Date: Sun, 31 May 2026 20:43:05 -0700 Subject: [PATCH] fix(reply): preserve sessions_send external routes (#88803) * fix(reply): preserve sessions_send external routes * fix(reply): preserve inherited route thread ids * fix(reply): keep sessions_send delivery single-owner * fix(reply): satisfy dispatch route lint * fix(reply): preserve inherited ACP route metadata * test(reply): type inherited route event assertions * test(ci): satisfy current lint rules * fix(reply): avoid stale inherited route threads * fix(reply): trust explicit inherited route threads * fix(reply): require trusted route thread sources --------- Co-authored-by: Peter Steinberger --- src/agents/openclaw-tools.sessions.test.ts | 5 + src/agents/tools/agent-step.test.ts | 5 +- src/agents/tools/agent-step.ts | 2 + src/agents/tools/sessions-send-tool.ts | 7 + .../reply/dispatch-acp-delivery.test.ts | 26 +++ src/auto-reply/reply/dispatch-acp-delivery.ts | 16 +- src/auto-reply/reply/dispatch-acp.ts | 4 + .../reply/dispatch-from-config.test.ts | 68 ++++++ src/auto-reply/reply/dispatch-from-config.ts | 15 +- .../reply/effective-reply-route.test.ts | 220 ++++++++++++++++++ src/auto-reply/reply/effective-reply-route.ts | 59 ++++- src/plugin-sdk/acp-runtime-backend.ts | 2 + src/plugins/hook-types.ts | 2 + 13 files changed, 419 insertions(+), 12 deletions(-) diff --git a/src/agents/openclaw-tools.sessions.test.ts b/src/agents/openclaw-tools.sessions.test.ts index 3ea858e740eb..6b19d8911a8d 100644 --- a/src/agents/openclaw-tools.sessions.test.ts +++ b/src/agents/openclaw-tools.sessions.test.ts @@ -1481,6 +1481,7 @@ describe("sessions tools", () => { isStreaming: () => true, isCompacting: () => false, supportsTranscriptCommitWait: true, + sourceReplyDeliveryMode: "message_tool_only", abort: () => {}, }, runScopedCallerKey, @@ -1533,6 +1534,7 @@ describe("sessions tools", () => { debounceMs: 0, deliveryTimeoutMs: 30_000, waitForTranscriptCommit: true, + sourceReplyDeliveryMode: "message_tool_only", }); await vi.waitFor(() => { @@ -1587,6 +1589,7 @@ describe("sessions tools", () => { queueMessage, isStreaming: () => true, isCompacting: () => false, + sourceReplyDeliveryMode: "message_tool_only", abort: () => {}, }, runScopedCallerKey, @@ -1622,6 +1625,7 @@ describe("sessions tools", () => { steeringMode: "all", debounceMs: 0, deliveryTimeoutMs: 30_000, + sourceReplyDeliveryMode: "message_tool_only", }); expect(calls.some((call) => call.method === "agent")).toBe(false); }); @@ -1638,6 +1642,7 @@ describe("sessions tools", () => { isStreaming: () => true, isCompacting: () => false, supportsTranscriptCommitWait: true, + sourceReplyDeliveryMode: "message_tool_only", abort: () => {}, }, runScopedCallerKey, diff --git a/src/agents/tools/agent-step.test.ts b/src/agents/tools/agent-step.test.ts index dc02470894bc..1bf24e508a6f 100644 --- a/src/agents/tools/agent-step.test.ts +++ b/src/agents/tools/agent-step.test.ts @@ -52,6 +52,7 @@ describe("runAgentStep", () => { message?: string; sessionKey?: string; deliver?: boolean; + sourceReplyDeliveryMode?: string; lane?: string; inputProvenance?: { kind?: string; sourceTool?: string }; } @@ -59,6 +60,7 @@ describe("runAgentStep", () => { expect(params?.message).toContain("[Inter-session message"); expect(params?.sessionKey).toBe("agent:main:subagent:child"); expect(params?.deliver).toBe(false); + expect(params?.sourceReplyDeliveryMode).toBe("message_tool_only"); expect(params?.lane).toBe("nested:agent:main:subagent:child"); expect(params?.inputProvenance?.kind).toBe("inter_session"); expect(params?.inputProvenance?.sourceTool).toBe("sessions_send"); @@ -119,10 +121,11 @@ describe("runAgentStep", () => { expect(gatewayCalls).toStrictEqual([]); expect(agentCommandFromIngress).toHaveBeenCalledTimes(1); const ingressCalls = agentCommandFromIngress.mock.calls as unknown as Array< - [{ message?: string; transcriptMessage?: string }] + [{ message?: string; sourceReplyDeliveryMode?: string; transcriptMessage?: string }] >; const ingress = ingressCalls[0]?.[0]; expect(ingress?.message).toContain("internal announce step"); + expect(ingress?.sourceReplyDeliveryMode).toBe("message_tool_only"); expect(ingress?.transcriptMessage).toBe(""); }); }); diff --git a/src/agents/tools/agent-step.ts b/src/agents/tools/agent-step.ts index ccd399da4721..3f1874b67b40 100644 --- a/src/agents/tools/agent-step.ts +++ b/src/agents/tools/agent-step.ts @@ -69,6 +69,7 @@ export async function runAgentStep(params: { transcriptMessage: params.transcriptMessage, sessionKey: params.sessionKey, deliver: false, + sourceReplyDeliveryMode: "message_tool_only", channel, lane, runId: stepIdem, @@ -89,6 +90,7 @@ export async function runAgentStep(params: { sessionKey: params.sessionKey, idempotencyKey: stepIdem, deliver: false, + sourceReplyDeliveryMode: "message_tool_only", channel, lane, extraSystemPrompt: params.extraSystemPrompt, diff --git a/src/agents/tools/sessions-send-tool.ts b/src/agents/tools/sessions-send-tool.ts index 87786ea5c468..c8d80963febc 100644 --- a/src/agents/tools/sessions-send-tool.ts +++ b/src/agents/tools/sessions-send-tool.ts @@ -226,11 +226,17 @@ async function startAgentRun(params: { const messageText = typeof params.sendParams.message === "string" ? params.sendParams.message : undefined; if (activeRunSessionId && fallbackSessionKey && messageText) { + const sourceReplyDeliveryMode = + params.sendParams.sourceReplyDeliveryMode === "automatic" || + params.sendParams.sourceReplyDeliveryMode === "message_tool_only" + ? params.sendParams.sourceReplyDeliveryMode + : undefined; const queueOptions: EmbeddedAgentQueueMessageOptions = { steeringMode: "all", debounceMs: 0, deliveryTimeoutMs: params.deliveryTimeoutMs, waitForTranscriptCommit: true, + ...(sourceReplyDeliveryMode ? { sourceReplyDeliveryMode } : {}), }; let queueOutcome = await queueEmbeddedAgentMessageWithOutcomeAsync( activeRunSessionId, @@ -557,6 +563,7 @@ export function createSessionsSendTool(opts?: { sessionKey: resolvedKey, idempotencyKey, deliver: false, + sourceReplyDeliveryMode: "message_tool_only" as const, channel: INTERNAL_MESSAGE_CHANNEL, lane: resolveNestedAgentLaneForSession(resolvedKey), extraSystemPrompt: agentMessageContext, diff --git a/src/auto-reply/reply/dispatch-acp-delivery.test.ts b/src/auto-reply/reply/dispatch-acp-delivery.test.ts index 65a75fcaa753..92181e52a604 100644 --- a/src/auto-reply/reply/dispatch-acp-delivery.test.ts +++ b/src/auto-reply/reply/dispatch-acp-delivery.test.ts @@ -780,6 +780,32 @@ describe("createAcpDispatchDeliveryCoordinator", () => { expect(routeParams.threadId).toBe("101.000"); }); + it("uses inherited account and thread metadata for routed ACP replies", async () => { + const coordinator = createAcpDispatchDeliveryCoordinator({ + cfg: createAcpTestConfig(), + ctx: buildTestCtx({ + Provider: "webchat", + Surface: "webchat", + SessionKey: "agent:main:feishu:direct:ou_123", + }), + dispatcher: createDispatcher(), + inboundAudio: false, + shouldRouteToOriginating: true, + originatingChannel: "feishu", + originatingTo: "user:ou_123", + originatingAccountId: "work", + originatingThreadId: "thread:om_123", + }); + + await coordinator.deliver("block", { text: "hello" }, { skipTts: true }); + + const [[routeParams]] = deliveryMocks.routeReply.mock.calls as unknown as Array< + [{ accountId?: string; threadId?: string | number }] + >; + expect(routeParams.accountId).toBe("work"); + expect(routeParams.threadId).toBe("thread:om_123"); + }); + it("routes ACP replies when cfg.channels is missing", async () => { await expectVisibleChatBlockRoutesToAccount({} as OpenClawConfig, undefined); }); diff --git a/src/auto-reply/reply/dispatch-acp-delivery.ts b/src/auto-reply/reply/dispatch-acp-delivery.ts index ec5e41e498b5..274ebd67696c 100644 --- a/src/auto-reply/reply/dispatch-acp-delivery.ts +++ b/src/auto-reply/reply/dispatch-acp-delivery.ts @@ -192,6 +192,8 @@ export function createAcpDispatchDeliveryCoordinator(params: { shouldRouteToOriginating: boolean; originatingChannel?: string; originatingTo?: string; + originatingAccountId?: string; + originatingThreadId?: string | number; onReplyStart?: () => Promise | void; abortSignal?: AbortSignal; runId?: string; @@ -199,7 +201,9 @@ export function createAcpDispatchDeliveryCoordinator(params: { const directChannel = normalizeOptionalLowercaseString(params.ctx.Provider ?? params.ctx.Surface); const routedChannel = normalizeOptionalLowercaseString(params.originatingChannel); const deliverySessionKey = normalizeOptionalString(params.sessionKey) ?? params.ctx.SessionKey; - const explicitAccountId = normalizeOptionalString(params.ctx.AccountId); + const explicitAccountId = + normalizeOptionalString(params.originatingAccountId) ?? + normalizeOptionalString(params.ctx.AccountId); const resolvedAccountId = explicitAccountId ?? normalizeOptionalString( @@ -404,10 +408,12 @@ export function createAcpDispatchDeliveryCoordinator(params: { routed: true, }); const { routeReply } = await loadRouteReplyRuntime(); - const threadId = resolveRoutedDeliveryThreadId({ - ctx: params.ctx, - sessionKey: deliverySessionKey, - }); + const threadId = + params.originatingThreadId ?? + resolveRoutedDeliveryThreadId({ + ctx: params.ctx, + sessionKey: deliverySessionKey, + }); const result = await routeReply({ payload: ttsPayload, channel: params.originatingChannel, diff --git a/src/auto-reply/reply/dispatch-acp.ts b/src/auto-reply/reply/dispatch-acp.ts index d4c4441f161e..f8fa2bdc74df 100644 --- a/src/auto-reply/reply/dispatch-acp.ts +++ b/src/auto-reply/reply/dispatch-acp.ts @@ -371,6 +371,8 @@ export async function tryDispatchAcpReply(params: { shouldRouteToOriginating: boolean; originatingChannel?: string; originatingTo?: string; + originatingAccountId?: string; + originatingThreadId?: string | number; shouldSendToolSummaries: boolean; shouldSendToolSummariesNow?: () => boolean; bypassForCommand: boolean; @@ -427,6 +429,8 @@ export async function tryDispatchAcpReply(params: { shouldRouteToOriginating: params.shouldRouteToOriginating, originatingChannel: params.originatingChannel, originatingTo: params.originatingTo, + originatingAccountId: params.originatingAccountId, + originatingThreadId: params.originatingThreadId, onReplyStart: params.onReplyStart, abortSignal: params.abortSignal, runId: params.runId, diff --git a/src/auto-reply/reply/dispatch-from-config.test.ts b/src/auto-reply/reply/dispatch-from-config.test.ts index 08efd40cb064..a09b4c52a7ff 100644 --- a/src/auto-reply/reply/dispatch-from-config.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.test.ts @@ -1351,7 +1351,9 @@ describe("dispatchReplyFromConfig", () => { const replyDispatchCall = firstMockCall(hookMocks.runner.runReplyDispatch, "reply dispatch") as | [ { + originatingAccountId?: unknown; originatingChannel?: unknown; + originatingThreadId?: unknown; originatingTo?: unknown; shouldRouteToOriginating?: unknown; }, @@ -1364,6 +1366,72 @@ describe("dispatchReplyFromConfig", () => { expect(typeof replyDispatchCall?.[1]).toBe("object"); }); + it("routes sessions_send internal webchat handoffs through persisted external delivery context", async () => { + setNoAbort(); + mocks.routeReply.mockClear(); + sessionStoreMocks.currentEntry = { + route: { + channel: "feishu", + accountId: "work", + target: { to: "user:ou_123" }, + thread: { id: "thread:om_123", source: "explicit" }, + }, + deliveryContext: { + channel: "feishu", + to: "user:ou_123", + accountId: "work", + threadId: "thread:om_123", + }, + lastChannel: "feishu", + lastTo: "user:ou_123", + lastAccountId: "work", + }; + const cfg = emptyConfig; + const dispatcher = createDispatcher(); + const ctx = buildTestCtx({ + Provider: "webchat", + Surface: "webchat", + SessionKey: "agent:main:feishu:direct:ou_123", + AccountId: undefined, + OriginatingChannel: "webchat", + OriginatingTo: "session:dashboard", + InputProvenance: { + kind: "inter_session", + sourceTool: "sessions_send", + sourceChannel: "webchat", + }, + }); + + const replyResolver = async () => ({ text: "hi" }) satisfies ReplyPayload; + await dispatchReplyFromConfig({ ctx, cfg, dispatcher, replyResolver }); + + expect(dispatcher.sendFinalReply).not.toHaveBeenCalled(); + const routeCall = firstRouteReplyCall() as + | { accountId?: unknown; channel?: unknown; threadId?: unknown; to?: unknown } + | undefined; + expect(routeCall?.channel).toBe("feishu"); + expect(routeCall?.to).toBe("user:ou_123"); + expect(routeCall?.accountId).toBe("work"); + expect(routeCall?.threadId).toBe("thread:om_123"); + const replyDispatchCall = firstMockCall(hookMocks.runner.runReplyDispatch, "reply dispatch") as + | [ + { + originatingAccountId?: unknown; + originatingChannel?: unknown; + originatingThreadId?: unknown; + originatingTo?: unknown; + shouldRouteToOriginating?: unknown; + }, + unknown, + ] + | undefined; + expect(replyDispatchCall?.[0]?.shouldRouteToOriginating).toBe(true); + expect(replyDispatchCall?.[0]?.originatingChannel).toBe("feishu"); + expect(replyDispatchCall?.[0]?.originatingTo).toBe("user:ou_123"); + expect(replyDispatchCall?.[0]?.originatingAccountId).toBe("work"); + expect(replyDispatchCall?.[0]?.originatingThreadId).toBe("thread:om_123"); + }); + it("routes exec-event replies using last route fields when delivery context is missing", async () => { setNoAbort(); mocks.routeReply.mockClear(); diff --git a/src/auto-reply/reply/dispatch-from-config.ts b/src/auto-reply/reply/dispatch-from-config.ts index 4606647e7b7a..b9714a7b9eae 100644 --- a/src/auto-reply/reply/dispatch-from-config.ts +++ b/src/auto-reply/reply/dispatch-from-config.ts @@ -1202,6 +1202,9 @@ export async function dispatchReplyFromConfig( ctx, sessionKey: acpDispatchSessionKey, }); + // Inherited sessions_send routes carry thread ids only when the stored route + // proves the thread came from an explicit target, not session normalization. + const routeReplyThreadId = replyRoute.threadId ?? routeThreadId; const inboundAudio = isInboundAudioContext(ctx); const sessionTtsAuto = normalizeTtsAutoMode(sessionStoreEntry.entry?.ttsAuto); const workspaceDir = resolveAgentWorkspaceDir(cfg, sessionAgentId); @@ -1400,10 +1403,12 @@ export async function dispatchReplyFromConfig( const normalizedProviderChannel = normalizeMessageChannel(ctx.Provider); const normalizedSurfaceChannel = normalizeMessageChannel(ctx.Surface); const normalizedCurrentSurface = normalizedProviderChannel ?? normalizedSurfaceChannel; + const effectiveExplicitDeliverRoute = + ctx.ExplicitDeliverRoute === true || replyRoute.inheritedExternalRoute === true; const isInternalWebchatTurn = normalizedCurrentSurface === INTERNAL_MESSAGE_CHANNEL && (normalizedSurfaceChannel === INTERNAL_MESSAGE_CHANNEL || !normalizedSurfaceChannel) && - ctx.ExplicitDeliverRoute !== true; + !effectiveExplicitDeliverRoute; const hasRouteReplyCandidate = Boolean( !suppressAcpChildUserDelivery && !isInternalWebchatTurn && @@ -1420,7 +1425,7 @@ export async function dispatchReplyFromConfig( } = resolveReplyRoutingDecision({ provider: ctx.Provider, surface: ctx.Surface, - explicitDeliverRoute: ctx.ExplicitDeliverRoute, + explicitDeliverRoute: effectiveExplicitDeliverRoute, originatingChannel: replyRoute.channel, originatingTo: replyRoute.to, suppressDirectUserDelivery: suppressAcpChildUserDelivery, @@ -1489,7 +1494,7 @@ export async function dispatchReplyFromConfig( requesterSenderName: ctx.SenderName, requesterSenderUsername: ctx.SenderUsername, requesterSenderE164: ctx.SenderE164, - threadId: routeThreadId, + threadId: routeReplyThreadId, cfg, abortSignal: options?.abortSignal, mirror: options?.mirror, @@ -2161,6 +2166,8 @@ export async function dispatchReplyFromConfig( shouldRouteToOriginating, originatingChannel: routeReplyChannel, originatingTo: routeReplyTo, + originatingAccountId: replyRoute.accountId, + originatingThreadId: routeReplyThreadId, shouldSendToolSummaries, sendPolicy, }), @@ -2797,6 +2804,8 @@ export async function dispatchReplyFromConfig( shouldRouteToOriginating, originatingChannel: routeReplyChannel, originatingTo: routeReplyTo, + originatingAccountId: replyRoute.accountId, + originatingThreadId: routeReplyThreadId, shouldSendToolSummaries, sendPolicy, isTailDispatch: true, diff --git a/src/auto-reply/reply/effective-reply-route.test.ts b/src/auto-reply/reply/effective-reply-route.test.ts index 4f13ae522dee..b7edcd7ab7a1 100644 --- a/src/auto-reply/reply/effective-reply-route.test.ts +++ b/src/auto-reply/reply/effective-reply-route.test.ts @@ -59,6 +59,226 @@ describe("resolveEffectiveReplyRoute", () => { }); }); + it("uses established external route for sessions_send internal webchat handoffs", () => { + expect( + resolveEffectiveReplyRoute({ + ctx: ctx({ + Provider: "webchat", + Surface: "webchat", + OriginatingChannel: "webchat", + OriginatingTo: "session:dashboard", + AccountId: "webchat-account", + InputProvenance: { + kind: "inter_session", + sourceTool: "sessions_send", + sourceChannel: "webchat", + }, + }), + entry: entry({ + deliveryContext: { + channel: "feishu", + to: "user:ou_123", + accountId: "work", + threadId: "thread:om_123", + }, + lastChannel: "webchat", + lastTo: "session:dashboard", + lastAccountId: "webchat-account", + }), + }), + ).toEqual({ + channel: "feishu", + to: "user:ou_123", + accountId: "work", + inheritedExternalRoute: true, + }); + }); + + it("keeps trusted inherited thread ids from explicit route metadata", () => { + expect( + resolveEffectiveReplyRoute({ + ctx: ctx({ + Provider: "webchat", + Surface: "webchat", + InputProvenance: { + kind: "inter_session", + sourceTool: "sessions_send", + }, + }), + entry: entry({ + route: { + channel: "feishu", + accountId: "work", + target: { to: "user:ou_123" }, + thread: { id: "thread:om_123", source: "explicit" }, + }, + deliveryContext: { + channel: "feishu", + to: "user:ou_123", + accountId: "work", + threadId: "thread:om_123", + }, + }), + }), + ).toEqual({ + channel: "feishu", + to: "user:ou_123", + accountId: "work", + threadId: "thread:om_123", + inheritedExternalRoute: true, + }); + }); + + it("drops inherited thread ids from session-normalized route metadata", () => { + expect( + resolveEffectiveReplyRoute({ + ctx: ctx({ + Provider: "webchat", + Surface: "webchat", + InputProvenance: { + kind: "inter_session", + sourceTool: "sessions_send", + }, + }), + entry: entry({ + route: { + channel: "feishu", + accountId: "work", + target: { to: "user:ou_123" }, + thread: { id: "thread:stale", source: "session" }, + }, + deliveryContext: { + channel: "feishu", + to: "user:ou_123", + accountId: "work", + threadId: "thread:stale", + }, + }), + }), + ).toEqual({ + channel: "feishu", + to: "user:ou_123", + accountId: "work", + inheritedExternalRoute: true, + }); + }); + + it("drops inherited thread ids from unmarked normalized route metadata", () => { + expect( + resolveEffectiveReplyRoute({ + ctx: ctx({ + Provider: "webchat", + Surface: "webchat", + InputProvenance: { + kind: "inter_session", + sourceTool: "sessions_send", + }, + }), + entry: entry({ + route: { + channel: "feishu", + accountId: "work", + target: { to: "user:ou_123" }, + thread: { id: "thread:stale" }, + }, + deliveryContext: { + channel: "feishu", + to: "user:ou_123", + accountId: "work", + threadId: "thread:stale", + }, + }), + }), + ).toEqual({ + channel: "feishu", + to: "user:ou_123", + accountId: "work", + inheritedExternalRoute: true, + }); + }); + + it("keeps plugin-owned external routes for runtime routability checks", () => { + expect( + resolveEffectiveReplyRoute({ + ctx: ctx({ + Provider: "webchat", + Surface: "webchat", + OriginatingChannel: "webchat", + OriginatingTo: "session:dashboard", + InputProvenance: { + kind: "inter_session", + sourceTool: "sessions_send", + }, + }), + entry: entry({ + deliveryContext: { + channel: "customer-chat", + to: "conversation:123", + accountId: "workspace-a", + }, + }), + }), + ).toEqual({ + channel: "customer-chat", + to: "conversation:123", + accountId: "workspace-a", + inheritedExternalRoute: true, + }); + }); + + it("keeps normal webchat turns on their live route", () => { + expect( + resolveEffectiveReplyRoute({ + ctx: ctx({ + Provider: "webchat", + Surface: "webchat", + OriginatingChannel: "webchat", + OriginatingTo: "session:dashboard", + }), + entry: entry({ + deliveryContext: { + channel: "feishu", + to: "user:ou_123", + accountId: "work", + }, + }), + }), + ).toEqual({ + channel: "webchat", + to: "session:dashboard", + accountId: undefined, + }); + }); + + it("ignores persisted webchat routes for sessions_send handoffs", () => { + expect( + resolveEffectiveReplyRoute({ + ctx: ctx({ + Provider: "webchat", + Surface: "webchat", + OriginatingChannel: "webchat", + OriginatingTo: "session:dashboard", + InputProvenance: { + kind: "inter_session", + sourceTool: "sessions_send", + }, + }), + entry: entry({ + deliveryContext: { + channel: "webchat", + to: "session:old-dashboard", + }, + lastChannel: "webchat", + lastTo: "session:old-dashboard", + }), + }), + ).toEqual({ + channel: "webchat", + to: "session:dashboard", + accountId: undefined, + }); + }); + it("prefers live origin context for exec-event replies", () => { expect( resolveEffectiveReplyRoute({ diff --git a/src/auto-reply/reply/effective-reply-route.ts b/src/auto-reply/reply/effective-reply-route.ts index 8dee0852bde1..c2a8e8b975dc 100644 --- a/src/auto-reply/reply/effective-reply-route.ts +++ b/src/auto-reply/reply/effective-reply-route.ts @@ -1,30 +1,84 @@ 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 { INTERNAL_MESSAGE_CHANNEL, normalizeMessageChannel } from "../../utils/message-channel.js"; import type { FinalizedMsgContext } from "../templating.js"; export type EffectiveReplyRouteContext = Pick< FinalizedMsgContext, - "Provider" | "OriginatingChannel" | "OriginatingTo" | "AccountId" + "Provider" | "Surface" | "OriginatingChannel" | "OriginatingTo" | "AccountId" | "InputProvenance" >; export type EffectiveReplyRouteEntry = Pick< SessionEntry, - "deliveryContext" | "lastChannel" | "lastTo" | "lastAccountId" + "deliveryContext" | "lastChannel" | "lastTo" | "lastAccountId" | "route" >; export type EffectiveReplyRoute = { channel?: string; to?: string; accountId?: string; + threadId?: string | number; + inheritedExternalRoute?: boolean; }; 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 resolveTrustedInheritedThreadId( + entry: EffectiveReplyRouteEntry | undefined, +): string | number | undefined { + const deliveryThreadId = entry?.deliveryContext?.threadId; + if (deliveryThreadId == null) { + return undefined; + } + const routeThread = entry?.route?.thread; + if ( + routeThread?.id != null && + (routeThread.source === "explicit" || + routeThread.source === "target" || + routeThread.source === "turn") && + stringifyRouteThreadId(routeThread.id) === stringifyRouteThreadId(deliveryThreadId) + ) { + return deliveryThreadId; + } + return undefined; +} + export function resolveEffectiveReplyRoute(params: { ctx: EffectiveReplyRouteContext; entry?: EffectiveReplyRouteEntry; }): EffectiveReplyRoute { + const currentSurface = + normalizeMessageChannel(params.ctx.Provider) ?? + normalizeMessageChannel(params.ctx.Surface) ?? + normalizeMessageChannel(params.ctx.OriginatingChannel); + const persistedDeliveryContext = params.entry?.deliveryContext; + const persistedDeliveryChannel = normalizeMessageChannel(persistedDeliveryContext?.channel); + if ( + isSessionsSendInterSessionHandoff(params.ctx.InputProvenance) && + currentSurface === INTERNAL_MESSAGE_CHANNEL && + persistedDeliveryChannel && + persistedDeliveryChannel !== INTERNAL_MESSAGE_CHANNEL && + persistedDeliveryContext?.to + ) { + const inheritedThreadId = resolveTrustedInheritedThreadId(params.entry); + return { + channel: persistedDeliveryChannel, + to: persistedDeliveryContext.to, + accountId: persistedDeliveryContext.accountId, + ...(inheritedThreadId !== undefined ? { threadId: inheritedThreadId } : {}), + inheritedExternalRoute: true, + }; + } if (!isSystemEventProvider(params.ctx.Provider)) { return { channel: params.ctx.OriginatingChannel, @@ -32,7 +86,6 @@ export function resolveEffectiveReplyRoute(params: { accountId: params.ctx.AccountId, }; } - const persistedDeliveryContext = params.entry?.deliveryContext; return { channel: params.ctx.OriginatingChannel ?? diff --git a/src/plugin-sdk/acp-runtime-backend.ts b/src/plugin-sdk/acp-runtime-backend.ts index d757b68bd190..d736e3d7b751 100644 --- a/src/plugin-sdk/acp-runtime-backend.ts +++ b/src/plugin-sdk/acp-runtime-backend.ts @@ -84,6 +84,8 @@ export async function tryDispatchAcpReplyHook( shouldRouteToOriginating: event.shouldRouteToOriginating, originatingChannel: event.originatingChannel, originatingTo: event.originatingTo, + originatingAccountId: event.originatingAccountId, + originatingThreadId: event.originatingThreadId, shouldSendToolSummaries: event.shouldSendToolSummaries, shouldSendToolSummariesNow: () => event.shouldSendToolSummaries, bypassForCommand, diff --git a/src/plugins/hook-types.ts b/src/plugins/hook-types.ts index e00c53811a89..125c8944dd90 100644 --- a/src/plugins/hook-types.ts +++ b/src/plugins/hook-types.ts @@ -447,6 +447,8 @@ export type PluginHookReplyDispatchEvent = { shouldRouteToOriginating: boolean; originatingChannel?: string; originatingTo?: string; + originatingAccountId?: string; + originatingThreadId?: string | number; shouldSendToolSummaries: boolean; sendPolicy: "allow" | "deny"; isTailDispatch?: boolean;