From 71e9af9df50172bca311c233216561f83a8c8ca9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B3=9B=E5=87=BD?= <71737444+Ivor-NCUT@users.noreply.github.com> Date: Thu, 30 Jul 2026 17:42:06 +0800 Subject: [PATCH 1/3] fix: acknowledge queued IM tasks immediately --- src/bot/channel.ts | 29 +++++++++++++++++++++-------- 1 file changed, 21 insertions(+), 8 deletions(-) diff --git a/src/bot/channel.ts b/src/bot/channel.ts index 6ca43ef0..cd126977 100644 --- a/src/bot/channel.ts +++ b/src/bot/channel.ts @@ -268,6 +268,7 @@ export async function startChannel(deps: StartChannelDeps): Promise>(); // 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 log.fail('intake', err)); }, @@ -577,6 +580,7 @@ interface IntakeDeps { logThreadModeOverride: LogThreadModeOverride; executor: RunExecutor; pool: ProcessPool; + workingReactions: Map>; } type LogThreadModeOverride = (input: { @@ -600,6 +604,7 @@ async function intakeMessage(deps: IntakeDeps): Promise { 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 @@ -731,10 +736,14 @@ async function intakeMessage(deps: IntakeDeps): Promise { }); 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 }); } @@ -1018,13 +1027,6 @@ async function runAgentBatch(deps: RunBatchDeps): Promise { } : {}; - // 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 +1240,6 @@ async function runAgentBatch(deps: RunBatchDeps): Promise { log.fail('stream', err); } finally { activePolicyFingerprints.delete(scope); - scheduleWorkingReactionCleanup(channel, lastMsg.messageId, reactionPromise); } } @@ -1748,6 +1749,18 @@ function scheduleWorkingReactionCleanup( })(); } +function scheduleBatchWorkingReactionCleanup( + channel: LarkChannel, + batch: NormalizedMessage[], + workingReactions: Map>, +): 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 { return new Promise((resolve) => setTimeout(resolve, ms)); } From 5c681b7f5d61a57d00fd9ba296dcf1d4fcbe84c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B3=9B=E5=87=BD?= <71737444+Ivor-NCUT@users.noreply.github.com> Date: Thu, 30 Jul 2026 17:42:08 +0800 Subject: [PATCH 2/3] test: cover intake reaction before agent start --- .../bot/markdown-stream-startup-failure.test.ts | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/tests/integration/bot/markdown-stream-startup-failure.test.ts b/tests/integration/bot/markdown-stream-startup-failure.test.ts index e4e8bacf..a66829fc 100644 --- a/tests/integration/bot/markdown-stream-startup-failure.test.ts +++ b/tests/integration/bot/markdown-stream-startup-failure.test.ts @@ -82,6 +82,20 @@ afterEach(async () => { }); describe('markdown stream startup failures', () => { + 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('does not leave the IM queue blocked when the agent exits before stream producer starts', async () => { const h = await createHarness(); await startTestBridge(h); From 8301a3aa2663224538486b18d5f9a1a1d5017c45 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B3=9B=E5=87=BD?= <71737444+Ivor-NCUT@users.noreply.github.com> Date: Wed, 5 Aug 2026 08:51:17 +0800 Subject: [PATCH 3/3] fix: acknowledge recovered mentions and startup failures --- src/bot/channel.ts | 21 ++++- src/bot/run-flow.ts | 23 +++++- .../integration/bot/claude-regression.test.ts | 3 +- .../markdown-stream-startup-failure.test.ts | 77 ++++++++++++++++++- 4 files changed, 118 insertions(+), 6 deletions(-) diff --git a/src/bot/channel.ts b/src/bot/channel.ts index cd126977..b8971589 100644 --- a/src/bot/channel.ts +++ b/src/bot/channel.ts @@ -659,6 +659,11 @@ async function intakeMessage(deps: IntakeDeps): Promise { 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) @@ -669,7 +674,7 @@ async function intakeMessage(deps: IntakeDeps): Promise { 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) }), ); @@ -689,7 +694,7 @@ async function intakeMessage(deps: IntakeDeps): Promise { 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; @@ -748,6 +753,18 @@ async function intakeMessage(deps: IntakeDeps): Promise { 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; diff --git a/src/bot/run-flow.ts b/src/bot/run-flow.ts index 884528d3..3d07024a 100644 --- a/src/bot/run-flow.ts +++ b/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 { // 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\''); }); }); diff --git a/tests/integration/bot/markdown-stream-startup-failure.test.ts b/tests/integration/bot/markdown-stream-startup-failure.test.ts index a66829fc..d660c151 100644 --- a/tests/integration/bot/markdown-stream-startup-failure.test.ts +++ b/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,48 @@ 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); @@ -96,6 +139,26 @@ describe('markdown stream startup failures', () => { 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); @@ -452,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; @@ -474,6 +538,7 @@ async function createHarness(options: { }, access: { allowedUsers: ['ou_user'], + allowedChats: ['oc_group'], }, codex: { binaryPath: '/usr/local/bin/codex', @@ -503,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); @@ -635,7 +705,11 @@ function createControls(profileConfig: ReturnType = {}, +): NormalizedMessage { return { messageId, chatId: 'oc_dm', @@ -647,6 +721,7 @@ function message(messageId: string, content: string): NormalizedMessage { resources: [], mentionedBot: false, createTime: 1760000001000, + ...overrides, } as unknown as NormalizedMessage; }