From bcafa51f5331ba7a57b62cf9574c2a4df523c330 Mon Sep 17 00:00:00 2001 From: imsobear Date: Mon, 12 Oct 2026 20:29:04 -0700 Subject: [PATCH] Answer Feishu mentions in a thread, and keep answering inside it. The bot replied to a mention with a quote in the group, so every answer added to the main chat. It now replies with reply_in_thread, and the group shows one line per question. An ordinary group's history only lists the first message of each thread, so a follow-up mention inside one was never seen. The loop's cursor now remembers each message the bot was asked in, looks up the thread its answer opened, and reads that thread on every look. A mention there is answered in the thread, with up to 20 earlier messages of it as context. A thread with no message for a week is let go, as is an ask that never got a thread within a day, and at most 50 threads are read per loop. Co-Authored-By: Claude Opus 5.5 --- docs/connectors.md | 2 +- src/connectors/feishu/api.test.ts | 1 + src/connectors/feishu/api.ts | 54 ++++++- src/connectors/feishu/runtime.test.ts | 127 ++++++++++++++++- src/connectors/feishu/runtime.ts | 195 ++++++++++++++++++++++---- 5 files changed, 346 insertions(+), 33 deletions(-) diff --git a/docs/connectors.md b/docs/connectors.md index ea2f427..322af21 100644 --- a/docs/connectors.md +++ b/docs/connectors.md @@ -92,4 +92,4 @@ im:message.group_msg im:message:send_as_bot ``` -Invite the bot to any group it should watch. A loop fires on an @mention in those groups and replies in the thread. Feishu does not list p2p chats for a bot, so DMs are not polled. +Invite the bot to any group it should watch. A loop fires on an @mention in those groups and answers in a thread under it, so the group itself stays readable. An @mention inside that thread is answered there too, with what was said earlier in the thread; threads go unread after a week without a message. Feishu does not list p2p chats for a bot, so DMs are not polled. diff --git a/src/connectors/feishu/api.test.ts b/src/connectors/feishu/api.test.ts index 3207896..c8d0721 100644 --- a/src/connectors/feishu/api.test.ts +++ b/src/connectors/feishu/api.test.ts @@ -177,6 +177,7 @@ describe("replyMessage", () => { expect(JSON.parse(String(init.body))).toEqual({ msg_type: "text", content: "{\"text\":\"Cause: redis timeout\"}", + reply_in_thread: true, }); return ok({ code: 0, diff --git a/src/connectors/feishu/api.ts b/src/connectors/feishu/api.ts index aa7572c..9863a72 100644 --- a/src/connectors/feishu/api.ts +++ b/src/connectors/feishu/api.ts @@ -31,6 +31,8 @@ export type FeishuMessage = { senderId?: string; senderType?: string; mentionIds: string[]; + /** The thread it is in, or the thread it starts. */ + threadId?: string; }; type Envelope = { @@ -178,17 +180,44 @@ function mentionsOf(raw: unknown): string[] { .filter(Boolean); } +/** + * A group's messages newer than `oldestMs`, oldest first. In an ordinary group + * that is only the first message of each thread: what is said inside one is + * read with `threadHistory`. + */ export async function history( credential: FeishuCredential, chatId: string, oldestMs: string, ): Promise { + return await messagesIn(credential, { type: "chat", id: chatId, chatId }, oldestMs); +} + +/** What was said in one thread after `oldestMs`, oldest first. */ +export async function threadHistory( + credential: FeishuCredential, + thread: { threadId: string; chatId: string }, + oldestMs: string, +): Promise { + return await messagesIn( + credential, + { type: "thread", id: thread.threadId, chatId: thread.chatId }, + oldestMs, + ); +} + +async function messagesIn( + credential: FeishuCredential, + container: { type: "chat" | "thread"; id: string; chatId: string }, + oldestMs: string, +): Promise { + const chatId = container.chatId; const parsed: FeishuMessage[] = []; let pageToken = ""; do { const query: Record = { - container_id_type: "chat", - container_id: chatId, + container_id_type: container.type, + container_id: container.id, sort_type: "ByCreateTimeAsc", page_size: "50", start_time: String(Math.floor(Number(oldestMs) / 1000)), @@ -209,6 +238,7 @@ export async function history( const message = raw as { message_id?: string; chat_id?: string; + thread_id?: string; create_time?: string; msg_type?: string; deleted?: boolean; @@ -225,6 +255,7 @@ export async function history( ...(message.sender?.id ? { senderId: message.sender.id } : {}), ...(message.sender?.sender_type ? { senderType: message.sender.sender_type } : {}), mentionIds: mentionsOf(message.mentions), + ...(message.thread_id ? { threadId: message.thread_id } : {}), }); } pageToken = data.has_more === true && typeof data.page_token === "string" ? data.page_token : ""; @@ -232,6 +263,23 @@ export async function history( return parsed.sort((a, b) => Number(a.createTime) - Number(b.createTime)); } +/** + * The thread a message is in or starts, or null while it has none. A message + * the bot was asked in only gets one once the answer is posted. + */ +export async function threadOf(credential: FeishuCredential, messageId: string): Promise { + const data = dataOf( + await call(credential.host, `/im/v1/messages/${messageId}`, { token: credential.tenantAccessToken }), + ); + const items = Array.isArray(data.items) ? data.items : []; + const message = (items[0] ?? {}) as { thread_id?: unknown }; + return typeof message.thread_id === "string" && message.thread_id ? message.thread_id : null; +} + +/** + * Answers in a thread under the message, or in the thread it is already in, + * so a conversation with the bot does not fill the group. + */ export async function replyMessage( credential: FeishuCredential, messageId: string, @@ -240,7 +288,7 @@ export async function replyMessage( const body = await call(credential.host, `/im/v1/messages/${messageId}/reply`, { token: credential.tenantAccessToken, method: "POST", - body: { msg_type: "text", content: JSON.stringify({ text }) }, + body: { msg_type: "text", content: JSON.stringify({ text }), reply_in_thread: true }, }); const data = dataOf(body); const id = typeof data.message_id === "string" ? data.message_id : ""; diff --git a/src/connectors/feishu/runtime.test.ts b/src/connectors/feishu/runtime.test.ts index 0e15f4f..fb50cbf 100644 --- a/src/connectors/feishu/runtime.test.ts +++ b/src/connectors/feishu/runtime.test.ts @@ -25,6 +25,8 @@ function givenFeishu(handlers: { chats?: Array<{ chat_id: string; name?: string }>; history?: Record>>; reply?: { message_id: string; chat_id: string }; + /** Messages looked up one at a time, by id. */ + messages?: Record>; send?: { message_id: string; chat_id: string }; }) { const sent: Call[] = []; @@ -57,6 +59,12 @@ function givenFeishu(handlers: { const chat = new URL(url).searchParams.get("container_id") ?? ""; return json({ code: 0, data: { items: handlers.history?.[chat] ?? [], has_more: false } }); } + const one = /^\/im\/v1\/messages\/([^/]+)$/.exec(path)?.[1]; + if (one && init?.method !== "POST") { + const message = handlers.messages?.[one]; + if (!message) return json({ code: 230011, msg: "message not found" }); + return json({ code: 0, data: { items: [message] } }); + } if (path.endsWith("/reply")) { return json({ code: 0, data: handlers.reply ?? { message_id: "om_reply", chat_id: "oc_eng" } }); } @@ -65,7 +73,10 @@ function givenFeishu(handlers: { return sent; } -afterEach(() => vi.unstubAllGlobals()); +afterEach(() => { + vi.unstubAllGlobals(); + vi.useRealTimers(); +}); describe("connectWithFields", () => { it("refuses a missing app id", async () => { @@ -145,7 +156,11 @@ describe("feishu poll", () => { const { signals, cursor } = await poll(null); expect(signals.map((signal) => [signal.key, signal.at])).toEqual([["oc_eng:om_ask", 1690000030000]]); - expect(cursor).toEqual({ oc_eng: "1690000030000", oc_quiet: "1690000000000" }); + expect(cursor).toEqual({ + oc_eng: "1690000030000", + oc_quiet: "1690000000000", + threads: { om_ask: { chatId: "oc_eng", messageId: "om_ask", last: "1690000030000" } }, + }); }); it("keeps a group mention of the bot and skips chat and its own replies", async () => { @@ -199,7 +214,112 @@ describe("feishu poll", () => { at: 1700000002000, }, ]); - expect(cursor).toEqual({ oc_eng: "1700000003000" }); + expect(cursor).toEqual({ + oc_eng: "1700000003000", + threads: { om_ask: { chatId: "oc_eng", messageId: "om_ask", last: "1700000002000" } }, + }); + }); +}); + +describe("feishu threads", () => { + const poll = (threads: Record) => + feishuRuntime.poll!({ + workflowId: "feishu.ask", + settings: {}, + credential: account, + cursor: { oc_eng: "1700000000000", threads } as never, + since: 1690000000000, + }); + + const said = (id: string, at: string, text: string, by: "ou_maya" | "bot", mention = false) => ({ + message_id: id, + create_time: at, + chat_id: "oc_eng", + thread_id: "omt_1", + msg_type: "text", + sender: by === "bot" ? { id: "cli_app", sender_type: "app" } : { id: by, sender_type: "user" }, + body: { content: JSON.stringify({ text }) }, + ...(mention ? { mentions: [{ id: "ou_bot" }] } : {}), + }); + + it("answers a mention inside the thread its answer opened, with what came before", async () => { + vi.useFakeTimers({ now: 1700000100000, toFake: ["Date"] }); + givenFeishu({ + chats: [{ chat_id: "oc_eng", name: "eng" }], + messages: { om_ask: { message_id: "om_ask", thread_id: "omt_1" } }, + history: { + omt_1: [ + said("om_ask", "1700000002000", "@_user_1 checkout 5xx", "ou_maya", true), + said("om_answer", "1700000003000", "Cause: redis timeout", "bot"), + said("om_more", "1700000004000", "@_user_1 since when?", "ou_maya", true), + said("om_chat", "1700000005000", "thanks all", "ou_maya"), + ], + }, + }); + + const { signals, cursor } = await poll({ + om_ask: { chatId: "oc_eng", messageId: "om_ask", last: "1700000002000" }, + }); + + expect(signals.map((signal) => signal.key)).toEqual(["oc_eng:om_more"]); + expect(signals[0]!.payload).toMatchObject({ + messageId: "om_more", + earlier: [ + { by: "person", text: "@_user_1 checkout 5xx" }, + { by: "bot", text: "Cause: redis timeout" }, + ], + }); + expect((cursor as { threads: unknown }).threads).toEqual({ + omt_1: { chatId: "oc_eng", messageId: "om_ask", threadId: "omt_1", last: "1700000005000" }, + }); + + const item = await feishuRuntime.resolveWorkItem!({ url: "", payload: signals[0]!.payload ?? null, credential: account }); + expect(item.context[0]!.body).toBe( + [ + "# Earlier in this thread", + "", + "**Them:** @_user_1 checkout 5xx", + "", + "**You:** Cause: redis timeout", + "", + "# What you were asked", + "", + "@_user_1 since when?", + "", + ].join("\n"), + ); + }); + + it("waits for a thread to open, and lets go of one that went quiet", async () => { + const day = 24 * 60 * 60 * 1000; + const now = 1700000100000; + vi.useFakeTimers({ now, toFake: ["Date"] }); + givenFeishu({ + chats: [], + messages: { om_waiting: { message_id: "om_waiting" } }, + }); + + const { signals, cursor } = await poll({ + om_waiting: { chatId: "oc_eng", messageId: "om_waiting", last: String(now - 60_000) }, + om_unanswered: { chatId: "oc_eng", messageId: "om_unanswered", last: String(now - 2 * day) }, + omt_old: { chatId: "oc_eng", messageId: "om_old", threadId: "omt_old", last: String(now - 8 * day) }, + }); + + expect(signals).toEqual([]); + expect((cursor as { threads: unknown }).threads).toEqual({ + om_waiting: { chatId: "oc_eng", messageId: "om_waiting", last: String(now - 60_000) }, + }); + }); + + it("stops reading a thread that is gone, without failing the look", async () => { + vi.useFakeTimers({ now: 1700000100000, toFake: ["Date"] }); + givenFeishu({ chats: [] }); + + const { cursor } = await poll({ + om_deleted: { chatId: "oc_eng", messageId: "om_deleted", last: "1700000000000" }, + }); + + expect((cursor as { threads: unknown }).threads).toEqual({}); }); }); @@ -225,6 +345,7 @@ describe("feishu reply", () => { expect(sent.find((call) => call.url.endsWith("/reply"))?.body).toEqual({ msg_type: "text", content: "{\"text\":\"Cause: redis timeout\"}", + reply_in_thread: true, }); expect(outcome.url).toBe("https://applink.feishu.cn/client/chat/open?openChatId=oc_eng"); }); diff --git a/src/connectors/feishu/runtime.ts b/src/connectors/feishu/runtime.ts index b48c5fd..22c4bd5 100644 --- a/src/connectors/feishu/runtime.ts +++ b/src/connectors/feishu/runtime.ts @@ -1,4 +1,5 @@ import type { JsonValue } from "#/lib/domain.ts"; +import { isTransient } from "../errors.ts"; import type { ActionSource, ConnectorRuntime, Signal, WorkItem } from "../types.ts"; import { botInfo, @@ -9,16 +10,39 @@ import { replyMessage, sendMessage, tenantAccessToken, + threadHistory, + threadOf, type FeishuCredential, + type FeishuMessage, } from "./api.ts"; +/** What was said in the thread before the bot was asked, oldest first. */ +type Said = { by: "bot" | "person"; text: string }; + type Asked = { chatId: string; messageId: string; user?: string; text: string; + earlier?: Said[]; }; +/** + * A thread the bot answers in, and how far into it this loop has read. Until + * the first answer is posted there is no thread yet, only the message it will + * hang from. + */ +type Watch = { chatId: string; messageId: string; threadId?: string; last: string }; + +/** A thread nobody has written in for this long is no longer read. */ +const WATCH_MS = 7 * 24 * 60 * 60 * 1000; +/** An ask that never got a thread was answered somewhere else, or not at all. */ +const PENDING_MS = 24 * 60 * 60 * 1000; +/** Every thread costs a request on every look, so only the liveliest are kept. */ +const WATCH_LIMIT = 50; +const EARLIER_LIMIT = 20; +const SAID_LIMIT = 2000; + function asAsked(value: JsonValue | null): Asked { const record = value as Partial | null; if (!record || typeof record.chatId !== "string" || typeof record.messageId !== "string") { @@ -29,9 +53,29 @@ function asAsked(value: JsonValue | null): Asked { messageId: record.messageId, user: typeof record.user === "string" ? record.user : undefined, text: typeof record.text === "string" ? record.text : "", + ...(Array.isArray(record.earlier) ? { earlier: record.earlier } : {}), }; } +function asWatches(value: JsonValue | null): Record { + const record = value && typeof value === "object" && !Array.isArray(value) ? value.threads : null; + if (!record || typeof record !== "object" || Array.isArray(record)) return {}; + const out: Record = {}; + for (const [key, entry] of Object.entries(record)) { + const watch = entry as Partial | null; + if (!watch || typeof watch.chatId !== "string" || typeof watch.messageId !== "string") continue; + if (typeof watch.last !== "string") continue; + out[key] = { + chatId: watch.chatId, + messageId: watch.messageId, + last: watch.last, + ...(typeof watch.threadId === "string" ? { threadId: watch.threadId } : {}), + }; + } + return out; +} + +/** Chats by id to the newest message read in each. Threads are kept apart. */ function asCursor(value: JsonValue | null): Record { if (!value || typeof value !== "object" || Array.isArray(value)) return {}; const out: Record = {}; @@ -57,6 +101,103 @@ function isAsk(input: { return input.mentionIds.includes(input.botOpenId); } +function signalFor(account: FeishuCredential, message: FeishuMessage, earlier?: Said[]): Signal { + const asked: Asked = { + chatId: message.chatId, + messageId: message.id, + user: message.senderId, + text: message.text, + ...(earlier?.length ? { earlier } : {}), + }; + return { + key: `${message.chatId}:${message.id}`, + kind: "message", + ref: message.id, + title: summarise(message.text), + url: permalink(account.host, message.chatId), + payload: asked as unknown as JsonValue, + at: Number(message.createTime), + }; +} + +function saidBy(account: FeishuCredential, message: FeishuMessage): Said { + const bot = message.senderType === "app" || message.senderId === account.botOpenId; + const text = message.text.length > SAID_LIMIT ? `${message.text.slice(0, SAID_LIMIT - 1)}…` : message.text; + return { by: bot ? "bot" : "person", text }; +} + +function asksIn(account: FeishuCredential, messages: FeishuMessage[], after: string): FeishuMessage[] { + return messages.filter( + (message) => + Number(message.createTime) > Number(after) && + isAsk({ + senderId: message.senderId, + senderType: message.senderType, + mentionIds: message.mentionIds, + botOpenId: account.botOpenId, + }), + ); +} + +/** + * Reads the threads the bot answers in, for the mentions an ordinary group's + * history leaves out. Returns the threads still worth reading. + */ +async function readThreads( + account: FeishuCredential, + watches: Record, + signals: Signal[], +): Promise> { + const now = Date.now(); + const kept: Record = {}; + for (const watch of Object.values(watches)) { + const idle = now - Number(watch.last); + if (idle > WATCH_MS || (!watch.threadId && idle > PENDING_MS)) continue; + try { + const threadId = watch.threadId ?? (await threadOf(account, watch.messageId)); + if (!threadId) { + kept[watch.messageId] = watch; + continue; + } + const thread = { threadId, chatId: watch.chatId }; + const messages = await threadHistory(account, thread, watch.last); + const asks = asksIn(account, messages, watch.last); + // Someone asking in a thread is carrying on a conversation, so the agent + // gets what came before it. Read once, only when there is an ask. + const whole = asks.length > 0 ? await threadHistory(account, thread, "0") : []; + for (const ask of asks) { + const before = whole.filter((message) => Number(message.createTime) < Number(ask.createTime)); + signals.push( + signalFor( + account, + ask, + before.slice(-EARLIER_LIMIT).map((message) => saidBy(account, message)), + ), + ); + } + const newest = messages.reduce( + (latest, message) => (Number(message.createTime) > Number(latest) ? message.createTime : latest), + watch.last, + ); + const existing = kept[threadId]; + kept[threadId] = { + ...watch, + threadId, + last: existing && Number(existing.last) > Number(newest) ? existing.last : newest, + }; + } catch (error) { + if (isTransient(error)) throw error; + // Deleted, or the bot was taken out of the group: nothing left to read. + } + } + return kept; +} + +function newest(watches: Record): Record { + const entries = Object.entries(watches).sort(([, a], [, b]) => Number(b.last) - Number(a.last)); + return Object.fromEntries(entries.slice(0, WATCH_LIMIT)); +} + function destination( target: Record, source: ActionSource, @@ -114,6 +255,9 @@ export const feishuRuntime: ConnectorRuntime = { const chats = await listChats(account); const next: Record = { ...previous }; const signals: Signal[] = []; + // Answers go in threads, and a group's history does not show what is + // said in them, so the threads are read on their own. + const watches = await readThreads(account, asWatches(cursor), signals); for (const chat of chats) { // A group seen for the first time is read from when the loop last @@ -125,37 +269,21 @@ export const feishuRuntime: ConnectorRuntime = { continue; } next[chat.id] = messages[messages.length - 1]!.createTime; - for (const message of messages) { - if (Number(message.createTime) <= Number(oldest)) continue; - if ( - !isAsk({ - senderId: message.senderId, - senderType: message.senderType, - mentionIds: message.mentionIds, - botOpenId: account.botOpenId, - }) - ) { - continue; - } - const asked: Asked = { + for (const message of asksIn(account, messages, oldest)) { + signals.push(signalFor(account, message)); + // The answer will open a thread under it, which is read from here on. + const key = message.threadId ?? message.id; + if (watches[key]) continue; + watches[key] = { chatId: message.chatId, messageId: message.id, - user: message.senderId, - text: message.text, + last: message.createTime, + ...(message.threadId ? { threadId: message.threadId } : {}), }; - signals.push({ - key: `${message.chatId}:${message.id}`, - kind: "message", - ref: message.id, - title: summarise(message.text), - url: permalink(account.host, message.chatId), - payload: asked as unknown as JsonValue, - at: Number(message.createTime), - }); } } - return { signals, cursor: next }; + return { signals, cursor: { ...next, threads: newest(watches) } as unknown as JsonValue }; }, async resolveWorkItem({ payload }): Promise { @@ -169,7 +297,22 @@ export const feishuRuntime: ConnectorRuntime = { context: [ { name: "MESSAGE.md", - body: [`# What you were asked`, ``, asked.text, ``].join("\n"), + body: [ + ...(asked.earlier?.length + ? [ + `# Earlier in this thread`, + ``, + ...asked.earlier.flatMap((said) => [ + `**${said.by === "bot" ? "You" : "Them"}:** ${said.text}`, + ``, + ]), + ] + : []), + `# What you were asked`, + ``, + asked.text, + ``, + ].join("\n"), }, ], };