fix: acknowledge IM tasks before debounce and queue wait

src/bot/channel.ts

@@ -268,6 +268,7 @@ export async function startChannel(deps: StartChannelDeps): Promise<BridgeChanne
 
   const channel = createLarkChannel(opts);
   const media = new MediaCache(channel, deps.appPaths?.mediaDir);
+  const workingReactions = new Map<string, Promise<string | undefined>>();
 
   // Pending → run handoff: while a run is active on a chat, block its pending
   // queue so messages keep accumulating without flushing. When the run ends,
@@ -320,6 +321,7 @@ export async function startChannel(deps: StartChannelDeps): Promise<BridgeChanne
       } catch (err) {
         log.fail('flush', err);
       } finally {
+        scheduleBatchWorkingReactionCleanup(channel, batch, workingReactions);
         pending.unblock(scope);
         log.info('flush', 'end');
       }
@@ -346,6 +348,7 @@ export async function startChannel(deps: StartChannelDeps): Promise<BridgeChanne
           logThreadModeOverride,
           executor,
           pool,
+          workingReactions,
         }),
       ).catch((err) => log.fail('intake', err));
     },
@@ -577,6 +580,7 @@ interface IntakeDeps {
   logThreadModeOverride: LogThreadModeOverride;
   executor: RunExecutor;
   pool: ProcessPool;
+  workingReactions: Map<string, Promise<string | undefined>>;
 }
 
 type LogThreadModeOverride = (input: {
@@ -600,6 +604,7 @@ async function intakeMessage(deps: IntakeDeps): Promise<void> {
     logThreadModeOverride,
     executor,
     pool,
+    workingReactions,
   } = deps;
   const preview = msg.content.length > 80 ? `${msg.content.slice(0, 80)}…` : msg.content;
   // Resolve scope (and underlying chat mode) once at intake — every
@@ -654,6 +659,11 @@ async function intakeMessage(deps: IntakeDeps): Promise<void> {
     resources: msg.resources.length,
   });
 
+  const mentionedBot = messageMentionsBot(emsg, channel.botIdentity);
+  if (mentionedBot && !emsg.mentionedBot) {
+    log.info('intake', 'mention-recovered', { scope, msgId: emsg.messageId });
+  }
+
   const accessDecision =
     msg.chatType === 'p2p'
       ? canUseDm(controls.profileConfig, controls, msg.senderId)
@@ -664,7 +674,7 @@ async function intakeMessage(deps: IntakeDeps): Promise<void> {
       sender: msg.senderId.slice(-6),
       reason: accessDecision.reason,
     });
-    if (msg.chatType !== 'p2p' && accessDecision.reason === 'denied-chat' && msg.mentionedBot) {
+    if (msg.chatType !== 'p2p' && accessDecision.reason === 'denied-chat' && mentionedBot) {
       void sendNonAllowedGroupHint(channel, msg.chatId, msg.messageId).catch((err) =>
         log.warn('intake', 'non-allowed-hint-failed', { err: String(err) }),
       );
@@ -684,7 +694,7 @@ async function intakeMessage(deps: IntakeDeps): Promise<void> {
   if (
     msg.chatType !== 'p2p' &&
     requireMentionForChat(controls.profileConfig, controls.cfg, msg.chatId) &&
-    !msg.mentionedBot
+    !mentionedBot
   ) {
     log.info('intake', 'skip-no-mention', { scope, chatType: msg.chatType });
     return;
@@ -731,14 +741,30 @@ async function intakeMessage(deps: IntakeDeps): Promise<void> {
   });
   if (handled) {
     const dropped = pending.cancel(scope);
+    scheduleBatchWorkingReactionCleanup(channel, dropped, workingReactions);
     log.info('intake', 'command', { scope, droppedPending: dropped.length });
     return;
   }
 
+  if (!workingReactions.has(emsg.messageId)) {
+    workingReactions.set(emsg.messageId, addWorkingReaction(channel, emsg.messageId));
+  }
   const size = pending.push(scope, emsg);
   log.info('intake', 'queued', { scope, queueSize: size, debounceMs: DEBOUNCE_MS });
 }
 
+function messageMentionsBot(
+  msg: NormalizedMessage,
+  botIdentity: LarkChannel['botIdentity'],
+): boolean {
+  if (msg.mentionedBot) return true;
+  return (msg.mentions ?? []).some(
+    (mention) =>
+      Boolean(botIdentity?.openId && mention.openId === botIdentity.openId) ||
+      Boolean(botIdentity?.userId && mention.userId === botIdentity.userId),
+  );
+}
+
 interface RunBatchDeps {
   channel: LarkChannel;
   executor: RunExecutor;
@@ -1018,13 +1044,6 @@ async function runAgentBatch(deps: RunBatchDeps): Promise<void> {
       }
     : {};
 
-  // For non-card modes Claude's output doesn't surface visually until either
-  // a first streamed token (markdown mode) or the whole run ends (text mode).
-  // Add a "Typing" reaction to the triggering message as an instant ack, but
-  // never let that outbound API call block agent event draining.
-  const reactionPromise =
-    cotEnabled || replyMode === 'card' ? undefined : addWorkingReaction(channel, lastMsg.messageId);
-
   try {
     if (cotEnabled) {
       const cotPublisher = new CotPublisher({
@@ -1238,7 +1257,6 @@ async function runAgentBatch(deps: RunBatchDeps): Promise<void> {
     log.fail('stream', err);
   } finally {
     activePolicyFingerprints.delete(scope);
-    scheduleWorkingReactionCleanup(channel, lastMsg.messageId, reactionPromise);
   }
 }
 
@@ -1748,6 +1766,18 @@ function scheduleWorkingReactionCleanup(
   })();
 }
 
+function scheduleBatchWorkingReactionCleanup(
+  channel: LarkChannel,
+  batch: NormalizedMessage[],
+  workingReactions: Map<string, Promise<string | undefined>>,
+): void {
+  for (const msg of batch) {
+    const reactionPromise = workingReactions.get(msg.messageId);
+    workingReactions.delete(msg.messageId);
+    scheduleWorkingReactionCleanup(channel, msg.messageId, reactionPromise);
+  }
+}
+
 function delay(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }

src/bot/run-flow.ts

@@ -16,7 +16,12 @@ import {
   type WorkingDirectoryResolveResult,
 } from '../policy/workspace';
 import type { RunExecution, RunExecutor } from '../runtime/run-executor';
-import { RunRejected, type RunRejectedCode } from '../runtime/errors';
+import {
+  RunRejected,
+  SpawnFailed,
+  type RunRejectedCode,
+  type SpawnFailedCode,
+} from '../runtime/errors';
 import type { SessionCatalog } from '../session/catalog';
 import type { SessionStore } from '../session/store';
 import type { WorkspaceStore } from '../workspace/store';
@@ -46,7 +51,11 @@ export interface StartRunFlowInput {
 export type RunFlowRejectCode =
   | WorkingDirectoryRejectReason
   | RunPolicyReject['rejectReason']['code']
-  | RunRejectedCode;
+  | RunRejectedCode
+  | SpawnFailedCode;
+
+export const RUN_START_FAILED_MESSAGE =
+  '任务启动失败，请稍后重试。如果问题持续，请联系管理员检查 Agent 配置。';
 
 export type StartRunFlowResult =
   | {
@@ -159,6 +168,16 @@ export async function startRunFlow(input: StartRunFlowInput): Promise<StartRunFl
       observability: input.observability,
     });
   } catch (err) {
+    if (err instanceof SpawnFailed) {
+      return {
+        ok: false,
+        rejectReason: {
+          code: err.code,
+          userVisible: RUN_START_FAILED_MESSAGE,
+        },
+        workspace,
+      };
+    }
     if (err instanceof RunRejected) {
       return {
         ok: false,

tests/integration/bot/claude-regression.test.ts

@@ -60,7 +60,8 @@ describe('Claude IM regression boundaries', () => {
     // The group-mention gate honors a per-chat override first, then the global
     // setting (both resolved by requireMentionForChat).
     expect(source).toContain('requireMentionForChat(controls.profileConfig, controls.cfg, msg.chatId)');
-    expect(source).toContain('!msg.mentionedBot');
+    expect(source).toContain('const mentionedBot = messageMentionsBot(emsg, channel.botIdentity)');
+    expect(source).toContain('!mentionedBot');
     expect(source).toContain('msg.chatType !== \'p2p\'');
   });
 });

tests/integration/bot/markdown-stream-startup-failure.test.ts

@@ -3,6 +3,7 @@ import { realpath } from 'node:fs/promises';
 import { join } from 'node:path';
 import { afterEach, describe, expect, it, vi } from 'vitest';
 import type { AgentEvent } from '../../../src/agent/types.js';
+import { RUN_START_FAILED_MESSAGE } from '../../../src/bot/run-flow.js';
 import type { FakeAgentEvents } from '../../helpers/fake-agent.js';
 import { createDefaultProfileConfig } from '../../../src/config/profile-schema.js';
 import { log } from '../../../src/core/logger.js';
@@ -82,6 +83,82 @@ afterEach(async () => {
 });
 
 describe('markdown stream startup failures', () => {
+  it('acknowledges an accepted group mention before the debounced agent run starts', async () => {
+    const h = await createHarness();
+    await startTestBridge(h);
+
+    await h.channel.handlers.message?.(
+      message('om_mention', 'run', {
+        chatId: 'oc_group',
+        chatType: 'group',
+        mentionedBot: true,
+      }),
+    );
+
+    expect(h.channel.rawClient.im.v1.messageReaction.create).toHaveBeenCalledWith({
+      path: { message_id: 'om_mention' },
+      data: { reaction_type: { emoji_type: 'Typing' } },
+    });
+    expect(h.agent.runOptions).toHaveLength(0);
+    await waitFor(() => h.agent.runOptions.length === 1);
+  });
+
+  it('recovers a structured bot mention when the SDK mentionedBot flag is false', async () => {
+    const h = await createHarness();
+    await startTestBridge(h);
+
+    await h.channel.handlers.message?.(
+      message('om_rich_mention', 'long post', {
+        chatId: 'oc_group',
+        chatType: 'group',
+        rawContentType: 'post',
+        mentionedBot: false,
+        mentions: [{ key: '@_user_1', openId: 'ou_bot', name: 'Bridge', isBot: true }],
+      }),
+    );
+
+    expect(h.channel.rawClient.im.v1.messageReaction.create).toHaveBeenCalledWith({
+      path: { message_id: 'om_rich_mention' },
+      data: { reaction_type: { emoji_type: 'Typing' } },
+    });
+    expect(h.agent.runOptions).toHaveLength(0);
+    await waitFor(() => h.agent.runOptions.length === 1);
+  });
+
+  it('adds the working reaction before the debounced agent run starts', async () => {
+    const h = await createHarness();
+    await startTestBridge(h);
+
+    await h.channel.handlers.message?.(message('om_first', 'first'));
+
+    expect(h.channel.rawClient.im.v1.messageReaction.create).toHaveBeenCalledWith({
+      path: { message_id: 'om_first' },
+      data: { reaction_type: { emoji_type: 'Typing' } },
+    });
+    expect(h.agent.runOptions).toHaveLength(0);
+    await waitFor(() => h.agent.runOptions.length === 1);
+  });
+
+  it('replies with a fixed message and clears the reaction when agent startup fails', async () => {
+    const h = await createHarness({ spawnError: new Error('private spawn detail') });
+    await startTestBridge(h);
+
+    await h.channel.handlers.message?.(
+      message('om_spawn_failure', 'run', {
+        chatId: 'oc_group',
+        chatType: 'group',
+        mentionedBot: true,
+      }),
+    );
+
+    expect(h.channel.rawClient.im.v1.messageReaction.create).toHaveBeenCalled();
+    await waitFor(() => h.channel.sent.length === 1);
+    expect(lastMarkdown(h.channel)).toBe(RUN_START_FAILED_MESSAGE);
+    expect(lastMarkdown(h.channel)).not.toContain('private spawn detail');
+    expect(h.channel.sent[0]?.options).toMatchObject({ replyTo: 'om_spawn_failure' });
+    await waitFor(() => h.channel.rawClient.im.v1.messageReaction.delete.mock.calls.length === 1);
+  });
+
   it('does not leave the IM queue blocked when the agent exits before stream producer starts', async () => {
     const h = await createHarness();
     await startTestBridge(h);
@@ -438,6 +515,7 @@ async function createHarness(options: {
   messageReply?: 'card' | 'markdown' | 'text';
   /** Codex holds its answer back for a dedicated final reply; Claude streams it. */
   agentKind?: 'claude' | 'codex';
+  spawnError?: Error;
 } = {}): Promise<{
   tmp: TmpProfile;
   channel: FakeLarkChannel;
@@ -460,6 +538,7 @@ async function createHarness(options: {
     },
     access: {
       allowedUsers: ['ou_user'],
+      allowedChats: ['oc_group'],
     },
     codex: {
       binaryPath: '/usr/local/bin/codex',
@@ -489,6 +568,11 @@ async function createHarness(options: {
       [{ type: 'done', terminationReason: 'normal' }],
     ],
   });
+  if (options.spawnError) {
+    vi.spyOn(agent, 'run').mockImplementation(() => {
+      throw options.spawnError;
+    });
+  }
   const channel = createFakeLarkChannel(options);
   sdkMock.channel = channel;
   const controls = createControls(profileConfig);
@@ -621,7 +705,11 @@ function createControls(profileConfig: ReturnType<typeof createDefaultProfileCon
   };
 }
 
-function message(messageId: string, content: string): NormalizedMessage {
+function message(
+  messageId: string,
+  content: string,
+  overrides: Partial<NormalizedMessage> = {},
+): NormalizedMessage {
   return {
     messageId,
     chatId: 'oc_dm',
@@ -633,6 +721,7 @@ function message(messageId: string, content: string): NormalizedMessage {
     resources: [],
     mentionedBot: false,
     createTime: 1760000001000,
+    ...overrides,
   } as unknown as NormalizedMessage;
 }