diff --git a/src/channels/registry-routes.ts b/src/channels/registry-routes.ts index cf45e622a7..dfe964c807 100644 --- a/src/channels/registry-routes.ts +++ b/src/channels/registry-routes.ts @@ -1,4 +1,4 @@ -import { getBackend } from "@/backend"; +import { type ConversationMessageListBody, getBackend } from "@/backend"; import { LEGACY_CHANNEL_ACCOUNT_ID } from "./accounts"; import type { ChannelRegistryEvent } from "./registry-events"; import { @@ -23,6 +23,26 @@ import type { WhatsAppChannelAccount, } from "./types"; +async function hasDurableUserMessage( + route: ChannelRoute, +): Promise { + try { + const page = await getBackend().listConversationMessages( + route.conversationId, + { + limit: 1, + order: "desc", + include_return_message_types: ["user_message"], + ...(route.agentId ? { agent_id: route.agentId } : {}), + } as ConversationMessageListBody, + ); + return page.getPaginatedItems().length > 0; + } catch { + // Only suppress or replay bootstrap context when the backend proves user-turn state. + return null; + } +} + export function createChannelRouteProvisioner(deps: { emitEvent: (event: ChannelRegistryEvent) => void; }) { @@ -123,26 +143,42 @@ export function createChannelRouteProvisioner(deps: { } if (route) { + let routeForDelivery = route; + let shouldPersistRoute = false; + let isFirstRouteTurn = false; + if (msg.chatType === "channel" && !route.bootstrapUserMessageSeenAt) { + const durableUserMessageExists = await hasDurableUserMessage(route); + if (durableUserMessageExists === true) { + routeForDelivery = { + ...routeForDelivery, + bootstrapUserMessageSeenAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + }; + shouldPersistRoute = true; + } else if (durableUserMessageExists === false) { + isFirstRouteTurn = true; + } + } if ( msg.chatType === "channel" && msg.isMention === true && - (route.outboundEnabled === false || route.detached === true) + (routeForDelivery.outboundEnabled === false || + routeForDelivery.detached === true) ) { - const updatedRoute: ChannelRoute = { - ...route, + routeForDelivery = { + ...routeForDelivery, outboundEnabled: true, detached: false, updatedAt: new Date().toISOString(), }; - addRoute(msg.channel, updatedRoute); - return { - route: updatedRoute, - isFirstRouteTurn: false, - }; + shouldPersistRoute = true; + } + if (shouldPersistRoute) { + addRoute(msg.channel, routeForDelivery); } return { - route, - isFirstRouteTurn: false, + route: routeForDelivery, + isFirstRouteTurn, }; } diff --git a/src/channels/routing.test.ts b/src/channels/routing.test.ts index 52c5196218..a8dd7de754 100644 --- a/src/channels/routing.test.ts +++ b/src/channels/routing.test.ts @@ -7,6 +7,7 @@ import { getAllRoutes, getRoute, getRoutesForChannel, + loadRoutes, removeRoute, removeRoutesForScope, } from "@/channels/routing"; @@ -114,6 +115,29 @@ describe("routing", () => { expect(slackRoutes).toHaveLength(0); }); + test("loadRoutes preserves Slack bootstrap user-message markers", () => { + __testOverrideLoadRoutes(() => [ + { + accountId: "slack-bot", + chatId: "C123", + chatType: "channel", + threadId: "1712790000.000050", + agentId: "agent-a", + conversationId: "conv-1", + enabled: true, + bootstrapUserMessageSeenAt: "2026-04-11T00:01:00.000Z", + createdAt: "2026-04-11T00:00:00.000Z", + }, + ]); + + loadRoutes("slack"); + + expect( + getRoute("slack", "C123", "slack-bot", "1712790000.000050") + ?.bootstrapUserMessageSeenAt, + ).toBe("2026-04-11T00:01:00.000Z"); + }); + test("getAllRoutes returns all routes", () => { addRoute("telegram", { chatId: "chat-1", diff --git a/src/channels/routing.ts b/src/channels/routing.ts index e47f0f22a5..9d4b32b997 100644 --- a/src/channels/routing.ts +++ b/src/channels/routing.ts @@ -69,6 +69,9 @@ export function loadRoutes(channelId: string): void { enabled: route.enabled !== false, outboundEnabled: route.outboundEnabled !== false, detached: route.detached === true, + ...(route.bootstrapUserMessageSeenAt + ? { bootstrapUserMessageSeenAt: route.bootstrapUserMessageSeenAt } + : {}), createdAt: route.createdAt ?? new Date().toISOString(), updatedAt: route.updatedAt ?? route.createdAt ?? new Date().toISOString(), @@ -101,6 +104,9 @@ export function loadRoutes(channelId: string): void { enabled: route.enabled !== false, outboundEnabled: route.outboundEnabled !== false, detached: route.detached === true, + ...(route.bootstrapUserMessageSeenAt + ? { bootstrapUserMessageSeenAt: route.bootstrapUserMessageSeenAt } + : {}), createdAt: route.createdAt ?? new Date().toISOString(), updatedAt: route.updatedAt ?? route.createdAt ?? new Date().toISOString(), diff --git a/src/channels/slack-registry.test.ts b/src/channels/slack-registry.test.ts index 20c760fdf9..82fe7017a1 100644 --- a/src/channels/slack-registry.test.ts +++ b/src/channels/slack-registry.test.ts @@ -29,14 +29,48 @@ import { clearTargetStores, } from "@/channels/targets"; import type { ChannelAdapter, InboundChannelMessage } from "@/channels/types"; +import { + createStartedSlackAdapter, + installSlackAdapterTestHooks, + resolveSlackThreadHistoryMock, + resolveSlackThreadStarterMock, +} from "./slack/adapter-test-harness"; + +installSlackAdapterTestHooks(); const createConversation = mock(async () => ({ id: "conv-slack" })); +function createPage(items: unknown[]) { + return { + getPaginatedItems: () => items, + }; +} + +function contentText(content: unknown): string { + if (!Array.isArray(content)) return String(content); + return content + .map((part) => { + if (part && typeof part === "object" && "text" in part) { + return String((part as { text?: unknown }).text ?? ""); + } + return JSON.stringify(part); + }) + .join("\n"); +} + +const listConversationMessages = mock( + async (_conversationId: string, _body?: unknown, _options?: unknown) => + createPage([{ id: "msg-existing", message_type: "user_message" }]), +); + mock.module("@/backend/api/client", () => ({ getServerUrl: () => "https://api.letta.com", getClient: async () => ({ conversations: { create: createConversation, + messages: { + list: listConversationMessages, + }, }, }), })); @@ -57,6 +91,10 @@ describe("slack channel registry", () => { __testOverrideSaveTargetStore(null); createConversation.mockReset(); createConversation.mockResolvedValue({ id: "conv-slack" }); + listConversationMessages.mockReset(); + listConversationMessages.mockResolvedValue( + createPage([{ id: "msg-existing", message_type: "user_message" }]), + ); } function createInboundMessage( @@ -150,6 +188,8 @@ describe("slack channel registry", () => { expect(createConversation).toHaveBeenCalledTimes(1); const route = getRoute("slack", "C123", "slack-bot", "1712790000.000050"); expect(route).toEqual(expect.objectContaining({ outboundEnabled: false })); + expect(route?.bootstrapUserMessageSeenAt).toBeUndefined(); + expect(listConversationMessages).not.toHaveBeenCalled(); expect(deliveries).toHaveLength(1); expect(deliveries[0]?.turnSources).toEqual([ expect.objectContaining({ @@ -173,6 +213,143 @@ describe("slack channel registry", () => { ).toBeNull(); }); + test("persisted orphan Slack thread route recovers formatted bootstrap context once", async () => { + const nonUserMessages = [ + { id: "summary-1", message_type: "summary_message" }, + { id: "assistant-1", message_type: "assistant_message" }, + ]; + listConversationMessages + .mockImplementationOnce(async (_conversationId, body) => { + const requestedTypes = + (body as { include_return_message_types?: string[] }) + .include_return_message_types ?? []; + return createPage( + requestedTypes.length === 0 + ? nonUserMessages + : nonUserMessages.filter((message) => + requestedTypes.includes(message.message_type), + ), + ); + }) + .mockResolvedValueOnce( + createPage([{ id: "msg-recovered", message_type: "user_message" }]), + ); + __testOverrideLoadRoutes(() => [ + { + accountId: "slack-bot", + chatId: "C123", + chatType: "channel", + threadId: "1712790000.000050", + agentId: "agent-1", + conversationId: "conv-recovered", + enabled: true, + outboundEnabled: true, + createdAt: "2026-04-11T00:00:00.000Z", + updatedAt: "2026-04-11T00:00:00.000Z", + }, + ]); + + const { ChannelRegistry } = await import("@/channels/registry"); + const registry = new ChannelRegistry(); + const adapter = await createStartedSlackAdapter({ + accountId: "slack-bot", + agentId: "agent-1", + defaultPermissionMode: "unrestricted", + dmPolicy: "open", + allowedUsers: [], + }); + registry.registerAdapter(adapter); + + const deliveries: Array<{ content: unknown; turnSources?: unknown[] }> = []; + registry.setMessageHandler((delivery) => { + deliveries.push(delivery); + }); + registry.setReady(); + + resolveSlackThreadStarterMock.mockResolvedValueOnce({ + text: "Original root problem statement", + userId: "U111", + ts: "1712790000.000050", + attachments: [ + { + id: "FROOT", + name: "root-screenshot.png", + mimeType: "image/png", + kind: "image", + localPath: "/tmp/root-screenshot.png", + }, + ], + }); + resolveSlackThreadHistoryMock.mockResolvedValueOnce([ + { + text: "Prior human investigation notes", + userId: "U222", + ts: "1712795000.000060", + }, + ]); + + await adapter.onMessage?.( + createInboundMessage({ + text: "bump", + messageId: "1712800001.000300", + }), + ); + + expect(createConversation).not.toHaveBeenCalled(); + expect(listConversationMessages).toHaveBeenCalledTimes(1); + expect(listConversationMessages.mock.calls[0]?.[0]).toBe("conv-recovered"); + expect(listConversationMessages.mock.calls[0]?.[1]).toEqual( + expect.objectContaining({ + limit: 1, + order: "desc", + include_return_message_types: ["user_message"], + agent_id: "agent-1", + }), + ); + expect( + getRoute("slack", "C123", "slack-bot", "1712790000.000050") + ?.bootstrapUserMessageSeenAt, + ).toBeUndefined(); + const recoveredContent = contentText(deliveries[0]?.content); + expect(recoveredContent).toContain("Original root problem statement"); + expect(recoveredContent).toContain("Prior human investigation notes"); + expect(recoveredContent).toContain("/tmp/root-screenshot.png"); + expect(recoveredContent).toContain("bump"); + + await adapter.onMessage?.( + createInboundMessage({ + text: "follow-up", + messageId: "1712800002.000400", + }), + ); + + expect(listConversationMessages).toHaveBeenCalledTimes(2); + const markedRoute = getRoute( + "slack", + "C123", + "slack-bot", + "1712790000.000050", + ); + expect(markedRoute?.bootstrapUserMessageSeenAt).toEqual(expect.any(String)); + const incrementalContent = contentText(deliveries[1]?.content); + expect(incrementalContent).toContain("follow-up"); + expect(incrementalContent).not.toContain("Original root problem statement"); + expect(incrementalContent).not.toContain("Prior human investigation notes"); + expect(incrementalContent).not.toContain("/tmp/root-screenshot.png"); + + await adapter.onMessage?.( + createInboundMessage({ + text: "post-marker", + messageId: "1712800003.000500", + }), + ); + + expect(listConversationMessages).toHaveBeenCalledTimes(2); + expect(contentText(deliveries[2]?.content)).toContain("post-marker"); + expect(resolveSlackThreadStarterMock).toHaveBeenCalledTimes(1); + expect(deliveries).toHaveLength(3); + }); + test("an explicit Slack mention upgrades a listen-only route for outbound replies", async () => { const { ChannelRegistry } = await import("@/channels/registry"); const registry = new ChannelRegistry(); diff --git a/src/channels/types.ts b/src/channels/types.ts index 559ddcbfb6..a3614e29ae 100644 --- a/src/channels/types.ts +++ b/src/channels/types.ts @@ -493,6 +493,11 @@ export interface ChannelRoute { outboundEnabled?: boolean; /** Slack-only: a detached thread stays silent until the app is mentioned again. */ detached?: boolean; + /** + * Slack-only: set after the backend proves this route has a durable user turn. + * Older routes without this marker may need one bootstrap recovery probe after restart. + */ + bootstrapUserMessageSeenAt?: string; /** ISO 8601 creation timestamp. */ createdAt: string; /** ISO 8601 update timestamp. */