package vip.mate.channel; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Component; import reactor.core.publisher.Flux; import vip.mate.agent.AgentService; import vip.mate.agent.context.ChatOrigin; import vip.mate.agent.model.AgentEntity; import vip.mate.approval.ApprovalWorkflowService; import vip.mate.approval.ResolveOutcome; import vip.mate.approval.PendingApproval; import vip.mate.channel.event.ChannelMessageReceivedEvent; import vip.mate.channel.model.ChannelEntity; import vip.mate.channel.notification.ApprovalNotificationService; import vip.mate.channel.service.ChannelService; import vip.mate.channel.web.AgentStreamAccumulator; import vip.mate.channel.web.ChatStreamTracker; import vip.mate.exception.MateClawException; import vip.mate.llm.model.ModelConfigEntity; import vip.mate.llm.service.ModelConfigService; import vip.mate.memory.event.ConversationCompletionPublisher; import vip.mate.tts.TtsService; import vip.mate.workspace.conversation.ConversationService; import vip.mate.workspace.conversation.model.ConversationEntity; import vip.mate.workspace.conversation.model.MessageContentPart; import vip.mate.workspace.core.service.ChatUploadLocationResolver; import vip.mate.workspace.conversation.model.MessageEntity; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.time.Duration; import java.util.HashMap; import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.concurrent.*; import java.util.concurrent.locks.ReentrantLock; /** * 渠道消息路由器 *

* 采用每渠道独立队列架构： * - 每渠道一个 BlockingQueue，N 个消费线程从队列取消息处理 * - 会话级锁保证同一 conversationId 串行处理 * - 500ms 防抖：同一会话的连续消息合并为一条 * - Web 渠道不走队列（有自己的 SSE 流程） * * @author MateClaw Team */ @Slf4j @Component public class ChannelMessageRouter { private final AgentService agentService; private final ConversationService conversationService; private final ChannelService channelService; private final ChannelSessionStore channelSessionStore; private final ApprovalWorkflowService approvalService; private final ApprovalNotificationService approvalNotificationService; private final ConversationCompletionPublisher completionPublisher; private final TtsService ttsService; private final ObjectMapper objectMapper; private final ChatStreamTracker streamTracker; private final ChannelChatOriginFactory chatOriginFactory; private final ChannelErrorClassifier errorClassifier; private final InboundMessageDeduplicator inboundDedup; /** Field-injected (rather than constructor) to avoid a signature * change that would ripple through every test that constructs the * router directly. Spring's stock publisher is always available. */ @Autowired(required = false) private ApplicationEventPublisher events; /** Field-injected for the same reason as {@link #events}: the chat-upload * resolver resolves the workspace-aware TTS output directory on the * voice-reply path. Optional so tests that build the router directly * still work; falls back to the legacy default dir when unset. */ @Autowired(required = false) private vip.mate.workspace.core.service.ChatUploadLocationResolver chatUploadLocationResolver; /** Field-injected for the same reason as {@link #events}: backs the * /model magic command (list + switch). Optional so tests that build * the router directly still work; when unset the command degrades to * a "service unavailable" reply instead of failing message intake. */ @Autowired(required = false) private ModelConfigService modelConfigService; /** Field-injected so the IM sync path can scrub hallucinated * {@code /api/v1/files/generated/{id}} URLs (LLM wrote a UUID-shaped * link without ever calling a render tool). The graph's FinalAnswerNode * already does this, but the IM sync path accumulates {@code delta.content()} * directly and bypasses FinalAnswerNode — without this scrub, the fake * URL reaches the IM channel as a clickable link that 404s. */ @Autowired(required = false) private vip.mate.tool.document.GeneratedFileCache generatedFileCache; /** 队列条目：封装消息及其路由上下文 */ private record QueueEntry(ChannelMessage message, ChannelAdapter adapter, ChannelEntity channelEntity) {} /** 每个渠道类型的消息队列 */ private final ConcurrentHashMap> channelQueues = new ConcurrentHashMap<>(); /** 每个渠道类型的消费线程池 */ private final ConcurrentHashMap channelExecutors = new ConcurrentHashMap<>(); /** 会话级别的锁：保证同一 conversationId 串行处理 */ private final ConcurrentHashMap sessionLocks = new ConcurrentHashMap<>(); /** 防抖调度器 */ private final ScheduledExecutorService debounceScheduler = Executors.newSingleThreadScheduledExecutor(r -> { Thread t = new Thread(r, "channel-debounce-scheduler"); t.setDaemon(true); return t; }); /** 防抖缓冲区：conversationId -> 待合并消息 */ private final ConcurrentHashMap pendingMessages = new ConcurrentHashMap<>(); /** 每个渠道的消费线程数 */ private static final int CONSUMERS_PER_CHANNEL = 4; /** 每个渠道的队列容量 */ private static final int QUEUE_CAPACITY = 1000; /** 防抖等待时间（毫秒）。Package-private for unit-test access. */ static final long DEBOUNCE_MS = 500; /** * Extended debounce window for suspected paste-split scenarios. WeCom * (and other IM clients) silently split a single pasted long prompt * into 2-4 separate messages when it exceeds the per-frame limit * (~2000 chars). The fragments arrive 0.5-2s apart, which means the * default {@link #DEBOUNCE_MS} flushes the first fragment before the * second one arrives — the agent then sees a torn context, calls the * LLM on a partial prompt, and gets re-triggered when the next * fragment lands. When merged content exceeds * {@link #LONG_TEXT_THRESHOLD} we extend the window so the merger has * time to absorb the rest. *

* Package-private for unit-test access. */ static final long LONG_DEBOUNCE_MS = 2500; /** * Content length (chars) above which we treat the message as a likely * paste-split fragment. 1500 sits below the typical ~2000-char IM * client split point while staying well above any normally-typed * message, so the long-debounce path doesn't penalize ordinary * chatting. A short typed "hello" still flushes in 500ms. *

* Package-private for unit-test access. */ static final int LONG_TEXT_THRESHOLD = 1500; /** * Pick the debounce window: extend to {@link #LONG_DEBOUNCE_MS} when * either the new arrival or the accumulated merged buffer looks like * a paste-split fragment, otherwise stay at {@link #DEBOUNCE_MS}. *

* Package-private + static so tests can pin the threshold without * spinning up the whole router (which has 12+ injected dependencies). */ static long pickDebounceMs(int currentMergedLength) { return currentMergedLength > LONG_TEXT_THRESHOLD ? LONG_DEBOUNCE_MS : DEBOUNCE_MS; } /** 是否已关闭 */ private volatile boolean shutdown = false; public ChannelMessageRouter(AgentService agentService, ConversationService conversationService, ChannelService channelService, ChannelSessionStore channelSessionStore, ApprovalWorkflowService approvalService, ApprovalNotificationService approvalNotificationService, ConversationCompletionPublisher completionPublisher, TtsService ttsService, ObjectMapper objectMapper, ChatStreamTracker streamTracker, ChannelChatOriginFactory chatOriginFactory, ChannelErrorClassifier errorClassifier, InboundMessageDeduplicator inboundDedup) { this.agentService = agentService; this.conversationService = conversationService; this.channelService = channelService; this.channelSessionStore = channelSessionStore; this.approvalService = approvalService; this.approvalNotificationService = approvalNotificationService; this.completionPublisher = completionPublisher; this.ttsService = ttsService; this.objectMapper = objectMapper; this.streamTracker = streamTracker; this.chatOriginFactory = chatOriginFactory; this.errorClassifier = errorClassifier; this.inboundDedup = inboundDedup; } // ==================== 防抖辅助类 ==================== /** * 防抖待合并消息 */ private static class PendingMessage { final ChannelAdapter adapter; final ChannelEntity channelEntity; final ChannelMessage firstMessage; final StringBuilder mergedContent; volatile ScheduledFuture timer; PendingMessage(ChannelMessage message, ChannelAdapter adapter, ChannelEntity channelEntity) { this.firstMessage = message; this.adapter = adapter; this.channelEntity = channelEntity; this.mergedContent = new StringBuilder(message.getContent() != null ? message.getContent() : ""); } synchronized void appendContent(String content) { if (content != null && !content.isBlank()) { if (!mergedContent.isEmpty()) { mergedContent.append('\n'); } mergedContent.append(content); } } synchronized String getMergedContent() { return mergedContent.toString(); } } // ==================== 入队（替代原 route 方法） ==================== /** * 将渠道消息入队到对应渠道的处理队列（防抖后入队）。 *

* Webhook 调用此方法后立即返回，不阻塞。 * * @param message 入站消息 * @param adapter 来源渠道适配器（用于回复） * @param channelEntity 渠道配置（含关联 agentId） */ public void enqueue(ChannelMessage message, ChannelAdapter adapter, ChannelEntity channelEntity) { // The adapter caches the ChannelEntity it was constructed with, so a // long-lived adapter (e.g. Feishu WS) keeps handing us a snapshot // that may be stale by the time the message arrives. Refresh from // the DB so a freshly-rebound agent (or any other routing-metadata // change applied without a restart) is honoured immediately. ChannelEntity fresh = freshChannelEntity(channelEntity); if (fresh == null) { // Channel deleted between adapter start and message arrival. // Skip everything — even the trigger publish, since the channel // no longer exists for downstream consumers to reference. return; } // Only drop on an EXPLICIT enabled=false. A null enabled (which the // production DB never returns but tests / hand-constructed entities // do) means "not declared", and treating it as disabled would // collapse every downstream behaviour into a silent drop — which is // exactly how the previous !Boolean.TRUE.equals(...) form regressed // mock-driven tests that don't bother seeding the flag. if (Boolean.FALSE.equals(fresh.getEnabled())) { log.warn("[{}] Channel {} (id={}) is disabled; dropping message from {}", adapter.getChannelType(), fresh.getName(), fresh.getId(), message.getSenderId()); return; } channelEntity = fresh; // Inbound idempotency, before ANY side effect (magic commands, trigger // fan-out, agent turn). IM platforms redeliver a message whose ack was // late, lost, or non-200; without this claim every redelivery runs its // own agent turn and the user gets the same answer again. Claiming here // rather than in each adapter means every channel — including the four // that never had dedup — is covered by one code path. if (!inboundDedup.claim(channelEntity.getId(), inboundIdentity(message))) { log.info("[{}] Duplicate inbound message (id={}) on channel {}; dropping", adapter.getChannelType(), inboundIdentity(message), channelEntity.getId()); return; } String conversationId = buildConversationId(message, channelEntity.getId()); if (handleMagicCommand(message, adapter, channelEntity, conversationId)) { return; } // Fan out to the trigger pipeline FIRST — channel_message and // content_match triggers fire on every received message regardless // of whether the channel has an agent attached. If we returned // early on a missing agent below without publishing, the workflow // side would silently lose every channel-event that doesn't also // route to a chat agent. publishChannelEvent(message, adapter, channelEntity); Long agentId = channelEntity.getAgentId(); if (agentId == null) { log.warn("Channel {} has no associated agent, ignoring message from {}", channelEntity.getName(), message.getSenderId()); return; } if (shutdown) { log.warn("Router is shutting down, rejecting message from {}", message.getSenderId()); return; } String channelType = adapter.getChannelType(); log.info("[{}] Enqueuing message: sender={}, conversationId={}, agentId={}", channelType, message.getSenderId(), conversationId, agentId); // Debounce + adaptive merge: same conversation messages within the // (500ms / 2.5s) window get concatenated into one. Adaptive: when // the merged buffer crosses the LONG_TEXT_THRESHOLD we extend to // LONG_DEBOUNCE_MS so paste-split fragments arrive together // instead of triggering one agent call per piece. synchronized (pendingMessages) { PendingMessage existing = pendingMessages.get(conversationId); if (existing != null) { // Sender boundary in groups: when a different user sends to the // same group within the debounce window, merging would attribute // both fragments to whoever sent first — the LLM then loses the // ability to tell who asked what. Flush the existing buffer // immediately so each user's text rides its own pending window. // Reentrant on `pendingMessages`, so the inner flushPending's // synchronized block re-acquires safely on the same thread. String existingSender = existing.firstMessage.getSenderId(); String incomingSender = message.getSenderId(); boolean sameSender = isSameSender(existingSender, incomingSender); if (!sameSender) { log.info("[{}] Sender boundary in conversation {}: flushing pending from sender={}, accepting new sender={}", channelType, conversationId, existingSender, incomingSender); if (existing.timer != null) { existing.timer.cancel(false); } flushPending(conversationId); // Fall through to create a fresh pending for the new sender. } else { // Same sender — original paste-split / rapid-follow merge path. if (existing.timer != null) { existing.timer.cancel(false); } existing.appendContent(message.getContent()); int mergedLen = existing.getMergedContent().length(); long debounceMs = pickDebounceMs(mergedLen); existing.timer = debounceScheduler.schedule( () -> flushPending(conversationId), debounceMs, TimeUnit.MILLISECONDS); if (debounceMs > DEBOUNCE_MS) { log.info("[{}] Long-text merger active: conversationId={}, mergedLen={}, debounce={}ms (paste-split suspected)", channelType, conversationId, mergedLen, debounceMs); } else { log.debug("[{}] Message merged with pending (debounce {}ms): conversationId={}", channelType, debounceMs, conversationId); } return; } } // 首条消息（或 sender boundary 之后的新 sender），创建 PendingMessage 并设定防抖定时器 PendingMessage pending = new PendingMessage(message, adapter, channelEntity); pendingMessages.put(conversationId, pending); int firstLen = message.getContent() != null ? message.getContent().length() : 0; long debounceMs = pickDebounceMs(firstLen); pending.timer = debounceScheduler.schedule( () -> flushPending(conversationId), debounceMs, TimeUnit.MILLISECONDS); if (debounceMs > DEBOUNCE_MS) { log.info("[{}] Long-text merger armed on first message: conversationId={}, len={}, debounce={}ms", channelType, conversationId, firstLen, debounceMs); } } } /** * Publish a {@link ChannelMessageReceivedEvent} so the trigger module's * bridge can fan the message out to channel_message + content_match * triggers. Best-effort — a publish failure must never block the * primary chat-routing path. {@code messageId} is used as the dedup * key downstream so repeated webhook deliveries can't double-fire * the same trigger. */ private void publishChannelEvent(ChannelMessage message, ChannelAdapter adapter, ChannelEntity channelEntity) { if (events == null || message == null || adapter == null || channelEntity == null) return; try { long ws = channelEntity.getWorkspaceId() == null ? 0L : channelEntity.getWorkspaceId(); String channelType = adapter.getChannelType(); // Reuse the identity the inbound claim uses so both agree on what // "the same message" is. Unlike the claim — which the deduplicator // scopes by channel id — this key travels to the trigger pipeline // unscoped, so anything we derive ourselves keeps the channelType // prefix: two channels can otherwise produce the same // sender+timestamp pair inside one workspace. String platformId = message.getMessageId(); String messageId = (platformId != null && !platformId.isBlank()) ? platformId : channelType + ":" + (inboundIdentity(message) != null ? inboundIdentity(message) : message.getSenderId() + "@" + System.currentTimeMillis()); events.publishEvent(new ChannelMessageReceivedEvent( ws, channelType, messageId, message.getSenderId(), message.getSenderName(), message.getChatId(), message.getContent())); } catch (Exception e) { log.warn("[ChannelMessageRouter] event publish failed for sender {}: {}", message.getSenderId(), e.getMessage()); } } /** * The stable identity of an inbound message, used both for the inbound * dedup claim and as the trigger pipeline's dedup key. * *

Prefers the platform message id — every adapter that has one puts it * on {@link ChannelMessage#getMessageId()}, and a redelivery carries the * same value. Adapters whose stable token is not the raw message id (WeCom * uses its {@code context_token}) put that token there instead. * *

Falls back to {@code sender@timestamp} when there is no id but the * platform stamped the message — still stable across redeliveries of the * same payload. Returns {@code null} when neither exists: there is nothing * to tell a redelivery apart from a fresh message, so the caller must fail * open rather than guess. * *

Package-private for unit-test access. */ static String inboundIdentity(ChannelMessage message) { if (message == null) { return null; } String messageId = message.getMessageId(); if (messageId != null && !messageId.isBlank()) { return messageId; } if (message.getTimestamp() == null) { return null; } return message.getSenderId() + "@" + message.getTimestamp(); } /** * Has this inbound message already been claimed? A peek, not a claim — * the authoritative claim happens once, in {@link #enqueue}. * *

For adapters to call before expensive inbound work (media download, * payload decryption) so a known redelivery costs nothing. Adapters reach * it through the router they already hold, which keeps the deduplicator * out of every adapter constructor. * * @param identity the same value the adapter will put on * {@link ChannelMessage#getMessageId()} */ public boolean isDuplicateInbound(Long channelId, String identity) { return inboundDedup.contains(channelId, identity); } /** * 防抖到期：将合并后的消息真正放入渠道队列 */ private void flushPending(String conversationId) { PendingMessage pending; synchronized (pendingMessages) { pending = pendingMessages.remove(conversationId); } if (pending == null) return; // 更新消息内容为合并后的文本 pending.firstMessage.setContent(pending.getMergedContent()); String channelType = pending.adapter.getChannelType(); LinkedBlockingQueue queue = channelQueues.computeIfAbsent(channelType, this::createChannelQueue); boolean offered = queue.offer(new QueueEntry(pending.firstMessage, pending.adapter, pending.channelEntity)); if (!offered) { log.error("[{}] Message queue full (capacity={}), dropping message from {}", channelType, QUEUE_CAPACITY, pending.firstMessage.getSenderId()); // Never handed off — give the claim back so the platform's own // retry can still get an answer. A turn that ran and *failed* // keeps its claim: the user already got the error reply, and a // retry would only produce a second one. inboundDedup.release(pending.channelEntity != null ? pending.channelEntity.getId() : null, inboundIdentity(pending.firstMessage)); try { String replyTarget = resolveReplyTarget(pending.firstMessage); pending.adapter.sendMessage(replyTarget, "系统繁忙，请稍后再试"); } catch (Exception e) { log.error("[{}] Failed to send busy message: {}", channelType, e.getMessage()); } } else { log.debug("[{}] Message flushed to queue: conversationId={}, queueSize={}", channelType, conversationId, queue.size()); } } // ==================== 消费线程 ==================== /** * 为渠道类型创建队列并启动消费线程 */ private LinkedBlockingQueue createChannelQueue(String channelType) { LinkedBlockingQueue queue = new LinkedBlockingQueue<>(QUEUE_CAPACITY); ExecutorService executor = Executors.newFixedThreadPool(CONSUMERS_PER_CHANNEL, new ThreadFactory() { private int counter = 0; @Override public Thread newThread(Runnable r) { Thread t = new Thread(r, "channel-consumer-" + channelType + "-" + (counter++)); t.setDaemon(true); return t; } }); for (int i = 0; i < CONSUMERS_PER_CHANNEL; i++) { executor.execute(() -> consumeLoop(channelType, queue)); } channelExecutors.put(channelType, executor); log.info("[{}] Created message queue (capacity={}) with {} consumer threads", channelType, QUEUE_CAPACITY, CONSUMERS_PER_CHANNEL); return queue; } /** * 消费线程循环：从队列取消息，加会话锁后串行处理 */ private void consumeLoop(String channelType, LinkedBlockingQueue queue) { log.info("[{}] Consumer thread started: {}", channelType, Thread.currentThread().getName()); while (!shutdown) { try { QueueEntry entry = queue.poll(1, TimeUnit.SECONDS); if (entry == null) { continue; // 超时，重新检查 shutdown 标志 } String conversationId = buildConversationId(entry.message(), entry.channelEntity().getId()); ReentrantLock lock = sessionLocks.computeIfAbsent(conversationId, k -> new ReentrantLock()); lock.lock(); try { processMessage(entry.message(), entry.adapter(), entry.channelEntity(), conversationId); } finally { lock.unlock(); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { log.error("[{}] Unexpected error in consumer loop: {}", channelType, e.getMessage(), e); } } log.info("[{}] Consumer thread stopped: {}", channelType, Thread.currentThread().getName()); } // ==================== 审批命令识别 ==================== private static final java.util.Set APPROVE_COMMANDS = java.util.Set.of( "approve", "/approve", "批准", "/批准"); private static final java.util.Set DENY_COMMANDS = java.util.Set.of( "deny", "/deny", "拒绝", "/拒绝"); /** 带 pendingId 的审批命令格式：/approve a1b2c3 */ private static final java.util.regex.Pattern APPROVE_WITH_ID = java.util.regex.Pattern.compile("^/?(approve|批准)\\s+([a-f0-9]{6,16})$", java.util.regex.Pattern.CASE_INSENSITIVE); private static final java.util.regex.Pattern DENY_WITH_ID = java.util.regex.Pattern.compile("^/?(deny|拒绝)\\s+([a-f0-9]{6,16})$", java.util.regex.Pattern.CASE_INSENSITIVE); private boolean isApproveCommand(String text) { String t = text.toLowerCase().strip(); return APPROVE_COMMANDS.contains(t) || APPROVE_WITH_ID.matcher(t).matches(); } private boolean isDenyCommand(String text) { String t = text.toLowerCase().strip(); return DENY_COMMANDS.contains(t) || DENY_WITH_ID.matcher(t).matches(); } /** * 从审批命令中提取 shortId（如 "/approve a1b2c3" → "a1b2c3"），无 id 则返回 null */ private String extractShortId(String text) { java.util.regex.Matcher m = APPROVE_WITH_ID.matcher(text.strip()); if (m.matches()) return m.group(2); m = DENY_WITH_ID.matcher(text.strip()); if (m.matches()) return m.group(2); return null; } /** * Identity gate shared by /approve and /deny (group-chat safety): only the * original human requester may resolve a pending. Agent/cron ("system") and * unattributed (null) approvals are fail-closed in IM — any group member * could otherwise approve OR deny/cancel a guarded action — and must be * handled from the admin console. Sends the rejection notice + logs and * returns {@code false} when the caller is not authorized. */ private boolean approvalResolveAuthorized(PendingApproval pending, ChannelMessage message, ChannelAdapter adapter, String replyTarget) { String originalRequester = pending.getUserId(); boolean systemOriginated = originalRequester == null || "system".equals(originalRequester); if (systemOriginated || !originalRequester.equals(message.getSenderId())) { adapter.sendMessage(replyTarget, systemOriginated ? "⚠️ 该审批由系统/定时任务发起，请在管理端处理。" : "⚠️ 只有原始请求者可以审批此操作。"); log.warn("[{}] Approval resolve rejected: sender={} != requester={} (systemOriginated={})", adapter.getChannelType(), message.getSenderId(), originalRequester, systemOriginated); return false; } return true; } /** * When an /approve or /deny command carries an explicit short pendingId * (e.g. "/deny a1b2c3"), verify it matches the conversation's current * pending before resolving — otherwise a stale or copy-pasted id would * silently act on the wrong pending. Sends the mismatch notice and returns * {@code true} (caller must abort) when the ids don't line up. */ private boolean pendingIdMismatch(String userText, PendingApproval pending, ChannelAdapter adapter, String replyTarget) { String shortId = extractShortId(userText); if (shortId != null && !pending.getPendingId().startsWith(shortId)) { adapter.sendMessage(replyTarget, "⚠️ 审批ID不匹配。当前待审批: " + pending.getPendingId().substring(0, Math.min(6, pending.getPendingId().length()))); return true; } return false; } // ==================== 消息处理（原 route 逻辑 + 审批拦截层） ==================== /** * 处理单条消息：保存 -> 调用 Agent -> 保存回复 -> 发送回复 *

* 当钉钉渠道启用 AI Card 时，走流式卡片路径。 */ private void processMessage(ChannelMessage message, ChannelAdapter adapter, ChannelEntity channelEntity, String conversationId) { // The snapshot captured at enqueue time can be stale: an admin may // have rebound, deleted, or disabled the channel between debounce- // queue and flush. Re-read here so the rest of this method sees the // current state, and fail closed on deletion / disable so we don't // process traffic for a channel the admin has shut down. ChannelEntity fresh = freshChannelEntity(channelEntity); if (fresh == null) { log.warn("[{}] Channel id={} not found at processing time; dropping message from {}", adapter.getChannelType(), channelEntity != null ? channelEntity.getId() : null, message.getSenderId()); return; } if (Boolean.FALSE.equals(fresh.getEnabled())) { log.warn("[{}] Channel {} (id={}) is disabled at processing time; dropping message from {}", adapter.getChannelType(), fresh.getName(), fresh.getId(), message.getSenderId()); return; } channelEntity = fresh; Long agentId = channelEntity.getAgentId(); log.info("[{}] Processing message: sender={}, conversationId={}, agentId={}", adapter.getChannelType(), message.getSenderId(), conversationId, agentId); try { // Magic commands run before the agent-binding check so /help and // /status still answer on a channel with no agent attached. if (handleMagicCommand(message, adapter, channelEntity, conversationId)) { return; } if (agentId == null) { log.warn("[{}] Channel {} has no associated agent at processing time; dropping message from {}", adapter.getChannelType(), channelEntity.getName(), message.getSenderId()); return; } // ======= 审批拦截层 ======= String userText = message.getContent() != null ? message.getContent().trim() : ""; PendingApproval pending = approvalService.findPendingByConversation(conversationId); if (pending != null) { String replyTarget = resolveReplyTarget(message); if (isApproveCommand(userText)) { // pendingId 校验：approve / deny 共用——命令带 shortId 时必须匹配当前 pending。 if (pendingIdMismatch(userText, pending, adapter, replyTarget)) { return; } // 身份校验：approve / deny 共用同一道门禁（群聊安全 + system/null fail-closed）。 if (!approvalResolveAuthorized(pending, message, adapter, replyTarget)) { return; } // Approve via IM: workflow.resolveAndConsume runs DB + metadata + memory atomically. ResolveOutcome consumeOutcome = approvalService.resolveAndConsume( pending.getPendingId(), message.getSenderId()); if (consumeOutcome.isAlreadyResolved()) { adapter.sendMessage(replyTarget, "⚠️ 审批记录已过期或已被处理。"); return; } PendingApproval consumed = consumeOutcome.consumedSnapshot(); log.info("[{}] Approval APPROVED via IM command: pendingId={}, tool={}, msgRewritten={}", adapter.getChannelType(), consumed.getPendingId(), consumed.getToolName(), consumeOutcome.messagesRewritten()); replayApprovedToolCall(consumed, conversationId, adapter, message, channelEntity); return; } else if (isDenyCommand(userText)) { // pendingId 校验：与 approve 一致——命令带 shortId 时必须匹配当前 pending， // 否则 /deny <其它ID> 会错误地拒绝当前 conversation 的 pending。 if (pendingIdMismatch(userText, pending, adapter, replyTarget)) { return; } // 身份校验：deny 与 approve 共用门禁。否则群里任意成员可拒绝/取消他人的 // pending，system/null 发起的审批也会被任意人 deny（取消审批、清 placeholder、 // 写入 denied 状态）；这类审批改到管理端处理。 if (!approvalResolveAuthorized(pending, message, adapter, replyTarget)) { return; } // Deny via IM: workflow.resolve owns the full state-machine transition. ResolveOutcome denyOutcome = approvalService.resolve( pending.getPendingId(), message.getSenderId(), "denied"); if (denyOutcome.isAlreadyResolved()) { adapter.sendMessage(replyTarget, "⚠️ 审批记录已过期或已被处理。"); return; } conversationService.removeApprovalPlaceholders(conversationId); String denyHint = "⛔ 已拒绝执行工具: " + pending.getToolName(); persistAndBroadcastApprovalHint(conversationId, denyHint, "denied", pending.getPendingId(), pending.getToolName()); adapter.sendMessage(replyTarget, denyHint); log.info("[{}] Approval DENIED via IM command: pendingId={}, tool={}, msgRewritten={}", adapter.getChannelType(), pending.getPendingId(), pending.getToolName(), denyOutcome.messagesRewritten()); return; } else if (adapter.usesInteractiveApprovalCards()) { // Channel approves via button-clicks on an interactive // card, NOT via /approve text. A casual follow-up // message from the user during the wait window MUST // NOT auto-cancel the pending — the button click is // the canonical decision path. Treat the new message // as a fresh turn; the pending stays alive until the // user clicks Approve / Deny, the GC TTL expires, or // the workflow explicitly resolves it. log.info("[{}] Non-approval message while pending exists; channel uses card buttons so NOT auto-cancelling pendingId={}", adapter.getChannelType(), pending.getPendingId()); // Fall through to process the new message normally. } else { // Non-approval message while a pending exists → treat as implicit deny. // Text-command channels rely on this: the user is told // "type /approve " and anything else is an implicit // change of mind. approvalService.resolve(pending.getPendingId(), message.getSenderId(), "denied"); conversationService.removeApprovalPlaceholders(conversationId); String cancelHint = "⛔ 审批已取消。将继续处理您的新消息。"; persistAndBroadcastApprovalHint(conversationId, cancelHint, "cancelled", pending.getPendingId(), pending.getToolName()); adapter.sendMessage(replyTarget, cancelHint); log.info("[{}] Approval auto-cancelled (non-approval message): pendingId={}", adapter.getChannelType(), pending.getPendingId()); // Fall through to process the new message normally. } } // ======= 审批拦截层结束 ======= // Ensure the conversation exists, seeded with the agent's // currently-configured default model so per-conversation model // selection works for IM channels too (issue #183). // // Two-part behaviour, both inside getOrCreateSharedConversation: // 1. Brand-new conversation → write defaultModelName so the // very first turn picks the right model; user can later // switch via the admin UI (updateConversationModel) and the // override sticks. // 2. Pre-existing conversation with model still null (legacy // rows created before #183 fix) → backfill once, then leave // alone. Already-pinned conversations are never overwritten. // // We pass provider=null because AgentEntity doesn't carry a // provider field — the downstream ProviderChatModelFactory // resolves provider from the model name. The seed logic in // ConversationService treats (null, name) as no-seed (both // fields must be non-blank to take effect), which is the // correct defensive behaviour: we only pin when we have a // complete (provider, model) pair from the admin UI. String agentDefaultModel = null; try { AgentEntity agentEntity = agentService.getAgent(agentId); agentDefaultModel = agentEntity.getModelName(); } catch (Exception e) { // Agent deleted / disabled mid-flight — don't block message // intake. Downstream agentService.chatStructuredStream will // surface the real error to the user. log.debug("[{}] Could not load agent {} for model-seed lookup: {}", adapter.getChannelType(), agentId, e.getMessage()); } conversationService.getOrCreateSharedConversation( conversationId, agentId, channelEntity.getWorkspaceId(), null, agentDefaultModel); // 更新渠道会话存储（用于主动推送） String replyTarget = resolveReplyTarget(message); if (replyTarget != null) { channelSessionStore.saveOrUpdate( conversationId, adapter.getChannelType(), replyTarget, message.getSenderId(), message.getSenderName(), channelEntity.getId() ); } else { log.warn("[{}] No reply target resolved for sender={}, skipping session store update", adapter.getChannelType(), message.getSenderId()); } // 保存用户消息（带 contentParts） // Group sender attribution: tag the persisted content + the // prompt with [@sender] in groups so the LLM can disambiguate // multiple users sharing one conversation. Single chats pass // through unchanged (chatId is null). List parts = message.getContentParts(); String attributedContent = applyGroupTag(message, message.getContent()); conversationService.saveMessage(conversationId, "user", attributedContent, parts); // 构建 prompt（语音输入时注入场景提示词） String promptText = buildPromptFromParts(message.getContent(), parts, message.getInputMode()); // Re-apply the tag in case the prompt was assembled from // non-text parts (image/file) where buildPromptFromParts // ignored `content`. Idempotent: skips when already prefixed. promptText = applyGroupTag(message, promptText); // 注册到 ChatStreamTracker：让 graph 节点广播的事件（phase / content_delta / tool_call_* 等） // 能被 ChatConsole observer 订阅到。不注册 → broadcast() 会因 state==null 短路丢弃。 // 同步 DB stream_status 让侧栏列表可以识别"该渠道对话正在运行"，从而触发 selectConversation 的 reconnect 分支。 streamTracker.register(conversationId); streamTracker.incrementFlux(conversationId); conversationService.updateStreamStatus(conversationId, "running"); // 捕获已保存 assistant 的 DB id，供 finally 的 done 事件带出来 —— 前端 useChat.done 处理器 // 依赖 data.assistantMessageId 把本地流式 placeholder 的 client-uuid 升级为 DB id，这样 // 紧接着 refreshCurrentConversationMessages 的 reconcile 能按 id 干净匹配，不会走"认领"兜底 // 导致气泡丢失。 Long savedAssistantId = null; try { // 流式路径：渠道实现了 StreamingChannelAdapter 则委托渠道渲染流式事件 // RFC-063r §2.5: build the ChatOrigin once per channel-message // so cron jobs created during this conversation inherit the // channel binding (Issue #25 root path). ChatOrigin chatOrigin = chatOriginFactory.from( channelEntity, message, conversationId, /* workspaceBasePath */ null); if (adapter instanceof StreamingChannelAdapter streamingAdapter) { savedAssistantId = processWithStreaming(message, streamingAdapter, conversationId, agentId, promptText, channelEntity, chatOrigin); } else { // Sync path for non-streaming IM adapters (weixin / slack / // discord / qq / telegram). We can't use agentService.chat() // because its collector filters out `delta.isEvent()` deltas — that // would silently drop the tool/plan events the Web Console // mirror and the persisted execution metadata both need. // Instead we consume chatStructuredStream directly through // the shared accumulator (reply text, metadata, live mirror). final String channelType = adapter.getChannelType(); // Channel-level toggle for relaying per-stage narration as // standalone messages mid-run. Shares the key the streaming // progress path uses so operators have one knob per channel. // Disabled → narration is dropped from the IM channel (it is // never part of the final reply either way; web observers // still see it via the live broadcast). final boolean relayNarration = channelConfigBoolean( channelEntity, "stream_progress", true); // Shared accumulator: builds the segments/toolCalls metadata // the Web console renders for history, mirrors events + // deltas to live Web observers, and captures token usage + // model attribution (_usage_final is consumed internally). // Reply text also comes from it — same semantics as the // legacy collector: persistOnly deltas included (DirectAnswerNode- // routed answers arrive as persistOnly when CONTENT_STREAMED=true // and IM channels still need the text for the outgoing reply), // segmentOnly narration excluded (issue #120). AgentStreamAccumulator accumulator = newAccumulator(); agentService.chatStructuredStream(agentId, promptText, conversationId, message.getSenderId(), chatOrigin) .doOnNext(delta -> { accumulator.accept(delta, conversationId); if (!delta.isEvent() && delta.segmentOnly()) { // Per-stage narration ("Let me look that up…"), emitted as // one complete delta per agent loop iteration. Relay it // immediately as its own outgoing message so the user sees // progress mid-run. String narration = delta.content() != null ? delta.content().trim() : ""; if (relayNarration && !narration.isEmpty() && replyTarget != null) { try { adapter.renderAndSend(replyTarget, narration); } catch (Exception sendErr) { // A failed progress send must not abort the agent // run — the final reply still goes out below. log.warn("[{}] Narration relay failed (non-fatal): {}", channelType, sendErr.getMessage()); } } } }) .blockLast(Duration.ofMinutes(10)); String reply = accumulator.getContent(); // The IM sync path bypasses FinalAnswerNode, so hallucinated // /api/v1/files/generated/{id} URLs (LLM wrote a fake link // without calling a render tool) reach here verbatim. Scrub // them to the user-visible warning so IM clients don't see // a clickable link that 404s. Real tool-produced URLs are // left intact for the channel adapter's scrubber to upgrade // into native attachments. if (generatedFileCache != null) { String scrubbed = generatedFileCache.scrubMissingReferences(reply); if (!scrubbed.equals(reply)) { log.info("[{}] Scrubbed hallucinated generated-file URL(s) from IM reply ({} -> {} chars)", adapter.getChannelType(), reply.length(), scrubbed.length()); reply = scrubbed; } } // 检查 chat 过程中是否产生了审批 pending PendingApproval newPending = approvalService.findPendingByConversation(conversationId); if (newPending != null) { // Channel-specific approval rendering: WeCom overrides // sendApprovalNotice to post a button_interaction card; // every other adapter falls back to the markdown-text path // on AbstractChannelAdapter (preserves PR-0 behavior for // non-WeCom channels). var notice = approvalNotificationService.buildNotice(newPending); adapter.sendApprovalNotice(replyTarget, notice); log.info("[{}] Approval triggered during chat, sent notice (NOT saved to DB): tool={}", adapter.getChannelType(), newPending.getToolName()); } else { // Tag error replies (matched by ChannelErrorClassifier — the // "[错误]" content prefix / Bad request: / LLM error templates) // with status='error' so BaseAgent.sanitizeForLlm drops them // from the next turn's LLM history, breaking the self-replicating // 400 loop. Only successful replies fire the ConversationCompletedEvent — // error turns must not pollute memory extraction. boolean isError = errorClassifier.isErrorReply(reply); String status = isError ? "error" : "completed"; // Persist the full execution record (parts + metadata) // so the Web console renders IM-routed turns exactly // like Web direct chats. The content column keeps the // scrubbed reply text that actually went out. MessageEntity saved = conversationService.saveMessage( conversationId, "assistant", reply, accumulator.toAssistantParts(), status, accumulator.getPromptTokens(), accumulator.getCompletionTokens(), accumulator.getCacheReadTokens(), accumulator.getCacheWriteTokens(), accumulator.getReasoningTokens(), blankToNull(accumulator.getRuntimeModelName()), blankToNull(accumulator.getRuntimeProviderId()), accumulator.toMetadataJson()); savedAssistantId = saved != null ? saved.getId() : null; if (!isError) { publishConversationCompletedEvent(agentId, conversationId, message.getContent(), reply, chatOrigin); } adapter.renderAndSend(replyTarget, reply); log.info("[{}] Reply sent to {}: {}chars", adapter.getChannelType(), replyTarget, reply.length()); // 给 observer 推送完整的 assistant 消息以便前端一次性渲染为气泡 // （在 content_delta 事件可能没开的同步路径下作为兜底） Map msgCompletePayload = new HashMap<>(); msgCompletePayload.put("conversationId", conversationId); msgCompletePayload.put("content", reply); if (savedAssistantId != null) { msgCompletePayload.put("assistantMessageId", savedAssistantId); } msgCompletePayload.put("timestamp", System.currentTimeMillis()); streamTracker.broadcastObject(conversationId, "message_complete", msgCompletePayload); // 语音回复：异步 TTS 合成并追加发送（先文本后语音，不阻塞） maybeGenerateVoiceReply(message, adapter, replyTarget, conversationId, reply, channelEntity); // Per-channel completion ack (e.g. Feishu ✅ reaction). // No-op for adapters that haven't overridden the hook. try { adapter.onAgentCompleted(message); } catch (Exception hookErr) { log.debug("[{}] onAgentCompleted hook failed (non-fatal): {}", adapter.getChannelType(), hookErr.getMessage()); } } } } finally { // 广播 done 事件让 observer 前端收尾（HashMap 允许 null 值缺失，Map.of 不允许） Map donePayload = new HashMap<>(); donePayload.put("conversationId", conversationId); donePayload.put("status", "completed"); if (savedAssistantId != null) { donePayload.put("assistantMessageId", savedAssistantId); } donePayload.put("timestamp", System.currentTimeMillis()); streamTracker.broadcastObject(conversationId, "done", donePayload); streamTracker.completeAndConsumeIfLast(conversationId); try { conversationService.updateStreamStatus(conversationId, "idle"); } catch (Exception e) { log.debug("Failed to reset stream_status for {}: {}", conversationId, e.getMessage()); } } } catch (Exception e) { log.error("[{}] Failed to process message from {}: {}", adapter.getChannelType(), message.getSenderId(), e.getMessage(), e); // 尝试发送错误提示 try { String errorTarget = resolveReplyTarget(message); adapter.sendMessage(errorTarget, "抱歉，处理消息时出现错误：" + e.getMessage()); } catch (Exception sendErr) { log.error("[{}] Failed to send error message: {}", adapter.getChannelType(), sendErr.getMessage()); } } } /** * Handle channel-native control commands before the message is persisted * or forwarded to the agent. A recognized command is terminal for this * inbound message: it never reaches the debounce queue or the LLM. */ private boolean handleMagicCommand(ChannelMessage message, ChannelAdapter adapter, ChannelEntity channelEntity, String conversationId) { String userText = message != null ? message.getContent() : null; ChannelMagicCommand.Parsed command = ChannelMagicCommand.parse(userText).orElse(null); if (command == null) { return false; } String replyTarget = resolveReplyTarget(message); String reply = switch (command.type()) { case CLEAR -> { cancelPending(conversationId); conversationService.clearMessages(conversationId); yield ChannelMagicCommand.clearConfirmation(); } case NEW -> { // Channel conversation ids are deterministic (channelType:chatId), // so "new session" cannot rotate the id — it clears the context // like CLEAR and only differs in the confirmation wording. cancelPending(conversationId); conversationService.clearMessages(conversationId); yield ChannelMagicCommand.newConfirmation(); } case STOP -> { cancelPending(conversationId); boolean stopped = streamTracker.requestStop(conversationId); yield stopped ? ChannelMagicCommand.stopConfirmation() : ChannelMagicCommand.stopNothingRunning(); } case HELP -> ChannelMagicCommand.helpText(); case STATUS -> buildStatusReply(channelEntity, conversationId); case MODEL -> handleModelCommand(channelEntity, conversationId, command.args()); }; if (replyTarget != null && reply != null) { // renderAndSend (not sendMessage) so adapters that pre-post a // "thinking..." placeholder on inbound (WeCom reply_stream) // consume it here: the confirmation overwrites the placeholder // bubble in place and the keepalive refresher is stopped. // Plain sendMessage would leave the placeholder dangling forever. adapter.renderAndSend(replyTarget, reply); } log.info("[{}] Magic command handled: {} conversationId={}, sender={}", adapter.getChannelType(), command.type(), conversationId, message != null ? message.getSenderId() : null); return true; } /** Build the /status reply; every lookup degrades gracefully to keep the command side-effect free. */ private String buildStatusReply(ChannelEntity channelEntity, String conversationId) { StringBuilder sb = new StringBuilder("📊 会话状态\n"); sb.append("- 会话: ").append(conversationId).append('\n'); Long agentId = channelEntity != null ? channelEntity.getAgentId() : null; String[] pinned = findPinnedModel(conversationId); if (agentId == null) { sb.append("- 智能体: 未绑定\n"); } else { try { AgentEntity agent = agentService.getAgent(agentId); if (agent != null) { sb.append("- 智能体: ").append(agent.getName()).append('\n'); // Conversation-pinned model wins over the agent default — // mirrors the resolution order in AgentService, so /status // never contradicts what /model just switched to. if (pinned != null) { sb.append("- 模型: ").append(pinned[0]).append(':').append(pinned[1]) .append("（会话指定）\n"); } else if (agent.getModelName() != null && !agent.getModelName().isBlank()) { sb.append("- 模型: ").append(agent.getModelName()).append('\n'); } } else { sb.append("- 智能体: 未找到（id=").append(agentId).append("）\n"); } } catch (Exception e) { log.warn("Failed to load agent {} for /status: {}", agentId, e.getMessage()); sb.append("- 智能体: 查询失败\n"); } } try { sb.append("- 历史消息数: ").append(conversationService.countMessages(conversationId)).append('\n'); } catch (Exception e) { log.warn("Failed to count messages for /status: {}", e.getMessage()); } boolean running = streamTracker.isRunning(conversationId); sb.append("- 当前任务: ").append(running ? "进行中（可用 /stop 停止）" : "空闲"); return sb.toString(); } /** * Handle the /model command: list enabled chat models, pin one on this * conversation, or reset to the agent default. Listing and resetting work * without a bound agent; switching requires one because the pinned pair * only takes effect when the agent graph is built. */ private String handleModelCommand(ChannelEntity channelEntity, String conversationId, String args) { if (modelConfigService == null) { return "⚠️ 模型管理服务不可用，请稍后再试。"; } String arg = args == null ? "" : args.trim(); if ("reset".equalsIgnoreCase(arg) || "恢复默认".equals(arg)) { conversationService.clearConversationModel(conversationId); return "✅ 已恢复默认模型（跟随智能体配置），下一条消息生效。"; } List models; try { models = modelConfigService.listEnabledModels(); } catch (Exception e) { log.warn("Failed to list models for /model on {}: {}", conversationId, e.getMessage()); return "⚠️ 查询模型列表失败，请稍后再试。"; } if (arg.isEmpty() || "list".equalsIgnoreCase(arg)) { return buildModelListReply(models, conversationId); } return switchConversationModel(channelEntity, conversationId, arg, models); } /** Max rows shown by /model list — a full catalog can exceed 180 rows, * which segments into several IM bubbles and buries the usage hint. */ private static final int MODEL_LIST_MAX_ROWS = 20; private String buildModelListReply(List models, String conversationId) { if (models.isEmpty()) { return "当前没有已启用的对话模型，请先在控制台配置。"; } String[] pinned = findPinnedModel(conversationId); StringBuilder sb = new StringBuilder("🧠 可用模型（/model <名称> 切换，/model reset 恢复默认）：\n"); int shown = 0; for (ModelConfigEntity m : models) { if (shown >= MODEL_LIST_MAX_ROWS) { break; } sb.append("- ").append(m.getProvider()).append(':').append(m.getModelName()); if (pinned != null && pinned[0].equalsIgnoreCase(String.valueOf(m.getProvider())) && pinned[1].equalsIgnoreCase(String.valueOf(m.getModelName()))) { sb.append(" ✅ 当前"); } sb.append('\n'); shown++; } if (models.size() > MODEL_LIST_MAX_ROWS) { sb.append("…共 ").append(models.size()) .append(" 个已启用模型，仅展示前 ").append(MODEL_LIST_MAX_ROWS) .append(" 个；发送 /model <关键词> 搜索其余模型。\n"); } sb.append(pinned == null ? "当前：跟随智能体默认模型" : "当前会话已指定：" + pinned[0] + ":" + pinned[1]); return sb.toString(); } private String switchConversationModel(ChannelEntity channelEntity, String conversationId, String arg, List models) { if (channelEntity == null || channelEntity.getAgentId() == null) { return "⚠️ 当前渠道未绑定智能体，请先在控制台绑定后再切换模型。"; } String wantedProvider = null; String wantedName = arg; int colon = arg.indexOf(':'); if (colon > 0 && colon < arg.length() - 1) { wantedProvider = arg.substring(0, colon).trim(); wantedName = arg.substring(colon + 1).trim(); } final String fProvider = wantedProvider; final String fName = wantedName; List matches = models.stream() .filter(m -> fName.equalsIgnoreCase(m.getModelName())) .filter(m -> fProvider == null || fProvider.equalsIgnoreCase(m.getProvider())) .toList(); if (matches.isEmpty()) { // No exact hit — treat the arg as a search keyword so users can // discover models the capped /model list didn't show. String keyword = fName.toLowerCase(Locale.ROOT); List fuzzy = models.stream() .filter(m -> String.valueOf(m.getModelName()).toLowerCase(Locale.ROOT).contains(keyword) || String.valueOf(m.getProvider()).toLowerCase(Locale.ROOT).contains(keyword)) .limit(MODEL_LIST_MAX_ROWS) .toList(); if (fuzzy.isEmpty()) { return "⚠️ 未找到已启用的模型「" + arg + "」，发送 /model 查看可用列表。"; } StringBuilder sb = new StringBuilder("未找到精确匹配「").append(arg) .append("」，相近的可用模型：\n"); for (ModelConfigEntity m : fuzzy) { sb.append("- /model ").append(m.getProvider()).append(':') .append(m.getModelName()).append('\n'); } return sb.toString().stripTrailing(); } if (matches.size() > 1) { StringBuilder sb = new StringBuilder("⚠️ 模型「").append(fName) .append("」在多个 provider 下存在，请带上前缀再试：\n"); for (ModelConfigEntity m : matches) { sb.append("- /model ").append(m.getProvider()).append(':').append(m.getModelName()).append('\n'); } return sb.toString().stripTrailing(); } ModelConfigEntity target = matches.get(0); try { // The magic-command layer runs before processMessage's // get-or-create, so a /model sent as the very first message must // create the conversation row itself — updateConversationModel // silently no-ops on a missing row. conversationService.getOrCreateSharedConversation( conversationId, channelEntity.getAgentId(), channelEntity.getWorkspaceId()); conversationService.updateConversationModel( conversationId, target.getProvider(), target.getModelName()); } catch (Exception e) { log.warn("Failed to pin model {} on {}: {}", arg, conversationId, e.getMessage()); return "⚠️ 切换失败，请稍后再试。"; } return "✅ 本会话模型已切换为 " + target.getProvider() + ":" + target.getModelName() + "，下一条消息生效。发送 /model reset 可恢复默认。"; } /** Conversation-pinned (provider, model) pair, or null when unpinned/unavailable. */ private String[] findPinnedModel(String conversationId) { try { ConversationEntity conv = conversationService.findByConversationId(conversationId); if (conv != null && conv.getModelProvider() != null && !conv.getModelProvider().isBlank() && conv.getModelName() != null && !conv.getModelName().isBlank()) { return new String[]{conv.getModelProvider(), conv.getModelName()}; } } catch (Exception e) { log.debug("Failed to load pinned model for {}: {}", conversationId, e.getMessage()); } return null; } private void cancelPending(String conversationId) { PendingMessage pending; synchronized (pendingMessages) { pending = pendingMessages.remove(conversationId); } if (pending != null && pending.timer != null) { pending.timer.cancel(false); } } /** * Build a per-turn accumulator wired to the stream tracker, so live Web * observers of an IM conversation receive the same event fan-out as Web * direct chats, and the persisted metadata matches byte-for-byte. */ private AgentStreamAccumulator newAccumulator() { return new AgentStreamAccumulator(objectMapper, new AgentStreamAccumulator.Sink() { @Override public void broadcast(String conversationId, String eventName, Object payload) { streamTracker.broadcastObject(conversationId, eventName, payload); } @Override public void updatePhase(String conversationId, String phase) { streamTracker.updatePhase(conversationId, phase); } }); } /** Map the accumulator's empty-string defaults back to SQL NULL. */ private static String blankToNull(String s) { return s == null || s.isBlank() ? null : s; } /** * 流式处理路径（渠道无关） *

* 事件流与渲染分离： * - Router 负责产生 StreamDelta 流（调用 AgentService） * - StreamingChannelAdapter 负责渲染（AI Card / 卡片更新 / 文本累积等） * - Router 负责后续的审批检查、消息持久化、事件发布 */ private Long processWithStreaming(ChannelMessage message, StreamingChannelAdapter streamingAdapter, String conversationId, Long agentId, String promptText, ChannelEntity channelEntity, ChatOrigin chatOrigin) { String channelType = streamingAdapter.getChannelType(); log.info("[{}] Streaming processing started: conversationId={}", channelType, conversationId); try { // Step 1: 产生事件流（forward ChatOrigin so tools see channelId） Flux stream = agentService.chatStructuredStream( agentId, promptText, conversationId, message.getSenderId(), chatOrigin); // Feed every delta through the shared accumulator before the // adapter consumes the Flux. The accumulator builds the same // segments/toolCalls metadata the Web SSE path persists (so the // console renders the execution timeline for IM-routed turns), // mirrors tool/plan/content events to any live Web observer of // this conversation, and captures token usage + model // attribution. Internal bookkeeping events (_usage_final, // _routing_decision) are consumed inside the accumulator and // never reach subscribers. AgentStreamAccumulator accumulator = newAccumulator(); Flux mirroredStream = stream.doOnNext(delta -> accumulator.accept(delta, conversationId)); // Step 2: 委托渠道渲染（渠道内部消费 Flux 并处理 UI 更新） String finalContent = streamingAdapter.processStream(mirroredStream, message, conversationId); // Step 3: 审批检查 + 持久化（渠道无关逻辑，由 Router 统一处理） PendingApproval newPending = approvalService.findPendingByConversation(conversationId); if (newPending != null) { String replyTarget = resolveReplyTarget(message); // Same polymorphic dispatch as the non-streaming path — WeCom // renders a card, others render text. See the buildNotice + // sendApprovalNotice pair at the non-streaming call site above. var notice = approvalNotificationService.buildNotice(newPending); streamingAdapter.sendApprovalNotice(replyTarget, notice); log.info("[{}] Approval triggered during streaming (NOT saved to DB): tool={}", channelType, newPending.getToolName()); } else if (finalContent != null && !finalContent.isBlank()) { boolean isError = errorClassifier.isErrorReply(finalContent); String status = isError ? "error" : "completed"; // Persist the full execution record — parts (text/thinking/ // tool_call) and metadata (segments/toolCalls/plan/…) — so // the Web console renders IM-routed turns exactly like Web // direct chats. The content column keeps the adapter's final // text (the adapter may have post-processed it). MessageEntity saved = conversationService.saveMessage( conversationId, "assistant", finalContent, accumulator.toAssistantParts(), status, accumulator.getPromptTokens(), accumulator.getCompletionTokens(), accumulator.getCacheReadTokens(), accumulator.getCacheWriteTokens(), accumulator.getReasoningTokens(), blankToNull(accumulator.getRuntimeModelName()), blankToNull(accumulator.getRuntimeProviderId()), accumulator.toMetadataJson()); if (!isError) { publishConversationCompletedEvent(agentId, conversationId, promptText, finalContent, chatOrigin); } log.info("[{}] Streaming completed: contentLen={}, isError={}", channelType, finalContent.length(), isError); // 流式回复完成后也触发语音回复 String replyTarget = resolveReplyTarget(message); if (replyTarget != null) { maybeGenerateVoiceReply(message, streamingAdapter, replyTarget, conversationId, finalContent, channelEntity); } try { streamingAdapter.onAgentCompleted(message); } catch (Exception hookErr) { log.debug("[{}] onAgentCompleted hook failed (non-fatal): {}", channelType, hookErr.getMessage()); } return saved != null ? saved.getId() : null; } } catch (Exception e) { log.error("[{}] Streaming processing failed: {}", channelType, e.getMessage(), e); // Persist an error placeholder (status='error') so that the next // turn's history does not show a user → user sequence (which some // providers reject with 400). sanitizeForLlm filters this row out // before the LLM sees it, so it costs nothing at the prompt layer. try { conversationService.saveMessage(conversationId, "assistant", "[错误] " + e.getMessage(), null, "error"); } catch (Exception persistErr) { log.warn("[{}] Failed to persist error placeholder: {}", channelType, persistErr.getMessage()); } try { String errorTarget = resolveReplyTarget(message); streamingAdapter.sendMessage(errorTarget, "抱歉，流式处理失败：" + e.getMessage()); } catch (Exception sendErr) { log.error("[{}] Failed to send streaming error message: {}", channelType, sendErr.getMessage()); } } return null; } // ==================== 审批重放 ==================== /** * 重放被审批阻塞的工具调用 *

* 接收已消费的审批记录（由 resolveAndConsume 原子获取），通过 AgentService.chatWithReplay 重新执行工具。 * 重放前清理 DB 中的审批占位消息，防止 LLM 看到残留文本后重新发起工具调用（死循环根因）。 */ private void replayApprovedToolCall(PendingApproval consumed, String conversationId, ChannelAdapter adapter, ChannelMessage triggerMessage, ChannelEntity channelEntity) { String replyTarget = resolveReplyTarget(triggerMessage); Long agentId = channelEntity.getAgentId(); // Notify the user that the approval went through. Persist + broadcast so a // Web mirror of the same conversationId sees the resolution; otherwise this // hint would only land in the IM channel and the Web admin console would // show the replay reply with no preceding "approved" marker. String approveHint = "✅ 已批准执行工具: " + consumed.getToolName(); persistAndBroadcastApprovalHint(conversationId, approveHint, "approved", consumed.getPendingId(), consumed.getToolName()); adapter.sendMessage(replyTarget, approveHint); // 清理 DB 中残留的审批占位消息 conversationService.removeApprovalPlaceholders(conversationId); // 简化 replay prompt（不重复工具名，防止 LLM 误解） String replayPrompt = "继续执行已批准的工具调用。"; try { // RFC-063r §2.12: prefer the persisted Memento (covers // cross-restart approval where the channel session changed) and // only fall back to rebuilding from the current inbound message // when no snapshot was captured (legacy rows from before this PR). ChatOrigin replayOrigin = approvalService.restoreChatOrigin(consumed.getChatOrigin()); if (replayOrigin == ChatOrigin.EMPTY) { replayOrigin = chatOriginFactory.from( channelEntity, triggerMessage, conversationId, /* workspaceBasePath */ null); } AgentService.ChatResult replayResult = agentService.chatWithReplayWithUsage( agentId, replayPrompt, conversationId, consumed.getToolCallPayload(), replayOrigin); String reply = replayResult.content(); // Persist the replay result. If the LLM 400'd during replay, // the error reply must also get status='error' — otherwise the // next turn's history would re-feed the error placeholder back // into the prompt and re-trigger the same failure. boolean isError = errorClassifier.isErrorReply(reply); conversationService.saveMessage(conversationId, "assistant", reply, null, isError ? "error" : "completed", replayResult.promptTokens(), replayResult.completionTokens(), replayResult.runtimeModel(), replayResult.runtimeProvider()); // 发送回复 adapter.renderAndSend(replyTarget, reply); log.info("[{}] Replay completed: tool={}, replyLen={}", adapter.getChannelType(), consumed.getToolName(), reply.length()); } catch (Exception e) { log.error("[approval-replay] Replay failed: {}", e.getMessage(), e); String errHint = "❌ 工具执行失败: " + e.getMessage(); persistAndBroadcastApprovalHint(conversationId, errHint, null, null, null); adapter.sendMessage(replyTarget, errHint); } } /** * Persist an approval-related hint as an assistant message and best-effort * broadcast it to any live SSE viewer of the conversation. *

* Without this, IM-driven approve/deny only reaches the originating IM * channel via {@code adapter.sendMessage(...)} — a Web mirror of the same * conversationId has no record of the resolution because nothing lands in * {@code mate_message} and no SSE event is emitted. The hint then "vanishes" * from the Web admin console even though it shows up on the user's phone. *

* Persistence is the load-bearing fix (Web reload picks it up). Broadcast * is best-effort: if no SSE stream is currently registered for the * conversation, the broadcast no-ops silently — that's the common case * since IM-driven clicks rarely race with an active web subscriber. * * @param conversationId conversation owning the hint * @param hint text to render as an assistant bubble * @param decision "approved" / "denied" / "cancelled" / null (skips the * structured resolved event when null, e.g. on replay error) * @param pendingId pending approval id; null when not applicable * @param toolName tool name for the structured event; null when not applicable */ private void persistAndBroadcastApprovalHint(String conversationId, String hint, String decision, String pendingId, String toolName) { try { conversationService.saveMessage(conversationId, "assistant", hint, null, "completed"); } catch (Exception e) { log.warn("[approval-hint] saveMessage failed for conv={}: {}", conversationId, e.getMessage()); } try { if (decision != null) { streamTracker.broadcastObject(conversationId, "tool_approval_resolved", Map.of( "pendingId", pendingId == null ? "" : pendingId, "decision", decision, "toolName", toolName == null ? "" : toolName, "timestamp", System.currentTimeMillis() )); } streamTracker.broadcastObject(conversationId, "message_start", Map.of("role", "assistant")); streamTracker.broadcastObject(conversationId, "content_delta", Map.of("delta", hint)); streamTracker.broadcastObject(conversationId, "message_complete", Map.of("status", "completed")); } catch (Exception e) { // Broadcast is best-effort; a missing run state is the common case. log.debug("[approval-hint] broadcast skipped/failed for conv={}: {}", conversationId, e.getMessage()); } } /** * 从 PendingApproval 元数据构建 IM 友好的审批通知（委托给 ApprovalNotificationService） */ private String buildApprovalNotice(PendingApproval pending) { return approvalNotificationService.buildApprovalText(pending); } /** * Publish the conversation-completed event (triggers async memory extraction). * Delegates to {@link ConversationCompletionPublisher} so the try/catch and * messageCount lookup no longer live here. */ private void publishConversationCompletedEvent(Long agentId, String conversationId, String userMessage, String assistantReply, ChatOrigin origin) { // Attribute the memory write to the same external sender the read path // recalled for, so per-sender IM memory is both written and recalled // under the same owner key. completionPublisher.publishForOrigin(agentId, conversationId, userMessage, assistantReply, "channel", origin); } // ==================== 流式处理（Web 渠道专用，不走队列） ==================== /** * 路由消息并使用流式处理（用于支持流式的渠道，如 Web） */ public Flux routeStream(ChannelMessage message, ChannelEntity channelEntity) { ChannelEntity fresh = freshChannelEntity(channelEntity); if (fresh == null) { return Flux.error(new IllegalStateException("Channel no longer exists")); } if (Boolean.FALSE.equals(fresh.getEnabled())) { return Flux.error(new IllegalStateException("Channel is disabled")); } channelEntity = fresh; Long agentId = channelEntity.getAgentId(); if (agentId == null) { return Flux.error(new IllegalStateException("Channel has no associated agent")); } String conversationId = buildConversationId(message, channelEntity.getId()); String username = message.getSenderName() != null ? message.getSenderName() : message.getSenderId(); conversationService.getOrCreateConversation(conversationId, agentId, username, channelEntity.getWorkspaceId()); List parts = message.getContentParts(); // Mirror processMessage's group attribution for the streaming path // (Web channel today; future streaming IM channels inherit it). String attributedContent = applyGroupTag(message, message.getContent()); conversationService.saveMessage(conversationId, "user", attributedContent, parts); String promptText = buildPromptFromParts(message.getContent(), parts, message.getInputMode()); promptText = applyGroupTag(message, promptText); // RFC-063r §2.5: forward ChatOrigin so tools created during this // streaming conversation inherit channel binding. ChatOrigin origin = chatOriginFactory.from( channelEntity, message, conversationId, /* workspaceBasePath */ null); return agentService.chatStream(agentId, promptText, conversationId, origin); } // ==================== 优雅关闭 ==================== /** * 优雅关闭：停止防抖调度器和所有消费线程 */ public void shutdown() { log.info("Shutting down ChannelMessageRouter..."); shutdown = true; // 1. 关闭防抖调度器 debounceScheduler.shutdownNow(); // 2. 清理残留的 pending 消息 synchronized (pendingMessages) { pendingMessages.forEach((convId, pending) -> { if (pending.timer != null) { pending.timer.cancel(false); } log.warn("Dropping pending debounced message for conversation: {}", convId); }); pendingMessages.clear(); } // 3. 关闭每个渠道的消费线程池：shutdown -> 等待 5 秒 -> shutdownNow channelExecutors.forEach((channelType, executor) -> { log.info("[{}] Shutting down consumer threads...", channelType); executor.shutdown(); try { if (!executor.awaitTermination(5, TimeUnit.SECONDS)) { log.warn("[{}] Consumer threads did not terminate in 5s, forcing shutdown", channelType); executor.shutdownNow(); } } catch (InterruptedException e) { executor.shutdownNow(); Thread.currentThread().interrupt(); } }); channelExecutors.clear(); channelQueues.clear(); sessionLocks.clear(); // 4. 关闭语音回复线程池 voiceReplyExecutor.shutdownNow(); log.info("ChannelMessageRouter shutdown complete"); } // ==================== 工具方法 ==================== /** * Re-read the channel row from the database so the rest of the message * pipeline sees current routing metadata (agentId, workspaceId, identityJson) * rather than the snapshot captured when the adapter was constructed. * *

Failure semantics: *

* *

{@code enabled=false} is NOT handled here — that's an admin decision * the callers check separately, with channel-type-specific logging. */ private ChannelEntity freshChannelEntity(ChannelEntity snapshot) { if (snapshot == null || snapshot.getId() == null) { return snapshot; } try { ChannelEntity latest = channelService.getChannel(snapshot.getId()); return latest != null ? latest : snapshot; } catch (MateClawException biz) { if ("err.channel.not_found".equals(biz.getMsgKey())) { log.warn("Channel id={} no longer exists; dropping incoming message", snapshot.getId()); return null; } log.debug("Transient channel lookup failure id={}, using snapshot: {}", snapshot.getId(), biz.getMessage()); return snapshot; } catch (Exception e) { log.debug("Failed to refresh ChannelEntity id={}, using snapshot: {}", snapshot.getId(), e.getMessage()); return snapshot; } } /** * 构建会话 ID * 格式：{channelType}:{chatId 或 senderId} * 格式采用 {channelType}:{identifier} 命名规则 */ /** * Build the conversation id for an inbound channel message. * *

The id is scoped by {@code channelId} so the same sender reaching two * different workspaces' same-type channels (e.g. two separate wecom channels) * no longer collapses into one shared conversation row. {@code channelId} is * the {@code ChannelEntity} primary key, which binds to exactly one workspace. * *

Format: {@code {channelType}:{channelId}:{chatId|senderId}}. When * {@code channelId} is null (defensive; the routed channel row always has an * id) the legacy {@code {channelType}:{identifier}} form is used so nothing * NPEs — those ids remain workspace-ambiguous but that path is not reachable * for a persisted channel. */ private String buildConversationId(ChannelMessage message, Long channelId) { String identifier = message.getChatId() != null ? message.getChatId() : message.getSenderId(); if (channelId == null) { return message.getChannelType() + ":" + identifier; } return message.getChannelType() + ":" + channelId + ":" + identifier; } /** * Read-only existence check for a conversation by its logical id. Used by * adapters that need to alias a legacy conversationId scheme to a new one * without rewriting stored rows (e.g. Feishu group session-id migration). */ public boolean conversationExists(String conversationId) { return conversationService.findByConversationId(conversationId) != null; } /** * Build a sender-attribution tag for group messages. Returns * {@code [@senderName]} when the message is from a multi-user channel * context (chatId is set), else {@code null} for 1:1 chats. * *

Without this tag, three users asking three different questions in * the same group conversation collapse into an unattributed wall of * "user:" turns and the LLM can no longer tell who is asking what — * it answers based on the most-recent text and ignores the rest. * Single chats are unaffected because chatId is null there. * *

Prefer {@code senderName} when populated; otherwise fall back to * {@code senderId}. WeCom currently sets both to the same opaque * openid which is still useful for disambiguation; future channels * (DingTalk, Slack) carry friendlier display names that flow through * unchanged. * * @return sender tag like {@code [@Alice]}, or {@code null} if the * message is not from a group context. */ static String buildGroupTag(ChannelMessage message) { if (message == null) return null; String chatId = message.getChatId(); if (chatId == null || chatId.isBlank()) return null; String name = (message.getSenderName() != null && !message.getSenderName().isBlank()) ? message.getSenderName() : message.getSenderId(); if (name == null || name.isBlank()) return null; return "[@" + name + "]"; } /** * Apply {@link #buildGroupTag} to {@code content}. Idempotent: if * {@code content} already starts with the tag (e.g. an upstream * adapter has pre-attributed it), returns it unchanged so we don't * double-stamp. No-op for single chats. */ static String applyGroupTag(ChannelMessage message, String content) { String tag = buildGroupTag(message); if (tag == null) return content; // Empty content: leave empty rather than persist or prompt with a // bare "[@Alice]" — the message had no payload to attribute. if (content == null || content.isEmpty()) return content; if (content.startsWith(tag)) return content; return tag + " " + content; } /** * Decision helper for the debounce merger: should an incoming message * from {@code incomingSender} merge into a pending buffer started by * {@code existingSender}? True only when the senders match — different * senders in the same conversation (a group context) must NOT merge, * else the second user's text gets attributed to the first. * *

Null-handling: a null {@code existingSender} means "no buffer to * merge into" so the answer is always false; a null * {@code incomingSender} (rare, but seen in test fixtures) is also * not allowed to silently merge — returning false routes to the * "create new pending" branch which is safe. */ static boolean isSameSender(String existingSender, String incomingSender) { if (existingSender == null || incomingSender == null) return false; return existingSender.equals(incomingSender); } /** * 确定回复目标 * 优先使用 replyToken（渠道特有的回复标识），其次 chatId，最后 senderId */ private String resolveReplyTarget(ChannelMessage message) { if (message.getReplyToken() != null) { return message.getReplyToken(); } return message.getChatId() != null ? message.getChatId() : message.getSenderId(); } /** * 从 contentParts 构建完整 prompt 文本。 * 文本直接拼接；媒体类型生成描述性占位符，让 Agent 知道用户发送了什么。 * 语音输入时注入场景提示词，引导 Agent 用简短口语化方式回复。 */ private String buildPromptFromParts(String fallbackContent, List parts, String inputMode) { if (parts == null || parts.isEmpty()) { return fallbackContent != null ? fallbackContent : ""; } StringBuilder sb = new StringBuilder(); for (MessageContentPart part : parts) { if (part == null || part.getType() == null) continue; switch (part.getType()) { case "text" -> appendLine(sb, part.getText()); case "image" -> appendLine(sb, "[用户发送了图片" + descMedia(part) + "]"); case "file" -> appendLine(sb, "[用户发送了文件: " + safe(part.getFileName()) + "]"); case "audio" -> appendLine(sb, "[用户发送了音频" + descMedia(part) + "]"); case "video" -> appendLine(sb, "[用户发送了视频" + descMedia(part) + "]"); default -> appendLine(sb, part.getText()); } } String result = sb.toString().trim(); if (result.isEmpty()) { return fallbackContent != null ? fallbackContent : ""; } // 语音场景：注入提示词让 Agent 回复更简短口语化 if ("voice".equals(inputMode)) { result = "[用户通过语音输入，请用简短口语化的方式回复]\n" + result; } return result; } private void appendLine(StringBuilder sb, String text) { if (text == null || text.isBlank()) return; if (!sb.isEmpty()) sb.append('\n'); sb.append(text); } private String descMedia(MessageContentPart part) { if (part.getFileName() != null && !part.getFileName().isBlank()) { return ": " + part.getFileName(); } return ""; } private String safe(String s) { return s == null ? "" : s; } // ==================== 语音回复（TTS）==================== /** TTS 异步工作线程池 */ private final ExecutorService voiceReplyExecutor = Executors.newFixedThreadPool(2, r -> { Thread t = new Thread(r, "voice-reply-worker"); t.setDaemon(true); return t; }); /** * 根据消息上下文和渠道配置，判断是否需要生成语音回复。 * 若需要，异步合成 TTS 并通过渠道发送音频文件。 *

* 设计原则（借鉴 OpenClaw）： * - 文本先行，语音异步追加，不阻塞用户体验 * - 短回复（<10字）跳过 TTS（不值得合成） * - TTS 失败静默降级，不影响已发出的文本回复 */ private void maybeGenerateVoiceReply(ChannelMessage message, ChannelAdapter adapter, String replyTarget, String conversationId, String replyText, ChannelEntity channelEntity) { if (!shouldGenerateVoiceReply(message, channelEntity, replyText)) { return; } voiceReplyExecutor.submit(() -> { try { // 读取渠道级语音配置 Map channelConfig = parseChannelConfig(channelEntity.getConfigJson()); String voiceName = channelConfig.getOrDefault("voice_name", "").toString(); double voiceSpeed = 1.0; Object speedObj = channelConfig.get("voice_speed"); if (speedObj instanceof Number n) voiceSpeed = n.doubleValue(); // 调用 TtsService 合成 Map result = ttsService.synthesize( conversationId, replyText, voiceName.isBlank() ? null : voiceName, voiceSpeed, "mp3"); if (!Boolean.TRUE.equals(result.get("success"))) { log.debug("[voice-reply] TTS synthesis failed: {}", result.get("error")); return; } // 构建音频 MessageContentPart String audioUrl = (String) result.get("audioUrl"); String fileName = Paths.get(audioUrl).getFileName().toString(); // TTS output may live under a workspace-scoped dir or the legacy // default dir — probe each candidate root to find the file. Path audioPath = resolveVoiceReplyAudio(conversationId, fileName); if (audioPath == null) { log.warn("[voice-reply] TTS output file not found for conversation {} ({})", conversationId, fileName); return; } MessageContentPart audioPart = new MessageContentPart(); audioPart.setType("audio"); audioPart.setFileName(fileName); audioPart.setPath(audioPath.toString()); audioPart.setContentType("audio/mpeg"); adapter.sendContentParts(replyTarget, List.of(audioPart)); log.info("[voice-reply] Sent to {} via {}: {} ({}KB)", replyTarget, adapter.getChannelType(), fileName, Files.size(audioPath) / 1024); } catch (Exception e) { // 静默降级：TTS 失败不影响已发出的文本回复 log.warn("[voice-reply] Failed for conversation {}: {}", conversationId, e.getMessage()); } }); } /** * Resolve the TTS audio file across every candidate upload root. Returns the * first existing match, or {@code null} when the file is absent under every * root. Used by the voice-reply path so workspace-scoped and legacy default * outputs are both found. */ private Path resolveVoiceReplyAudio(String conversationId, String fileName) { if (chatUploadLocationResolver != null) { // Probes every candidate root and both layouts (flat + date // sub-directories), so TTS files written under a per-day dir resolve. Path found = chatUploadLocationResolver.resolveExistingFile(conversationId, fileName); if (found != null) { return found; } } // Fallback to the legacy default dir when the resolver is absent // (e.g. direct-construction unit tests). Sanitize the id for the path // segment so it matches the write side. Path legacy = Paths.get("data", "chat-uploads", ChatUploadLocationResolver.sanitizeSegment(conversationId), fileName); return Files.exists(legacy) ? legacy : null; } /** * 判断是否需要为此消息生成语音回复 */ private boolean shouldGenerateVoiceReply(ChannelMessage message, ChannelEntity channelEntity, String replyText) { // 1. TTS 全局开关 if (!ttsService.isTtsEnabled()) return false; // 2. 回复内容过短或为空，跳过 TTS（借鉴 OpenClaw: <10字不合成） if (replyText == null || replyText.trim().length() < 10) return false; // 3. 读取渠道级语音回复模式 Map channelConfig = parseChannelConfig(channelEntity.getConfigJson()); String voiceMode = channelConfig.getOrDefault("voice_reply_mode", "off").toString(); if ("off".equals(voiceMode)) return false; if ("always".equals(voiceMode)) return true; // 4. auto 模式：仅当用户通过语音输入时回语音 return "auto".equals(voiceMode) && "voice".equals(message.getInputMode()); } /** * Boolean lookup on the channel's configJson. Accepts Boolean or String * values, mirroring the adapter-side config parsing rules, so the router * and the adapters read the same key identically. */ private boolean channelConfigBoolean(ChannelEntity channelEntity, String key, boolean defaultValue) { Object value = parseChannelConfig(channelEntity.getConfigJson()).get(key); if (value instanceof Boolean b) return b; if (value instanceof String s && !s.isBlank()) return Boolean.parseBoolean(s.trim()); return defaultValue; } /** * 解析 Channel 的 configJson 为 Map */ private Map parseChannelConfig(String configJson) { if (configJson == null || configJson.isBlank()) return Map.of(); try { return objectMapper.readValue(configJson, new TypeReference<>() {}); } catch (Exception e) { log.debug("[voice-reply] Failed to parse channel config: {}", e.getMessage()); return Map.of(); } } }