+ * Default implementation degrades to the two-arg variant, dropping the + * owner key. External providers that need per-owner recall (e.g. Mem0) + * should override this to use {@code ownerKey} as their per-user identifier. + * + * @param agentId the agent ID + * @param userQuery the current user message + * @param ownerKey memory owner key (e.g. {@code "user:42"}), or null if unknown + * @return context text to inject, or empty string + */ + default String prefetch(Long agentId, String userQuery, String ownerKey) { + return prefetch(agentId, userQuery); + } + /** * Post-turn sync. Called after LLM response is available. * Should be non-blocking (async). @@ -62,6 +80,28 @@ public interface PluginMemoryProvider { String userMessage, String assistantReply) { } + /** + * Post-turn sync with per-owner isolation. Called by the platform with the + * same {@code ownerKey} that was resolved for this turn's prefetch, so + * providers can persist the turn under the same per-user identifier they + * recall by. + *
+ * Default implementation degrades to the four-arg variant, dropping the
+ * owner key. External providers that isolate memory per end-user should
+ * override this so that written memories stay reachable by owner-scoped
+ * recall.
+ *
+ * @param agentId the agent ID
+ * @param conversationId the conversation ID
+ * @param userMessage user's message text
+ * @param assistantReply assistant's reply text
+ * @param ownerKey memory owner key (e.g. {@code "user:42"}), or null if unknown
+ */
+ default void syncTurn(Long agentId, String conversationId,
+ String userMessage, String assistantReply, String ownerKey) {
+ syncTurn(agentId, conversationId, userMessage, assistantReply);
+ }
+
/**
* Tool beans this provider wants to expose to the agent.
*/
diff --git a/mateclaw-plugin-mem0/pom.xml b/mateclaw-plugin-mem0/pom.xml
new file mode 100644
index 00000000..d9c75a1e
--- /dev/null
+++ b/mateclaw-plugin-mem0/pom.xml
@@ -0,0 +1,74 @@
+
+
+ * Covers the two endpoints used by {@link Mem0Provider}: + *
Failure semantics: every call either returns a parsed result or throws
+ * {@link Mem0Exception}. Callers are expected to catch and degrade gracefully
+ * (return empty recall / log sync failures).
+ *
+ * @author MateClaw Team
+ */
+class Mem0Client {
+
+ private final Mem0Config config;
+ private final HttpClient http;
+ private final ObjectMapper mapper = new ObjectMapper();
+
+ Mem0Client(Mem0Config config) {
+ this.config = config;
+ this.http = HttpClient.newBuilder()
+ .connectTimeout(Duration.ofMillis(config.timeoutMs()))
+ .build();
+ }
+
+ /**
+ * Push a conversation turn to Mem0 for extraction.
+ *
+ * @param userId Mem0 user_id, typically MateClaw's ownerKey
+ * @param agentId Mem0 agent_id, typically MateClaw's agentId
+ * @param conversationId optional conversation identifier (stored as metadata)
+ * @param userMessage user's message text
+ * @param assistantReply assistant's reply text
+ */
+ void addMemories(String userId, String agentId, String conversationId,
+ String userMessage, String assistantReply) {
+ ObjectNode body = mapper.createObjectNode();
+ body.put("user_id", userId);
+ if (agentId != null && !agentId.isBlank()) {
+ body.put("agent_id", agentId);
+ }
+ ArrayNode messages = body.putArray("messages");
+ if (userMessage != null && !userMessage.isBlank()) {
+ ObjectNode m = messages.addObject();
+ m.put("role", "user");
+ m.put("content", userMessage);
+ }
+ if (assistantReply != null && !assistantReply.isBlank()) {
+ ObjectNode m = messages.addObject();
+ m.put("role", "assistant");
+ m.put("content", assistantReply);
+ }
+ if (conversationId != null && !conversationId.isBlank()) {
+ ObjectNode meta = body.putObject("metadata");
+ meta.put("conversation_id", conversationId);
+ }
+
+ post("/memories/", body);
+ }
+
+ /**
+ * Semantic recall.
+ *
+ * @param userId Mem0 user_id (ownerKey)
+ * @param agentId Mem0 agent_id
+ * @param query user query text
+ * @return list of memory strings, possibly empty; never null
+ */
+ List
+ * Read once from {@link vip.mate.plugin.api.PluginContext#getConfig} at plugin
+ * load time and passed to {@link Mem0Client} / {@link Mem0Provider}. Snapshot
+ * semantics — config changes require a plugin reload.
+ *
+ * @param baseUrl Mem0 REST API base URL, e.g. {@code http://localhost:8080}
+ * @param apiKey optional bearer token; null/blank means no Authorization header
+ * @param searchEnabled whether prefetch should query Mem0 /memories/search/
+ * @param syncEnabled whether syncTurn should POST to Mem0 /memories/
+ * @param maxResults cap on memories returned per recall
+ * @param timeoutMs HTTP timeout for both recall and sync
+ * @author MateClaw Team
+ */
+record Mem0Config(
+ String baseUrl,
+ String apiKey,
+ boolean searchEnabled,
+ boolean syncEnabled,
+ int maxResults,
+ int timeoutMs
+) {
+ static final int DEFAULT_MAX_RESULTS = 5;
+ static final int DEFAULT_TIMEOUT_MS = 3000;
+
+ /**
+ * Whether this provider should participate at all.
+ * Mem0 without a base URL is unusable; treat as unavailable.
+ */
+ boolean isUsable() {
+ return baseUrl != null && !baseUrl.isBlank();
+ }
+
+ /**
+ * Strip trailing slashes from the base URL to avoid double-slash in path joins.
+ */
+ String normalizedBaseUrl() {
+ String url = baseUrl;
+ while (url.endsWith("/")) {
+ url = url.substring(0, url.length() - 1);
+ }
+ return url;
+ }
+}
diff --git a/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Exception.java b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Exception.java
new file mode 100644
index 00000000..064fffa3
--- /dev/null
+++ b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Exception.java
@@ -0,0 +1,21 @@
+package vip.mate.plugin.mem0;
+
+/**
+ * Raised when a Mem0 REST call fails (non-2xx response, IO error, timeout).
+ *
+ * Caught and logged by {@link Mem0Provider} so that Mem0 outages degrade
+ * gracefully (empty recall / dropped sync) without affecting the agent's
+ * response path.
+ *
+ * @author MateClaw Team
+ */
+class Mem0Exception extends RuntimeException {
+
+ Mem0Exception(String message) {
+ super(message);
+ }
+
+ Mem0Exception(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Plugin.java b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Plugin.java
new file mode 100644
index 00000000..62696403
--- /dev/null
+++ b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Plugin.java
@@ -0,0 +1,107 @@
+package vip.mate.plugin.mem0;
+
+import org.slf4j.Logger;
+import vip.mate.plugin.api.MateClawPlugin;
+import vip.mate.plugin.api.PluginContext;
+
+import java.net.URI;
+
+/**
+ * MateClaw plugin entrypoint that registers {@link Mem0Provider} with the
+ * platform's memory subsystem.
+ *
+ * Lifecycle:
+ * This plugin is NOT part of the default stack. Users must:
+ *
+ * Behavior matrix:
+ * Per-owner isolation: {@code ownerKey} (e.g. {@code "user:42"}) is passed
+ * verbatim as Mem0's {@code user_id}; {@code agentId} as Mem0's {@code agent_id}.
+ * When {@code ownerKey} is null/blank, both recall and sync are skipped — Mem0
+ * requires {@code user_id}.
+ *
+ * Asynchronous sync: a single-thread daemon executor is used
+ * so that bursts of turns don't pile up on the platform's request thread.
+ *
+ * @author MateClaw Team
+ */
+class Mem0Provider implements PluginMemoryProvider {
+
+ static final String ID = "mem0";
+
+ private final Mem0Config config;
+ private final Mem0Client client;
+ private final Logger log;
+ private final Executor async;
+
+ Mem0Provider(Mem0Config config, Mem0Client client, Logger log) {
+ this.config = config;
+ this.client = client;
+ this.log = log;
+ // Single-thread executor is enough — syncTurn calls are sequential per
+ // agent and not latency-sensitive; the platform's request thread must
+ // not be blocked. A bounded single-thread queue keeps memory footprint
+ // predictable even under burst load.
+ this.async = Executors.newSingleThreadExecutor(r -> {
+ Thread t = new Thread(r, "mem0-sync");
+ t.setDaemon(true);
+ return t;
+ });
+ }
+
+ @Override
+ public String id() {
+ return ID;
+ }
+
+ @Override
+ public int order() {
+ // Same as the SPI default; declared explicitly for clarity.
+ return 200;
+ }
+
+ @Override
+ public boolean isAvailable() {
+ // Provider is "available" if at least one of recall/sync can fire.
+ return config.isUsable() && (config.searchEnabled() || config.syncEnabled());
+ }
+
+ @Override
+ public String systemPromptBlock(Long agentId) {
+ return "";
+ }
+
+ @Override
+ public String prefetch(Long agentId, String userQuery) {
+ // Two-arg variant: no owner key → cannot isolate per-user → skip.
+ // Mem0 requires user_id; without it the call would either fail or
+ // return global memories breaking per-owner isolation.
+ return "";
+ }
+
+ @Override
+ public String prefetch(Long agentId, String userQuery, String ownerKey) {
+ if (!config.searchEnabled()) {
+ return "";
+ }
+ if (ownerKey == null || ownerKey.isBlank()) {
+ return "";
+ }
+ if (userQuery == null || userQuery.isBlank()) {
+ return "";
+ }
+ try {
+ List
+ * The {@code [Mem0 Recall]} label is intentional: it lets the LLM
+ * distinguish this block from the local providers' output and avoid
+ * treating it as authoritative PROFILE.md content.
+ */
+ private String formatRecallBlock(List Only the fresh-turn entries ({@code chat} / {@code chatStream} /
+ * {@code chatStructuredStream} / {@code execute}) call this. The
+ * approval-replay entries ({@code chatWithReplay*}) resume the SAME
+ * logical turn after a tool approval and must keep the safety net for
+ * work already done before the pause.
+ */
+ private void clearAutoRecordedForNewTurn(String conversationId) {
+ if (progressLedgerService == null || conversationId == null || conversationId.isBlank()) {
+ return;
+ }
+ try {
+ progressLedgerService.clearAutoRecorded(conversationId);
+ } catch (Exception e) {
+ // Ledger housekeeping must never block the chat itself.
+ log.warn("Failed to clear auto-recorded ledger entries for {}: {}",
+ conversationId, e.getMessage());
+ }
+ }
+
public String chat(Long agentId, String message, String conversationId) {
return chat(agentId, message, conversationId, ChatOrigin.EMPTY);
}
@@ -251,6 +299,7 @@ public class AgentService {
* down to {@code @Tool} methods via Spring AI {@link org.springframework.ai.chat.model.ToolContext}.
*/
public String chat(Long agentId, String message, String conversationId, ChatOrigin origin) {
+ clearAutoRecordedForNewTurn(conversationId);
memoryRecallTracker.trackRecalls(agentId, message);
BaseAgent agent = getOrBuildAgentForConversation(agentId, conversationId);
ChatOriginHolder.set(origin != null ? origin : ChatOrigin.EMPTY);
@@ -287,6 +336,7 @@ public class AgentService {
}
public Flux
+ * The {@link ChatOrigin} carried in a tool's {@code ToolContext} usually already
+ * holds the workspaceId (populated at the web / channel entry point). Some paths
+ * — notably approval replay — carry a conversationId but a {@code null}
+ * workspaceId; there we fall back to {@link WorkspaceLookupCache}, which maps a
+ * conversationId to its owning workspace. When neither yields a workspace, the
+ * result is {@code null}: callers must treat that conservatively (resolve only
+ * builtin / global skills, never another workspace's skill).
+ */
+@Component
+@RequiredArgsConstructor
+public class AgentWorkspaceResolver {
+
+ private final WorkspaceLookupCache workspaceLookupCache;
+
+ /** Best-effort workspace id for the given origin; {@code null} if unresolved. */
+ @Nullable
+ public Long resolve(@Nullable ChatOrigin origin) {
+ if (origin == null) {
+ return null;
+ }
+ if (origin.workspaceId() != null) {
+ return origin.workspaceId();
+ }
+ String conversationId = origin.conversationId();
+ return conversationId != null ? workspaceLookupCache.resolveByConversation(conversationId) : null;
+ }
+}
diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java
index 3aacd086..ba5ecd35 100644
--- a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java
+++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java
@@ -8,6 +8,8 @@ import org.springframework.ai.chat.messages.UserMessage;
import org.springframework.ai.chat.model.ChatModel;
import org.springframework.ai.chat.model.ChatResponse;
import org.springframework.ai.chat.prompt.Prompt;
+import org.springframework.http.HttpHeaders;
+import org.springframework.web.client.RestClientResponseException;
import org.springframework.web.reactive.function.client.WebClientResponseException;
import vip.mate.channel.web.ChatStreamTracker;
import vip.mate.llm.chatmodel.AssistantThinkingRelay;
@@ -15,6 +17,10 @@ import vip.mate.llm.chatmodel.ReasoningContentCache;
import reactor.core.Disposable;
+import java.time.Instant;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -193,6 +199,16 @@ public class NodeStreamingChatHelper {
else healthTracker.recordFailure(primaryProviderId);
}
+ /**
+ * Record a primary failure carrying a provider-stated retry window so the
+ * health tracker can start a cooldown of exactly that length. No-op under
+ * the same conditions as {@link #recordPrimary}.
+ */
+ private void recordPrimaryFailure(long cooldownOverrideMs) {
+ if (healthTracker == null || primaryProviderId == null) return;
+ healthTracker.recordFailure(primaryProviderId, cooldownOverrideMs);
+ }
+
/**
* Map an {@link ErrorType} to the matching pool
* {@link vip.mate.llm.failover.AvailableProviderPool.RemovalSource} for
@@ -207,7 +223,11 @@ public class NodeStreamingChatHelper {
* {@link vip.mate.llm.failover.ProviderHealthTracker}'s cooldown instead. Priority: {@code Retry-After} (delta-seconds or HTTP-date) →
+ * Anthropic RFC-3339 reset instants → OpenAI-style duration resets. For
+ * multi-bucket reset headers the earliest future instant wins —
+ * optimistic, because a premature retry just re-records the hint, while
+ * over-waiting silently costs the user the whole window. Each constant carries four policy attributes so the retry loop, the
+ * fallback-chain router, the pool eviction hook, and the health tracker
+ * all read one source of truth instead of maintaining parallel
+ * per-type branch chains: Two types additionally have side-effectful recovery steps that
+ * cannot be expressed as attributes and keep explicit branches in the
+ * loop: {@link #PROMPT_TOO_LONG} (report server-stated window, return to
+ * node for compaction) and {@link #THINKING_BLOCK_ERROR} (strip stale
+ * thinking blocks from the prompt, then retry once). Defaults to file-read tools that already cap their own output internally.
* Configurable so deployments can add more retrieval-style tools (e.g.,
* MCP-provided readers) without code changes. {@code readSkillFile} / {@code load_skill} are included because they
+ * deliberately return the full SKILL.md — the skill's usage contract —
+ * and spilling it down to a preview makes the model act on incomplete
+ * instructions (e.g. wrong API parameter names). Their references/scripts
+ * reads are already self-paginated to a bounded size. Auto-recorded entries are a safety net against context trimming
+ * within one turn's tool loop. Letting them survive into the next
+ * user turn is harmful: the snapshot renders them as DONE alongside the
+ * "已完成的步骤不要重复执行" instruction, which stops the agent from
+ * re-running read-only / status-query tools when the user repeats a
+ * question that needs fresh data (e.g. "看下会议室有没有人"), and the
+ * frozen 120-char result note tempts it to answer from stale output.
+ *
+ * LLM-authored regular entries and pinned skill constraints are
+ * untouched — multi-turn task tracking keeps working.
+ */
+ public void clearAutoRecorded(String conversationId) {
+ if (conversationId == null || conversationId.isBlank()) {
+ return;
+ }
+ ReentrantLock lock = upsertLocks.computeIfAbsent(conversationId, k -> new ReentrantLock());
+ lock.lock();
+ try {
+ LedgerWrapper wrapper = loadWrapper(conversationId);
+ boolean removed = wrapper.entries.keySet().removeIf(
+ k -> k != null && k.startsWith(ProgressLedger.AUTO_RECORDED_PREFIX));
+ if (removed) {
+ persistWrapper(conversationId, wrapper);
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+
/**
* Auto-record a completed tool call as a ledger entry (B5). Uses the
* {@link ProgressLedger#AUTO_RECORDED_PREFIX} on the key so the renderer
diff --git a/mateclaw-server/src/main/java/vip/mate/approval/grant/controller/ApprovalGrantController.java b/mateclaw-server/src/main/java/vip/mate/approval/grant/controller/ApprovalGrantController.java
index 6074c80b..42ded287 100644
--- a/mateclaw-server/src/main/java/vip/mate/approval/grant/controller/ApprovalGrantController.java
+++ b/mateclaw-server/src/main/java/vip/mate/approval/grant/controller/ApprovalGrantController.java
@@ -11,6 +11,7 @@ import org.springframework.security.core.Authentication;
import org.springframework.web.bind.annotation.*;
import vip.mate.approval.grant.entity.ApprovalGrant;
import vip.mate.approval.grant.entity.ApprovalResolutionLog;
+import vip.mate.agent.repository.AgentMapper;
import vip.mate.approval.grant.repository.ApprovalGrantMapper;
import vip.mate.approval.grant.repository.ApprovalResolutionLogMapper;
import vip.mate.approval.grant.service.ApprovalGrantService;
@@ -55,6 +56,7 @@ public class ApprovalGrantController {
private static final long DEFAULT_WORKSPACE_ID = 1L;
private final ApprovalGrantService grantService;
+ private final AgentMapper agentMapper;
private final ApprovalGrantMapper grantMapper;
private final ApprovalResolutionLogMapper resolutionMapper;
private final AuthService authService;
@@ -295,6 +297,7 @@ public class ApprovalGrantController {
}
}
case ApprovalGrant.ScopeType.AGENT -> {
+ requireExistingAgent(body.scopeId);
if (toolNull) {
requireAdminPlusPassword(isAdmin, body.password, actorId);
} else if (!isAdmin) {
@@ -304,6 +307,15 @@ public class ApprovalGrantController {
}
}
case ApprovalGrant.ScopeType.WORKSPACE -> {
+ // WORKSPACE-scope matching requires scope_id == the invocation's
+ // workspaceId AND the grant row's workspace_id (tenant column) to
+ // equal that same workspace — a scopeId pointing anywhere else can
+ // never fire. Reject the dead configuration outright.
+ if (!String.valueOf(workspaceId).equals(body.scopeId)) {
+ throw new MateClawException("err.approval.workspace_scope_mismatch", 400,
+ "WORKSPACE-scope scopeId must equal the current workspace id ("
+ + workspaceId + "); a cross-workspace grant can never match");
+ }
workspaceService.requirePermission(workspaceId, actorId, "admin");
if (toolNull) {
requireAdminPlusPassword(true, body.password, actorId);
@@ -314,6 +326,24 @@ public class ApprovalGrantController {
}
}
+ /**
+ * AGENT-scope scopeId must reference an existing agent — a workspace or
+ * conversation id pasted here compiles into a grant that never matches.
+ */
+ private void requireExistingAgent(String scopeId) {
+ Long agentId;
+ try {
+ agentId = Long.parseLong(scopeId);
+ } catch (NumberFormatException e) {
+ throw new MateClawException("err.approval.agent_not_found", 400,
+ "AGENT-scope scopeId must be a numeric agent id: " + scopeId);
+ }
+ if (agentMapper.selectById(agentId) == null) {
+ throw new MateClawException("err.approval.agent_not_found", 400,
+ "AGENT-scope scopeId does not reference an existing agent: " + scopeId);
+ }
+ }
+
private void requireAdminPlusPassword(boolean isAdmin, String rawPassword, Long actorId) {
if (!isAdmin) {
throw new MateClawException("err.approval.admin_required", 403, "admin role required");
diff --git a/mateclaw-server/src/main/java/vip/mate/approval/grant/repository/ApprovalGrantMapper.java b/mateclaw-server/src/main/java/vip/mate/approval/grant/repository/ApprovalGrantMapper.java
index f041de68..d7c1f888 100644
--- a/mateclaw-server/src/main/java/vip/mate/approval/grant/repository/ApprovalGrantMapper.java
+++ b/mateclaw-server/src/main/java/vip/mate/approval/grant/repository/ApprovalGrantMapper.java
@@ -49,6 +49,21 @@ public interface ApprovalGrantMapper extends BaseMapper
+ * 卡片式流式渠道不经过 {@link #renderAndSend}（它们自己管理消息长度和
+ * 卡片更新节奏），如果不在流式收尾处调用本方法，
+ * {@code filter_thinking} / {@code filter_tool_messages} 两个开关
+ * 在这些路径上就完全不生效。
+ *
+ * @param content 原始文本
+ * @return 过滤后的文本（入参为空时返回空串）
+ */
+ protected String filterOutboundContent(String content) {
+ return ChannelMessageRenderer.applyFilters(content,
+ getConfigBoolean("filter_thinking", true),
+ getConfigBoolean("filter_tool_messages", true));
+ }
+
/**
* Approval notice rendering — primary implementation position.
*
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelDedupProperties.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelDedupProperties.java
new file mode 100644
index 00000000..08227874
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelDedupProperties.java
@@ -0,0 +1,79 @@
+package vip.mate.channel;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import java.time.Duration;
+
+/**
+ * Tunables for inbound channel-message deduplication.
+ *
+ * IM platforms redeliver the same message when an acknowledgement is late,
+ * lost, or answered with a non-200 — DingTalk, WeCom and Feishu all do this.
+ * Every redelivery that reaches the router starts a full, independent agent
+ * turn, so the user sees the same answer twice (and the conversation gains a
+ * duplicate user/assistant pair). {@link InboundMessageDeduplicator} keeps a
+ * short-lived record of the message identities already claimed so a
+ * redelivery is dropped instead of answered again.
+ *
+ * 入站渠道消息去重配置。平台重投同一条消息时，若不去重则每次重投都会跑一轮完整
+ * 的 Agent 回合，用户看到重复答复。
+ *
+ * Must comfortably exceed the platforms' redelivery windows (seconds to
+ * low minutes) while staying short enough that a user who genuinely resends
+ * the identical payload later is not silenced. Note that a resend carries a
+ * fresh platform message id in every channel we support, so the TTL only
+ * matters for the id-less fallback identity.
+ */
+ private Duration ttl = Duration.ofMinutes(5);
+
+ /**
+ * Hard cap on tracked identities. Reached only under sustained traffic
+ * within one TTL window; the oldest claims are dropped first.
+ */
+ private int maxSize = 2000;
+
+ public boolean isEnabled() {
+ return enabled;
+ }
+
+ public void setEnabled(boolean enabled) {
+ this.enabled = enabled;
+ }
+
+ public Duration getTtl() {
+ return ttl;
+ }
+
+ public void setTtl(Duration ttl) {
+ this.ttl = ttl;
+ }
+
+ public int getMaxSize() {
+ return maxSize;
+ }
+
+ public void setMaxSize(int maxSize) {
+ this.maxSize = maxSize;
+ }
+}
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMagicCommand.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMagicCommand.java
new file mode 100644
index 00000000..3e3e0e4f
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMagicCommand.java
@@ -0,0 +1,134 @@
+package vip.mate.channel;
+
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * User-typed channel control commands that should be handled by the platform
+ * instead of being sent to the agent as normal prompt text.
+ *
+ * Matching rules:
+ *
+ * 卡片式流式渠道（钉钉 AI Card、飞书 CardKit）自己管理长度限制，
+ * 但同样需要遵守渠道的消息过滤配置，因此把过滤部分单独暴露出来。
+ *
+ * @param content 原始内容
+ * @param filterThinking 是否过滤 thinking 标签
+ * @param filterToolMessages 是否过滤工具调用信息
+ * @return 过滤后的内容（入参为空时返回空串）
+ */
+ public static String applyFilters(String content,
+ boolean filterThinking,
+ boolean filterToolMessages) {
+ if (content == null || content.isBlank()) {
+ return "";
+ }
+
String rendered = content;
// 1. 过滤 thinking
@@ -89,14 +113,7 @@ public final class ChannelMessageRenderer {
}
// 3. 清理多余空行
- rendered = rendered.replaceAll("\n{3,}", "\n\n").trim();
-
- if (rendered.isEmpty()) {
- return List.of("");
- }
-
- // 4. 按平台限制分割
- return truncateForPlatform(rendered, maxLength);
+ return rendered.replaceAll("\n{3,}", "\n\n").trim();
}
// ==================== 过滤方法 ====================
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 397f6766..23db1dd8 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java
@@ -15,11 +15,15 @@ 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;
@@ -33,6 +37,7 @@ 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.*;
@@ -65,6 +70,7 @@ public class ChannelMessageRouter {
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. */
@@ -78,6 +84,22 @@ public class ChannelMessageRouter {
@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) {}
@@ -148,30 +170,6 @@ public class ChannelMessageRouter {
return currentMergedLength > LONG_TEXT_THRESHOLD ? LONG_DEBOUNCE_MS : DEBOUNCE_MS;
}
- /**
- * Plan-Execute SSE events that the Web Console mirror needs to see when
- * a conversation runs through an IM channel.
- *
- * The agent emits these via {@code GraphEventPublisher} and they ride on
- * the {@code chatStructuredStream} Flux as {@code StreamDelta.event(...)}.
- * Web direct chats already broadcast them via the ChatController
- * accumulator. IM channels (DingTalk + the seven sync-path adapters)
- * historically dropped them — DingTalk's {@code processStreamAsText}
- * only consumes {@code delta.content()}, and the sync {@code chat()}
- * collector explicitly filters {@code delta.isEvent()} out. The whitelist
- * is applied in the IM stream path so PlanStepsPanel renders correctly
- * when an operator monitors an IM conversation in the Web Console.
- *
- * Whitelist (not pass-through) so Web-side accumulator-internal events
- * like {@code _usage_final} or future agent-internal markers don't leak
- * to subscribers.
- */
- private static final Set 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);
+ }
+
/**
* 防抖到期：将合并后的消息真正放入渠道队列
*/
@@ -415,6 +481,12 @@ public class ChannelMessageRouter {
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, "系统繁忙，请稍后再试");
@@ -467,7 +539,7 @@ public class ChannelMessageRouter {
continue; // 超时，重新检查 shutdown 标志
}
- String conversationId = buildConversationId(entry.message());
+ String conversationId = buildConversationId(entry.message(), entry.channelEntity().getId());
ReentrantLock lock = sessionLocks.computeIfAbsent(conversationId, k -> new ReentrantLock());
lock.lock();
@@ -592,15 +664,21 @@ public class ChannelMessageRouter {
}
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);
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);
@@ -780,46 +858,72 @@ public class ChannelMessageRouter {
if (adapter instanceof StreamingChannelAdapter streamingAdapter) {
savedAssistantId = processWithStreaming(message, streamingAdapter, conversationId, agentId, promptText, channelEntity, chatOrigin);
} else {
- // Sync path for non-streaming IM adapters (feishu / wecom / weixin /
- // slack / discord / qq / telegram). We can't use agentService.chat()
+ // 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 plan_created / plan_step_* events that the Web
- // Console mirror needs to render PlanStepsPanel. Instead we consume
- // chatStructuredStream directly: content gets accumulated for the IM
- // reply, and whitelisted plan events are mirrored to ChatStreamTracker
- // for any Web SSE viewer of the same conversationId.
- StringBuilder replyAccumulator = new StringBuilder();
+ // 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();
- // Token usage + model attribution: capture _usage_final event emitted at stream end
- final int[] usage = {0, 0, 0, 0, 0}; // [prompt, completion, cacheRead, cacheWrite, reasoning]
- final String[] modelInfo = {null, null}; // [runtimeModel, runtimeProvider]
+ // 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 -> {
- if (delta.isEvent()) {
- if ("_usage_final".equals(delta.eventType())) {
- Map
@@ -918,30 +1298,6 @@ public class ChannelMessageRouter {
* - StreamingChannelAdapter 负责渲染（AI Card / 卡片更新 / 文本累积等）
* - Router 负责后续的审批检查、消息持久化、事件发布
*/
- /**
- * Forward whitelisted Plan-Execute SSE events to ChatStreamTracker so a
- * Web Console viewer of an IM-routed conversation sees PlanStepsPanel.
- *
- * Bounded to {@link #MIRRORED_PLAN_EVENTS} — see the constant's javadoc
- * for why this is a whitelist rather than a pass-through. Failures here
- * are best-effort and never propagate, since dropping a UI update is
- * preferable to derailing the channel reply.
- */
- private void mirrorPlanEventToTracker(String conversationId,
- AgentService.StreamDelta delta,
- String channelTypeForLog) {
- String eventType = delta.eventType();
- if (eventType == null || !MIRRORED_PLAN_EVENTS.contains(eventType)) {
- return;
- }
- try {
- streamTracker.broadcastObject(conversationId, eventType, delta.eventData());
- } catch (Exception ex) {
- log.debug("[{}] Failed to mirror plan event {}: {}",
- channelTypeForLog, eventType, ex.getMessage());
- }
- }
-
private Long processWithStreaming(ChannelMessage message, StreamingChannelAdapter streamingAdapter,
String conversationId, Long agentId, String promptText,
ChannelEntity channelEntity, ChatOrigin chatOrigin) {
@@ -949,33 +1305,22 @@ public class ChannelMessageRouter {
log.info("[{}] Streaming processing started: conversationId={}", channelType, conversationId);
try {
- // Step 1: 产生事件流（RFC-063r §2.5: forward ChatOrigin so tools see channelId）
+ // Step 1: 产生事件流（forward ChatOrigin so tools see channelId）
Flux 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();
- return message.getChannelType() + ":" + identifier;
+ if (channelId == null) {
+ return message.getChannelType() + ":" + identifier;
+ }
+ return message.getChannelType() + ":" + channelId + ":" + identifier;
}
/**
@@ -1577,6 +1950,18 @@ public class ChannelMessageRouter {
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
*/
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelSessionStore.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelSessionStore.java
index d6a85032..aed2c5c7 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelSessionStore.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelSessionStore.java
@@ -8,6 +8,7 @@ import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;
import vip.mate.channel.model.ChannelSessionEntity;
import vip.mate.channel.repository.ChannelSessionMapper;
+import vip.mate.workspace.conversation.event.ConversationDeletedEvent;
import java.time.LocalDateTime;
import java.util.Comparator;
@@ -56,6 +57,30 @@ public class ChannelSessionStore {
log.info("ChannelSessionStore initialized: loaded {} sessions from DB", sessions.size());
}
+ /**
+ * Drop the cached session for a conversation the user just deleted.
+ *
+ * {@code deleteConversation} removes the {@code mate_channel_session}
+ * row inside its DB cascade, but the cache is this class's private state
+ * and no DB delete can reach it. Without this listener the entry survives
+ * as a phantom: the next inbound message takes the "update existing" branch
+ * and calls {@code updateById} against a primary key that no longer exists,
+ * which affects 0 rows and never re-inserts — so the channel session stays
+ * missing and proactive push / cron channel resolution silently degrade
+ * after the next restart.
+ *
+ * Runs after the DB cascade commits — see {@link ConversationDeletedEvent}.
+ *
+ * 会话被删除后清理内存缓存，避免留下指向已删除行的幽灵条目。
+ */
+ @EventListener
+ public void onConversationDeleted(ConversationDeletedEvent event) {
+ if (cache.remove(event.conversationId()) != null) {
+ log.info("[ChannelSession] Evicted cached session for deleted conversation {}",
+ event.conversationId());
+ }
+ }
+
/**
* 保存或更新会话标识（收到用户消息时调用）
*
@@ -78,40 +103,51 @@ public class ChannelSessionStore {
existing.setSenderName(senderName);
existing.setChannelId(channelId);
existing.setLastActiveTime(now);
- sessionMapper.updateById(existing);
- log.debug("Updated channel session: conversationId={}, targetId={}", conversationId, targetId);
- } else {
- // 先查 DB（可能是上次启动后的新记录）
- ChannelSessionEntity dbEntity = sessionMapper.selectOne(
- new LambdaQueryWrapper Use this rather than the mapper: this class owns the cache, so a
+ * caller that deletes the row directly leaves a phantom entry behind —
+ * every later {@code saveOrUpdate} then updates a primary key that no
+ * longer exists and the session is never re-created.
+ *
+ * The conversation-delete cascade does not come through here: it removes
+ * the row inside its own transaction and lets
+ * {@link #onConversationDeleted} drop the cache after commit, so the cache
+ * is never cleared for a delete that later rolls back.
+ *
+ * 删除会话（内存 + DB 双层）。
+ *
+ * @return number of DB rows removed
*/
- public void remove(String conversationId) {
- ChannelSessionEntity removed = cache.remove(conversationId);
- if (removed != null) {
- sessionMapper.deleteById(removed.getId());
+ public int remove(String conversationId) {
+ cache.remove(conversationId);
+ int deleted = sessionMapper.delete(new LambdaQueryWrapper One shared implementation for every channel. Before this existed, four
+ * adapters carried four hand-rolled variants (a 500-entry LRU, an unbounded
+ * set halved on overflow, an access-ordered map) and four adapters carried
+ * none at all — DingTalk among them, which is why a redelivered DingTalk
+ * message produced a second full answer.
+ *
+ * A message is identified by {@code channelId + identity}, where identity is
+ * the platform message id (see
+ * {@link ChannelMessageRouter#inboundIdentity(ChannelMessage)}). Scoping by
+ * channel keeps two channels of the same type from colliding on a platform id
+ * that is only unique per app.
+ *
+ * Three operations, matching the three things a caller needs:
+ * Fail-open by design: a blank identity means "this platform gave us
+ * nothing stable to dedup on", and the message is always let through. Dropping
+ * a real message is worse than answering a redelivery twice.
+ *
+ * 入站消息去重登记表（TTL + 容量双约束），全渠道共用一份实现。
+ */
+@Slf4j
+@Component
+@EnableConfigurationProperties(ChannelDedupProperties.class)
+public class InboundMessageDeduplicator {
+
+ private final ChannelDedupProperties props;
+
+ /**
+ * Claimed identity -> claim timestamp (epoch millis). Insertion-ordered so
+ * the eldest entries sit at the head and overflow trimming is a head scan.
+ * Guarded by its own monitor — claims are short, contended only by the
+ * channel intake threads.
+ */
+ private final LinkedHashMap
+ * {@code segmentOnly} 的 delta 携带的是每轮 ReAct 的旁白（"我来查一下…"），
+ * 共享累加器刻意不把它写进 {@code mate_message.content}。适配器如果直接
+ * 累加 {@code delta.content()}，就会把每轮旁白拼进外发文本 —— 而旁白通常
+ * 是对答案的复述，用户就会把同一段内容读到两三遍。被污染的文本还会回写
+ * 持久化并在下一轮作为历史重放，重复量随轮次增长，而不是稳定在 2 倍。
+ *
+ * 旁白要不要露出，由渠道的 {@code stream_progress} 开关决定：想露出就作为
+ * 独立的进度消息下发，而不是混进最终答案。
+ *
+ * @param delta 流式片段
+ * @return true 表示该片段的文本应计入最终回复
+ */
+ static boolean contributesToFinalContent(StreamDelta delta) {
+ return delta != null
+ && !delta.isEvent()
+ && !delta.segmentOnly()
+ && delta.content() != null;
+ }
}
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/dingtalk/DingTalkChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/dingtalk/DingTalkChannelAdapter.java
index 8144479e..1cd177b2 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/dingtalk/DingTalkChannelAdapter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/dingtalk/DingTalkChannelAdapter.java
@@ -27,6 +27,8 @@ import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
/**
* 钉钉渠道适配器
@@ -62,6 +64,12 @@ public class DingTalkChannelAdapter extends AbstractChannelAdapter implements St
/** AI Card 管理器（message_type=card 时初始化） */
private DingTalkAICardManager aiCardManager;
+ /**
+ * Off-callback worker for inbound parsing, so the Stream frame is acked
+ * immediately. See {@link #dispatchInbound}.
+ */
+ private volatile ExecutorService inboundExecutor;
+
/** 钉钉媒体上传器（doStart 时初始化） */
private DingTalkMediaUploader mediaUploader;
@@ -118,6 +126,11 @@ public class DingTalkChannelAdapter extends AbstractChannelAdapter implements St
// 启动 Stream 模式或 Webhook 模式
if (isStreamMode()) {
+ this.inboundExecutor = Executors.newSingleThreadExecutor(r -> {
+ Thread t = new Thread(r, "dingtalk-inbound-" + channelEntity.getId());
+ t.setDaemon(true);
+ return t;
+ });
startStreamMode(clientId, clientSecret);
} else {
log.info("[dingtalk] Webhook mode: waiting for callbacks at /api/v1/channels/webhook/dingtalk");
@@ -234,12 +247,44 @@ public class DingTalkChannelAdapter extends AbstractChannelAdapter implements St
return;
}
- handleWebhook(payload);
+ dispatchInbound(payload);
} catch (Exception e) {
log.error("[dingtalk-stream] Failed to parse stream message: {}", e.getMessage(), e);
}
}
+ /**
+ * Hand the parsed payload to a worker and return, so the SDK can ack the
+ * Stream frame immediately.
+ *
+ * {@link #handleWebhook} resolves media inline — each attachment costs a
+ * download-URL call plus a byte fetch against DingTalk. Running that on the
+ * callback thread delays the ack by however long the downloads take, and a
+ * late ack makes DingTalk redeliver the message: the user gets the same
+ * answer once per redelivery. Acking first removes the cause; the router's
+ * inbound claim is the second line of defence for redeliveries we can't
+ * prevent.
+ *
+ * Single-threaded on purpose — the agent turn itself already runs on the
+ * router's queue, so this thread only parses, and keeping it serial
+ * preserves the arrival order of a sender's messages.
+ */
+ private void dispatchInbound(Map 飞书 SDK 投递的 mention 里，bot 的标识可能是群内自定义别名（{@code ou_357e...} / 自定义名称），
@@ -460,7 +457,6 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
this.botName = null;
this.botOpenIdLastFailureMs = 0L;
}
- this.processedMessageIds.clear();
this.chatBotAliases.clear();
this.mentionTracker.clear();
this.nicknameCache.clear();
@@ -1212,7 +1208,8 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
if (isGroup && chatId != null) {
shortSuffix = resolveGroupSessionSuffix(chatId);
}
- String conversationId = buildConversationId(shortSuffix, senderOpenId, isGroup);
+ String conversationId = buildConversationId(shortSuffix, senderOpenId, isGroup,
+ channelEntity != null ? channelEntity.getId() : null);
String stagedUploadPath = null;
if (isFileMessage) {
@@ -1232,12 +1229,14 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
log.warn("[feishu] require_mention=true but bot open_id unavailable; allowing messageId={}", messageId);
}
- // 消息去重
- if (messageId != null && !processedMessageIds.add(messageId)) {
+ // Early duplicate gate. The authoritative claim happens once, in
+ // ChannelMessageRouter.enqueue; this peek only spares a redelivery the
+ // side effects below (the "received" reaction, media downloads) that
+ // would otherwise fire again before the router ever sees the message.
+ if (messageRouter.isDuplicateInbound(channelEntity.getId(), messageId)) {
log.debug("[feishu] Duplicate message_id: {}, skipping", messageId);
return;
}
- cleanupProcessedIds();
// 添加消息反应（非阻塞，表示"已收到"）
if (messageId != null && getConfigBoolean("enable_reaction", true)) {
@@ -1298,21 +1297,6 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
onMessage(channelMessage);
}
- /**
- * 清理旧的去重记录：超过 1000 条时保留最近添加的（移除最早的一半）
- */
- private void cleanupProcessedIds() {
- if (processedMessageIds.size() > 1000) {
- int toRemove = processedMessageIds.size() / 2;
- var iterator = processedMessageIds.iterator();
- while (iterator.hasNext() && toRemove > 0) {
- iterator.next();
- iterator.remove();
- toRemove--;
- }
- }
- }
-
// ==================== 消息反应 ====================
/**
@@ -1731,14 +1715,22 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
* {@code senderId} is the full open id. Mirror that exactly:
* {@code groups → feishu:{shortSuffix}}, {@code DMs → feishu:{senderOpenId}}.
*/
- static String buildConversationId(String shortSuffix, String senderOpenId, boolean isGroup) {
+ static String buildConversationId(String shortSuffix, String senderOpenId, boolean isGroup,
+ Long channelId) {
// The routed ChannelMessage carries chatId = (isGroup ? shortSuffix : null);
// the router then falls back to senderId when that chatId is null. Mirror both
// steps so the storage id matches the runtime id in every case (including the
// degenerate group-with-no-suffix path).
String routedChatId = isGroup ? shortSuffix : null;
String identifier = routedChatId != null ? routedChatId : senderOpenId;
- return identifier != null ? CHANNEL_TYPE + ":" + identifier : null;
+ if (identifier == null) {
+ return null;
+ }
+ // Mirror ChannelMessageRouter#buildConversationId: scope the id by channelId so
+ // the same sender on two workspaces' feishu channels never shares a conversation.
+ return channelId != null
+ ? CHANNEL_TYPE + ":" + channelId + ":" + identifier
+ : CHANNEL_TYPE + ":" + identifier;
}
// ==================== Per-chat recent file cache ====================
@@ -2613,7 +2605,10 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
StringBuilder accumulator = new StringBuilder();
try {
stream.doOnNext(delta -> {
- if (delta.content() != null) {
+ // segmentOnly narration is skipped: appending every
+ // ReAct iteration's "我来查一下…" into the card text is
+ // what makes the answer read as if it were sent twice.
+ if (StreamingChannelAdapter.contributesToFinalContent(delta)) {
accumulator.append(delta.content());
streamingCardManager.appendContent(sessionKey, delta.content(), false);
}
@@ -2626,8 +2621,12 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
.blockLast(Duration.ofMinutes(5));
String finalContent = accumulator.toString();
- if (finalContent.isBlank()) {
- finalContent = "（无回复内容）";
+ // Card streaming never touches renderAndSend, so the channel's
+ // message-filter config has to be applied here — otherwise
+ // filter_thinking / filter_tool_messages are inert on this path.
+ String cardContent = filterOutboundContent(finalContent);
+ if (cardContent.isBlank()) {
+ cardContent = "（无回复内容）";
}
// Strip any /api/v1/files/generated/{id} URLs out of the card
// text (replacing each with a "📎 filename" marker) AND send
@@ -2636,11 +2635,11 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
// the user sees a broken-looking download link instead of the
// actual file. Cache-miss URLs fall back to the user-facing
// retry hint that GeneratedFileScrubber emits.
- String renderedContent = scrubAndSendAttachments(receiveId, finalContent);
+ String renderedContent = scrubAndSendAttachments(receiveId, cardContent);
streamingCardManager.finishCard(sessionKey, renderedContent);
log.info("[feishu-stream] Card streaming completed: sessionKey={}, contentLen={}",
sessionKey, renderedContent.length());
- return finalContent;
+ return finalContent.isBlank() ? cardContent : finalContent;
} catch (Exception e) {
log.error("[feishu-stream] Card streaming failed: sessionKey={}, err={}",
@@ -2671,13 +2670,17 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter implements Stre
private String processStreamAsText(Flux Caller passes the resulting Map to a {@code CallBackCard}
* with {@code type="raw"} (NOT {@code card_json}).
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/AgentStreamAccumulator.java b/mateclaw-server/src/main/java/vip/mate/channel/web/AgentStreamAccumulator.java
new file mode 100644
index 00000000..4cc6ba2d
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/channel/web/AgentStreamAccumulator.java
@@ -0,0 +1,530 @@
+package vip.mate.channel.web;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import vip.mate.agent.AgentService;
+import vip.mate.agent.GraphEventPublisher;
+import vip.mate.workspace.conversation.model.MessageContentPart;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * 流式累积器 — 收集 StreamDelta 事件，持久化到 DB。
+ *
+ * 维护两份数据：
+ *
+ * Shared by the Web SSE path ({@code ChatController}) and the IM channel
+ * router — live fan-out side effects go through the injected {@link Sink}
+ * so each caller keeps its own broadcast semantics. Internal bookkeeping
+ * events ({@code _usage_final}, {@code _routing_decision}) are consumed
+ * here and never reach the sink.
+ */
+@Slf4j
+public final class AgentStreamAccumulator {
+
+ /**
+ * Live fan-out hooks. The accumulator itself only builds the persisted
+ * metadata/parts; anything a subscriber should see in real time is
+ * delegated here.
+ */
+ public interface Sink {
+ /** Broadcast a named event to live subscribers of the conversation. */
+ void broadcast(String conversationId, String eventName, Object payload);
+
+ /** Update the conversation's current phase indicator. */
+ void updatePhase(String conversationId, String phase);
+ }
+
+ /** Markdown link pointing at a generated-file download URL. Used to
+ * surface generated artifacts in the run-overview rail. */
+ private static final Pattern GENERATED_FILE_LINK_PATTERN =
+ Pattern.compile("\\[([^\\]]+)\\]\\(((?:https?://[^/\\s)\\]]+)?/api/v1/files/generated/[A-Za-z0-9-]+)\\)");
+
+ private final ObjectMapper objectMapper;
+ private final Sink sink;
+
+ private final StringBuilder content = new StringBuilder();
+ private final StringBuilder thinking = new StringBuilder();
+ private final List
- * 维护两份数据：
- *
- * 语义:找到会话最后一条 {@code role=user} 消息 → stop 当前流(如有)→ 删除最后一条
- * {@code role=assistant} 消息 → 用 last user message 重新启动 agent turn。
- * 实际启动复用 {@link #chatStream},它会重新 saveMessage user(新消息 id,内容相同)。
- * 这样不重复 100 行 SSE 代码,代价是用户消息多一条(语义上等同"重发")。
+ * Semantics: stop any in-flight stream, rewind the conversation to the last
+ * {@code role=user} message (removing the trailing assistant reply), then
+ * re-run the agent turn from that message. The restart reuses
+ * {@link #chatStream} with {@code internalSkipUserPersist} set, so the
+ * existing user row is used as the seed and no duplicate user message is
+ * inserted.
*
- * 没有任何 user 消息时返回 400(无内容可重新生成)。
+ * Returns an error when the conversation has no user message to regenerate from.
*/
@Operation(summary = "重新生成最后一条助手回复")
@PostMapping(value = "/sessions/regenerate", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
@@ -1425,29 +1432,27 @@ public class WebChatController {
// right disposable; multi-node is a separate epic.
streamTracker.requestStop(conversationId);
- MessageEntity lastAssistant = conversationService.findLastMessageByRole(conversationId, "assistant");
- if (lastAssistant != null) {
- conversationService.deleteMessageById(lastAssistant.getId());
- }
- MessageEntity lastUser = conversationService.findLastMessageByRole(conversationId, "user");
- if (lastUser == null) {
+ ConversationService.RegenerateSeed seed = conversationService.prepareRegenerate(conversationId);
+ if (seed == null) {
sendErrorAndComplete(emitter, "No user message to regenerate from");
return emitter;
}
log.info("[WebChat] Regenerate: conversationId={}, visitor={}, seedMessageId={}",
- conversationId, visitorId, lastUser.getId());
+ conversationId, visitorId, seed.seedMessageId());
audit(channel, visitorId, "webchat.regenerate-session", conversationId,
- "{\"sessionId\":\"" + sid + "\",\"seedMessageId\":" + lastUser.getId() + "}");
+ "{\"sessionId\":\"" + sid + "\",\"seedMessageId\":" + seed.seedMessageId() + "}");
// Reuse chatStream: it'll resolve the agent again (cheap), re-derive
- // conversationId, saveMessage user (new id, same content), and start
- // the agent turn. visitorId echoes through to keep the visitor-scoped
- // memory owner consistent.
+ // conversationId and start the agent turn. The seed user row is reused
+ // as-is — internalSkipUserPersist stops chatStream from inserting a
+ // duplicate user row. visitorId echoes through to keep the
+ // visitor-scoped memory owner consistent.
WebChatRequest req = new WebChatRequest();
- req.setMessage(lastUser.getContent());
+ req.setMessage(seed.content());
req.setVisitorId(visitorId);
req.setSessionId(sid);
+ req.setInternalSkipUserPersist(true);
return chatStream(apiKey, req);
}
@@ -1656,7 +1661,7 @@ public class WebChatController {
return full;
}
return "webchat:" + key8 + ":#"
- + sha256Hex(visitorId + " " + (sessionId == null ? "" : sessionId)).substring(0, 40);
+ + sha256Hex(visitorId + "\0" + (sessionId == null ? "" : sessionId)).substring(0, 40);
}
/**
@@ -1917,6 +1922,13 @@ public class WebChatController {
* for this conversation. Metadata is resolved server-side; unknown / foreign / expired
* ids are dropped. */
private List
+ * 渲染策略：复用入站时发出的 "🤔 思考中..." reply_stream 气泡，随
+ * agent 事件（思考、工具调用、计划步骤、内容产出）持续覆写为实时进度，
+ * 流结束后原地渐变为最终答案（走 renderAndSend 的分段逻辑）。
+ *
+ * The narration goes through the channel renderer (thinking/tool-tag
+ * filters, table formatting, length split): the first segment overwrites
+ * the current bubble with {@code finish=true}, overflow segments ride
+ * plain messages. The replacement context is registered in
+ * {@link #replyContexts} so {@code renderAndSend} / approval cards keep
+ * working against the newest bubble, and keepalive restarts on the new
+ * stream with the live progress snapshot.
+ *
+ * @param progress live progress renderer, or {@code null} when the stream
+ * has already finished (the replacement bubble is then a
+ * bare slot for {@code renderAndSend}, with no keepalive)
+ */
+ private WeComReplyContext rollProgressBubble(String replyTarget, WeComReplyContext ctx,
+ String narration, WeComProgressRenderer progress) {
+ boolean filterThinking = getConfigBoolean("filter_thinking", true);
+ boolean filterToolMessages = getConfigBoolean("filter_tool_messages", true);
+ String format = getConfigString("message_format", "auto");
+ int maxLen = vip.mate.channel.ChannelMessageRenderer.PLATFORM_LIMITS
+ .getOrDefault(getChannelType(), 2048);
+ List Used for reply segments after the first when a live
+ * {@link WeComReplyContext} exists: the platform rejects
+ * {@code aibot_send_msg} in group chats, so segments pushed actively
+ * would silently vanish there. Riding the inbound frame's reply slot
+ * works in both group and single chats, and keeps segment ordering
+ * behind the stream bubble (same per-reqId serial worker queue).
+ *
+ * Failures are logged, not propagated — one bad segment must not
+ * abort the remaining segments of a long reply.
+ */
+ private void replyMarkdown(String frameReqId, String content) {
+ if (frameReqId == null || frameReqId.isBlank()
+ || content == null || content.isBlank()) {
+ return;
+ }
+ try {
+ Map
- * AES-256-CBC 解密：base64 decode aesKey → IV = 前 16 字节 → PKCS#7 去填充
- *
- * @param url 文件下载 URL
- * @param aesKey Base64 编码的 AES-256 密钥
- * @param msgId 消息 ID（用于生成文件名）
- * @param fileNameHint 文件名提示
- * @return 本地文件路径，失败返回 null
- */
- private String downloadAndDecryptMedia(String url, String aesKey, String msgId, String fileNameHint) {
- try {
- String mediaDir = getConfigString("media_dir", "data/media");
- Path mediaDirPath = Path.of(mediaDir);
- Files.createDirectories(mediaDirPath);
-
- // 1. HTTP GET 下载文件
- HttpRequest request = HttpRequest.newBuilder()
- .uri(URI.create(url))
- .timeout(Duration.ofSeconds(30))
- .GET()
- .build();
-
- HttpResponse
- * 1. Base64 decode aesKey（自动补齐 padding）
- * 2. IV = decoded key 前 16 字节
- * 3. AES-256-CBC 解密
- * 4. PKCS#7 去填充
+ * 1. Base64 decode aesKey (auto-fix missing padding)
+ * 2. IV = first 16 bytes of the decoded key
+ * 3. AES-256-CBC decrypt
+ * 4. Strip PKCS#7 padding
*/
private byte[] decryptAes256Cbc(byte[] encryptedData, String aesKeyBase64) throws Exception {
// 补齐 Base64 padding
@@ -3131,19 +3512,6 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
return prefix + "_" + System.currentTimeMillis() + "_" + reqIdCounter.incrementAndGet();
}
- /**
- * MD5 哈希（hex 字符串）
- */
- private String md5Hex(String input) {
- try {
- MessageDigest md = MessageDigest.getInstance("MD5");
- byte[] hash = md.digest(input.getBytes());
- return bytesToHex(hash);
- } catch (Exception e) {
- return Integer.toHexString(input.hashCode());
- }
- }
-
/**
* MD5 哈希（字节数组输入）
*/
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComKeepaliveScheduler.java b/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComKeepaliveScheduler.java
index b53e6db6..7724d836 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComKeepaliveScheduler.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComKeepaliveScheduler.java
@@ -9,6 +9,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
/**
* Periodically refreshes a WeCom AI Bot {@code stream} reply with the
@@ -47,9 +48,17 @@ public class WeComKeepaliveScheduler {
/** Hard ceiling — after this many seconds, force-finish the stream. */
static final long MAX_DURATION_SECONDS = 180;
- /** Placeholder text written on every refresh tick + on force-finish. */
+ /** Placeholder text written on every refresh tick (when no live progress supplier is attached). */
static final String PROCESSING_TEXT = "🤔 思考中...";
+ /**
+ * Text written when the 180s ceiling force-finishes the stream. The real
+ * answer will arrive later as a separate pushed bubble (the reply context
+ * is invalidated below), so the sealed bubble must tell the user the
+ * reply is still coming — freezing it on "思考中..." reads as a hang.
+ */
+ static final String FORCE_FINISH_TEXT = "⏳ 任务耗时较长，仍在处理中，结果稍后送达";
+
/** One-shot state per active stream. Held by reference inside the scheduled task. */
private static final class StreamState {
final WeComChannelAdapter adapter;
@@ -58,6 +67,10 @@ public class WeComKeepaliveScheduler {
final String replyToken;
final long startedAt;
volatile ScheduledFuture future;
+ /** Optional live progress text source; when set, refresh ticks write
+ * its current snapshot instead of the static placeholder so the
+ * bubble keeps showing elapsed time / tool state between events. */
+ volatile Supplier
+ * Mutations arrive from the stream-consuming thread; {@link #snapshot()} may
+ * also be called from the keepalive scheduler thread, so all state access is
+ * synchronized on this instance.
+ */
+final class WeComProgressRenderer {
+
+ /** Max completed/running tool lines rendered before collapsing to a counter. */
+ private static final int MAX_TOOL_LINES = 3;
+
+ /** Rolling window (chars) of reasoning text when thinking display is on. */
+ private static final int THINKING_WINDOW = 500;
+
+ /** Tail window (chars) of the streamed answer kept inside the bubble.
+ * The full answer is delivered by the final render pass; the bubble only
+ * needs enough to show live progress while staying under the 2048 limit. */
+ private static final int ANSWER_WINDOW = 1200;
+
+ private record ToolLine(String callId, String name, long startedAt,
+ Long finishedAt, boolean success) {
+ }
+
+ private final long startedAtMillis;
+ private final boolean showThinking;
+ private final boolean showToolTrace;
+
+ private final Deque
+ *
+ *
+ *
+ *
+ *
+ * @author MateClaw Team
+ */
+public class Mem0Plugin implements MateClawPlugin {
+
+ private static final String CONFIG_BASE_URL = "baseUrl";
+ private static final String CONFIG_API_KEY = "apiKey";
+ private static final String CONFIG_SEARCH_ENABLED = "searchEnabled";
+ private static final String CONFIG_SYNC_ENABLED = "syncEnabled";
+ private static final String CONFIG_MAX_RESULTS = "maxResults";
+ private static final String CONFIG_TIMEOUT_MS = "timeoutMs";
+
+ private Logger log;
+
+ @Override
+ public void onLoad(PluginContext context) {
+ this.log = context.getLogger();
+
+ Mem0Config config = readConfig(context);
+ if (!config.isUsable()) {
+ log.warn("Mem0 plugin loaded without baseUrl — provider will stay unavailable. "
+ + "Configure 'baseUrl' in the plugin config to enable.");
+ }
+
+ Mem0Client client = new Mem0Client(config);
+ Mem0Provider provider = new Mem0Provider(config, client, log);
+ context.registerMemoryProvider(provider);
+
+ log.info("Mem0 plugin loaded: baseUrl={}, searchEnabled={}, syncEnabled={}, maxResults={}, timeoutMs={}",
+ maskUrl(config.baseUrl()), config.searchEnabled(), config.syncEnabled(),
+ config.maxResults(), config.timeoutMs());
+ }
+
+ @Override
+ public void onEnable() {
+ if (log != null) log.info("Mem0 plugin enabled");
+ }
+
+ @Override
+ public void onDisable() {
+ if (log != null) log.info("Mem0 plugin disabled");
+ }
+
+ private Mem0Config readConfig(PluginContext ctx) {
+ String baseUrl = ctx.getConfig(CONFIG_BASE_URL, String.class);
+ String apiKey = ctx.getConfig(CONFIG_API_KEY, String.class);
+ Boolean searchEnabled = ctx.getConfig(CONFIG_SEARCH_ENABLED, Boolean.class);
+ Boolean syncEnabled = ctx.getConfig(CONFIG_SYNC_ENABLED, Boolean.class);
+ Integer maxResults = ctx.getConfig(CONFIG_MAX_RESULTS, Integer.class);
+ Integer timeoutMs = ctx.getConfig(CONFIG_TIMEOUT_MS, Integer.class);
+
+ return new Mem0Config(
+ baseUrl,
+ apiKey,
+ searchEnabled == null ? true : searchEnabled,
+ syncEnabled == null ? true : syncEnabled,
+ maxResults == null ? Mem0Config.DEFAULT_MAX_RESULTS : maxResults,
+ timeoutMs == null ? Mem0Config.DEFAULT_TIMEOUT_MS : timeoutMs
+ );
+ }
+
+ /**
+ * Mask credentials in the URL when logging. Keeps the scheme + host,
+ * strips any user info and path.
+ */
+ private static String maskUrl(String url) {
+ if (url == null || url.isBlank()) return "(unset)";
+ try {
+ URI u = URI.create(url);
+ String host = u.getHost();
+ int port = u.getPort();
+ return u.getScheme() + "://" + host + (port > 0 ? ":" + port : "");
+ } catch (Exception e) {
+ return "(malformed)";
+ }
+ }
+}
diff --git a/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Provider.java b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Provider.java
new file mode 100644
index 00000000..fc42efd6
--- /dev/null
+++ b/mateclaw-plugin-mem0/src/main/java/vip/mate/plugin/mem0/Mem0Provider.java
@@ -0,0 +1,177 @@
+package vip.mate.plugin.mem0;
+
+import org.slf4j.Logger;
+import vip.mate.plugin.api.memory.PluginMemoryProvider;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+
+/**
+ * Memory provider that bridges MateClaw's per-turn lifecycle to a self-hosted
+ * Mem0 service.
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
*/
private static final Set
+ * mate:
+ * channel:
+ * dedup:
+ * enabled: true
+ * ttl: 5m
+ * max-size: 2000
+ *
+ */
+@ConfigurationProperties(prefix = "mate.channel.dedup")
+public class ChannelDedupProperties {
+
+ /**
+ * Master switch. When false every message is treated as new — only useful
+ * when debugging a suspected false-positive drop.
+ */
+ private boolean enabled = true;
+
+ /**
+ * How long a claimed message identity keeps suppressing redeliveries.
+ *
+ *
+ *
+ */
+final class ChannelMagicCommand {
+
+ /** Platform-level command kinds, dispatched by {@link ChannelMessageRouter}. */
+ enum Type { CLEAR, NEW, HELP, STATUS, STOP, MODEL }
+
+ /** A recognized command plus its raw (possibly empty) argument string. */
+ record Parsed(Type type, String args) {
+ }
+
+ /**
+ * Alias token → command type. LinkedHashMap keeps registration ordering
+ * stable. Every bare alias also registers its "/"-prefixed twin.
+ */
+ private static final Map
+ *
+ *
+ *
+ *
+ * 两份数据从同一事件流构建，保证一致。segments 保留了 thinking → tools → content
+ * 的真实交错顺序，toolCalls 是 segments 中 tool_call 类型的平铺视图。
+ *
- *
- * 两份数据从同一事件流构建，保证一致。segments 保留了 thinking → tools → content
- * 的真实交错顺序，toolCalls 是 segments 中 tool_call 类型的平铺视图。
+ * Build a per-stream accumulator wired to this controller's SSE
+ * broadcast and phase tracking. Kept as a factory so every stream gets
+ * its own instance while the fan-out semantics stay in one place.
*/
- private final class StreamAccumulator {
- private final StringBuilder content = new StringBuilder();
- private final StringBuilder thinking = new StringBuilder();
- private final List
+ *
+ */
+ @Override
+ public String processStream(Flux