Merge pull request #131 from odefun/feat/lark-status-post-edit-limit-177193

fix: rotate Lark status message after edit cap and use post markdown

Author: LIU9293

packages/core/runtime.ts

@@ -148,6 +148,7 @@ export function createCoreRuntime(deps: RuntimeDeps) {
     const finalChunks = splitResultMessage(text);
     const singleChunk = finalChunks[0] ?? text;
     const statusRateLimited = runtimeDeps.im.wasRateLimited?.(channelId, statusTs) ?? false;
+    const statusRateLimitError = runtimeDeps.im.getRateLimitError?.(channelId, statusTs);
 
     if (finalChunks.length > 1) {
       if (statusFormat !== "aggressive" && !statusRateLimited) {
@@ -157,6 +158,7 @@ export function createCoreRuntime(deps: RuntimeDeps) {
           channelId,
           threadId,
           statusTs,
+          ...(statusRateLimitError ? { error: statusRateLimitError } : {}),
         });
       }
 
@@ -176,6 +178,7 @@ export function createCoreRuntime(deps: RuntimeDeps) {
         channelId,
         threadId,
         statusTs,
+        ...(statusRateLimitError ? { error: statusRateLimitError } : {}),
       });
       await runtimeDeps.im.sendMessage(channelId, threadId, singleChunk, true);
       return;

packages/core/runtime/event-stream.ts

@@ -2,6 +2,7 @@ import {
   clearPendingQuestion,
   getPendingQuestion,
   setPendingQuestion,
+  updateActiveRequest,
   type ActiveRequest,
   type TrackedTodo,
   type TrackedTool,
@@ -159,7 +160,7 @@ export async function startEventStreamWatcher(
       onUpdate();
 
       void (async () => {
-        await deps.im.updateMessage(
+        const updatedStatusTs = await deps.im.updateMessage(
           request.channelId,
           request.statusMessageTs,
           buildStatusMessageForAgent({
@@ -171,6 +172,10 @@ export async function startEventStreamWatcher(
           }),
           false
         );
+        if (typeof updatedStatusTs === "string" && updatedStatusTs !== request.statusMessageTs) {
+          request.statusMessageTs = updatedStatusTs;
+          updateActiveRequest(request.channelId, request.threadId, { statusMessageTs: updatedStatusTs });
+        }
         setPendingQuestion(request.channelId, request.threadId, {
           requestId,
           sessionId: properties.sessionID ?? request.sessionId,

packages/core/runtime/message-updates.ts

@@ -7,7 +7,7 @@ type QueuedUpdate = {
   messageTs: string;
   text: string;
   asMarkdown: boolean;
-  resolve: () => void;
+  resolve: (messageTs?: string) => void;
 };
 
 function isRateLimitError(error: unknown): boolean {
@@ -23,6 +23,7 @@ export function createRateLimitedImAdapter(
   const queue: QueuedUpdate[] = [];
   let processing = false;
   const rateLimitedMessages = new Set<string>();
+  const rateLimitErrors = new Map<string, string>();
 
   function key(channelId: string, messageTs: string): string {
     return `${channelId}:${messageTs}`;
@@ -43,10 +44,13 @@ export function createRateLimitedImAdapter(
 
       globalLastUpdateAt = Date.now();
       try {
-        await im.updateMessage(item.channelId, item.messageTs, item.text, item.asMarkdown);
+        const maybeUpdatedTs = await im.updateMessage(item.channelId, item.messageTs, item.text, item.asMarkdown);
+        item.resolve(typeof maybeUpdatedTs === "string" ? maybeUpdatedTs : undefined);
       } catch (error) {
         if (isRateLimitError(error)) {
-          rateLimitedMessages.add(key(item.channelId, item.messageTs));
+          const rateLimitKey = key(item.channelId, item.messageTs);
+          rateLimitedMessages.add(rateLimitKey);
+          rateLimitErrors.set(rateLimitKey, String(error));
           log.warn("IM message update hit rate limit (429)", {
             channelId: item.channelId,
             messageTs: item.messageTs,
@@ -58,9 +62,8 @@ export function createRateLimitedImAdapter(
           messageTs: item.messageTs,
           error: String(error),
         });
+        item.resolve();
       }
-
-      item.resolve();
     }
 
     processing = false;
@@ -74,12 +77,19 @@ export function createRateLimitedImAdapter(
       }
       return rateLimitedMessages.has(key(channelId, messageTs));
     },
+    getRateLimitError: (channelId: string, messageTs: string): string | undefined => {
+      if (typeof im.getRateLimitError === "function") {
+        const upstream = im.getRateLimitError(channelId, messageTs);
+        if (upstream) return upstream;
+      }
+      return rateLimitErrors.get(key(channelId, messageTs));
+    },
     updateMessage: async (
       channelId: string,
       messageTs: string,
       text: string,
       asMarkdown = true
-    ): Promise<void> => {
+    ): Promise<string | undefined> => {
       for (let i = queue.length - 1; i >= 0; i--) {
         const queued = queue[i];
         if (queued && queued.channelId === channelId && queued.messageTs === messageTs) {
@@ -88,7 +98,7 @@ export function createRateLimitedImAdapter(
         }
       }
 
-      return new Promise<void>((resolve) => {
+      return new Promise<string | undefined>((resolve) => {
         queue.push({ channelId, messageTs, text, asMarkdown, resolve });
         void processQueue();
       });

packages/core/runtime/open-request.ts

@@ -57,17 +57,18 @@ export async function runOpenRequest(params: {
 
   const providerLabel = deps.agent.getDisplayNameForSession(sessionId);
 
-  const statusTs = await deps.im.sendMessage(
+  const initialStatusTs = await deps.im.sendMessage(
     context.channelId,
     context.replyThreadId,
     `${providerLabel} is running...`,
     false
   );
 
-  if (!statusTs) {
+  if (!initialStatusTs) {
     log.error("Failed to send status message");
     return null;
   }
+  let statusTs = initialStatusTs;
 
   const request = createActiveRequest(
     sessionId,
@@ -97,7 +98,6 @@ export async function runOpenRequest(params: {
   const result = await runTrackedRequest({
     deps,
     request,
-    statusTs,
     workingPath: cwd,
     stateMachine,
     liveEventHistory,
@@ -126,9 +126,14 @@ export async function runOpenRequest(params: {
         statusMessageFormat: resolveStatusMessageFormat(),
       });
       if (!request.statusFrozen) {
-        await deps.im.updateMessage(context.channelId, statusTs, statusText, false);
+        const updatedStatusTs = await deps.im.updateMessage(context.channelId, statusTs, statusText, false);
+        if (typeof updatedStatusTs === "string" && updatedStatusTs !== statusTs) {
+          statusTs = updatedStatusTs;
+          request.statusMessageTs = updatedStatusTs;
+        }
       }
       updateActiveRequest(context.channelId, context.threadId, {
+        statusMessageTs: request.statusMessageTs,
         currentText: request.currentText,
         tools: request.tools,
         todos: request.todos,

packages/core/runtime/request-runner.ts

@@ -15,7 +15,6 @@ type RunnerDeps = {
 export type RunTrackedRequestParams = {
   deps: RunnerDeps;
   request: ActiveRequest;
-  statusTs: string;
   workingPath: string;
   stateMachine: CoreStateMachine;
   liveEventHistory: Map<string, SessionEvent[]>;
@@ -43,7 +42,6 @@ export async function runTrackedRequest(
   const {
     deps,
     request,
-    statusTs,
     workingPath,
     stateMachine,
     liveEventHistory,
@@ -161,7 +159,7 @@ export async function runTrackedRequest(
     liveParsedState.delete(getStatusMessageKey(request));
 
     const errorStatus = `Error: ${message}\n_${suggestion}_`;
-    await deps.im.updateMessage(request.channelId, statusTs, errorStatus, false);
+    await deps.im.updateMessage(request.channelId, request.statusMessageTs, errorStatus, false);
     onFail(message);
     return { responses: null };
   }

packages/core/runtime/selection-reply.ts

@@ -6,6 +6,7 @@ import {
   loadSession,
   markMessageProcessed,
   saveSession,
+  updateActiveRequest,
 } from "@/config/local/sessions";
 import { resolveStatusMessageFormat } from "@/config";
 import { buildMessageOptions } from "@/core/runtime/message-options";
@@ -72,11 +73,12 @@ export async function handleSelectionReply(params: HandleSelectionReplyParams):
   const providerId = deps.agent.getProviderForSession(sessionId);
   const providerLabel = deps.agent.getDisplayNameForSession(sessionId);
 
-  const statusTs = await deps.im.sendMessage(channelId, threadId, `${providerLabel} is running...`, false);
-  if (!statusTs) {
+  const initialStatusTs = await deps.im.sendMessage(channelId, threadId, `${providerLabel} is running...`, false);
+  if (!initialStatusTs) {
     log.error("Failed to send status message for button selection");
     return;
   }
+  let statusTs = initialStatusTs;
 
   const request = createActiveRequest(sessionId, channelId, threadId, threadId, statusTs, selection);
   const statusMessageKey = getStatusMessageKey(request);
@@ -119,7 +121,6 @@ export async function handleSelectionReply(params: HandleSelectionReplyParams):
   await runTrackedRequest({
     deps,
     request,
-    statusTs,
     workingPath: cwd,
     stateMachine: getStateMachine({ channelId, threadId }),
     liveEventHistory: state.liveEventHistory,
@@ -141,7 +142,12 @@ export async function handleSelectionReply(params: HandleSelectionReplyParams):
         state: state.liveParsedState.get(statusMessageKey),
         statusMessageFormat: resolveStatusMessageFormat(),
       });
-      await deps.im.updateMessage(channelId, statusTs, statusText, false);
+      const updatedStatusTs = await deps.im.updateMessage(channelId, statusTs, statusText, false);
+      if (typeof updatedStatusTs === "string" && updatedStatusTs !== statusTs) {
+        statusTs = updatedStatusTs;
+        request.statusMessageTs = updatedStatusTs;
+        updateActiveRequest(channelId, threadId, { statusMessageTs: updatedStatusTs });
+      }
     },
     onComplete: () => {
       completeActiveRequest(channelId, threadId);

packages/core/test/request-runner.test.ts

@@ -36,7 +36,6 @@ describe("runTrackedRequest", () => {
         } as any,
       },
       request: buildRequest(),
-      statusTs: "100.2",
       workingPath: "/tmp/project",
       stateMachine: new CoreStateMachine("C1:T1"),
       liveEventHistory: new Map(),
@@ -72,7 +71,6 @@ describe("runTrackedRequest", () => {
         } as any,
       },
       request: buildRequest(),
-      statusTs: "100.2",
       workingPath: "/tmp/project",
       stateMachine: new CoreStateMachine("C1:T1"),
       liveEventHistory: new Map(),

packages/core/types.ts

@@ -52,8 +52,14 @@ export type AgentStatusMessageParams = {
 export interface IMAdapter {
   maxEditableMessageChars?: number;
   sendMessage(channelId: string, threadId: string, text: string, asMarkdown?: boolean): Promise<string | undefined>;
-  updateMessage(channelId: string, messageTs: string, text: string, asMarkdown?: boolean): Promise<void>;
+  updateMessage(
+    channelId: string,
+    messageTs: string,
+    text: string,
+    asMarkdown?: boolean
+  ): Promise<string | undefined | void>;
   wasRateLimited?(channelId: string, messageTs: string): boolean;
+  getRateLimitError?(channelId: string, messageTs: string): string | undefined;
   deleteMessage(channelId: string, messageTs: string): Promise<void>;
   fetchThreadHistory(channelId: string, threadId: string, messageId: string): Promise<string | null>;
   buildAgentContext(params: AgentContextBuilderParams): Promise<OpenCodeMessageContext>;