diff --git a/src/auto-reply/reply/dispatch-from-config.test.ts b/src/auto-reply/reply/dispatch-from-config.test.ts index 5a55bebcad43..d0b27293655f 100644 --- a/src/auto-reply/reply/dispatch-from-config.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.test.ts @@ -480,6 +480,16 @@ vi.mock("../../infra/outbound/session-binding-service.js", () => ({ unbind: vi.fn(async () => []), }), })); +vi.mock("../../bindings/records.js", () => ({ + resolveConversationBindingRecord: (conversation: { + channel: string; + accountId: string; + conversationId: string; + parentConversationId?: string; + }) => sessionBindingMocks.resolveByConversation(conversation), + touchConversationBindingRecord: (...args: [bindingId: string, at?: number]) => + sessionBindingMocks.touch(...args), +})); vi.mock("../../infra/agent-events.js", () => ({ emitAgentEvent: (params: unknown) => agentEventMocks.emitAgentEvent(params), onAgentEvent: (listener: unknown) => agentEventMocks.onAgentEvent(listener), @@ -594,6 +604,11 @@ const automaticGroupReplyConfig = { }, }, } as const satisfies OpenClawConfig; +const automaticDirectReplyConfig = { + messages: { + visibleReplies: "automatic", + }, +} as const satisfies OpenClawConfig; let dispatchReplyFromConfig: typeof import("./dispatch-from-config.js").dispatchReplyFromConfig; let dispatchFromConfigTesting: typeof import("./dispatch-from-config.js").testing; let resetInboundDedupe: typeof import("./inbound-dedupe.js").resetInboundDedupe; @@ -854,6 +869,28 @@ function firstRouteReplyCall(): Record { return call as Record; } +function installThreadingTestPlugin(params: { defaultAccountId?: string; id: string }) { + const plugin = createChannelTestPluginBase({ id: params.id }); + const defaultAccountId = params.defaultAccountId; + setActivePluginRegistry( + createTestRegistry([ + { + pluginId: params.id, + source: "test", + plugin: { + ...plugin, + config: defaultAccountId + ? { ...plugin.config, defaultAccountId: () => defaultAccountId } + : plugin.config, + threading: { + resolveReplyToMode: () => "all", + }, + }, + }, + ]), + ); +} + function requireToolResultHandler( handler: GetReplyOptions["onToolResult"] | undefined, ): NonNullable { @@ -945,6 +982,25 @@ describe("dispatchReplyFromConfig", () => { ), }, }; + const passiveThreadingTestPlugins = [ + "slack", + "telegram", + "feishu", + "mattermost", + "imessage", + ].map((id) => { + const plugin = createChannelTestPluginBase({ id }); + return { + pluginId: id, + source: "test" as const, + plugin: { + ...plugin, + threading: { + resolveReplyToMode: () => "all" as const, + }, + }, + }; + }); setActivePluginRegistry( createTestRegistry([ { @@ -957,6 +1013,7 @@ describe("dispatchReplyFromConfig", () => { source: "test", plugin: signalTestPlugin, }, + ...passiveThreadingTestPlugins, ]), ); clearApprovalNativeRouteStateForTest(); @@ -1209,7 +1266,8 @@ describe("dispatchReplyFromConfig", () => { it("does not route when Provider matches OriginatingChannel (even if Surface is missing)", async () => { setNoAbort(); mocks.routeReply.mockClear(); - const cfg = emptyConfig; + installThreadingTestPlugin({ id: "slack", defaultAccountId: "work" }); + const cfg = automaticDirectReplyConfig; const dispatcher = createDispatcher(); const ctx = buildTestCtx({ Provider: "slack", @@ -1227,11 +1285,23 @@ describe("dispatchReplyFromConfig", () => { expect(mocks.routeReply).not.toHaveBeenCalled(); expect(dispatcher.sendFinalReply).toHaveBeenCalledTimes(1); + const replyDispatchCall = firstMockCall(hookMocks.runner.runReplyDispatch, "reply dispatch") as + | [ + { + originatingAccountId?: unknown; + shouldRouteToOriginating?: unknown; + }, + unknown, + ] + | undefined; + expect(replyDispatchCall?.[0]?.shouldRouteToOriginating).toBe(false); + expect(replyDispatchCall?.[0]?.originatingAccountId).toBe("work"); }); it("mirrors ownerless same-channel Slack finals after successful delivery", async () => { setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "slack" }); const dispatcher = createDispatcher(); const ctx = buildTestCtx({ Provider: "slack", @@ -1273,6 +1343,7 @@ describe("dispatchReplyFromConfig", () => { it("mirrors reset acknowledgements into the canonically prepared Slack session", async () => { setNoAbort(); + hookMocks.runner.hasHooks.mockReturnValue(false); const dispatcher = createDispatcher(); const sessionKey = "Agent:Main:Slack:Channel:C123"; const preparedSessionKey = "agent:main:slack:channel:c123"; @@ -1352,6 +1423,8 @@ describe("dispatchReplyFromConfig", () => { setNoAbort(); const dispatcher = createDispatcher(); mocks.routeReply.mockClear(); + hookMocks.runner.hasHooks.mockReturnValue(false); + installThreadingTestPlugin({ id: "telegram", defaultAccountId: "default" }); const result = await dispatchReplyFromConfig({ ctx: buildTestCtx({ @@ -1359,9 +1432,10 @@ describe("dispatchReplyFromConfig", () => { Surface: "slack", OriginatingChannel: "telegram", OriginatingTo: "telegram:999", + AccountId: "default", SessionKey: "agent:main:telegram:group:999", }), - cfg: emptyConfig, + cfg: automaticDirectReplyConfig, dispatcher, replyResolver: async () => setReplyPayloadMetadata( @@ -1581,6 +1655,7 @@ describe("dispatchReplyFromConfig", () => { it("keeps non-Slack routed direct turns behind the active reply operation", async () => { setNoAbort(); + installThreadingTestPlugin({ id: "telegram" }); const sessionKey = "agent:main:telegram:direct:1"; const activeOperation = createReplyOperation({ sessionKey, @@ -1627,6 +1702,7 @@ describe("dispatchReplyFromConfig", () => { it("routes when OriginatingChannel differs from Provider", async () => { setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "telegram" }); const cfg = emptyConfig; const dispatcher = createDispatcher(); const ctx = buildTestCtx({ @@ -1667,6 +1743,7 @@ describe("dispatchReplyFromConfig", () => { it("routes exec-event replies using persisted session delivery context when current turn has no originating route", async () => { setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "telegram" }); sessionStoreMocks.currentEntry = { deliveryContext: { channel: "telegram", @@ -1724,6 +1801,7 @@ describe("dispatchReplyFromConfig", () => { it("routes sessions_send internal webchat handoffs through persisted external delivery context", async () => { setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "feishu" }); sessionStoreMocks.currentEntry = { route: { channel: "feishu", @@ -1835,6 +1913,7 @@ describe("dispatchReplyFromConfig", () => { it("honors sendPolicy deny for recovered exec-event delivery channel", async () => { setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "telegram" }); sessionStoreMocks.currentEntry = { deliveryContext: { channel: "telegram", @@ -1908,6 +1987,7 @@ describe("dispatchReplyFromConfig", () => { it("uses Slack DM TransportThreadId when ReplyToId is the current message", async () => { setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "slack" }); const cfg = emptyConfig; const dispatcher = createDispatcher(); const ctx = buildTestCtx({ @@ -1935,6 +2015,7 @@ describe("dispatchReplyFromConfig", () => { it("does not resurrect a cleared route thread from origin metadata", async () => { setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "mattermost" }); // Simulate the real store: lastThreadId and deliveryContext.threadId may be normalised from // origin.threadId on read, but a non-thread session key must still route to channel root. sessionStoreMocks.currentEntry = { @@ -1975,6 +2056,7 @@ describe("dispatchReplyFromConfig", () => { it("forces suppressTyping when routing to a different originating channel", async () => { setNoAbort(); + installThreadingTestPlugin({ id: "telegram" }); const cfg = emptyConfig; const dispatcher = createDispatcher(); const ctx = buildTestCtx({ @@ -2015,6 +2097,7 @@ describe("dispatchReplyFromConfig", () => { it("routes when provider is webchat but surface carries originating channel metadata", async () => { setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "telegram" }); const cfg = emptyConfig; const dispatcher = createDispatcher(); const ctx = buildTestCtx({ @@ -2036,6 +2119,7 @@ describe("dispatchReplyFromConfig", () => { it("routes Feishu replies when provider is webchat and origin metadata points to Feishu", async () => { setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "feishu" }); const cfg = emptyConfig; const dispatcher = createDispatcher(); const ctx = buildTestCtx({ @@ -2076,6 +2160,7 @@ describe("dispatchReplyFromConfig", () => { it("does not route external origin replies when current surface is internal webchat without explicit delivery", async () => { setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "imessage" }); const cfg = emptyConfig; const dispatcher = createDispatcher(); const ctx = buildTestCtx({ @@ -2099,6 +2184,7 @@ describe("dispatchReplyFromConfig", () => { it("routes external origin replies for internal webchat turns when explicit delivery is set", async () => { setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "imessage" }); const cfg = emptyConfig; const dispatcher = createDispatcher(); const ctx = buildTestCtx({ @@ -2128,6 +2214,7 @@ describe("dispatchReplyFromConfig", () => { it("routes media-only tool results when summaries are suppressed", async () => { setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "telegram" }); const cfg = automaticGroupReplyConfig; const dispatcher = createDispatcher(); const ctx = buildTestCtx({ @@ -5497,6 +5584,7 @@ describe("dispatchReplyFromConfig", () => { it("deduplicates same-agent inbound replies across main and direct session keys", async () => { setNoAbort(); + hookMocks.runner.hasHooks.mockReturnValue(false); const cfg = emptyConfig; const replyResolver = vi.fn(async () => ({ text: "hi" }) as ReplyPayload); const baseCtx = buildTestCtx({ @@ -5530,6 +5618,7 @@ describe("dispatchReplyFromConfig", () => { it("emits message_received hook with originating channel metadata", async () => { setNoAbort(); hookMocks.runner.hasHooks.mockReturnValue(true); + installThreadingTestPlugin({ id: "telegram" }); const cfg = emptyConfig; const dispatcher = createDispatcher(); const ctx = buildTestCtx({ @@ -5789,6 +5878,7 @@ describe("dispatchReplyFromConfig", () => { // would receive divergent keys on every native redirect. setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "telegram" }); const cfg = emptyConfig; const dispatcher = createDispatcher(); const ctx = buildTestCtx({ @@ -5825,6 +5915,7 @@ describe("dispatchReplyFromConfig", () => { // generalization of the native-redirect branch. setNoAbort(); mocks.routeReply.mockClear(); + installThreadingTestPlugin({ id: "telegram" }); const cfg = emptyConfig; const dispatcher = createDispatcher(); const ctx = buildTestCtx({ @@ -7669,6 +7760,7 @@ describe("before_dispatch hook", () => { it("uses canonical hook metadata and shared routed final delivery", async () => { ttsMocks.state.synthesizeFinalAudio = true; hookMocks.runner.runBeforeDispatch.mockResolvedValue({ handled: true, text: "Blocked" }); + installThreadingTestPlugin({ id: "telegram" }); const dispatcher = createDispatcher(); const ctx = createHookCtx({ Body: "raw body", diff --git a/src/auto-reply/reply/dispatch-from-config.ts b/src/auto-reply/reply/dispatch-from-config.ts index 29f468b38518..3b9018cfe05f 100644 --- a/src/auto-reply/reply/dispatch-from-config.ts +++ b/src/auto-reply/reply/dispatch-from-config.ts @@ -1606,10 +1606,12 @@ export async function dispatchReplyFromConfig( }); const routeReplyTo = replyRoute.to; const deliveryChannel = shouldRouteToOriginating ? routeReplyChannel : currentSurface; - const routedReplyAccountId = routeReplyChannel + const shouldPrepareRoutedReplyDelivery = shouldRouteToOriginating && Boolean(routeReplyChannel); + const replyContextAccountId = routeReplyChannel ? resolveReplyDeliveryAccountId(cfg, routeReplyChannel, replyRoute.accountId) : undefined; - const routedReplyDelivery = routeReplyChannel + const routedReplyAccountId = shouldPrepareRoutedReplyDelivery ? replyContextAccountId : undefined; + const routedReplyDelivery = shouldPrepareRoutedReplyDelivery ? createReplyDeliveryContext( resolveReplyToMode(cfg, routeReplyChannel, routedReplyAccountId, replyRoute.chatType), replyRoute.chatType, @@ -1630,7 +1632,7 @@ export async function dispatchReplyFromConfig( sessionKey: acpDispatchSessionKey, workspaceDir, messageProvider: deliveryChannel, - accountId: routedReplyAccountId, + accountId: replyContextAccountId, groupId, groupChannel: ctx.GroupChannel, groupSpace: ctx.GroupSpace, @@ -2500,7 +2502,7 @@ export async function dispatchReplyFromConfig( shouldRouteToOriginating, originatingChannel: routeReplyChannel, originatingTo: routeReplyTo, - originatingAccountId: routedReplyAccountId, + originatingAccountId: replyContextAccountId, originatingThreadId: routeReplyThreadId, originatingChatType: replyRoute.chatType, shouldSendToolSummaries, @@ -3216,7 +3218,7 @@ export async function dispatchReplyFromConfig( shouldRouteToOriginating, originatingChannel: routeReplyChannel, originatingTo: routeReplyTo, - originatingAccountId: routedReplyAccountId, + originatingAccountId: replyContextAccountId, originatingThreadId: routeReplyThreadId, originatingChatType: replyRoute.chatType, shouldSendToolSummaries, diff --git a/src/commands/agent.test.ts b/src/commands/agent.test.ts index 57b78075bbd4..d0b7cd32a584 100644 --- a/src/commands/agent.test.ts +++ b/src/commands/agent.test.ts @@ -25,7 +25,11 @@ import { import type { PluginProviderRegistration } from "../plugins/registry.js"; import { resetPluginRuntimeStateForTest, setActivePluginRegistry } from "../plugins/runtime.js"; import type { RuntimeEnv } from "../runtime.js"; -import { createTestRegistry } from "../test-utils/channel-plugins.js"; +import { + createDirectOutboundTestAdapter, + createOutboundTestPlugin, + createTestRegistry, +} from "../test-utils/channel-plugins.js"; import { agentCommand, agentCommandFromIngress, testing as agentCommandTesting } from "./agent.js"; import { createThrowingTestRuntime } from "./test-runtime-config-helpers.js"; @@ -63,6 +67,10 @@ vi.mock("../agents/auth-profiles/store.js", () => { }; }); +vi.mock("../agents/auth-profiles/source-check.js", () => ({ + hasAnyAuthProfileStoreSource: vi.fn(() => false), +})); + vi.mock("../agents/command/session-store.runtime.js", () => { return { updateSessionStoreAfterAgentRun: vi.fn(async () => undefined), @@ -336,8 +344,8 @@ function mockModelCatalogOnce(entries: ReturnType[0] = []) { + const registry = createTestRegistry(channels); registry.providers = ["anthropic", "codex", "ollama", "openai", "openrouter"].map( (providerId): PluginProviderRegistration => ({ pluginId: providerId, @@ -1311,6 +1319,16 @@ describe("agentCommand", () => { }, }); mockConfig(home, store); + installThinkingTestProviders([ + { + pluginId: "telegram", + source: "test", + plugin: createOutboundTestPlugin({ + id: "telegram", + outbound: createDirectOutboundTestAdapter({ channel: "telegram" }), + }), + }, + ]); await agentCommand( { message: "hi", to: sessionKey, deliver: true, channel: "telegram" }, diff --git a/ui/src/ui/views/skills.ts b/ui/src/ui/views/skills.ts index 5e4364d142e7..cdacc6ea31ac 100644 --- a/ui/src/ui/views/skills.ts +++ b/ui/src/ui/views/skills.ts @@ -257,7 +257,7 @@ export function renderSkills(props: SkillsProps) { ${agents.length > 0 ? html`