From 3ae4498f38a3029fd2a5691abda371276dcb6223 Mon Sep 17 00:00:00 2001 From: matevip Date: Tue, 26 May 2026 09:40:49 +0800 Subject: [PATCH] fix(channel,agent,chat): unify channel binding / conversation agent / model pin state sources --- .../vip/mate/agent/AgentGraphBuilder.java | 45 ++++++-- .../mate/channel/ChannelMessageRouter.java | 95 +++++++++++++++ .../channel/controller/ChannelController.java | 44 ++++++- .../conversation/ConversationService.java | 19 ++- mateclaw-ui/src/i18n/locales/en-US.ts | 1 + mateclaw-ui/src/i18n/locales/zh-CN.ts | 1 + mateclaw-ui/src/views/ChatConsole.vue | 109 +++++++++++++++++- 7 files changed, 300 insertions(+), 14 deletions(-) diff --git a/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java b/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java index 5deb483e..bc725646 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/AgentGraphBuilder.java @@ -168,6 +168,23 @@ public class AgentGraphBuilder { return modelConfigService.resolveModel(agentModelName); } + /** + * True iff the caller passed a complete (provider, model) pin AND that + * pair resolves to an enabled model row. Used by {@link #build} to decide + * whether the explicit pick should bypass capability-driven routing. + */ + private boolean pinResolvesToEnabledModel(String modelProvider, String modelName) { + if (modelProvider == null || modelProvider.isBlank() + || modelName == null || modelName.isBlank()) { + return false; + } + try { + return modelConfigService.findEnabledModel(modelProvider, modelName) != null; + } catch (Exception e) { + return false; + } + } + /** * 根据 AgentEntity 构建完整的 Agent 实例。 * @@ -201,22 +218,34 @@ public class AgentGraphBuilder { // looks up enabled-only models and silently degrades an unmatched pin / // override to the global default, preserving the legacy behaviour for // Agents and conversations without an explicit choice. - // providerRouter.selectPrimary below may still swap this for a model - // that satisfies a bound skill's requires-model constraint. ModelConfigEntity globalDefault; + boolean explicitPinHonoured; try { + explicitPinHonoured = pinResolvesToEnabledModel(modelProvider, modelName); globalDefault = resolveRuntimeBaseModel(modelProvider, modelName, entity.getModelName()); } catch (Exception e) { throw new MateClawException("err.agent.no_default_model", "无法构建 Agent：请先在「设置 → 模型」中配置并启用默认模型"); } ModelConfigEntity runtimeModel; - try { - runtimeModel = providerRouter.selectPrimary(entity.getId(), globalDefault); - if (runtimeModel == null) runtimeModel = globalDefault; - } catch (Exception e) { - log.debug("[ProviderRouter] primary selection failed, falling back to global default: {}", - e.getMessage()); + if (explicitPinHonoured) { + // The caller (admin UI / chat console) handed us a concrete + // (provider, model) pin and it points to an enabled row. Honour + // it verbatim — running providerRouter.selectPrimary here would + // silently swap to a different model whenever a bound skill + // advertised a capability gap, which is exactly the "I switched + // model but the agent kept using the old one" surface. The + // diagnostic below still surfaces capability gaps in the logs + // so operators can see if the pinned model misses a need. runtimeModel = globalDefault; + } else { + try { + runtimeModel = providerRouter.selectPrimary(entity.getId(), globalDefault); + if (runtimeModel == null) runtimeModel = globalDefault; + } catch (Exception e) { + log.debug("[ProviderRouter] primary selection failed, falling back to global default: {}", + e.getMessage()); + runtimeModel = globalDefault; + } } // Even after the upgrade, log a WARN when the chosen primary // still doesn't satisfy needs (e.g. no preferred provider was diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java index 4a9c5bc6..81398692 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java @@ -16,6 +16,7 @@ import vip.mate.channel.model.ChannelEntity; import vip.mate.channel.notification.ApprovalNotificationService; import vip.mate.channel.service.ChannelService; import vip.mate.channel.web.ChatStreamTracker; +import vip.mate.exception.MateClawException; import vip.mate.memory.event.ConversationCompletionPublisher; import vip.mate.tts.TtsService; import vip.mate.workspace.conversation.ConversationService; @@ -237,6 +238,25 @@ public class ChannelMessageRouter { * @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; + } + if (!Boolean.TRUE.equals(fresh.getEnabled())) { + log.warn("[{}] Channel {} (id={}) is disabled; dropping message from {}", + adapter.getChannelType(), fresh.getName(), fresh.getId(), message.getSenderId()); + return; + } + channelEntity = fresh; + // 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 @@ -538,7 +558,31 @@ public class ChannelMessageRouter { */ 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.TRUE.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(); + if (agentId == null) { + log.warn("[{}] Channel {} has no associated agent at processing time; dropping message from {}", + adapter.getChannelType(), channelEntity.getName(), message.getSenderId()); + return; + } log.info("[{}] Processing message: sender={}, conversationId={}, agentId={}", adapter.getChannelType(), message.getSenderId(), conversationId, agentId); @@ -1120,6 +1164,14 @@ public class ChannelMessageRouter { * 路由消息并使用流式处理（用于支持流式的渠道，如 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.TRUE.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")); @@ -1194,6 +1246,49 @@ public class ChannelMessageRouter { // ==================== 工具方法 ==================== + /** + * 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} diff --git a/mateclaw-server/src/main/java/vip/mate/channel/controller/ChannelController.java b/mateclaw-server/src/main/java/vip/mate/channel/controller/ChannelController.java index eb5ac1b5..771e69d3 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/controller/ChannelController.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/controller/ChannelController.java @@ -97,11 +97,53 @@ public class ChannelController { channel.setId(id); channel.setWorkspaceId(existing.getWorkspaceId()); ChannelEntity updated = channelService.updateChannel(channel); - channelManager.restartChannel(id); + // Restart only when a field the adapter consumes BEFORE the router + // takes over has changed — channel type, enabled toggle, configJson + // (app credentials, connection_mode, domain, …), and botPrefix + // (consumed by AbstractChannelAdapter.shouldProcess / cleanBotPrefix + // before enqueue, so a per-message DB refresh in the router can't + // catch it). Pure router-visible metadata (bound agent, display + // name, description, identityJson) is re-read on every message via + // ChannelMessageRouter.freshChannelEntity, so it doesn't justify + // dropping the live connection — for Feishu WS that would mean a + // multi-second blackout where inbound messages never reach the bot. + if (transportConfigChanged(existing, updated)) { + channelManager.restartChannel(id); + } auditEventService.record("UPDATE", "CHANNEL", String.valueOf(id), updated.getName(), null); return R.ok(updated); } + /** + * True iff a field the adapter consumes BEFORE the router takes over (or + * that gates the adapter lifecycle entirely) has changed. + * + *

{@code agentId}, {@code name}, {@code description}, {@code identityJson} + * stay excluded — those are routing metadata read on every message via + * {@code ChannelMessageRouter.freshChannelEntity()}. + * + *

{@code botPrefix} IS included even though it's "just routing metadata" + * conceptually: {@code AbstractChannelAdapter.shouldProcess()} and + * {@code cleanBotPrefix()} run inside the adapter before the message + * reaches the router, and they read from the adapter's cached + * {@code channelEntity}. A prefix edit without restart would still filter + * and strip with the old prefix until the adapter is recreated. + */ + private boolean transportConfigChanged(ChannelEntity oldRow, ChannelEntity newRow) { + if (!java.util.Objects.equals(oldRow.getChannelType(), newRow.getChannelType())) { + return true; + } + if (!java.util.Objects.equals(oldRow.getEnabled(), newRow.getEnabled())) { + return true; + } + if (!java.util.Objects.equals(oldRow.getBotPrefix(), newRow.getBotPrefix())) { + return true; + } + String oldCfg = oldRow.getConfigJson() == null ? "" : oldRow.getConfigJson(); + String newCfg = newRow.getConfigJson() == null ? "" : newRow.getConfigJson(); + return !oldCfg.equals(newCfg); + } + @RequireWorkspaceRole("admin") @Operation(summary = "删除渠道") @DeleteMapping("/{id}") diff --git a/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java b/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java index c1a4167e..1c71606e 100644 --- a/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java +++ b/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java @@ -366,7 +366,24 @@ public class ConversationService { conv.setUsername(SYSTEM_USER); changed = true; } - if (conv.getAgentId() == null && agentId != null) { + // Shared conversations (IM channel sessions, cron job-specific rows) + // take their agent from the caller's current authoritative binding — + // the channel's bound agent for IM, the job's bound agent for cron. + // Sync so the admin sidebar / dashboard / context resolution all see + // the same agent the runtime is dispatching to; otherwise an admin + // who rebinds a channel from A to B leaves every existing + // conversation pointing at the old A. + // + // Exception: Web-origin cron uses {@code tasks_} as a + // single aggregate conversation for ALL of the workspace's web cron + // runs (see CronConversationResolver). Many jobs with different + // bound agents land in that same row; overwriting agentId per run + // would make the header / avatar / model selector flicker to + // whichever cron fired last. The aggregate has no single "owner + // agent" — leave its agentId alone (the original first-runner value + // is fine; UI treats this conversation specially anyway). + boolean isCronAggregate = conversationId != null && conversationId.startsWith("tasks_"); + if (!isCronAggregate && agentId != null && !agentId.equals(conv.getAgentId())) { conv.setAgentId(agentId); changed = true; } diff --git a/mateclaw-ui/src/i18n/locales/en-US.ts b/mateclaw-ui/src/i18n/locales/en-US.ts index 776f6ec7..968f19cc 100644 --- a/mateclaw-ui/src/i18n/locales/en-US.ts +++ b/mateclaw-ui/src/i18n/locales/en-US.ts @@ -183,6 +183,7 @@ export default { startNewChat: 'Start a new chat above', messages: '{count} messages', configModel: 'Configure Model', + modelSaveFailed: 'Model switch was not saved — the next message may still use the previous model', openSessions: 'Session Admin', clearMessages: 'Clear Messages', goToModelSettings: 'Go to Model Settings', diff --git a/mateclaw-ui/src/i18n/locales/zh-CN.ts b/mateclaw-ui/src/i18n/locales/zh-CN.ts index 1b2c633f..bf19e665 100644 --- a/mateclaw-ui/src/i18n/locales/zh-CN.ts +++ b/mateclaw-ui/src/i18n/locales/zh-CN.ts @@ -183,6 +183,7 @@ export default { startNewChat: '开始新对话吧', messages: '{count} 条消息', configModel: '配置模型', + modelSaveFailed: '模型切换未保存，下条消息可能仍走原模型', openSessions: '会话管理', clearMessages: '清空消息', goToModelSettings: '前往模型设置', diff --git a/mateclaw-ui/src/views/ChatConsole.vue b/mateclaw-ui/src/views/ChatConsole.vue index e3d5ab14..8914ea96 100644 --- a/mateclaw-ui/src/views/ChatConsole.vue +++ b/mateclaw-ui/src/views/ChatConsole.vue @@ -353,6 +353,12 @@ const selectedAgentId = ref('') const currentConversationId = ref('') const inputText = ref('') const modelSaving = ref(false) +// Monotonic counter for in-flight setModel PUTs. The finally handler +// only clears modelSaving when its captured seq is still the latest, so a +// stale-finishing earlier PUT can't unlock the selector while a newer one +// is still in flight, and (crucially) switching conversations mid-PUT +// can't permanently lock the selector by leaving modelSaving stuck true. +let modelSaveSeq = 0 // Issue #81 v2 R2: split the single showModelPrompt boolean into two flags so // the chat surface can either hard-block (blockingPrompt) or warn but let the // backend fallback chain take over (recoverablePrompt). Driven by @@ -418,15 +424,71 @@ function selectModel(value: string) { const [providerId, model] = value.split('::') if (!providerId || !model) return // Per-conversation model: switching here only affects THIS conversation. - // The backend pins it onto the conversation row when the next message is - // sent (see sendChatMessage payload); we also patch the local list entry so - // re-opening the conversation restores the choice without a round-trip. - activeModels.value = { activeLlm: { providerId, model } } + // We update the selector + the local list entry immediately so the UI is + // responsive, then persist the pin to the server right away IF the + // conversation already exists. Without the eager persist, IM channels + // (Feishu / DingTalk / WeCom …) keep using whatever the conversation row + // last had — they don't see the /chat/stream payload that the web path + // pins on send — so the user "switches model in the chat box" but the + // next IM inbound message still picks the old / default model. + // + // Snapshot the previous selection BEFORE the optimistic update so a + // failed PUT can roll the UI back instead of stranding the user with a + // model the backend isn't using. + const prevLlm = activeModels.value?.activeLlm + const prevActive: ActiveModelsInfo | null = prevLlm?.providerId && prevLlm?.model + ? { activeLlm: { providerId: prevLlm.providerId, model: prevLlm.model } } + : null const conv = conversations.value.find(c => c.conversationId === currentConversationId.value) + const prevConvProvider = conv?.modelProvider + const prevConvModel = conv?.modelName + + activeModels.value = { activeLlm: { providerId, model } } if (conv) { conv.modelProvider = providerId conv.modelName = model } + // Only persist when the conversation is already in the server-side list. + // A brand-new chat (newConversation() generated a local id that hasn't + // been sent through /chat/stream yet) has no row to PUT against; that + // case still relies on the first /chat/stream call writing the pin. + if (conv && currentConversationId.value) { + const cid = currentConversationId.value + const mySeq = ++modelSaveSeq + modelSaving.value = true + conversationApi.setModel(cid, providerId, model) + .catch((e: any) => { + console.warn('[ChatConsole] Failed to persist model pin:', e) + mcToast.warning(t('chat.modelSaveFailed')) + // Roll back the visible selector + the cached conv pin so the UI + // doesn't keep claiming a model the backend isn't using. Only do + // it when the user is still on the same conversation AND hasn't + // picked yet another model — otherwise we'd corrupt the more + // recent state with this PUT's snapshot. + const liveConv = conversations.value.find(c => c.conversationId === cid) + if (liveConv && liveConv.modelProvider === providerId && liveConv.modelName === model) { + liveConv.modelProvider = prevConvProvider + liveConv.modelName = prevConvModel + } + if (currentConversationId.value !== cid) return + const stillShowingFailedPick = + activeModels.value?.activeLlm?.providerId === providerId + && activeModels.value?.activeLlm?.model === model + if (!stillShowingFailedPick) return + activeModels.value = prevActive + }) + .finally(() => { + // Only the LATEST in-flight PUT clears the saving flag. An earlier + // PUT finishing late must not flip saving to false while a newer + // one is still pending (the selector would unlock during a live + // request); and a conversation switch mid-PUT must not strand the + // flag at true forever (which would lock the selector across + // every conversation — the original bug this seq counter fixes). + if (mySeq === modelSaveSeq) { + modelSaving.value = false + } + }) + } } /** @@ -441,6 +503,44 @@ function applyConversationModel(conv?: Conversation | null) { } } +/** + * After a poll refreshes the conversation list, the currently-open + * conversation may have drifted server-side: an admin may have rebound the + * channel to a different agent, or pinned a different model via another + * tab / API call. Pull the new server-side state into the local selector + + * agent header so the chat surface doesn't keep claiming the old binding. + * + * Skipped while a turn is generating — yanking the model / agent mid-stream + * would orphan the active SSE subscription. + */ +function reconcileCurrentConversation() { + if (!currentConversationId.value) return + if (isGenerating.value) return + // Don't fight an in-flight setModel write — the poll cycle may run BEFORE + // the PUT lands, in which case the server still reports the old pin and + // we'd flicker the UI back. Wait for the next tick. + if (modelSaving.value) return + const fresh = conversations.value.find(c => c.conversationId === currentConversationId.value) + if (!fresh) return + if (fresh.agentId != null && String(fresh.agentId) !== String(selectedAgentId.value)) { + selectedAgentId.value = fresh.agentId + } + const pickedProvider = activeModels.value?.activeLlm?.providerId + const pickedModel = activeModels.value?.activeLlm?.model + const serverHasPin = !!(fresh.modelProvider && fresh.modelName) + if (serverHasPin) { + if (fresh.modelProvider !== pickedProvider || fresh.modelName !== pickedModel) { + activeModels.value = { activeLlm: { providerId: fresh.modelProvider!, model: fresh.modelName! } } + } + } else if (pickedProvider || pickedModel) { + // Server-side pin was cleared (admin reset, model deleted, …) but the + // local selector still shows the old pick. Drop back to whatever the + // global default resolves to — applyConversationModel does the right + // thing when conv has no pin. + applyConversationModel(fresh) + } +} + // 拖拽上传 — useFileDrop owns the hover/counter state; the directory-aware // payload handling (electron paths vs web FileSystem entries) stays here. const { isDragging, onDragEnter, onDragLeave, onDrop } = useFileDrop(processDroppedItems) @@ -949,6 +1049,7 @@ async function pollActivity() { try { try { await loadConversations() + reconcileCurrentConversation() } catch { // 静默失败，下一轮再试 }