mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
fix(channel,agent,chat): unify channel binding / conversation agent / model pin state sources
This commit is contained in:
matevip
parent
7f5652b2f0
commit
3ae4498f38
@ -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
|
||||
|
||||
@ -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<String> 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.
|
||||
*
|
||||
* <p>Failure semantics:
|
||||
* <ul>
|
||||
* <li><b>Channel deleted</b> — {@link ChannelService#getChannel} throws
|
||||
* a {@link MateClawException} with {@code msgKey="err.channel.not_found"}.
|
||||
* We return {@code null} so the caller drops the message: the channel
|
||||
* no longer exists, routing the message would land it against a row
|
||||
* that's been removed.</li>
|
||||
* <li><b>Transient lookup failure</b> — any other exception (DB blip,
|
||||
* NPE in mapper, …). We fall back to the snapshot so an isolated
|
||||
* infrastructure hiccup doesn't black-hole live traffic.</li>
|
||||
* </ul>
|
||||
*
|
||||
* <p>{@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}
|
||||
|
||||
@ -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.
|
||||
*
|
||||
* <p>{@code agentId}, {@code name}, {@code description}, {@code identityJson}
|
||||
* stay excluded — those are routing metadata read on every message via
|
||||
* {@code ChannelMessageRouter.freshChannelEntity()}.
|
||||
*
|
||||
* <p>{@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}")
|
||||
|
||||
@ -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_<workspaceId>} 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;
|
||||
}
|
||||
|
||||
@ -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',
|
||||
|
||||
@ -183,6 +183,7 @@ export default {
|
||||
startNewChat: '开始新对话吧',
|
||||
messages: '{count} 条消息',
|
||||
configModel: '配置模型',
|
||||
modelSaveFailed: '模型切换未保存，下条消息可能仍走原模型',
|
||||
openSessions: '会话管理',
|
||||
clearMessages: '清空消息',
|
||||
goToModelSettings: '前往模型设置',
|
||||
|
||||
@ -353,6 +353,12 @@ const selectedAgentId = ref<string | number>('')
|
||||
const currentConversationId = ref<string>('')
|
||||
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 {
|
||||
// 静默失败，下一轮再试
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user