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),

},

},

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

return call as Record<string, unknown>;

}

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<GetReplyOptions["onToolResult"]> {

),

},

};

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([

{

source: "test",

plugin: signalTestPlugin,

},

...passiveThreadingTestPlugins,

]),

);

clearApprovalNativeRouteStateForTest();

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

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

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

setNoAbort();

const dispatcher = createDispatcher();

mocks.routeReply.mockClear();

hookMocks.runner.hasHooks.mockReturnValue(false);

installThreadingTestPlugin({ id: "telegram", defaultAccountId: "default" });

const result = await dispatchReplyFromConfig({

ctx: buildTestCtx({

Provider: "slack",

Surface: "slack",

OriginatingChannel: "telegram",

OriginatingTo: "telegram:999",

AccountId: "default",

SessionKey: "agent:main:telegram:group:999",

}),

cfg: emptyConfig,

cfg: automaticDirectReplyConfig,

dispatcher,

replyResolver: async () =>

setReplyPayloadMetadata(

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,

it("routes when OriginatingChannel differs from Provider", async () => {

setNoAbort();

mocks.routeReply.mockClear();

installThreadingTestPlugin({ id: "telegram" });

const cfg = emptyConfig;

const dispatcher = createDispatcher();

const ctx = buildTestCtx({

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

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

it("honors sendPolicy deny for recovered exec-event delivery channel", async () => {

setNoAbort();

mocks.routeReply.mockClear();

installThreadingTestPlugin({ id: "telegram" });

sessionStoreMocks.currentEntry = {

deliveryContext: {

channel: "telegram",

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({

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 = {

it("forces suppressTyping when routing to a different originating channel", async () => {

setNoAbort();

installThreadingTestPlugin({ id: "telegram" });

const cfg = emptyConfig;

const dispatcher = createDispatcher();

const ctx = buildTestCtx({

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({

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({

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({

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({

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({

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({

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({

// would receive divergent keys on every native redirect.

setNoAbort();

mocks.routeReply.mockClear();

installThreadingTestPlugin({ id: "telegram" });

const cfg = emptyConfig;

const dispatcher = createDispatcher();

const ctx = buildTestCtx({

// generalization of the native-redirect branch.

setNoAbort();

mocks.routeReply.mockClear();

installThreadingTestPlugin({ id: "telegram" });

const cfg = emptyConfig;

const dispatcher = createDispatcher();

const ctx = buildTestCtx({

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