Answer Feishu mentions in a thread, and keep answering inside it

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.

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,

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<FeishuMessage[]> {
+  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<FeishuMessage[]> {
+  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<FeishuMessage[]> {
+  const chatId = container.chatId;
   const parsed: FeishuMessage[] = [];
   let pageToken = "";
   do {
     const query: Record<string, string> = {
-      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,13 +255,31 @@ 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 : "";
   } while (pageToken);
   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<string | null> {
+  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 : "";

src/connectors/feishu/runtime.test.ts

@@ -25,6 +25,8 @@ function givenFeishu(handlers: {
   chats?: Array<{ chat_id: string; name?: string }>;
   history?: Record<string, Array<Record<string, unknown>>>;
   reply?: { message_id: string; chat_id: string };
+  /** Messages looked up one at a time, by id. */
+  messages?: Record<string, Record<string, unknown>>;
   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<string, unknown>) =>
+    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");
   });