@@ -1584,6 +1611,7 @@ public class AgentGraphBuilder {
private RestClient.Builder applyHttpTimeouts(RestClient.Builder builder, Integer readTimeoutOverride) {
HttpClient httpClient = HttpClient.newBuilder()
.connectTimeout(vip.mate.llm.chatmodel.HttpTimeouts.CONNECT_TIMEOUT)
+ .version(HttpClient.Version.HTTP_1_1)
.build();
JdkClientHttpRequestFactory rf = new JdkClientHttpRequestFactory(httpClient);
rf.setReadTimeout(vip.mate.llm.chatmodel.HttpTimeouts.resolveReadTimeout(readTimeoutOverride));
@@ -1614,8 +1642,13 @@ public class AgentGraphBuilder {
* {@link #applyHttpTimeouts(RestClient.Builder, Integer)}.
*/
private WebClient.Builder applyHttpTimeoutsToWebClient(WebClient.Builder builder, Integer readTimeoutOverride) {
+ // Pin HTTP/1.1: many self-hosted OpenAI-compatible servers (vLLM, lmstudio,
+ // llama.cpp, ollama — all uvicorn/ASGI based) only speak HTTP/1.1 over
+ // cleartext and slam the socket on the JDK client's default H2C upgrade
+ // probe, surfacing as "header parser received no bytes" with no body sent.
HttpClient httpClient = HttpClient.newBuilder()
.connectTimeout(vip.mate.llm.chatmodel.HttpTimeouts.CONNECT_TIMEOUT)
+ .version(HttpClient.Version.HTTP_1_1)
.build();
org.springframework.http.client.reactive.JdkClientHttpConnector connector =
new org.springframework.http.client.reactive.JdkClientHttpConnector(httpClient);
diff --git a/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java b/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java
index 235c25e0..c101eb6c 100644
--- a/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java
+++ b/mateclaw-server/src/main/java/vip/mate/agent/AgentService.java
@@ -3,12 +3,15 @@ package vip.mate.agent;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import vip.mate.agent.context.ChatOrigin;
import vip.mate.agent.context.ChatOriginHolder;
+import vip.mate.agent.event.AgentLifecycleEvent;
import vip.mate.agent.model.AgentEntity;
import vip.mate.agent.repository.AgentMapper;
import vip.mate.exception.MateClawException;
@@ -43,6 +46,11 @@ public class AgentService {
private final MemoryLifecycleMediator lifecycleMediator;
private final MemoryProperties memoryProperties;
+ /** Field-injected publisher for agent_lifecycle trigger events; the
+ * trigger module's bridge listens and forwards into ingest. */
+ @Autowired(required = false)
+ private ApplicationEventPublisher events;
+
/** 运行时 Agent 实例缓存（agentId -> BaseAgent） */
private final Map The wire shape is the project-wide R<T> envelope: HTTP status
+ * stays 200 (per the convention in {@code R.fail} and the axios
+ * interceptor in {@code mateclaw-ui/src/api/index.ts}); the 409 lives in
+ * the response body's {@code code} field so the front-end can branch
+ * without breaking on an axios error. Without this pre-check the
+ * duplicate save would surface as an opaque
+ * {@code DataIntegrityViolation} stack trace.
+ *
+ * @param excludeId when non-null, skip this row in the lookup so
+ * {@link #updateAgent} doesn't mistake the row for its
+ * own duplicate.
+ */
+ private void requireUniqueName(AgentEntity agent, Long excludeId) {
+ if (agent.getName() == null || agent.getName().isBlank()) {
+ throw new MateClawException("err.agent.name_required", 400, "Agent 名称不能为空");
+ }
+ Long workspaceId = agent.getWorkspaceId() == null ? 1L : agent.getWorkspaceId();
+ LambdaQueryWrapper Algorithm: forward scan with a {@code seenIssuedIds} set. Leading
+ * {@link SystemMessage}s (boundary rows, system prompts) pass through
+ * untouched but contribute no ids. The first {@link AssistantMessage} or
+ * {@link UserMessage} we hit stops the repair walk — by that point we're
+ * out of head-orphan territory. Every {@link ToolResponseMessage} we
+ * encounter before that stop is checked against {@code seenIssuedIds};
+ * if every response id is unseen, the message is dropped and the scan
+ * re-examines the new head. A response whose ids are all in the seen
+ * set (e.g. {@code [system, assistant(X), toolResponse(X), ...]} when
+ * the assistant fell at index 1 of the slice) is left in place.
+ *
+ * Mixed responses (some ids matched, some not) inside a single
+ * leading {@code ToolResponseMessage} are dropped wholesale rather than
+ * surgically rewritten — the provider would reject partially-broken
+ * sequences anyway, and the mixed case implies an upstream invariant
+ * violation that surfaces in logs.
+ *
+ * Package-private + static so unit tests can drive it without standing
+ * up a full BaseAgent subclass.
+ */
+ static int stripHeadOrphanToolResponses(List Sibling to {@link #EVENT_FINISH_REASON} (which only carries the
+ * machine-readable reason). The two are kept separate so legacy
+ * consumers of {@code finish_reason} don't have to learn a new
+ * payload shape — and so a future graph branch (e.g. evidence-
+ * insufficient → "rerun with the listed files attached") can emit
+ * feedback affordances without abusing the finish_reason channel.
+ */
+ public static final String EVENT_FEEDBACK = "feedback_event";
+
+ /**
+ * Multimodal sidecar routing decision for the current turn. Emitted once
+ * per turn before the graph starts streaming; the channel-side accumulator
+ * stores it under {@code metadata.routing} so the chat UI can show which
+ * sidecar (if any) was invoked. Underscore-prefixed name keeps it out of
+ * IM channel rebroadcast (see {@code ChannelMessageRouter}).
+ */
+ public static final String EVENT_ROUTING_DECISION = "_routing_decision";
+
/**
* 事件记录
*/
@@ -217,6 +242,30 @@ public final class GraphEventPublisher {
), ts);
}
+ /**
+ * Emit a recovery-affordance event for the frontend. {@code errorType}
+ * mirrors the {@code NodeStreamingChatHelper.ErrorType} value (e.g.
+ * {@code AUTH_ERROR}, {@code BILLING}, {@code MODEL_NOT_FOUND}, or
+ * the generic {@code UNKNOWN}); {@code errorMessage} is the
+ * user-friendly text already displayed in the bubble; {@code actions}
+ * is the ordered list of buttons to render. Default offering is the
+ * standard {@code retry / regenerate / report} triad — call sites
+ * can narrow this if a category has limitations (e.g. AUTH_ERROR
+ * shouldn't offer "retry" until the key is fixed).
+ */
+ public static GraphEvent feedback(String errorType, String errorMessage,
+ java.util.List Three skill id flavors to handle:
+ * Most {@code mate_skill} rows currently sit in the default workspace
+ * (id=1) because skill creation doesn't yet honor the
+ * {@code X-Workspace-Id} header; the real-skill branch is therefore
+ * defense-in-depth right now and flips on automatically the moment
+ * workspace-scoped skill creation lands. ACP enforcement is live today.
+ *
+ * @throws MateClawException 404 if the agent or skill doesn't exist;
+ * 403 on a workspace mismatch.
+ */
+ private void requireSameWorkspace(Long agentId, Long skillId) {
+ if (agentId == null) {
+ throw new MateClawException("err.agent.not_found", 404, "Agent ID is required");
+ }
+ if (skillId == null) {
+ throw new MateClawException("err.skill.not_found", 404, "Skill ID is required");
+ }
+ AgentEntity agent = agentMapper.selectById(agentId);
+ if (agent == null) {
+ throw new MateClawException("err.agent.not_found", 404, "Agent 不存在: " + agentId);
+ }
+ // MCP virtual: no workspace on McpServerEntity — globally bindable.
+ if (McpSkillBridge.isVirtualMcpSkillId(skillId)) {
+ return;
+ }
+ SkillEntity skill;
+ if (AcpSkillBridge.isVirtualAcpSkillId(skillId)) {
+ // ACP virtual: synthesize from the bridge so workspace_id flows
+ // through from mate_acp_endpoint. A null reply here means the
+ // backing endpoint was deleted or disabled between picker render
+ // and save — same surface as a deleted real skill.
+ skill = acpSkillBridge.findEntityById(skillId);
+ if (skill == null) {
+ throw new MateClawException("err.skill.not_found", 404,
+ "ACP endpoint backing skill " + skillId + " is gone or disabled");
+ }
+ } else {
+ skill = skillMapper.selectById(skillId);
+ if (skill == null) {
+ throw new MateClawException("err.skill.not_found", 404, "Skill 不存在: " + skillId);
+ }
+ }
+ long agentWs = agent.getWorkspaceId() == null ? 1L : agent.getWorkspaceId();
+ long skillWs = skill.getWorkspaceId() == null ? 1L : skill.getWorkspaceId();
+ if (agentWs != skillWs) {
+ throw new MateClawException("err.skill.cross_workspace_binding", 403,
+ "Skill " + skillId + " (workspace=" + skillWs
+ + ") cannot be bound to Agent " + agentId
+ + " (workspace=" + agentWs + ")");
+ }
+ }
+
// ==================== Tool Bindings ====================
public List Auto-included on every non-null result, in addition to the bound
+ * tools and skill-expanded tools:
+ * Validation rule for each incoming name:
+ * Package-private so unit tests can assert on the marker.
+ */
+ static final String ANCHOR_PREFIX = "[Original goal]\n";
+
// ==================== 序列化截断参数 ====================
private static final int CONTENT_MAX = 6000;
private static final int CONTENT_HEAD = 4000;
private static final int CONTENT_TAIL = 1500;
- private static final int OLD_TOOL_RESULT_SUMMARY_THRESHOLD = 500;
+
+ /**
+ * Minimum body size at which the duplicate-output placeholder is preferred
+ * over keeping the verbatim copy. Below this size the placeholder text
+ * (~80 chars) is comparable to the body itself, so deduplication only
+ * complicates the prompt without saving meaningful tokens. Above this
+ * size the dedup placeholder is a real win.
+ */
+ private static final int DEDUP_MIN_CHARS = 500;
/**
* Tool names whose results must never be compacted into a one-line
@@ -99,6 +118,35 @@ public class ConversationWindowManager {
private final MemoryManager memoryManager;
private final ConversationService conversationService;
+ /**
+ * Optional spill store, injected via setter so unit tests and the two
+ * existing 3-arg constructor callers in tests stay source-compatible.
+ * When {@code null}, prune falls back to "keep originals verbatim" — no
+ * lossy summary rewrite is ever applied. Spring autowires this when
+ * {@link ToolResultStorage} is on the context.
+ */
+ private ToolResultStorage toolResultStorage;
+
+ @org.springframework.beans.factory.annotation.Autowired(required = false)
+ public void setToolResultStorage(ToolResultStorage toolResultStorage) {
+ this.toolResultStorage = toolResultStorage;
+ }
+
+ /**
+ * Optional stream tracker for broadcasting {@code compact_status}
+ * SSE events. Wired via setter so unit tests can leave it {@code null}
+ * without dragging in the channel layer. When present, every
+ * compaction emits start/skipped/summarize/done events so the
+ * frontend can render a boundary card and a status line in real
+ * time.
+ */
+ private vip.mate.channel.web.ChatStreamTracker streamTracker;
+
+ @org.springframework.beans.factory.annotation.Autowired(required = false)
+ public void setStreamTracker(vip.mate.channel.web.ChatStreamTracker streamTracker) {
+ this.streamTracker = streamTracker;
+ }
+
// ==================== 状态 ====================
/** 摘要缓存：key = "conversationId:oldMessageCount" */
@@ -133,7 +181,7 @@ public class ConversationWindowManager {
Integer maxInputTokens, ChatModel chatModel,
String conversationId, Long agentId) {
return fitToWindow(messages, systemPrompt, currentUserMessage,
- maxInputTokens, chatModel, conversationId, agentId, null);
+ maxInputTokens, chatModel, conversationId, agentId, null, null);
}
/**
@@ -149,10 +197,33 @@ public class ConversationWindowManager {
Integer maxInputTokens, ChatModel chatModel,
String conversationId, Long agentId,
java.util.Collection When {@code workspaceBasePath} is {@code null}, spill files land in
+ * the configured base dir, or the JVM tmpdir as last resort (see
+ * {@link ToolResultStorage#resolveBaseDir(String)}). Workspace-aware
+ * callers should always pass the path so historical spill files stay
+ * grouped with the workspace that produced them.
+ */
+ public List Walks forward, collecting every {@code tool_call_id}'s assistant
+ * index and the indices of its matching responses. Whenever an
+ * assistant in the prefix has at least one response in the tail, the
+ * cut moves backward to that assistant — pulling the whole cluster
+ * into the tail. The walk repeats until convergence because moving
+ * the cut can expose pairs that were previously fully in the tail.
+ *
+ * The method preserves pair integrity above any other concern. If
+ * the cut collapses all the way to {@code headEnd}, callers must
+ * interpret the return as "skip compaction this turn" — splitting a
+ * pair would produce HTTP 400 on every OpenAI-compatible provider,
+ * which is a worse failure mode than letting context grow by one turn.
+ *
+ * An orphan {@code ToolResponseMessage} (id matching no
+ * assistant in scope) does not trigger movement; the upstream code
+ * paths should never produce one, and logging at WARN gives us a
+ * breadcrumb if they ever do.
+ *
+ * @return adjusted cut index, or {@code headEnd} when no pair-safe
+ * cut larger than {@code headEnd} can be produced.
+ */
+ // Package-private so unit tests in the same package can drive it directly
+ // without standing up a ChatModel + the rest of the compactMessages pipeline.
+ int enforcePairSafeBoundary(List Sizing rules:
+ * Always returns a {@link UserMessage}. {@code null} when anchoring
+ * is disabled, no real first user exists in the prefix, or the body is
+ * blank.
+ *
+ * Package-private for direct unit testing — the surrounding
+ * {@link #compactMessages} path needs a ChatModel and the whole
+ * structured-summary pipeline, which the anchor logic does not.
+ */
+ Message buildFirstUserAnchor(List The {@link #PRUNE_EXEMPT_TOOLS} set still bypasses everything:
+ * sub-agent delegations are not replayable, so their full transcript
+ * stays in context.
+ *
+ * @param messages full conversation in chronological order
+ * @param conversationId used to scope spill files; {@code null} disables spill
+ * @param workspaceBasePath used to locate the spill directory; {@code null}
+ * falls back through the storage's resolveBaseDir chain
+ */
+ public List Spill-marker responses are left untouched so their on-disk pointer
+ * survives intact across compaction.
*/
- private int softTrimToolResults(List Spill-marker responses are left untouched so the on-disk pointer
+ * survives — a placeholder here would force the model to abandon a
+ * tool output it could otherwise recover via {@code read_file}.
*/
- private int hardClearToolResults(List Spill-marker responses are left untouched so the summary input
+ * still has the on-disk path the model might cite back in its summary.
*/
- private int prePruneForSummary(List {@code AgentService#getOrBuildAgent} also checks the flag, but only on
+ * a cache miss — once the {@code BaseAgent} instance is warm, a flip to
+ * disabled would silently keep serving requests until something else
+ * invalidates the cache. Enforcing here at the controller closes that gap
+ * for every external entry point.
+ */
+ private void verifyAgentEnabled(AgentEntity agent) {
+ if (agent != null && !Boolean.TRUE.equals(agent.getEnabled())) {
+ throw new MateClawException("err.agent.disabled", "Agent 已禁用: " + agent.getName());
+ }
+ }
+
private Long resolveUserId(Authentication auth) {
if (auth == null) {
throw new MateClawException("err.auth.unauthenticated", 401, "Not authenticated");
diff --git a/mateclaw-server/src/main/java/vip/mate/agent/event/AgentLifecycleEvent.java b/mateclaw-server/src/main/java/vip/mate/agent/event/AgentLifecycleEvent.java
new file mode 100644
index 00000000..b7fecd88
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/agent/event/AgentLifecycleEvent.java
@@ -0,0 +1,25 @@
+package vip.mate.agent.event;
+
+/**
+ * Spring application event fired when an agent's lifecycle state changes.
+ * The trigger module subscribes via {@code @EventListener} and forwards
+ * the payload through {@code TriggerEventIngestService} so triggers of
+ * pattern type {@code agent_lifecycle} can fan out to workflows.
+ *
+ * {@code phase} matches the matcher's vocabulary: {@code spawned} for
+ * a fresh create, {@code enabled} / {@code disabled} for a flag flip,
+ * {@code terminated} for a delete. {@code crashed} is reserved for v1
+ * once the agent runtime grows a structured error hook.
+ *
+ * The dedup key downstream is {@code phase + ":" + agentId + ":" +
+ * timestamp}; that's stable across retries of the same operation but
+ * lets the same agent flip enabled/disabled repeatedly without the
+ * trigger pipeline collapsing the events.
+ */
+public record AgentLifecycleEvent(
+ long workspaceId,
+ long agentId,
+ String agentName,
+ String phase,
+ long timestamp
+) {}
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 bf3d37cb..7b3fba58 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
@@ -1,5 +1,6 @@
package vip.mate.agent.graph;
+import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.ai.chat.messages.AssistantMessage;
import org.springframework.ai.chat.messages.Message;
@@ -283,6 +284,35 @@ public class NodeStreamingChatHelper {
*/
private static final int THINKING_ONLY_HARD_CAP_CHARS = 32768;
+ /**
+ * Narrow content-repetition guard — fires when the buffer ends with
+ * the same period-sized chunk repeated {@link
+ * #CONTENT_REPEAT_MAX_OCCURRENCES}+ times in a row. Picked to catch
+ * the specific failure mode where reasoning-mode models (qwen3.6,
+ * deepseek-r1) get into a "Wait, I should X. → 写答案 → Wait, I
+ * should Y. → 写同一份答案 → …" self-arguing loop and emit the same
+ * final-answer paragraph dozens of times until {@code max_tokens}
+ * runs out.
+ *
+ * Tests probe sizes from {@link #CONTENT_REPEAT_MIN_PERIOD} up
+ * to {@link #CONTENT_REPEAT_MAX_PERIOD}; the smallest period that
+ * yields the required consecutive copies trips the guard. 4
+ * verbatim consecutive copies of any 24+ char unit is a near-
+ * impossible coincidence in real text, so false positives are very
+ * rare. Not as exhaustive as the previous {@code RepetitionDetector}
+ * (removed at 42d406ff for being brittle on legitimate long-form
+ * content), just the cheap specific check that catches this loop.
+ */
+ public static final int CONTENT_REPEAT_MIN_PERIOD = 24;
+ public static final int CONTENT_REPEAT_MAX_PERIOD = 240;
+ private static final int CONTENT_REPEAT_MAX_OCCURRENCES = 4;
+ /**
+ * Re-scan every N chars of new content. Smaller = faster reaction,
+ * larger = less CPU. The probe loop is O(period_range × occurrences)
+ * char comparisons per scan — cheap even at 400-char intervals.
+ */
+ private static final int CONTENT_REPEAT_CHECK_INTERVAL = 200;
+
private static final int MAX_RETRIES = 5;
// RATE_LIMIT: fail fast to failover chain — staying on the same
// provider during a rate-limit window wastes time without recovery.
@@ -291,6 +321,8 @@ public class NodeStreamingChatHelper {
private static final long BACKOFF_BASE_MS = 3000;
private static final long BACKOFF_CAP_MS = 60_000;
+ private static final ObjectMapper TOOL_ARG_JSON_MAPPER = new ObjectMapper();
+
/**
* 判断错误是否可重试（基于状态码/异常类型）
*/
@@ -373,11 +405,31 @@ public class NodeStreamingChatHelper {
|| msg.contains("invalid_request_error") || msg.contains("unsupported")) {
return ErrorType.CLIENT_ERROR;
}
- // Server errors
+ // Server errors and transient TLS / socket-level network hiccups.
+ // Without the TLS-specific patterns, a single SSL fatal alert
+ // (e.g. bad_record_mac during long-running streams) falls through to
+ // UNKNOWN — non-retryable — so one transient handshake glitch surfaces
+ // to the user as "LLM 调用失败" with no recovery attempt. These are
+ // network-layer transients that almost always succeed on retry, so
+ // they belong in the same retryable bucket as 5xx/timeouts.
if (msg.contains("500") || msg.contains("502") || msg.contains("503") || msg.contains("504")
|| msg.contains("APITimeoutError") || msg.contains("APIConnectionError")
|| msg.contains("Connection reset") || msg.contains("Connection refused")
- || msg.contains("timeout") || msg.contains("Timeout")) {
+ || msg.contains("timeout") || msg.contains("Timeout")
+ // TLS-layer transients: bad_record_mac (RFC 5246 §7.2.2 fatal
+ // alert 20), aborted handshakes, mid-stream protocol errors.
+ || msg.contains("SSLException") || msg.contains("SSLHandshakeException")
+ || msg.contains("SSLProtocolException") || msg.contains("bad_record_mac")
+ // Socket-level transients: a peer closing the TCP connection
+ // mid-response, or the OS reporting a half-closed pipe.
+ || msg.contains("SocketException") || msg.contains("Broken pipe")
+ || msg.contains("Premature close") || msg.contains("PrematureCloseException")
+ || msg.contains("Connection prematurely closed")
+ || msg.contains("Connection closed prematurely")
+ // Reactor Netty wraps the raw socket cause in WebClientRequestException;
+ // surface that wrapper too so retries fire even when the cause chain
+ // string is "WebClientRequestException ...; nested ... SSLException".
+ || msg.contains("WebClientRequestException")) {
return ErrorType.SERVER_ERROR;
}
return ErrorType.UNKNOWN;
@@ -718,6 +770,16 @@ public class NodeStreamingChatHelper {
// 仅保留 thinking-only 这条体积兜底，处理 volcengine-plan 等 provider
// 在 thinking 通道堆字符不出 content 的死循环（生产 trace c1eefa45）。
AtomicBoolean thinkingOnlyCapTriggered = new AtomicBoolean(false);
+ // Content-repetition guard: trips when the same paragraph-sized
+ // suffix appears CONTENT_REPEAT_MAX_OCCURRENCES+ times in
+ // contentAccum. The outer poll loop disposes the upstream
+ // subscription within 500ms once flipped — same pattern as the
+ // thinking-only cap above.
+ AtomicBoolean contentRepeatCapTriggered = new AtomicBoolean(false);
+ // Last contentAccum length at which we ran the repetition scan.
+ // Throttles the O(n) substring scan so it runs at most once per
+ // CONTENT_REPEAT_CHECK_INTERVAL chars, not on every chunk.
+ AtomicInteger lastContentRepeatCheckLen = new AtomicInteger(0);
// Lifecycle events emitted at most once per call so consumers can
// pivot the UI between "thinking" and "drafting" without inspecting
@@ -760,7 +822,7 @@ public class NodeStreamingChatHelper {
lastAssistantMessage.set(msg);
// thinking-only soft cap 已触发 → 跳过一切处理（等外层 dispose）
- if (thinkingOnlyCapTriggered.get()) {
+ if (thinkingOnlyCapTriggered.get() || contentRepeatCapTriggered.get()) {
return;
}
@@ -847,6 +909,36 @@ public class NodeStreamingChatHelper {
return;
}
+ // 5. Content-repetition guard. Some reasoning-mode models
+ // (qwen3.6, deepseek-r1) get stuck in a "Wait, I should X
+ // → 写答案 → Wait, I should Y → 写同一份答案 → ..." loop
+ // and emit the same final-answer paragraph dozens of times
+ // until max_tokens runs out. Without this, the user sees a
+ // wall of duplicated text and the bot never actually finishes.
+ // Throttled to one scan per CONTENT_REPEAT_CHECK_INTERVAL
+ // chars of new content — the probe loop is cheap but no
+ // need to run on every chunk.
+ int currentLen = contentAccum.length();
+ int floor = CONTENT_REPEAT_MIN_PERIOD * CONTENT_REPEAT_MAX_OCCURRENCES;
+ if (currentLen >= floor
+ && currentLen - lastContentRepeatCheckLen.get() >= CONTENT_REPEAT_CHECK_INTERVAL) {
+ lastContentRepeatCheckLen.set(currentLen);
+ if (hasRepeatingSuffix(contentAccum, CONTENT_REPEAT_MIN_PERIOD,
+ CONTENT_REPEAT_MAX_PERIOD,
+ CONTENT_REPEAT_MAX_OCCURRENCES)) {
+ log.warn("[{}] Content-repetition cap reached " +
+ "({} chars, tail repeated {}+ times) " +
+ "— disposing stream for conversation {}",
+ phase, currentLen, CONTENT_REPEAT_MAX_OCCURRENCES,
+ conversationId);
+ broadcastContentTruncated(conversationId,
+ "content_repetition",
+ currentLen);
+ contentRepeatCapTriggered.set(true);
+ return;
+ }
+ }
+
// 4. 提取 token usage（通常最后一个 chunk 携带完整 usage）
if (chatResponse.getMetadata() != null && chatResponse.getMetadata().getUsage() != null) {
var usage = chatResponse.getMetadata().getUsage();
@@ -884,6 +976,20 @@ public class NodeStreamingChatHelper {
// dispose 后 latch 可能不会 countDown，直接跳出
break;
}
+ if (contentRepeatCapTriggered.get()) {
+ // Same dispose pattern as thinking-only cap. The
+ // accumulated content is preserved (it's the looping
+ // text — at least the user gets the FIRST occurrence
+ // as a partial answer instead of waiting for max_tokens).
+ log.warn("[{}] Stream guard tripped (content_repetition), disposing " +
+ "upstream subscription for conversation {}", phase, conversationId);
+ subscription.dispose();
+ if (broadcast) {
+ broadcastDelta(conversationId, "warning",
+ buildDeltaJson("检测到回答内容反复重复，已自动截断"));
+ }
+ break;
+ }
if (streamTracker.isStopRequested(conversationId)) {
// 用户主动停止 — 也 dispose 上游
subscription.dispose();
@@ -979,21 +1085,26 @@ public class NodeStreamingChatHelper {
conversationId, phase, errorType);
}
- // ===== 成功（检查是否因 thinking-only 软上限被截断） =====
+ // ===== 成功（检查是否因 thinking-only 软上限或内容重复被截断） =====
boolean truncatedByThinkingCap = thinkingOnlyCapTriggered.get();
+ boolean truncatedByContentRepeat = contentRepeatCapTriggered.get();
+ boolean truncated = truncatedByThinkingCap || truncatedByContentRepeat;
if (truncatedByThinkingCap) {
log.warn("[{}] LLM stream disposed: thinking-only soft cap reached for conversation {}",
phase, conversationId);
+ } else if (truncatedByContentRepeat) {
+ log.warn("[{}] LLM stream disposed: content-repetition cap reached for conversation {}",
+ phase, conversationId);
}
// RFC-009: guard against silent empty responses. Some providers return
// HTTP 200 with an empty body under soft-failure conditions (rate-limit
// capacity, context filter, upstream overload). Treat this as a failure
// signal so streamCallInternal can hand off to the fallback chain.
- // Only fire when the thinking-only cap didn't fire (which deliberately
- // produces thinking-only output) and there are no tool calls
+ // Only fire when neither truncation cap fired (those deliberately
+ // produce non-empty output) and there are no tool calls
// (tool-only responses are legitimately empty-text).
- if (!truncatedByThinkingCap
+ if (!truncated
&& contentAccum.length() == 0
&& thinkingAccum.length() == 0
&& toolCallAccumulators.isEmpty()) {
@@ -1001,11 +1112,14 @@ public class NodeStreamingChatHelper {
return buildErrorResultWithType("LLM 返回空响应", conversationId, phase, ErrorType.EMPTY_RESPONSE);
}
+ String truncationReason = truncatedByThinkingCap ? "thinking_only_no_content"
+ : truncatedByContentRepeat ? "content_repetition"
+ : null;
return assembleResult(contentAccum, thinkingAccum, toolCallAccumulators,
promptTokens.get(), completionTokens.get(),
cacheReadTokens.get(), cacheWriteTokens.get(), phase,
- truncatedByThinkingCap,
- truncatedByThinkingCap ? "thinking_only_no_content" : null);
+ truncated,
+ truncationReason);
}
/** 组装 stopped partial 结果（用户主动停止，有已累积内容） */
@@ -1516,6 +1630,99 @@ public class NodeStreamingChatHelper {
}
}
+ /**
+ * Collapse a content buffer's trailing run of verbatim repeats to a
+ * single copy. Used to clean up the persisted final answer after
+ * {@link #hasRepeatingSuffix} fires — the streamed text already
+ * contains the duplicates (SSE chunks can't be unsent), but the
+ * DB-persisted message and the IM channel reply should show ONE
+ * clean copy of the looping unit, not a wall.
+ *
+ * Algorithm: find the smallest period in {@code [minPeriod,
+ * maxPeriod]} where the buffer ends with that unit repeated 2+
+ * times consecutively, then return everything up to (and including)
+ * the FIRST copy of that unit. Conservative — if no period yields
+ * 2+ consecutive matches, returns the buffer unchanged.
+ *
+ * Public for unit-testing alongside {@link #hasRepeatingSuffix}.
+ */
+ public static String dedupTrailingRepeats(String content, int minPeriod, int maxPeriod) {
+ if (content == null || content.isEmpty()) return content;
+ int len = content.length();
+ if (minPeriod <= 0 || maxPeriod < minPeriod) return content;
+ int periodCap = Math.min(maxPeriod, len / 2);
+ for (int p = minPeriod; p <= periodCap; p++) {
+ int unitStart = len - p;
+ // Walk backward as far as the unit keeps matching.
+ int copies = 1;
+ int blockStart = unitStart - p;
+ while (blockStart >= 0
+ && content.regionMatches(blockStart, content, unitStart, p)) {
+ copies++;
+ blockStart -= p;
+ }
+ if (copies >= 2) {
+ // Keep prefix + ONE copy. The first copy starts at
+ // (blockStart + p) since the loop walked back one step
+ // past the last match.
+ int firstCopyStart = blockStart + p;
+ int trimEnd = firstCopyStart + p;
+ return content.substring(0, trimEnd);
+ }
+ }
+ return content;
+ }
+
+ /**
+ * Detect whether {@code accum} ends with the same {@code period}-sized
+ * unit repeated at least {@code minOccurrences} times consecutively,
+ * for some {@code period} in {@code [minPeriod, maxPeriod]}. Returns
+ * true when the model is stuck in a "self-arguing" loop emitting the
+ * same final-answer chunk over and over.
+ *
+ * Algorithm: probe period sizes from small to large. For each
+ * candidate period {@code p}, take the last {@code p} chars as the
+ * unit and check whether the {@code minOccurrences-1} preceding
+ * blocks of length {@code p} are byte-identical. The smallest period
+ * that yields the required consecutive copies trips the guard. We
+ * iterate small→large because tighter periods are more specific:
+ * a 30-char unit repeated 4× is a stronger signal than a 200-char
+ * unit happening to appear once.
+ *
+ * Cost: O(periodRange × occurrences × period) char comparisons.
+ * For default thresholds (~200 × 4 × 100) that's ~80K comparisons
+ * per scan — microseconds against an LLM call. Throttled by the
+ * caller via {@code lastContentRepeatCheckLen} so the scan amortizes.
+ *
+ * Package-private + static for unit-testing the threshold without
+ * spinning up a full {@code StreamResult}.
+ */
+ static boolean hasRepeatingSuffix(CharSequence accum, int minPeriod, int maxPeriod,
+ int minOccurrences) {
+ if (accum == null) return false;
+ int len = accum.length();
+ if (minPeriod <= 0 || minOccurrences <= 1 || maxPeriod < minPeriod) return false;
+ if (len < minPeriod * minOccurrences) return false;
+ String s = accum.toString();
+ int periodCap = Math.min(maxPeriod, len / minOccurrences);
+ for (int p = minPeriod; p <= periodCap; p++) {
+ // Unit = last p chars. Check prior (minOccurrences - 1)
+ // blocks of length p match the unit byte-for-byte.
+ int unitStart = len - p;
+ boolean allMatch = true;
+ for (int k = 2; k <= minOccurrences; k++) {
+ int blockStart = len - k * p;
+ if (blockStart < 0) { allMatch = false; break; }
+ if (!s.regionMatches(blockStart, s, unitStart, p)) {
+ allMatch = false;
+ break;
+ }
+ }
+ if (allMatch) return true;
+ }
+ return false;
+ }
+
/**
* Best-effort character count of the outbound prompt for the
* {@code context_prepared} event. Cheaper than tokenizing and only used
@@ -1637,11 +1844,49 @@ public class NodeStreamingChatHelper {
acc.id,
acc.type != null ? acc.type : "function",
acc.name,
- acc.arguments.toString()));
+ sanitizeToolCallArguments(acc.name, acc.arguments.toString())));
}
return result;
}
+ /**
+ * Ensure {@code function.arguments} is always a well-formed JSON string.
+ *
+ * Some providers (e.g. aliyun-codingplan) reject the entire follow-up
+ * request with HTTP 400 when the assistant message in history carries a
+ * tool call whose {@code arguments} is not parseable JSON. Streaming
+ * accumulation can produce such payloads when:
+ * Two-level budget chain (RFC-008 / RFC-06 D-5):
+ * Per-tool-result handling chain:
* Package-private + static so the spill/truncate decision is unit
+ * testable in isolation from the rest of the executor.
+ *
+ * @param storage spill store; {@code null} skips the spill attempt
+ * @param maxTruncateChars fallback inline hard cap
+ * @param result raw tool output (may be {@code null})
+ * @param toolName used in the spill preview header
+ * @param toolUseId unique within the conversation; becomes the file name
+ * @param conversationId spill files are scoped per conversation; blank/null falls back to "unknown"
+ * @param workspaceBasePath where the spill directory lives when set
+ * @return the SPILL_MARKER preview when spill succeeded, otherwise the
+ * original string (when ≤ threshold) or the inline-truncated string.
+ */
+ static String spillRawOrTruncate(ToolResultStorage storage, int maxTruncateChars,
+ String result, String toolName, String toolUseId,
+ String conversationId, String workspaceBasePath) {
+ if (result == null) return null;
+ if (storage != null) {
+ String safeConv = conversationId != null && !conversationId.isEmpty()
+ ? conversationId : "unknown";
+ String candidate = storage.persistIfOversized(
+ result, toolName, toolUseId, safeConv, workspaceBasePath);
+ if (candidate != null && candidate.startsWith(ToolResultStorage.SPILL_MARKER_PREFIX)) {
+ return candidate;
+ }
+ }
+ return truncateToolResult(result, maxTruncateChars);
+ }
+
/** 尾部错误模式检测 */
private static final java.util.regex.Pattern ERROR_TAIL_PATTERN = java.util.regex.Pattern.compile(
"(?i)\\b(error|exception|traceback|failed|fatal|panic|stack.?trace|errno)\\b");
@@ -443,6 +491,17 @@ public class ToolExecutionExecutor {
}
ToolCallback callback = toolCallbackMap.get(toolName);
if (callback == null) {
+ SkillRedirect redirect = tryAutoRedirectSkillCall(toolName, arguments, safeOrigin);
+ if (redirect != null) {
+ // Auto-redirect succeeds with success=true on the SSE event so the
+ // model treats the SKILL.md content as the answer to a different,
+ // valid question (rather than as another failed call to recover from).
+ events.add(GraphEventPublisher.toolComplete(
+ toolCall.id(), toolName, redirect.response(), true));
+ allResponses.add(new ToolResponseMessage.ToolResponse(
+ toolCall.id(), toolName, redirect.response()));
+ continue;
+ }
String msg = skillAwareNotFoundMessage(toolName);
log.warn("[ToolExecutor] {}", msg);
events.add(GraphEventPublisher.toolComplete(toolCall.id(), toolName, msg, false));
@@ -521,6 +580,18 @@ public class ToolExecutionExecutor {
ToolCallback callback = toolCallbackMap.get(toolName);
if (callback == null) {
+ // Same auto-redirect for pre-approved replays — a stale skill-as-tool
+ // approval shouldn't dead-end the conversation either.
+ ChatOrigin replayOriginForRedirect = ChatOrigin.EMPTY
+ .withConversationId(conversationId)
+ .withWorkspace(null, workspaceBasePath);
+ SkillRedirect redirect = tryAutoRedirectSkillCall(toolName, callArguments, replayOriginForRedirect);
+ if (redirect != null) {
+ events.add(GraphEventPublisher.toolComplete(
+ toolCall.id(), toolName, redirect.response(), true));
+ return new ToolResponseMessage.ToolResponse(
+ toolCall.id(), toolName, redirect.response());
+ }
String msg = skillAwareNotFoundMessage(toolName);
log.warn("[ToolExecutor] Pre-approved {}", msg);
events.add(GraphEventPublisher.toolComplete(toolCall.id(), toolName, msg, false));
@@ -558,16 +629,13 @@ public class ToolExecutionExecutor {
toolCall.id(), toolName, DIRECT_TOOL_PLACEHOLDER);
}
- // RFC-008 Layer 1 first, then Layer 2 — match the non-replay path
- // in executeSingleTool so behavior stays symmetric across approval
- // replays. The caller-supplied conversationId scopes spill files
- // into the same per-conversation directory layout.
- result = truncateToolResult(result, MAX_TOOL_RESULT_CHARS);
- if (resultStorage != null && result != null) {
- String spillConv = conversationId != null && !conversationId.isEmpty() ? conversationId : "unknown";
- result = resultStorage.persistIfOversized(
- result, toolName, toolCall.id(), spillConv, workspaceBasePath);
- }
+ // Raw-first spill, inline truncate as fallback. Symmetric with the
+ // non-replay path in executeSingleTool. The caller-supplied
+ // conversationId scopes spill files into the per-conversation
+ // directory layout. See spillRawOrTruncate javadoc for why the
+ // order matters.
+ result = spillRawOrTruncate(resultStorage, MAX_TOOL_RESULT_CHARS,
+ result, toolName, toolCall.id(), conversationId, workspaceBasePath);
log.info("[ToolExecutor] Pre-approved tool {} returned {} chars{}", toolName, rawLen,
result != null && result.length() < rawLen ? " (now " + result.length() + " after spill/truncate)" : "");
events.add(GraphEventPublisher.toolComplete(toolCall.id(), toolName, result, true));
@@ -783,20 +851,16 @@ public class ToolExecutionExecutor {
}
}
- // RFC-008 Layer 1: hard truncation cap to prevent oversized results
- // from inflating the prompt. Runs FIRST (before spill) so the spill
- // store doesn't need to handle multi-MB writes for run-of-the-mill
- // greps that happen to spit out a long stdout.
- result = truncateToolResult(result, MAX_TOOL_RESULT_CHARS);
- // RFC-008 Layer 2: spill oversized results to disk and replace
- // with preview + path. Falls back to truncation when spilling is
- // disabled or fails. Spill preserves the full output (read_file can
- // retrieve it); the Layer 1 truncation above already capped the
- // inline portion, so this layer mostly catches near-cap residues.
- if (resultStorage != null && result != null) {
- result = resultStorage.persistIfOversized(
- result, toolName, pc.toolCall.id(), pc.conversationId, pc.workspaceBasePath);
- }
+ // Raw-first spill: write the full output to disk and replace
+ // with preview + path so the model can call read_file for the
+ // ground truth. Fall back to inline truncate only when spilling
+ // is disabled, the tool is on the exclusion list, the body is
+ // already under the spill threshold, or the disk write fails.
+ // Truncating before spilling would persist a pre-shortened body
+ // to disk and silently lose data the model could otherwise
+ // recover.
+ result = spillRawOrTruncate(resultStorage, MAX_TOOL_RESULT_CHARS,
+ result, toolName, pc.toolCall.id(), pc.conversationId, pc.workspaceBasePath);
log.info("[ToolExecutor] Tool {} returned {} chars{}", toolName, rawLen,
result != null && result.length() < rawLen ? " (now " + result.length() + " after spill/truncate)" : "");
events.add(GraphEventPublisher.toolComplete(pc.toolCall.id(), toolName, result, true));
@@ -1049,6 +1113,79 @@ public class ToolExecutionExecutor {
return "Tool not found: " + toolName;
}
+ /**
+ * Holder for an auto-redirect outcome: the SKILL.md content (wrapped
+ * with a one-line nudge) that we substitute as the tool response when
+ * the LLM mistakenly calls a skill name as if it were a tool.
+ *
+ * {@code success=true} on the substituted response so the model
+ * doesn't read it as "tool failed, try harder" — semantically we
+ * answered a different question than the one it asked, and we want
+ * the model to follow the redirect rather than thrash.
+ */
+ private record SkillRedirect(String response) {}
+
+ /**
+ * When the LLM calls a skill name as if it were a tool, transparently
+ * fetch its SKILL.md and return that as the tool response. Smaller
+ * models (qwen-turbo et al.) often can't act on a "not a tool — go
+ * read X first" hint; they keep emitting the same wrong call until the
+ * iteration cap. With auto-redirect, the model receives runnable
+ * instructions on the very first attempt and can copy the runSkillScript
+ * shape from SKILL.md verbatim.
+ *
+ * Returns {@code null} if {@code toolName} isn't a registered skill,
+ * if {@code readSkillFile} isn't available in this agent's tool set, or
+ * if the redirect call itself errored — the caller then falls through
+ * to the usual {@code skillAwareNotFoundMessage} hint.
+ */
+ private SkillRedirect tryAutoRedirectSkillCall(String toolName, String originalArgs, ChatOrigin origin) {
+ if (skillRuntimeService == null || toolName == null || toolName.isBlank()) return null;
+ try {
+ boolean isSkill = skillRuntimeService.getActiveSkills().stream()
+ .anyMatch(s -> s.getName() != null && s.getName().equalsIgnoreCase(toolName));
+ if (!isSkill) return null;
+ } catch (Exception e) {
+ log.debug("[ToolExecutor] auto-redirect skill lookup failed: {}", e.getMessage());
+ return null;
+ }
+
+ ToolCallback readSkillFile = toolCallbackMap.get("readSkillFile");
+ if (readSkillFile == null) {
+ log.debug("[ToolExecutor] readSkillFile not bound to this agent — cannot auto-redirect '{}'", toolName);
+ return null;
+ }
+
+ String redirectArgs = "{\"skillName\":\""
+ + jsonStringEscape(toolName)
+ + "\",\"filePath\":\"SKILL.md\"}";
+ String skillMd;
+ try {
+ ToolContext ctx = (origin != null ? origin : ChatOrigin.EMPTY).toToolContext();
+ skillMd = readSkillFile.call(redirectArgs, ctx);
+ } catch (Exception e) {
+ log.warn("[ToolExecutor] Auto-redirect readSkillFile failed for '{}': {}", toolName, e.getMessage());
+ return null;
+ }
+
+ log.info("[ToolExecutor] Auto-redirected skill-as-tool call '{}' → readSkillFile (returned {} chars)",
+ toolName, skillMd != null ? skillMd.length() : 0);
+
+ String safeArgs = originalArgs == null || originalArgs.isBlank() ? "{}" : originalArgs;
+ String response = String.format(
+ "[auto-redirect] You called '%s' as a tool, but it's a Skill (documentation package). "
+ + "Its SKILL.md is loaded below — read the script invocation example, then call "
+ + "`runSkillScript(skillName=\"%s\", scriptPath=\"scripts/ Aligned with {@code ToolExecutionExecutor.MAX_TOOL_RESULT_CHARS}
+ * (8000): the executor now tries to spill the RAW result first; only
+ * when spilling is disabled, the tool is on {@link #excludedTools}, the
+ * body is under this threshold, or the disk write fails, does it fall
+ * back to truncating inline to 8000 chars. Keeping the threshold equal
+ * to the truncate cap yields a single semantic ladder — above the
+ * threshold means "preserved on disk", at-or-below means "stays inline
+ * verbatim".
+ *
+ * If you want to keep more text inline before spilling, raise this
+ * value AND raise the executor's hard cap together; otherwise the
+ * 8000-char fallback truncate would silently shorten anything between
+ * this threshold and 8000 even when spill is disabled, defeating the
+ * intent.
*/
- private int perResultThresholdChars = 16000; // was 4000 — prevents WebSearch spill-to-disk
+ private int perResultThresholdChars = 8000;
/**
* Layer 3 — aggregate cap on combined response size in one tool turn.
@@ -83,6 +97,28 @@ public class ToolResultProperties {
*/
private List Default 0 means time-based cleanup is disabled — spill files
+ * stay on disk until the owning conversation is explicitly deleted (which
+ * fires {@code purgeConversation} via {@code ConversationService}).
+ * This preserves the "recoverable" invariant: a summary or preview that
+ * cites a spill path will keep working for the whole life of the
+ * conversation, no matter how long it sits dormant.
+ * Set to a positive value if disk pressure outweighs recoverability
+ * for your deployment. The scheduled sweep will then delete files whose
+ * mtime falls outside the retention horizon.
+ */
+ private int retentionDays = 0;
+
+ /**
+ * Cron expression for the spill-cleanup task. Defaults to once a day at
+ * 03:00 server-local time so cleanup runs during quiet hours. Set this
+ * to a Spring-recognised value (six-field cron) or change the bean
+ * wiring to disable it entirely.
+ */
+ private String cleanupCron = "0 0 3 * * ?";
+
public boolean isEnabled() { return enabled; }
public void setEnabled(boolean enabled) { this.enabled = enabled; }
@@ -116,6 +152,14 @@ public class ToolResultProperties {
this.excludedTools = excludedTools == null ? List.of() : excludedTools;
}
+ public int getRetentionDays() { return retentionDays; }
+ public void setRetentionDays(int retentionDays) { this.retentionDays = retentionDays; }
+
+ public String getCleanupCron() { return cleanupCron; }
+ public void setCleanupCron(String cleanupCron) {
+ this.cleanupCron = cleanupCron == null ? "" : cleanupCron;
+ }
+
/** O(1) membership test for the exclusion list, used on every tool result. */
public Set The cron expression comes from
+ * {@link ToolResultProperties#getCleanupCron()} (default {@code 0 0 3 * * ?},
+ * i.e. once a day at 03:00 server-local time). The retention horizon comes
+ * from {@link ToolResultProperties#getRetentionDays()}.
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class ToolResultRetentionScheduler {
+
+ private final ToolResultStorage storage;
+ private final ToolResultProperties props;
+
+ /**
+ * Cron-fired hook. Failures are logged at WARN so they show up in
+ * standard log scrapes without aborting the scheduler thread — losing
+ * a single sweep is fine, the next one will catch the same files.
+ */
+ @Scheduled(cron = "${mate.agent.tool-result.cleanup-cron:0 0 3 * * ?}")
+ public void cleanup() {
+ if (props.getRetentionDays() <= 0) {
+ log.debug("[ToolResultRetentionScheduler] retentionDays<=0, skipping sweep");
+ return;
+ }
+ try {
+ int deleted = storage.cleanupExpired();
+ if (deleted > 0) {
+ log.info("[ToolResultRetentionScheduler] sweep deleted {} spill file(s) older than {} days",
+ deleted, props.getRetentionDays());
+ }
+ } catch (Exception e) {
+ log.warn("[ToolResultRetentionScheduler] sweep failed: {}", e.getMessage(), e);
+ }
+ }
+}
diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java
index 558d43bb..88aeb301 100644
--- a/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java
+++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/executor/ToolResultStorage.java
@@ -58,6 +58,15 @@ public class ToolResultStorage {
/** D-6: monotonically increasing spill counter for observability. */
private final java.util.concurrent.atomic.AtomicLong spillCount = new java.util.concurrent.atomic.AtomicLong();
+ /**
+ * Workspace roots observed during this JVM's lifetime. Populated every
+ * time a successful spill resolves a base directory; consulted by the
+ * scheduled retention sweep and by {@link #purgeConversation} so we
+ * don't have to query the database for every workspace path. Cross-JVM
+ * orphans are not covered — that is documented in the cleanup javadoc.
+ */
+ private final java.util.Set Workspaces that never received a spill in this JVM's lifetime are
+ * not covered. Persisting an observed-roots registry across restarts
+ * could fix that, but is intentionally out of scope — the operator-side
+ * remedy is to run a one-off cleanup with {@code storage-base-dir}
+ * pointed at the historical workspace.
+ */
+ public int cleanupExpired() {
+ if (props.getRetentionDays() <= 0) {
+ return 0;
+ }
+ long cutoffEpochMillis = System.currentTimeMillis()
+ - (long) props.getRetentionDays() * 24L * 60L * 60L * 1000L;
+
+ java.util.Set Silently no-ops when nothing matches — a conversation that never
+ * spilled, or one whose workspace root was never observed in this JVM,
+ * is simply left alone. Returns the number of files deleted.
+ */
+ public int purgeConversation(String conversationId) {
+ if (conversationId == null || conversationId.isEmpty()) {
+ return 0;
+ }
+ String safeConv = sanitize(conversationId);
+ java.util.Set Failure contract — read carefully.
+ * Reads the workspace off the just-persisted {@link AgentEntity}
+ * rather than a separate parameter so the lookup and the validator
+ * inside {@code AgentBindingService.requireSameWorkspace} can never
+ * disagree on which workspace they're talking about.
+ */
+ private void applyDefaultSkillBindings(TemplateDTO template, AgentEntity created) {
+ List Failure contract mirrors {@link #applyDefaultSkillBindings}:
+ * picker outage and unknown names are dropped with a warning; an
+ * exception from {@code setToolBindings} itself still propagates and
+ * rolls back the hire.
+ */
+ private void applyDefaultToolBindings(TemplateDTO template, AgentEntity created) {
+ List Returned by {@code GET /api/v1/agents/{id}/capabilities}. Computed on
+ * each request — cheap because everything is cached service-side and we only
+ * read at most three rows. Not persisted on {@code mate_agent}; sidecar
+ * configuration is system-wide and {@code modelCapabilities} is derived from
+ * {@code mate_model_config.modalities}.
+ */
+@Data
+@Builder
+@AllArgsConstructor
+public class AgentCapabilitiesVO {
+ private Long agentId;
+ private String modelName;
+ private String providerId;
+ /** Resolved modality set: any of {@code TEXT / VISION / VIDEO / AUDIO}. */
+ private List
+ * Only {@code ApprovalWorkflowService} should call this.
+ *
+ * @return number of entries removed
+ */
+ int removeAllByConversation(String conversationId) {
+ if (conversationId == null) return 0;
+ int removed = 0;
+ var iter = pendingMap.entrySet().iterator();
+ while (iter.hasNext()) {
+ var entry = iter.next();
+ if (conversationId.equals(entry.getValue().getConversationId())) {
+ iter.remove();
+ removed++;
+ }
+ }
+ return removed;
+ }
+
/**
* INTERNAL — register a {@link PendingApproval} reconstructed from DB during JVM startup.
* Bypasses id generation and pre-existing-entry checks; the snapshot's {@code pendingId}
diff --git a/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java b/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java
index 8a664032..343b5d62 100644
--- a/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java
+++ b/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java
@@ -8,8 +8,11 @@ import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.event.EventListener;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
@@ -17,11 +20,13 @@ import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import vip.mate.agent.context.ChatOrigin;
import vip.mate.agent.context.ChatOriginHolder;
+import vip.mate.approval.event.WorkflowApprovalResolvedEvent;
import vip.mate.approval.model.ToolApprovalEntity;
import vip.mate.approval.repository.ToolApprovalMapper;
import vip.mate.tool.guard.model.GuardEvaluation;
import vip.mate.tool.guard.model.GuardFinding;
import vip.mate.workspace.conversation.ConversationService;
+import vip.mate.workspace.conversation.event.ConversationDeletedEvent;
import java.time.Instant;
import java.time.LocalDateTime;
@@ -50,6 +55,12 @@ public class ApprovalWorkflowService implements ApplicationRunner {
private final ToolApprovalMapper approvalMapper;
private final ObjectMapper objectMapper;
private final ConversationService conversationService;
+ /** Optional — injected only in full Spring context. The workflow
+ * module listens for {@link WorkflowApprovalResolvedEvent}; in tests
+ * that don't wire the workflow runtime this stays null and the
+ * publish is a no-op. */
+ @Autowired(required = false)
+ private ApplicationEventPublisher events;
/**
* GC scheduler — owns the 5-minute clock for the entire approval state machine
@@ -82,6 +93,26 @@ public class ApprovalWorkflowService implements ApplicationRunner {
}
}
+ /**
+ * Drop in-memory approval state for a deleted conversation. The cascade in
+ * {@link ConversationService#deleteConversation} already removed the
+ * {@code mate_tool_approval} rows; this listener clears the parallel
+ * {@code pendingMap} entries so {@code findPendingByConversation} cannot
+ * keep returning a ghost approval that points at a non-existent
+ * conversation row.
+ *
+ * Runs after the DB cascade commits — see
+ * {@link ConversationDeletedEvent}.
+ */
+ @EventListener
+ public void onConversationDeleted(ConversationDeletedEvent event) {
+ int removed = approvalService.removeAllByConversation(event.conversationId());
+ if (removed > 0) {
+ log.info("[ApprovalWorkflow] Dropped {} in-memory pending entries for deleted conversation {}",
+ removed, event.conversationId());
+ }
+ }
+
/**
* Reconstruct in-memory pending approvals from DB at startup, preserving the
* original {@code pendingId} and {@code createdAt} so subsequent resolve / GC
@@ -238,6 +269,78 @@ public class ApprovalWorkflowService implements ApplicationRunner {
toolCallPayload, siblingToolCalls, agentId, null);
}
+ /**
+ * Workflow-scoped approval request — creates a {@code mate_tool_approval}
+ * row keyed to a workflow run + step instead of a conversation, so an
+ * {@code await_approval} step is visible in the same approval inbox the
+ * tool-approval flow uses. Returns the row's auto-generated long id; the
+ * caller (typically {@code AwaitApprovalStepAdapter}) writes that id back
+ * onto {@code mate_workflow_run_pause.external_approval_id} so a future
+ * approval-resolve callback can map "approval X resolved → resume run Y".
+ *
+ * The approval row's {@code conversationId} is set to
+ * {@code "workflow:run:{runId}"} as a synthetic key — that lets the
+ * existing {@link ApprovalService#findPendingByConversation} surface the
+ * workflow approval to operator UIs without needing a parallel query
+ * surface. {@code toolName} is set to {@code "workflow:{kind}"} so the
+ * inbox can group / filter workflow approvals from tool approvals.
+ *
+ * v0 keeps the resume path through {@code WorkflowResumeController}
+ * with the pauseToken; this method does not yet wire a resolve→resume
+ * callback. The approval row's purpose for v0 is operator visibility
+ * and a stable foreign key for the pause record.
+ */
+ public Long requestWorkflowApproval(long workspaceId,
+ long runId,
+ Long stepId,
+ String approvalKind,
+ String approvalMessage,
+ java.util.List Without this bridge, an operator who clicks "approve" in the
+ * approval inbox would only flip the {@code mate_tool_approval} row to
+ * APPROVED — the workflow run would stay paused forever until someone
+ * separately POSTed the pause token to the resume endpoint. That's the
+ * "approval is just a visibility surface, not an actual approval"
+ * trap RFC §3.4 calls out.
+ *
+ * {@code approvalRowId} is the {@code mate_tool_approval.id} long key,
+ * NOT the {@code pendingId} string. Pause rows store the long id in
+ * {@code external_approval_id}, so the listener can find the right
+ * pause with a single equality query.
+ *
+ * {@code decision} mirrors the resolve vocabulary so the listener
+ * can route to the right {@code WorkflowResumer.ResumeOutcome}:
+ * Subclasses that support a native card surface (WeCom
+ * {@code button_interaction}, DingTalk {@code ActionCard}, etc.)
+ * override this method and may call
+ * {@code super.sendApprovalNotice(...)} to fall back to the text
+ * path on render failure / payload-too-large / etc.
+ *
+ * Lives on the abstract class rather than as an interface
+ * default method so the {@code super.x(...)} call from subclasses
+ * resolves cleanly via Java's normal class inheritance — see
+ * RFC-32 §2.0.4 (C-4 fix).
+ */
+ @Override
+ public void sendApprovalNotice(String targetId,
+ vip.mate.channel.notification.ApprovalNotice notice) {
+ sendMessage(targetId,
+ vip.mate.channel.notification.ApprovalNotificationService.staticBuildText(notice));
+ }
+
// ==================== 模板方法（子类实现） ====================
/**
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/AsyncTaskMediaDispatcher.java b/mateclaw-server/src/main/java/vip/mate/channel/AsyncTaskMediaDispatcher.java
new file mode 100644
index 00000000..d0ea75f3
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/channel/AsyncTaskMediaDispatcher.java
@@ -0,0 +1,117 @@
+package vip.mate.channel;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import vip.mate.channel.model.ChannelSessionEntity;
+import vip.mate.workspace.conversation.model.MessageContentPart;
+
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Forward async-task results (image / video / music / 3D) generated by tool
+ * pipelines to the IM channel that originated the conversation.
+ *
+ * Without this dispatcher, completion bytes only land in
+ * {@code mate_message} + a Web SSE broadcast — IM users (WeCom / DingTalk /
+ * Feishu / Telegram / etc.) see nothing arrive in their chat client because
+ * the tool pipeline doesn't know about channel adapters. This dispatcher
+ * closes that loop: look up the conversation's bound channel session, get
+ * the live adapter from {@link ChannelManager}, and call
+ * {@link ChannelAdapter#sendContentParts} so the same bytes ride the
+ * channel-native attachment protocol.
+ *
+ * Web / webchat conversations are intentionally skipped because their SSE
+ * stream already carries the result; double-dispatching would render the
+ * image twice.
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class AsyncTaskMediaDispatcher {
+
+ private final ChannelSessionStore channelSessionStore;
+ private final ChannelManager channelManager;
+
+ /**
+ * Channel types that handle their own UX via SSE (no IM forward needed).
+ * Everything not in this set is treated as an IM channel and gets the
+ * generated parts pushed via the adapter.
+ */
+ private static final Set
+ * Best-effort: missing session, missing adapter, or adapter exception
+ * are all logged at debug/warn and never propagate. The caller has
+ * already persisted the message to {@code mate_message} and broadcast
+ * to Web SSE before invoking this — IM forwarding is additive.
+ *
+ * @param conversationId the conversation id used by the agent (e.g.
+ * {@code wecom:XuZhanFu}, {@code dingtalk:cid_xxx},
+ * {@code conv_xxx} for Web)
+ * @param parts assistant content parts to dispatch (typically a
+ * single image / video / audio / file part)
+ */
+ public void forwardToImIfBound(String conversationId, List Default implementation ignores {@code ctx} and falls back to
+ * {@link #renderAndSend(String, String)}, so existing channel
+ * adapters and callers see no behavior change. Channels that want
+ * to consume {@code SendContext} fields override this overload.
+ *
+ * This was introduced as part of PR-0 (RFC-32 §2.0.3) to give
+ * {@code ChannelMessageRouter} a way to thread the pre-allocated
+ * feedback id (registered against the persisted
+ * {@code mate_message.id}) down to the WeCom adapter without
+ * widening the legacy two-arg signature.
+ */
+ default void renderAndSend(String targetId, String content, SendContext ctx) {
+ renderAndSend(targetId, content);
+ }
+
+ /**
+ * Render and deliver an approval notice. Channels that support a
+ * native interactive surface (WeCom {@code button_interaction},
+ * DingTalk {@code ActionCard}, etc.) override this to skip the
+ * text path entirely.
+ *
+ * Primary implementation lives on
+ * {@link AbstractChannelAdapter}, which keeps the bytewise
+ * fallback (markdown text → {@link #sendMessage}). Adapters that
+ * inherit from {@code AbstractChannelAdapter} can call
+ * {@code super.sendApprovalNotice(...)} to fall back; the default
+ * here is just a safety net for adapters that, for some reason,
+ * implement {@link ChannelAdapter} directly.
+ *
+ * Introduced in PR-0 (RFC-32 §2.0.3) so the router does not
+ * need to know which channel renders cards vs text:
+ * Return {@code true} when the underlying transport rejects multiple
+ * concurrent connections from the same credentials — e.g. a bot WebSocket
+ * gateway that enforces a per-app connection cap, or a long-polling
+ * endpoint where multiple consumers would steal updates from each other.
+ * The channel manager will gate {@link #start()} on a distributed lease
+ * so only one node connects at a time, and failover to another node when
+ * the lease holder dies.
+ *
+ * Webhook-based channels (DingTalk, WeCom, Slack, …) should leave this
+ * at the default {@code false}: inbound HTTP traffic is fanned out by the
+ * load balancer, so every node may safely subscribe.
+ *
+ * Scope: this hook is honored by the framework for DB-backed
+ * channels registered via {@code ChannelManager.startChannel}. For
+ * plugin-registered channels the framework can only gate the initial
+ * register attempt — there is no follower retry, no hot-swap, and no
+ * disable-detection (plugins have a register/unregister lifecycle, not
+ * a DB-driven one). Plugin authors needing full single-leader semantics
+ * should depend on {@code ChannelLeaderElection} directly.
+ */
+ default boolean requiresSingleLeader() {
+ return false;
+ }
+
/**
* RFC-024 Change 2：本 adapter 认为"多久没活动就视作 stale 需要重启"的阈值。
*
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java
index b7805453..ca3bfffa 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java
@@ -10,6 +10,8 @@ import org.springframework.stereotype.Component;
import vip.mate.channel.dingtalk.DingTalkChannelAdapter;
import vip.mate.channel.discord.DiscordChannelAdapter;
import vip.mate.channel.feishu.FeishuChannelAdapter;
+import vip.mate.channel.leader.ChannelLeaderElection;
+import vip.mate.channel.leader.LeaderLease;
import vip.mate.channel.model.ChannelEntity;
import vip.mate.channel.qq.QQChannelAdapter;
import vip.mate.channel.service.ChannelService;
@@ -17,7 +19,9 @@ import vip.mate.channel.telegram.TelegramChannelAdapter;
import vip.mate.channel.web.WebChannelAdapter;
import vip.mate.channel.wecom.WeComChannelAdapter;
import vip.mate.channel.weixin.WeixinChannelAdapter;
+import vip.mate.exception.MateClawException;
+import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReadWriteLock;
@@ -46,12 +50,91 @@ public class ChannelManager {
private final ObjectMapper objectMapper;
private final vip.mate.tool.document.GeneratedFileCache generatedFileCache;
+ /**
+ * Approval notification renderer — used by WeCom adapter (PR-0
+ * threading; PR-1 wired the WeCom override to render a
+ * {@code button_interaction} card via this service's card builder).
+ * Other adapters keep using the text path on
+ * {@link AbstractChannelAdapter}, which calls
+ * {@code ApprovalNotificationService.staticBuildText} so this
+ * field is currently consumed only by WeCom.
+ */
+ private final vip.mate.channel.notification.ApprovalNotificationService approvalNotificationService;
+
+ /**
+ * WeCom interactive card dispatcher (PR-1). Drives the
+ * {@code button_interaction} approval card render + the inbound
+ * {@code template_card_event} routing.
+ */
+ private final vip.mate.channel.wecom.cards.WeComCardDispatcher weComCardDispatcher;
+
+ /**
+ * WeCom keepalive scheduler (PR-1). Refreshes the "🤔 思考中..."
+ * placeholder every 20s and force-finishes after 180s so long-
+ * running agent tasks don't lose their stream slot.
+ */
+ private final vip.mate.channel.wecom.WeComKeepaliveScheduler weComKeepaliveScheduler;
+
+ /**
+ * Distributed leader election. Channels whose adapter reports
+ * {@link ChannelAdapter#requiresSingleLeader()} are gated on a lease so
+ * only one node opens the upstream WebSocket / long-poll at a time.
+ */
+ private final ChannelLeaderElection leaderElection;
+
/** 运行中的渠道适配器：channelId -> adapter */
private final Map Caller must hold the adapter write lock.
+ */
+ private void attemptLeaderStart(ChannelEntity channel, ChannelAdapter adapter) {
+ String key = channel.getChannelType() + ":" + channel.getId();
+ Optional Package-private for unit testing — callers should rely on the
+ * heartbeat scheduler invoking this on its tick.
+ */
+ void reconcileChannel(Long channelId, String channelName) {
+ ChannelEntity current;
+ try {
+ current = channelService.getChannel(channelId);
+ } catch (MateClawException e) {
+ if (e.getMsgKey() != null && e.getMsgKey().startsWith("err.channel.not_found")) {
+ log.info("[reconcile] Channel id={} no longer exists — stopping local adapter and releasing lease",
+ channelId);
+ stopChannel(channelId);
+ } else {
+ log.debug("[reconcile] Channel lookup failed for id={}: {}", channelId, e.getMessage());
+ }
+ return;
+ } catch (Exception e) {
+ // Transient DB issue; skip this tick and try again on the next heartbeat.
+ log.debug("[reconcile] Channel lookup failed for id={}: {}", channelId, e.getMessage());
+ return;
+ }
+
+ if (!Boolean.TRUE.equals(current.getEnabled())) {
+ log.info("[reconcile] Channel {} (id={}) is now disabled — stopping local adapter",
+ channelName, channelId);
+ stopChannel(channelId);
+ return;
+ }
+
+ LocalDateTime previousSeen;
+ adapterLock.readLock().lock();
+ try {
+ previousSeen = lastSeenChannelUpdateTime.get(channelId);
+ } finally {
+ adapterLock.readLock().unlock();
+ }
+ LocalDateTime currentUpdateTime = current.getUpdateTime();
+ if (currentUpdateTime != null && previousSeen != null
+ && currentUpdateTime.isAfter(previousSeen)) {
+ log.info("[reconcile] Channel {} (id={}) config changed ({} → {})",
+ channelName, channelId, previousSeen, currentUpdateTime);
+ applyConfigChange(channelId, current);
+ }
+ }
+
+ /**
+ * Apply a detected config change to a locally-running channel.
+ *
+ * The fast path is the in-place swap that preserves the lease —
+ * but it is only valid when we are the current leader (we already
+ * hold {@code activeLeases[channelId]}). Without that gate, a
+ * non-leader node observing a {@code webhook → websocket} flip
+ * would call {@code newAdapter.start()} directly inside the swap
+ * and open a duplicate upstream connection, defeating the leader
+ * election. Every other transition — including
+ * {@code non-leader → leader-required}, {@code leader-required →
+ * non-leader}, and plain non-leader config updates — must go
+ * through {@code stopChannel} + {@code startChannel} so the lease
+ * is correctly released or acquired and follower retry is
+ * scheduled when election is lost.
+ *
+ * Package-private for unit testing — see {@link #reconcileChannel}.
+ */
+ void applyConfigChange(Long channelId, ChannelEntity newChannel) {
+ ChannelAdapter probe = createAdapter(newChannel);
+ boolean newRequiresLeader = probe.requiresSingleLeader();
+ boolean weHaveLease;
+ adapterLock.readLock().lock();
+ try {
+ weHaveLease = activeLeases.containsKey(channelId);
+ } finally {
+ adapterLock.readLock().unlock();
+ }
+
+ if (newRequiresLeader && weHaveLease) {
+ // Case A: same leader-required mode and we are the current leader
+ // — preserve the lease across the adapter swap.
+ swapAdapterPreservingLease(channelId, newChannel);
+ return;
+ }
+
+ // All other cases: tear down local state and route through
+ // startChannel so leader election runs, lease is released, or both.
+ log.info("[reconcile] Channel {} (id={}) config change (newRequiresLeader={}, weHaveLease={}) — stop+start",
+ newChannel.getName(), channelId, newRequiresLeader, weHaveLease);
+ stopChannel(channelId);
+ try {
+ startChannel(newChannel);
+ } catch (Exception e) {
+ log.error("[reconcile] Restart after config change failed for channel {} (id={}): {}",
+ newChannel.getName(), channelId, e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Swap to a freshly-built adapter using the new config, while keeping
+ * the leadership lease and heartbeat in place. The lease is only
+ * released if the new adapter fails to start, in which case we fall
+ * back to follower mode so another node can try.
+ */
+ private void swapAdapterPreservingLease(Long channelId, ChannelEntity newChannel) {
+ ChannelAdapter oldAdapter;
+ adapterLock.writeLock().lock();
+ try {
+ oldAdapter = activeAdapters.remove(channelId);
+ } finally {
+ adapterLock.writeLock().unlock();
+ }
+ if (oldAdapter != null) {
+ stopAdapterSafely(oldAdapter, "reconcile-swap");
+ }
+
+ ChannelAdapter newAdapter = createAdapter(newChannel);
+ boolean started = false;
+ Exception startError = null;
+ try {
+ newAdapter.start();
+ started = true;
+ } catch (Exception e) {
+ startError = e;
+ }
+
+ LeaderLease leaseToRelease = null;
+ adapterLock.writeLock().lock();
+ try {
+ if (started) {
+ activeAdapters.put(channelId, newAdapter);
+ lastSeenChannelUpdateTime.put(channelId, newChannel.getUpdateTime());
+ } else {
+ leaseToRelease = activeLeases.remove(channelId);
+ lastSeenChannelUpdateTime.remove(channelId);
+ cancelHeartbeatLocked(channelId);
+ scheduleFollowerRetryLocked(channelId);
+ }
+ } finally {
+ adapterLock.writeLock().unlock();
+ }
+
+ if (!started) {
+ log.error("[reconcile] New adapter start failed for channel {} (id={}): {} — released lease, entering follower mode",
+ newChannel.getName(), channelId,
+ startError != null ? startError.getMessage() : "unknown");
+ if (leaseToRelease != null) {
+ leaseToRelease.release();
+ }
+ }
+ }
+
+ /**
+ * Drop the local adapter (without releasing the already-lost lease)
+ * and start follower retry so we'll attempt to reclaim leadership
+ * once the current owner stops renewing.
+ */
+ private void handleLeadershipLoss(Long channelId) {
+ ChannelAdapter local;
+ adapterLock.writeLock().lock();
+ try {
+ local = activeAdapters.remove(channelId);
+ activeLeases.remove(channelId); // already lost; do not call release()
+ cancelHeartbeatLocked(channelId);
+ scheduleFollowerRetryLocked(channelId);
+ } finally {
+ adapterLock.writeLock().unlock();
+ }
+ if (local != null) {
+ stopAdapterSafely(local, "leadership-loss");
+ }
+ }
+
+ /**
+ * Schedule periodic follower retry. Caller must hold the adapter
+ * write lock.
+ */
+ private void scheduleFollowerRetryLocked(Long channelId) {
+ if (followerRetryFutures.containsKey(channelId)) {
+ return;
+ }
+ ScheduledFuture f = leaderScheduler.scheduleAtFixedRate(
+ () -> followerRetry(channelId),
+ FOLLOWER_RETRY_INTERVAL_SECONDS, FOLLOWER_RETRY_INTERVAL_SECONDS, TimeUnit.SECONDS);
+ followerRetryFutures.put(channelId, f);
+ }
+
+ /**
+ * One follower-retry tick. Re-reads the channel from the DB (it may
+ * have been disabled or deleted) and attempts to start it. The retry
+ * cancels itself once we successfully become leader.
+ */
+ /** Package-private for unit testing — see {@link #reconcileChannel}. */
+ void followerRetry(Long channelId) {
+ ChannelEntity current;
+ try {
+ current = channelService.getChannel(channelId);
+ } catch (MateClawException e) {
+ // Channel was deleted on another node — cancel the retry so we
+ // don't leak a scheduled task forever. Other exception codes
+ // (e.g. transient DB errors) fall through to the generic catch
+ // and let the retry continue.
+ if (e.getMsgKey() != null && e.getMsgKey().startsWith("err.channel.not_found")) {
+ log.info("[leader] Follower retry: channel id={} no longer exists — cancelling retry", channelId);
+ adapterLock.writeLock().lock();
+ try {
+ cancelFollowerRetryLocked(channelId);
+ } finally {
+ adapterLock.writeLock().unlock();
+ }
+ return;
+ }
+ log.debug("[leader] Follower retry: lookup failed for channel id={}: {}", channelId, e.getMessage());
+ return;
+ } catch (Exception e) {
+ log.debug("[leader] Follower retry: lookup failed for channel id={}: {}", channelId, e.getMessage());
+ return;
+ }
+ if (!Boolean.TRUE.equals(current.getEnabled())) {
+ adapterLock.writeLock().lock();
+ try {
+ cancelFollowerRetryLocked(channelId);
+ } finally {
+ adapterLock.writeLock().unlock();
+ }
+ return;
+ }
+ try {
+ startChannel(current);
+ } catch (Exception e) {
+ log.debug("[leader] Follower retry: startChannel failed for id={}: {}", channelId, e.getMessage());
+ }
+ }
+
+ /** Package-private for unit testing. */
+ boolean hasFollowerRetry(Long channelId) {
+ adapterLock.readLock().lock();
+ try {
+ return followerRetryFutures.containsKey(channelId);
+ } finally {
+ adapterLock.readLock().unlock();
+ }
+ }
+
+ private void cancelHeartbeatLocked(Long channelId) {
+ ScheduledFuture f = heartbeatFutures.remove(channelId);
+ if (f != null) {
+ f.cancel(false);
+ }
+ }
+
+ private void cancelFollowerRetryLocked(Long channelId) {
+ ScheduledFuture f = followerRetryFutures.remove(channelId);
+ if (f != null) {
+ f.cancel(false);
+ }
+ }
+
+ /**
+ * Schedule the cross-node reconcile ticker for a non-leader active
+ * adapter. Caller must hold the adapter write lock.
+ */
+ private void scheduleReconcileLocked(Long channelId, String channelName) {
+ cancelReconcileLocked(channelId);
+ ScheduledFuture f = leaderScheduler.scheduleAtFixedRate(
+ () -> reconcileChannel(channelId, channelName),
+ FOLLOWER_RETRY_INTERVAL_SECONDS, FOLLOWER_RETRY_INTERVAL_SECONDS, TimeUnit.SECONDS);
+ reconcileFutures.put(channelId, f);
+ }
+
+ private void cancelReconcileLocked(Long channelId) {
+ ScheduledFuture f = reconcileFutures.remove(channelId);
+ if (f != null) {
+ f.cancel(false);
+ }
}
/**
@@ -161,6 +657,50 @@ public class ChannelManager {
return;
}
+ // Leader-required channels: if we don't already own the local adapter
+ // (i.e. we are a follower, or this is a brand-new channel), there is
+ // nothing to hot-swap. Fall through to startChannel which handles
+ // lease acquisition and follower retry. Hot-swap (which briefly opens
+ // a second upstream connection) is also avoided here so we don't
+ // double-occupy the bot's connection quota during a restart.
+ boolean weHaveAdapter;
+ boolean weHaveLease;
+ adapterLock.readLock().lock();
+ try {
+ weHaveAdapter = activeAdapters.containsKey(channelId);
+ weHaveLease = activeLeases.containsKey(channelId);
+ } finally {
+ adapterLock.readLock().unlock();
+ }
+ ChannelAdapter probe = createAdapter(channel);
+ if (probe.requiresSingleLeader()) {
+ log.info("[hot-swap] Channel {} requires single-leader; stop+start instead of hot-swap (weHaveAdapter={}, weHaveLease={})",
+ channel.getName(), weHaveAdapter, weHaveLease);
+ stopChannel(channelId);
+ startChannel(channel);
+ return;
+ }
+ // Mode flip: we currently hold a lease but the new config is no
+ // longer leader-required (e.g. Feishu WS → webhook). The lease,
+ // heartbeat, and lastSeenUpdateTime must all be torn down before
+ // the new non-leader adapter starts — the in-place hot-swap path
+ // below would leave them behind until the next heartbeat tick
+ // noticed and re-restarted, causing a redundant restart and a
+ // window where this node is silently holding a lease nobody else
+ // can grab. Stop+start handles all the cleanup in one shot.
+ if (weHaveLease) {
+ log.info("[hot-swap] Channel {} flipping leader-required → non-leader; stop+start to release lease",
+ channel.getName());
+ stopChannel(channelId);
+ startChannel(channel);
+ return;
+ }
+ if (!weHaveAdapter) {
+ log.info("[hot-swap] No local adapter for channel {}, delegating to startChannel", channel.getName());
+ startChannel(channel);
+ return;
+ }
+
log.info("[hot-swap] Starting hot-swap for channel: {} (type={}, id={})",
channel.getName(), channel.getChannelType(), channelId);
@@ -182,6 +722,10 @@ public class ChannelManager {
adapterLock.writeLock().lock();
try {
oldAdapter = activeAdapters.put(channelId, newAdapter);
+ // Mark the version we've now applied so the reconcile ticker
+ // (running for non-leader adapters) doesn't immediately fire a
+ // redundant swap on its next tick.
+ lastSeenChannelUpdateTime.put(channelId, channel.getUpdateTime());
log.info("[hot-swap] Adapter reference swapped for channel: {} (old={})",
channel.getName(), oldAdapter != null ? "present" : "none");
} finally {
@@ -203,17 +747,65 @@ public class ChannelManager {
*/
public void stopAll() {
List If the adapter reports {@link ChannelAdapter#requiresSingleLeader()},
+ * the framework gates the local register on a distributed lease keyed by
+ * {@code plugin:{pluginName}}. When another node already owns the lease
+ * this node skips registration (its plugin instance is loaded but inert
+ * locally) — see the scope note on {@link ChannelAdapter#requiresSingleLeader()}.
+ *
* @param pluginName the plugin name (used as key for unregistration)
* @param adapter the channel adapter
*/
public void registerPluginChannel(String pluginName, ChannelAdapter adapter) {
- try {
- adapter.start();
- pluginChannels.put(pluginName, adapter);
- log.info("Plugin channel registered: {} (type={})", pluginName, adapter.getChannelType());
- } catch (Exception e) {
- log.error("Failed to start plugin channel {}: {}", pluginName, e.getMessage(), e);
+ synchronized (pluginLifecycleLock) {
+ LeaderLease lease = null;
+ if (adapter.requiresSingleLeader()) {
+ Optional Package-private + non-final so unit tests can substitute a stub
+ * adapter without spinning up real WebSocket / HTTP clients.
*/
- private ChannelAdapter createAdapter(ChannelEntity channel) {
+ ChannelAdapter createAdapter(ChannelEntity channel) {
String type = channel.getChannelType();
return switch (type) {
case "web" -> new WebChannelAdapter(channel, messageRouter, objectMapper);
@@ -455,7 +1143,9 @@ public class ChannelManager {
case "feishu" -> new FeishuChannelAdapter(channel, messageRouter, objectMapper);
case "telegram" -> new TelegramChannelAdapter(channel, messageRouter, objectMapper);
case "discord" -> new DiscordChannelAdapter(channel, messageRouter, objectMapper);
- case "wecom" -> new WeComChannelAdapter(channel, messageRouter, objectMapper);
+ case "wecom" -> new WeComChannelAdapter(channel, messageRouter, objectMapper,
+ approvalNotificationService, weComCardDispatcher, weComKeepaliveScheduler,
+ generatedFileCache);
case "qq" -> new QQChannelAdapter(channel, messageRouter, objectMapper);
case "weixin" -> new WeixinChannelAdapter(channel, messageRouter, objectMapper);
case "slack" -> new vip.mate.channel.slack.SlackChannelAdapter(channel, messageRouter, objectMapper);
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 f1bc8772..549c4d70 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java
@@ -1,6 +1,8 @@
package vip.mate.channel;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import vip.mate.agent.AgentService;
@@ -8,6 +10,7 @@ import vip.mate.agent.context.ChatOrigin;
import vip.mate.approval.ApprovalWorkflowService;
import vip.mate.approval.ResolveOutcome;
import vip.mate.approval.PendingApproval;
+import vip.mate.channel.event.ChannelMessageReceivedEvent;
import vip.mate.channel.model.ChannelEntity;
import vip.mate.channel.notification.ApprovalNotificationService;
import vip.mate.channel.service.ChannelService;
@@ -59,6 +62,11 @@ public class ChannelMessageRouter {
private final ChatStreamTracker streamTracker;
private final ChannelChatOriginFactory chatOriginFactory;
private final ChannelErrorClassifier errorClassifier;
+ /** Field-injected (rather than constructor) to avoid a signature
+ * change that would ripple through every test that constructs the
+ * router directly. Spring's stock publisher is always available. */
+ @Autowired(required = false)
+ private ApplicationEventPublisher events;
/** 队列条目：封装消息及其路由上下文 */
private record QueueEntry(ChannelMessage message, ChannelAdapter adapter, ChannelEntity channelEntity) {}
@@ -88,8 +96,47 @@ public class ChannelMessageRouter {
/** 每个渠道的队列容量 */
private static final int QUEUE_CAPACITY = 1000;
- /** 防抖等待时间（毫秒） */
- private static final long DEBOUNCE_MS = 500;
+ /** 防抖等待时间（毫秒）。Package-private for unit-test access. */
+ static final long DEBOUNCE_MS = 500;
+
+ /**
+ * Extended debounce window for suspected paste-split scenarios. WeCom
+ * (and other IM clients) silently split a single pasted long prompt
+ * into 2-4 separate messages when it exceeds the per-frame limit
+ * (~2000 chars). The fragments arrive 0.5-2s apart, which means the
+ * default {@link #DEBOUNCE_MS} flushes the first fragment before the
+ * second one arrives — the agent then sees a torn context, calls the
+ * LLM on a partial prompt, and gets re-triggered when the next
+ * fragment lands. When merged content exceeds
+ * {@link #LONG_TEXT_THRESHOLD} we extend the window so the merger has
+ * time to absorb the rest.
+ *
+ * Package-private for unit-test access.
+ */
+ static final long LONG_DEBOUNCE_MS = 2500;
+
+ /**
+ * Content length (chars) above which we treat the message as a likely
+ * paste-split fragment. 1500 sits below the typical ~2000-char IM
+ * client split point while staying well above any normally-typed
+ * message, so the long-debounce path doesn't penalize ordinary
+ * chatting. A short typed "hello" still flushes in 500ms.
+ *
+ * Package-private for unit-test access.
+ */
+ static final int LONG_TEXT_THRESHOLD = 1500;
+
+ /**
+ * Pick the debounce window: extend to {@link #LONG_DEBOUNCE_MS} when
+ * either the new arrival or the accumulated merged buffer looks like
+ * a paste-split fragment, otherwise stay at {@link #DEBOUNCE_MS}.
+ *
+ * Package-private + static so tests can pin the threshold without
+ * spinning up the whole router (which has 12+ injected dependencies).
+ */
+ static long pickDebounceMs(int currentMergedLength) {
+ return currentMergedLength > LONG_TEXT_THRESHOLD ? LONG_DEBOUNCE_MS : DEBOUNCE_MS;
+ }
/**
* Plan-Execute SSE events that the Web Console mirror needs to see when
@@ -189,6 +236,14 @@ public class ChannelMessageRouter {
* @param channelEntity 渠道配置（含关联 agentId）
*/
public void enqueue(ChannelMessage message, ChannelAdapter adapter, ChannelEntity channelEntity) {
+ // Fan out to the trigger pipeline FIRST — channel_message and
+ // content_match triggers fire on every received message regardless
+ // of whether the channel has an agent attached. If we returned
+ // early on a missing agent below without publishing, the workflow
+ // side would silently lose every channel-event that doesn't also
+ // route to a chat agent.
+ publishChannelEvent(message, adapter, channelEntity);
+
Long agentId = channelEntity.getAgentId();
if (agentId == null) {
log.warn("Channel {} has no associated agent, ignoring message from {}",
@@ -207,27 +262,101 @@ public class ChannelMessageRouter {
log.info("[{}] Enqueuing message: sender={}, conversationId={}, agentId={}",
channelType, message.getSenderId(), conversationId, agentId);
- // 防抖：同一会话 500ms 内的连续消息合并
+ // Debounce + adaptive merge: same conversation messages within the
+ // (500ms / 2.5s) window get concatenated into one. Adaptive: when
+ // the merged buffer crosses the LONG_TEXT_THRESHOLD we extend to
+ // LONG_DEBOUNCE_MS so paste-split fragments arrive together
+ // instead of triggering one agent call per piece.
synchronized (pendingMessages) {
PendingMessage existing = pendingMessages.get(conversationId);
if (existing != null) {
- // 合并到已有的 pending 消息
- if (existing.timer != null) {
- existing.timer.cancel(false);
+ // Sender boundary in groups: when a different user sends to the
+ // same group within the debounce window, merging would attribute
+ // both fragments to whoever sent first — the LLM then loses the
+ // ability to tell who asked what. Flush the existing buffer
+ // immediately so each user's text rides its own pending window.
+ // Reentrant on `pendingMessages`, so the inner flushPending's
+ // synchronized block re-acquires safely on the same thread.
+ String existingSender = existing.firstMessage.getSenderId();
+ String incomingSender = message.getSenderId();
+ boolean sameSender = isSameSender(existingSender, incomingSender);
+ if (!sameSender) {
+ log.info("[{}] Sender boundary in conversation {}: flushing pending from sender={}, accepting new sender={}",
+ channelType, conversationId, existingSender, incomingSender);
+ if (existing.timer != null) {
+ existing.timer.cancel(false);
+ }
+ flushPending(conversationId);
+ // Fall through to create a fresh pending for the new sender.
+ } else {
+ // Same sender — original paste-split / rapid-follow merge path.
+ if (existing.timer != null) {
+ existing.timer.cancel(false);
+ }
+ existing.appendContent(message.getContent());
+ int mergedLen = existing.getMergedContent().length();
+ long debounceMs = pickDebounceMs(mergedLen);
+ existing.timer = debounceScheduler.schedule(
+ () -> flushPending(conversationId), debounceMs, TimeUnit.MILLISECONDS);
+ if (debounceMs > DEBOUNCE_MS) {
+ log.info("[{}] Long-text merger active: conversationId={}, mergedLen={}, debounce={}ms (paste-split suspected)",
+ channelType, conversationId, mergedLen, debounceMs);
+ } else {
+ log.debug("[{}] Message merged with pending (debounce {}ms): conversationId={}",
+ channelType, debounceMs, conversationId);
+ }
+ return;
}
- existing.appendContent(message.getContent());
- existing.timer = debounceScheduler.schedule(
- () -> flushPending(conversationId), DEBOUNCE_MS, TimeUnit.MILLISECONDS);
- log.debug("[{}] Message merged with pending (debounce): conversationId={}",
- channelType, conversationId);
- return;
}
- // 首条消息，创建 PendingMessage 并设定防抖定时器
+ // 首条消息（或 sender boundary 之后的新 sender），创建 PendingMessage 并设定防抖定时器
PendingMessage pending = new PendingMessage(message, adapter, channelEntity);
pendingMessages.put(conversationId, pending);
+ int firstLen = message.getContent() != null ? message.getContent().length() : 0;
+ long debounceMs = pickDebounceMs(firstLen);
pending.timer = debounceScheduler.schedule(
- () -> flushPending(conversationId), DEBOUNCE_MS, TimeUnit.MILLISECONDS);
+ () -> flushPending(conversationId), debounceMs, TimeUnit.MILLISECONDS);
+ if (debounceMs > DEBOUNCE_MS) {
+ log.info("[{}] Long-text merger armed on first message: conversationId={}, len={}, debounce={}ms",
+ channelType, conversationId, firstLen, debounceMs);
+ }
+ }
+ }
+
+ /**
+ * Publish a {@link ChannelMessageReceivedEvent} so the trigger module's
+ * bridge can fan the message out to channel_message + content_match
+ * triggers. Best-effort — a publish failure must never block the
+ * primary chat-routing path. {@code messageId} is used as the dedup
+ * key downstream so repeated webhook deliveries can't double-fire
+ * the same trigger.
+ */
+ private void publishChannelEvent(ChannelMessage message, ChannelAdapter adapter,
+ ChannelEntity channelEntity) {
+ if (events == null || message == null || adapter == null || channelEntity == null) return;
+ try {
+ long ws = channelEntity.getWorkspaceId() == null ? 0L : channelEntity.getWorkspaceId();
+ String channelType = adapter.getChannelType();
+ // messageId may be null for adapters that don't surface one;
+ // fall back to a sender+timestamp composite so the dedup key
+ // is at least deterministic-ish per webhook delivery.
+ String messageId = message.getMessageId();
+ if (messageId == null || messageId.isBlank()) {
+ messageId = channelType + ":" + message.getSenderId() + ":"
+ + (message.getTimestamp() == null ? System.currentTimeMillis()
+ : message.getTimestamp());
+ }
+ events.publishEvent(new ChannelMessageReceivedEvent(
+ ws,
+ channelType,
+ messageId,
+ message.getSenderId(),
+ message.getSenderName(),
+ message.getChatId(),
+ message.getContent()));
+ } catch (Exception e) {
+ log.warn("[ChannelMessageRouter] event publish failed for sender {}: {}",
+ message.getSenderId(), e.getMessage());
}
}
@@ -416,7 +545,10 @@ public class ChannelMessageRouter {
ResolveOutcome denyOutcome = approvalService.resolve(
pending.getPendingId(), message.getSenderId(), "denied");
conversationService.removeApprovalPlaceholders(conversationId);
- adapter.sendMessage(replyTarget, "⛔ 已拒绝执行工具: " + pending.getToolName());
+ String denyHint = "⛔ 已拒绝执行工具: " + pending.getToolName();
+ persistAndBroadcastApprovalHint(conversationId, denyHint,
+ "denied", pending.getPendingId(), pending.getToolName());
+ adapter.sendMessage(replyTarget, denyHint);
log.info("[{}] Approval DENIED via IM command: pendingId={}, tool={}, msgRewritten={}",
adapter.getChannelType(), pending.getPendingId(), pending.getToolName(),
denyOutcome.messagesRewritten());
@@ -426,7 +558,10 @@ public class ChannelMessageRouter {
// Non-approval message while a pending exists → treat as implicit deny.
approvalService.resolve(pending.getPendingId(), message.getSenderId(), "denied");
conversationService.removeApprovalPlaceholders(conversationId);
- adapter.sendMessage(replyTarget, "⛔ 审批已取消。将继续处理您的新消息。");
+ String cancelHint = "⛔ 审批已取消。将继续处理您的新消息。";
+ persistAndBroadcastApprovalHint(conversationId, cancelHint,
+ "cancelled", pending.getPendingId(), pending.getToolName());
+ adapter.sendMessage(replyTarget, cancelHint);
log.info("[{}] Approval auto-cancelled (non-approval message): pendingId={}",
adapter.getChannelType(), pending.getPendingId());
// Fall through to process the new message normally.
@@ -454,11 +589,20 @@ public class ChannelMessageRouter {
}
// 保存用户消息（带 contentParts）
+ // Group sender attribution: tag the persisted content + the
+ // prompt with [@sender] in groups so the LLM can disambiguate
+ // multiple users sharing one conversation. Single chats pass
+ // through unchanged (chatId is null).
List
+ * Without this, IM-driven approve/deny only reaches the originating IM
+ * channel via {@code adapter.sendMessage(...)} — a Web mirror of the same
+ * conversationId has no record of the resolution because nothing lands in
+ * {@code mate_message} and no SSE event is emitted. The hint then "vanishes"
+ * from the Web admin console even though it shows up on the user's phone.
+ *
+ * Persistence is the load-bearing fix (Web reload picks it up). Broadcast
+ * is best-effort: if no SSE stream is currently registered for the
+ * conversation, the broadcast no-ops silently — that's the common case
+ * since IM-driven clicks rarely race with an active web subscriber.
+ *
+ * @param conversationId conversation owning the hint
+ * @param hint text to render as an assistant bubble
+ * @param decision "approved" / "denied" / "cancelled" / null (skips the
+ * structured resolved event when null, e.g. on replay error)
+ * @param pendingId pending approval id; null when not applicable
+ * @param toolName tool name for the structured event; null when not applicable
+ */
+ private void persistAndBroadcastApprovalHint(String conversationId, String hint,
+ String decision, String pendingId,
+ String toolName) {
+ try {
+ conversationService.saveMessage(conversationId, "assistant", hint, null, "completed");
+ } catch (Exception e) {
+ log.warn("[approval-hint] saveMessage failed for conv={}: {}",
+ conversationId, e.getMessage());
+ }
+ try {
+ if (decision != null) {
+ streamTracker.broadcastObject(conversationId, "tool_approval_resolved", Map.of(
+ "pendingId", pendingId == null ? "" : pendingId,
+ "decision", decision,
+ "toolName", toolName == null ? "" : toolName,
+ "timestamp", System.currentTimeMillis()
+ ));
+ }
+ streamTracker.broadcastObject(conversationId, "message_start",
+ Map.of("role", "assistant"));
+ streamTracker.broadcastObject(conversationId, "content_delta",
+ Map.of("delta", hint));
+ streamTracker.broadcastObject(conversationId, "message_complete",
+ Map.of("status", "completed"));
+ } catch (Exception e) {
+ // Broadcast is best-effort; a missing run state is the common case.
+ log.debug("[approval-hint] broadcast skipped/failed for conv={}: {}",
+ conversationId, e.getMessage());
}
}
@@ -776,9 +989,13 @@ public class ChannelMessageRouter {
conversationService.getOrCreateConversation(conversationId, agentId, username, channelEntity.getWorkspaceId());
List Without this tag, three users asking three different questions in
+ * the same group conversation collapse into an unattributed wall of
+ * "user:" turns and the LLM can no longer tell who is asking what —
+ * it answers based on the most-recent text and ignores the rest.
+ * Single chats are unaffected because chatId is null there.
+ *
+ * Prefer {@code senderName} when populated; otherwise fall back to
+ * {@code senderId}. WeCom currently sets both to the same opaque
+ * openid which is still useful for disambiguation; future channels
+ * (DingTalk, Slack) carry friendlier display names that flow through
+ * unchanged.
+ *
+ * @return sender tag like {@code [@Alice]}, or {@code null} if the
+ * message is not from a group context.
+ */
+ static String buildGroupTag(ChannelMessage message) {
+ if (message == null) return null;
+ String chatId = message.getChatId();
+ if (chatId == null || chatId.isBlank()) return null;
+ String name = (message.getSenderName() != null && !message.getSenderName().isBlank())
+ ? message.getSenderName() : message.getSenderId();
+ if (name == null || name.isBlank()) return null;
+ return "[@" + name + "]";
+ }
+
+ /**
+ * Apply {@link #buildGroupTag} to {@code content}. Idempotent: if
+ * {@code content} already starts with the tag (e.g. an upstream
+ * adapter has pre-attributed it), returns it unchanged so we don't
+ * double-stamp. No-op for single chats.
+ */
+ static String applyGroupTag(ChannelMessage message, String content) {
+ String tag = buildGroupTag(message);
+ if (tag == null) return content;
+ // Empty content: leave empty rather than persist or prompt with a
+ // bare "[@Alice]" — the message had no payload to attribute.
+ if (content == null || content.isEmpty()) return content;
+ if (content.startsWith(tag)) return content;
+ return tag + " " + content;
+ }
+
+ /**
+ * Decision helper for the debounce merger: should an incoming message
+ * from {@code incomingSender} merge into a pending buffer started by
+ * {@code existingSender}? True only when the senders match — different
+ * senders in the same conversation (a group context) must NOT merge,
+ * else the second user's text gets attributed to the first.
+ *
+ * Null-handling: a null {@code existingSender} means "no buffer to
+ * merge into" so the answer is always false; a null
+ * {@code incomingSender} (rare, but seen in test fixtures) is also
+ * not allowed to silently merge — returning false routes to the
+ * "create new pending" branch which is safe.
+ */
+ static boolean isSameSender(String existingSender, String incomingSender) {
+ if (existingSender == null || incomingSender == null) return false;
+ return existingSender.equals(incomingSender);
+ }
+
/**
* 确定回复目标
* 优先使用 replyToken（渠道特有的回复标识），其次 chatId，最后 senderId
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/SendContext.java b/mateclaw-server/src/main/java/vip/mate/channel/SendContext.java
new file mode 100644
index 00000000..d72b9ad8
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/channel/SendContext.java
@@ -0,0 +1,63 @@
+package vip.mate.channel;
+
+import java.util.Map;
+
+/**
+ * Per-send context used to thread optional metadata from
+ * {@link ChannelMessageRouter} down into channel-specific renderers
+ * without having to expand {@code renderAndSend} signatures every time
+ * a new channel needs a side-channel value.
+ *
+ * Currently used by:
+ * Adapters that don't need any of this can ignore the parameter:
+ * the default {@code renderAndSend(targetId, content, ctx)} on
+ * {@link ChannelAdapter} delegates to the legacy two-arg version and
+ * drops {@code ctx} entirely.
+ *
+ * @param feedbackId optional like/dislike correlation id; null if
+ * the channel does not collect feedback or the
+ * assistant message could not be persisted
+ * @param savedMessageId persisted {@code mate_message.id} for the
+ * assistant reply; null in error / streaming
+ * passthrough paths where no row was created
+ * @param extra open-ended map; must never be null — use
+ * {@link #empty()} if no extras
+ */
+public record SendContext(
+ String feedbackId,
+ Long savedMessageId,
+ Map {@code messageId} doubles as the dedup key — repeated webhook
+ * deliveries of the same message can't double-fire downstream triggers
+ * because the {@code mate_trigger_event} unique constraint catches the
+ * second insert.
+ */
+public record ChannelMessageReceivedEvent(
+ long workspaceId,
+ String channelType,
+ String messageId,
+ String senderId,
+ String senderName,
+ String chatId,
+ String content
+) {}
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java
index 602251bf..992937ed 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/feishu/FeishuChannelAdapter.java
@@ -1435,4 +1435,19 @@ public class FeishuChannelAdapter extends AbstractChannelAdapter {
public String getChannelType() {
return CHANNEL_TYPE;
}
+
+ /**
+ * WebSocket mode opens a long-lived connection to Lark's gateway, which
+ * caps concurrent connections per bot app (~2). In a multi-instance
+ * deployment every node would race for that quota and reconnect-loop on
+ * {@code 1000040350: the number of connections exceeded the limit}.
+ * The leader gate ensures only one node holds the connection at a time.
+ *
+ * Webhook mode is exempt: callbacks are HTTP-fanned by the load
+ * balancer, so all nodes can safely subscribe.
+ */
+ @Override
+ public boolean requiresSingleLeader() {
+ return "websocket".equals(getConfigString("connection_mode", "websocket"));
+ }
}
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/leader/ChannelLeaderElection.java b/mateclaw-server/src/main/java/vip/mate/channel/leader/ChannelLeaderElection.java
new file mode 100644
index 00000000..8951cb08
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/channel/leader/ChannelLeaderElection.java
@@ -0,0 +1,84 @@
+package vip.mate.channel.leader;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import net.javacrumbs.shedlock.core.LockConfiguration;
+import net.javacrumbs.shedlock.core.LockProvider;
+import net.javacrumbs.shedlock.core.SimpleLock;
+import org.springframework.stereotype.Component;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Optional;
+
+/**
+ * Distributed leader election for channel adapters whose upstream
+ * service rejects multiple concurrent connections from the same bot
+ * credentials.
+ *
+ * Typical examples are WebSocket-mode IM channels: Feishu's Lark
+ * SDK enforces a per-app connection cap (~2) and QQ's bot gateway
+ * rejects duplicate {@code IDENTIFY} sessions. Without coordination,
+ * every node of a multi-instance deployment trips that cap on startup
+ * and reconnects in a tight loop.
+ *
+ * Backed by ShedLock's {@link LockProvider} (already wired for cron
+ * coordination), so single-node deployments incur no extra
+ * infrastructure — the lock is acquired trivially on the only node.
+ *
+ * Semantics:
+ * Wraps a ShedLock {@link SimpleLock} so callers don't depend on the
+ * underlying lock provider. The lease must be periodically extended via
+ * {@link #extend(Duration)} or it expires automatically, at which point
+ * another node can claim leadership.
+ *
+ * Threading: a single lease instance is not safe for concurrent
+ * {@link #extend(Duration)} / {@link #release()} calls. The owning
+ * scheduler is expected to serialize them.
+ */
+@Slf4j
+public class LeaderLease {
+
+ private final String name;
+ private volatile SimpleLock current;
+ private volatile boolean released;
+
+ LeaderLease(String name, SimpleLock initial) {
+ this.name = name;
+ this.current = initial;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ /**
+ * Try to extend this lease for another {@code lockAtMostFor} window.
+ *
+ * @return true if the lease is still ours; false if it has been lost
+ * (e.g. the previous window expired before extend ran and
+ * another node acquired the lock — the caller should treat
+ * this as a leadership loss and stop the protected resource).
+ */
+ public boolean extend(Duration lockAtMostFor) {
+ if (released) {
+ return false;
+ }
+ try {
+ Optional Logic is identical to {@link #buildApprovalText(ApprovalNotice)};
+ * the instance method delegates here so the two paths can never
+ * drift.
+ */
+ public static String staticBuildText(ApprovalNotice notice) {
StringBuilder sb = new StringBuilder();
sb.append("🔐 **工具需要审批**\n\n");
sb.append("**工具名称**: ").append(notice.toolName()).append("\n");
- // 风险等级
+ // Risk severity
if (notice.maxSeverity() != null) {
- sb.append("**风险等级**: ").append(severityLabel(notice.maxSeverity())).append("\n");
+ sb.append("**风险等级**: ").append(staticSeverityLabel(notice.maxSeverity())).append("\n");
}
- // 摘要
+ // Summary
if (notice.summary() != null && !notice.summary().isEmpty()) {
sb.append("**摘要**: ").append(notice.summary()).append("\n");
}
- // 参数预览
+ // Args preview
if (notice.argumentsPreview() != null && !notice.argumentsPreview().isEmpty()) {
sb.append("**参数**: `").append(notice.argumentsPreview()).append("`\n");
}
- // Findings 摘要（最多显示 3 条）
+ // Findings (top 3)
if (notice.findings() != null && !notice.findings().isEmpty()) {
sb.append("\n**发现的问题**:\n");
int shown = 0;
@@ -100,6 +113,18 @@ public class ApprovalNotificationService {
return sb.toString();
}
+ private static String staticSeverityLabel(String severity) {
+ if (severity == null) return "";
+ return switch (severity) {
+ case "CRITICAL" -> "🔴 CRITICAL";
+ case "HIGH" -> "🟠 HIGH";
+ case "MEDIUM" -> "🟡 MEDIUM";
+ case "LOW" -> "🔵 LOW";
+ case "INFO" -> "⚪ INFO";
+ default -> severity;
+ };
+ }
+
/**
* 构建 Web SSE 事件数据
*/
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/qq/QQChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/qq/QQChannelAdapter.java
index bddde1ce..72bdc0b9 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/qq/QQChannelAdapter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/qq/QQChannelAdapter.java
@@ -199,6 +199,17 @@ public class QQChannelAdapter extends AbstractChannelAdapter {
return CHANNEL_TYPE;
}
+ /**
+ * The QQ bot gateway rejects duplicate {@code IDENTIFY} sessions for
+ * the same app credentials. Multiple nodes connecting simultaneously
+ * trip the connection cap and reconnect-loop forever. The leader gate
+ * ensures only one node holds the WebSocket at a time.
+ */
+ @Override
+ public boolean requiresSingleLeader() {
+ return true;
+ }
+
// ==================== Access Token 管理 ====================
/**
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/slack/SlackChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/slack/SlackChannelAdapter.java
index eb1b2737..7634cb5b 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/slack/SlackChannelAdapter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/slack/SlackChannelAdapter.java
@@ -7,6 +7,7 @@ import com.slack.api.bolt.AppConfig;
import com.slack.api.bolt.socket_mode.SocketModeApp;
import com.slack.api.methods.SlackApiException;
import com.slack.api.methods.response.chat.ChatPostMessageResponse;
+import com.slack.api.methods.response.files.FilesUploadV2Response;
import com.slack.api.model.event.MessageEvent;
import lombok.extern.slf4j.Slf4j;
import vip.mate.channel.AbstractChannelAdapter;
@@ -14,8 +15,16 @@ import vip.mate.channel.ChannelMessage;
import vip.mate.channel.ChannelMessageRouter;
import vip.mate.channel.ExponentialBackoff;
import vip.mate.channel.model.ChannelEntity;
+import vip.mate.workspace.conversation.model.MessageContentPart;
import java.io.IOException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
@@ -295,4 +304,199 @@ public class SlackChannelAdapter extends AbstractChannelAdapter {
result = result.replaceAll("(?m)^#{1,6}\\s+(.+)$", "*$1*");
return result;
}
+
+ /**
+ * Lazily-initialised JDK HTTP client for fetching media bytes from
+ * fully-qualified {@code fileUrl} fields. Only used when
+ * {@link MessageContentPart#getPath()} isn't set.
+ */
+ private volatile HttpClient httpClient;
+
+ private HttpClient getHttpClient() {
+ HttpClient hc = httpClient;
+ if (hc == null) {
+ synchronized (this) {
+ hc = httpClient;
+ if (hc == null) {
+ hc = HttpClient.newBuilder()
+ .connectTimeout(Duration.ofSeconds(10))
+ .build();
+ httpClient = hc;
+ }
+ }
+ }
+ return hc;
+ }
+
+ /**
+ * Dispatch a list of {@link MessageContentPart}s to Slack. Text parts
+ * fall through to the existing {@link #sendMessage} chat-post path;
+ * media parts (image / audio / video / file / model3d) ride
+ * {@code filesUploadV2} so users see a native file card with thumbnail
+ * preview rather than an unopenable markdown link.
+ *
+ * Wired by {@link vip.mate.channel.AsyncTaskMediaDispatcher} so async
+ * tool results (image generation / video generation / etc.) reach
+ * Slack channels the same way they reach WeCom / DingTalk / Feishu.
+ */
+ @Override
+ public void sendContentParts(String targetId, List Package-private so {@link #requiresSingleLeader()} can mirror the
+ * same predicate — the two answers must stay in lockstep, otherwise a
+ * single change to mode detection here would silently mis-classify the
+ * channel for multi-node coordination.
*/
- private boolean resolveWebhookMode() {
+ boolean resolveWebhookMode() {
String connectionMode = getConfigString("connection_mode");
String webhookUrl = getConfigString("webhook_url");
boolean hasWebhookUrl = webhookUrl != null && !webhookUrl.isBlank();
@@ -794,6 +799,21 @@ public class TelegramChannelAdapter extends AbstractChannelAdapter {
return CHANNEL_TYPE;
}
+ /**
+ * Long-polling mode runs a {@code getUpdates(offset=…)} loop that
+ * acknowledges each delivered update; multiple nodes polling the same
+ * bot token would steal updates from each other (whichever node calls
+ * {@code getUpdates} next consumes the queue, and the others get
+ * nothing). The leader gate ensures only one node polls at a time.
+ *
+ * Webhook mode is exempt: Telegram POSTs to a public URL fanned
+ * by the load balancer, so every node may safely receive callbacks.
+ */
+ @Override
+ public boolean requiresSingleLeader() {
+ return !resolveWebhookMode();
+ }
+
/** RFC-025 Change 4 入站文本净化上限（防止 caption 含超长二进制撑爆 prompt）。 */
private static final int INBOUND_TEXT_MAX = 4096;
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java
index f31411f4..19c79358 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java
@@ -1416,6 +1416,14 @@ public class ChatController {
payload.put("status", status);
if (savedAssistant != null && savedAssistant.getId() != null) {
payload.put("assistantMessageId", savedAssistant.getId());
+ // Surface runtime model attribution so the chat bubble can show
+ // which model produced this reply without waiting for a history reload.
+ if (savedAssistant.getRuntimeModel() != null && !savedAssistant.getRuntimeModel().isBlank()) {
+ payload.put("runtimeModel", savedAssistant.getRuntimeModel());
+ }
+ if (savedAssistant.getRuntimeProvider() != null && !savedAssistant.getRuntimeProvider().isBlank()) {
+ payload.put("runtimeProvider", savedAssistant.getRuntimeProvider());
+ }
}
if (promptTokens > 0) payload.put("promptTokens", promptTokens);
if (completionTokens > 0) payload.put("completionTokens", completionTokens);
@@ -1638,10 +1646,26 @@ public class ChatController {
* having to guess from text. Empty string until the event arrives.
*/
private String finishReason = "";
+ /**
+ * Recovery affordance payload from {@link
+ * vip.mate.agent.GraphEventPublisher#feedback}. Persisted into
+ * {@code metadata.feedbackEvent} so a page reload still surfaces
+ * the retry/regenerate/report card on the failed assistant
+ * bubble. Null when the turn ended cleanly.
+ */
+ private Map {@code closed} 是防御性标志位：worker 在 idle compute 退出时会把 entry
+ * 从 map 删掉，所以正常路径上 enqueue 看不到一个"closed=true 还在 map 里"的
+ * state；保留这个标志为未来重构兜底，避免任何破坏"close = remove entry"
+ * 耦合的改动让 silent drop 复活。
+ */
+ private final ConcurrentHashMap 必须由 transport-ready 信号 触发置 true（即认证成功后的
+ * {@link #markReady()}），而不是 executor-ready（{@link #ensureReplyExecutor()}）。
+ * 否则会出现"executor 活的、accepting=true、但 webSocket=null"的窗口——
+ * worker 调 {@link #sendFrame} 看到 {@code webSocket==null} 就 warn 后默默 return，
+ * 让 caller 等 5s 假超时（RFC-32 §2.4.1 a-1 / R-7 修正）。
+ */
+ private final AtomicBoolean replyQueueAccepting = new AtomicBoolean(false);
+
+ /**
+ * Idle-timeout (ms) for the per-reqId worker's {@code queue.poll}.
+ * Default 60s in production; tests in the same package may lower
+ * this to the millisecond range to surface idle-close vs late-offer
+ * races without waiting a real minute (RFC-32 §3.0 S-3 stress).
+ *
+ * Package-private on purpose — not exposed via getter or
+ * setter; tests assign it directly. Production code never writes
+ * to this field.
+ */
+ @SuppressWarnings("PackageVisibleField")
+ volatile long workerIdleTimeoutMs = 60_000L;
+
+ /**
+ * Per-reqId 回复队列状态。
+ *
+ * @param queue 串行回复任务队列
+ * @param closed 防御性 closed 标志（详见 {@link #replyQueues} 注释）
+ */
+ private record ReplyQueueState(
+ LinkedBlockingQueue
+ * The WeCom AI Bot platform blocks {@code aibot_send_msg} in
+ * group chats — proactive pushes (cron summaries,
+ * image-generation completions, TTS audio, async-task forwards) must
+ * ride {@code aibot_respond_msg} bound to some prior frame
+ * id. Without this cache every group push silently failed:
+ * {@code sendMessageToChat} fell through to {@code aibot_send_msg},
+ * the platform rejected it, the user saw nothing.
+ *
+ * Populated for group inbound frames only — single-chat
+ * {@code aibot_send_msg} still works, so we don't need a cached
+ * reqId there. {@link #pickGroupReplyReqId(String)} returns the
+ * cached id (or null when there's never been a group inbound), and
+ * the proactive send paths fall through to {@code aibot_send_msg}
+ * when null.
+ *
+ * Bounded LRU at {@link #LAST_CHAT_REQ_IDS_MAX_SIZE} via insertion-
+ * order eviction — long-lived bots in many groups don't unbounded-grow.
+ */
+ private final ConcurrentHashMap Routes outbound approval notices to a {@code button_interaction}
+ * card via tool_guard renderer, and inbound {@code template_card_event}
+ * frames to the matching handler by task_id prefix. Null-tolerant for
+ * test contexts (the {@link #sendApprovalNotice} override falls back
+ * to the abstract-class text path when the dispatcher is missing).
+ */
+ private final vip.mate.channel.wecom.cards.WeComCardDispatcher cardDispatcher;
+
+ /**
+ * Refreshes the "🤔 思考中..." processing-stream chunk every 20s and
+ * force-finishes after 180s, so WeCom's server-side stream slot
+ * doesn't drop while a long-running agent task is still computing
+ * (RFC-32 §2.1.2 / R-7 / B-5). Null-tolerant: if missing (test DI
+ * gap), placeholder still appears once but is not refreshed.
+ */
+ private final WeComKeepaliveScheduler keepaliveScheduler;
+
+ /**
+ * In-memory cache of bytes generated by tools like
+ * {@code DocxRenderTool} / {@code PptxRenderTool}. The agent emits a
+ * {@code /api/v1/files/generated/{id}} URL referencing this cache; the
+ * channel layer resolves that URL back to bytes and uploads them as a
+ * native WeCom file message so the user actually receives a tappable
+ * document instead of an unopenable link. Null-tolerant: if missing
+ * (older constructor / test DI gap), URL stays inline as plain markdown
+ * which renders as a non-interactive link in the bubble.
+ */
+ private final vip.mate.tool.document.GeneratedFileCache generatedFileCache;
+
public WeComChannelAdapter(ChannelEntity channelEntity,
ChannelMessageRouter messageRouter,
- ObjectMapper objectMapper) {
+ ObjectMapper objectMapper,
+ vip.mate.channel.notification.ApprovalNotificationService approvalNotificationService,
+ vip.mate.channel.wecom.cards.WeComCardDispatcher cardDispatcher,
+ WeComKeepaliveScheduler keepaliveScheduler) {
+ this(channelEntity, messageRouter, objectMapper, approvalNotificationService,
+ cardDispatcher, keepaliveScheduler, null);
+ }
+
+ public WeComChannelAdapter(ChannelEntity channelEntity,
+ ChannelMessageRouter messageRouter,
+ ObjectMapper objectMapper,
+ vip.mate.channel.notification.ApprovalNotificationService approvalNotificationService,
+ vip.mate.channel.wecom.cards.WeComCardDispatcher cardDispatcher,
+ WeComKeepaliveScheduler keepaliveScheduler,
+ vip.mate.tool.document.GeneratedFileCache generatedFileCache) {
super(channelEntity, messageRouter, objectMapper);
+ this.approvalNotificationService = approvalNotificationService;
+ this.cardDispatcher = cardDispatcher;
+ this.keepaliveScheduler = keepaliveScheduler;
+ this.generatedFileCache = generatedFileCache;
// Default to 8 bounded attempts (~4 minutes total at 2s..30s exponential)
// so the UI eventually settles in ERROR instead of getting stuck in
// RECONNECTING forever. User config still overrides (-1 = infinite).
@@ -185,6 +332,11 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
.connectTimeout(Duration.ofSeconds(10))
.build();
+ // Build the reply-queue worker pool BEFORE the WS handshake kicks off, so
+ // any inbound auth_succeed → markReady → openReplyQueue path finds a live
+ // executor to schedule against. The gate stays closed until markReady runs.
+ ensureReplyExecutor();
+
connectWebSocket(botId, secret);
log.info("[wecom] WeCom bot channel initialized: botId={}, maxReconnectAttempts={}",
@@ -211,6 +363,11 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
.connectTimeout(Duration.ofSeconds(10))
.build();
+ // Re-arm the reply-queue worker pool BEFORE attempting the new
+ // handshake. accepting flag stays false until the new connection's
+ // auth_succeed fires markReady → openReplyQueue.
+ ensureReplyExecutor();
+
String botId = getConfigString("bot_id");
String secret = getConfigString("secret");
connectWebSocket(botId, secret);
@@ -229,6 +386,16 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
* "Disable + Enable" did to recover.
*/
private void releaseConnectionResources(String reason) {
+ // ============================================================================
+ // RFC-32 §2.4.1 a-3 / R-6 + R-8 修正：必须按 step 0~4 顺序，不是尾部追加。
+ // step 0 (replyQueueAccepting=false) 必须在 ws.close()/wsThread.join() 之前；
+ // 否则在 ws teardown 期间还会有 keepalive / 最终回复 / proactiveSend 漏进 enqueue。
+ // ============================================================================
+
+ // ---- Step 0：先关 lifecycle gate，让任何后续 sendFrameWithAck 立刻 fast-fail ----
+ replyQueueAccepting.set(false);
+
+ // ---- 现有的 ws/heartbeat teardown（功能未变；插在 step 0 之后、step 1 之前） ----
if (heartbeatFuture != null) {
heartbeatFuture.cancel(false);
heartbeatFuture = null;
@@ -253,17 +420,173 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
}
wsThread = null;
}
- pendingAcks.forEach((k, f) ->
- f.completeExceptionally(new RuntimeException("Channel " + reason)));
- pendingAcks.clear();
+
+ // ---- Step 1：第一次 drain replyQueues ----
+ // forEach 是 weakly-consistent 迭代器，可能错过 step 0 之前刚提交但还没出 compute
+ // 的 enqueue —— step 3 会再 drain 一次兜底。
+ replyQueues.forEach((rid, state) -> {
+ state.closed().set(true);
+ ReplyTask t;
+ while ((t = state.queue().poll()) != null) {
+ if (!t.future().isDone()) {
+ t.future().completeExceptionally(new IllegalStateException("Channel " + reason));
+ }
+ }
+ });
+
+ // ---- Step 2：shutdownNow 中断 worker 阻塞中的 poll(60s) + 拒绝后续 submit ----
+ ExecutorService oldExecutor = this.replyExecutor;
+ if (oldExecutor != null) {
+ oldExecutor.shutdownNow();
+ this.replyExecutor = null;
+ }
+
+ // ---- Step 3：second drain，捕获 step 1 与 step 2 之间的窗口期残留 ----
+ // 此刻 shutdownNow 已经把任何新 fresh state 的 worker 拒掉，drain 是它们唯一退路。
+ replyQueues.forEach((rid, state) -> {
+ state.closed().set(true);
+ ReplyTask t;
+ while ((t = state.queue().poll()) != null) {
+ if (!t.future().isDone()) {
+ t.future().completeExceptionally(new IllegalStateException("Channel " + reason));
+ }
+ }
+ });
replyQueues.clear();
+
+ // ---- Step 4：pendingAcks 残留 ----
+ pendingAcks.forEach((k, f) -> {
+ if (!f.isDone()) {
+ f.completeExceptionally(new IllegalStateException("Channel " + reason));
+ }
+ });
+ pendingAcks.clear();
+
+ // ---- 其他 per-connection 状态 ----
pendingFrames.clear();
replyContexts.clear();
+ streamLastContent.clear();
+ if (keepaliveScheduler != null) {
+ keepaliveScheduler.shutdownAll();
+ }
missedPongCount.set(0);
this.httpClient = null;
}
+ // ====================================================================
+ // RFC-32 §2.0.5 / §2.4.1 a-1: lifecycle gate plumbing
+ // ====================================================================
+
+ /**
+ * (Re)build the worker pool. Called from {@link #doStart()} and
+ * {@link #doReconnect()}. Does not touch the {@link #replyQueueAccepting}
+ * gate — that flag is controlled by the transport-ready signal
+ * ({@link #markReady()}). See §2.4.1 a-1 / R-7.
+ */
+ private void ensureReplyExecutor() {
+ if (replyExecutor == null || replyExecutor.isShutdown()) {
+ replyExecutor = Executors.newCachedThreadPool(r -> {
+ Thread t = new Thread(r, "wecom-reply");
+ t.setDaemon(true);
+ return t;
+ });
+ }
+ }
+
+ /**
+ * Open the {@link #replyQueueAccepting} lifecycle gate. Only
+ * called from {@link #markReady()} after auth_succeed. Until this
+ * runs, every {@link #sendFrameWithAck} call fast-fails the caller's
+ * future with {@link IllegalStateException}.
+ */
+ private void openReplyQueue() {
+ replyQueueAccepting.set(true);
+ }
+
+ /**
+ * Per-reqId serial worker. Started lazily by
+ * {@link #sendFrameWithAck} when a fresh {@link ReplyQueueState} is
+ * created. Exits when:
+ * The compute-based idle-close fixes the TOCTOU race called out
+ * in RFC-32 §2.4.1 a-2 / R-5: enqueue's {@code compute} and
+ * worker's idle-close {@code compute} share the same bin lock,
+ * so offer and remove never interleave on the same key.
+ */
+ private void reqIdWorker(String reqId, ReplyQueueState state) {
+ while (running.get() && !Thread.currentThread().isInterrupted()) {
+ ReplyTask task;
+ try {
+ task = state.queue().poll(workerIdleTimeoutMs, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ break; // fall through to drainStateExceptionally + return
+ }
+
+ if (task == null) {
+ // Atomic close — serialized against sendFrameWithAck.compute on
+ // the same reqId by ConcurrentHashMap's bin lock.
+ ReplyQueueState afterClose = replyQueues.compute(reqId, (k, current) -> {
+ if (current != state) return current; // (c) replaced — defensive exit
+ if (!current.queue().isEmpty()) return current; // (b) late offer — stay alive
+ current.closed().set(true); // (a) truly idle — close
+ return null; // (a) remove entry
+ });
+ if (afterClose != state) return; // (a) or (c) — exit
+ continue; // (b) — keep going
+ }
+
+ try {
+ pendingAcks.put(reqId, task.future());
+ // orTimeout 5s 兜底，whenComplete 在完成时清 pendingAcks。
+ // 用 (key, value) 双参 remove 避免误删后续 task 的注册。
+ task.future().orTimeout(REPLY_ACK_TIMEOUT_MS, TimeUnit.MILLISECONDS)
+ .whenComplete((r, ex) -> pendingAcks.remove(reqId, task.future()));
+ sendFrame(task.frame());
+ task.future().join(); // serialize: don't dequeue next until this is done
+ } catch (CompletionException ce) {
+ // join() 抛的是 orTimeout 注入的异常（典型：TimeoutException）——
+ // task.future 已经 complete，无需手动 fail
+ log.debug("[wecom] reply task ACK failed for reqId={}: {}", reqId, ce.getCause());
+ } catch (Exception e) {
+ // sendFrame 同步抛 → ACK 永远不会到 → 必须显式 fail，否则 caller future 永久 pending
+ if (!task.future().isDone()) {
+ task.future().completeExceptionally(e);
+ }
+ pendingAcks.remove(reqId, task.future());
+ log.debug("[wecom] reply task send failed for reqId={}: {}", reqId, e.getMessage());
+ }
+ }
+
+ // running=false / interrupted: mark closed + drain leftover
+ state.closed().set(true);
+ drainStateExceptionally(reqId, state, "channel stopped");
+ }
+
+ /**
+ * Drain remaining tasks in a {@link ReplyQueueState} and best-effort
+ * remove the entry from {@link #replyQueues}. Used by worker exit
+ * paths (running=false / interrupt). For {@link #releaseConnectionResources}
+ * the drain is inlined (step 1 / step 3) to keep the ordering proof local.
+ */
+ private void drainStateExceptionally(String reqId, ReplyQueueState state, String reason) {
+ ReplyTask t;
+ while ((t = state.queue().poll()) != null) {
+ if (!t.future().isDone()) {
+ t.future().completeExceptionally(new IllegalStateException(reason));
+ }
+ }
+ replyQueues.remove(reqId, state);
+ }
+
// ==================== WebSocket 连接 ====================
/**
@@ -354,6 +677,10 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
reconnectFuture = null;
}
disconnectInflight.set(false);
+ // RFC-32 §2.4.1 a-1 / R-7: only NOW does sendFrameWithAck start
+ // accepting tasks — auth_succeed has just been observed and the
+ // WS is the canonical "transport ready" anchor.
+ openReplyQueue();
}
/**
@@ -622,15 +949,9 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
Map 5-second window: WeCom requires the
+ * {@code aibot_respond_update_msg} for this event to be sent inside
+ * 5s. Handlers therefore run synchronously here; the heavy work
+ * (e.g. agent re-execution) is deferred to the router's normal
+ * processMessage path via {@link #injectSyntheticMessage}.
+ */
+ @SuppressWarnings("unchecked")
+ private void handleTemplateCardEvent(Map Falls back to {@code super.sendApprovalNotice} (markdown text)
+ * in three failure modes:
+ * The card is sent via {@link #replyTemplateCard} bound to the
+ * inbound frame's {@code req_id} that
+ * {@link #handleMessageCallback} stashed in {@link #replyContexts}.
+ */
+ @Override
+ public void sendApprovalNotice(String targetId,
+ vip.mate.channel.notification.ApprovalNotice notice) {
+ if (cardDispatcher == null) {
+ super.sendApprovalNotice(targetId, notice);
+ return;
+ }
+ WeComReplyContext ctx = replyContexts.get(targetId);
+ if (ctx == null || ctx.frameReqId() == null || ctx.frameReqId().isBlank()) {
+ // No bound reply context — fall back to text. Most common in
+ // proactive paths (cron-triggered approvals) which WeCom AI
+ // Bot rejects for cards anyway.
+ super.sendApprovalNotice(targetId, notice);
+ return;
+ }
+ var kindOpt = cardDispatcher.lookupByMessageType(
+ vip.mate.channel.wecom.cards.tool_guard.ToolGuardCardKindFactory.MESSAGE_TYPE);
+ if (kindOpt.isEmpty()) {
+ super.sendApprovalNotice(targetId, notice);
+ return;
+ }
+ try {
+ Map
+ * Group fallback: WeCom AI Bot platform rejects {@code aibot_send_msg}
+ * in group chats. When {@code chatId} matches a known group (via
+ * {@link #pickGroupReplyReqId}), ride a cached inbound reqId via
+ * {@code aibot_respond_msg} instead. Single chats still use
+ * {@code aibot_send_msg} (which the platform allows).
*/
private void sendMessageToChat(String chatId, String content) {
if (webSocket == null || content == null || content.isBlank()) return;
+ Map
+ * Centralised so {@link #sendMessageToChat} (text) and
+ * {@link #sendMediaMessage} (image/file/voice/video) share the same
+ * group-vs-single dispatch — without this, every new outbound path
+ * had to remember the group rule, and several didn't.
+ */
+ private void sendOutboundFrame(String chatId, Map
+ * Package-private for test access.
+ */
+ String pickGroupReplyReqId(String chatId) {
+ if (chatId == null || chatId.isBlank()) return null;
+ return lastChatReqIds.get(chatId);
+ }
+
/**
* 覆写 renderAndSend：如果有 processing_stream_id 则用 reply_stream 覆盖"思考中..."
*/
@@ -842,6 +1424,21 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
public void renderAndSend(String targetId, String content) {
// 消费回复上下文（如果有的话）
WeComReplyContext ctx = replyContexts.remove(targetId);
+ // Stop keepalive before we send the real reply: avoids racing the next
+ // refresh tick against this finish=true chunk on the same stream.
+ // No-op if force-finish already evicted the entry.
+ if (keepaliveScheduler != null && ctx != null && ctx.processingStreamId() != null) {
+ keepaliveScheduler.stop(ctx.processingStreamId());
+ }
+
+ // Sniff `/api/v1/files/generated/{id}` URLs out of the agent's text
+ // BEFORE rendering. Each hit gets upgraded to a native WeCom file
+ // message via the chunked upload API; the URL in the text is replaced
+ // with a "📎 filename" marker so the bubble doesn't repeat itself.
+ // Without this, a generated docx/pptx would arrive as a markdown link
+ // the user can't open inside WeCom (no public access + JWT required).
+ List
- * WeCom 原生语音消息要求 AMR 格式。TTS 输出为 MP3，
- * Phase 1 以 file 类型发送（用户可点击播放），避免引入 AMR 转码依赖。
- * 非 AMR 格式走 file 类型而非 voice 类型，避免企微语音播放兼容问题。
+ * WeCom 原生语音消息要求 AMR 格式 + ≤ 2 MB。预校验里非 AMR 或超 2 MB
+ * 自动降级为 file（文件卡片，可点击下载播放）+ 给用户一行说明
+ * 提示——避免用户期待"语音气泡"但收到一个 .mp3 文件却不知道为啥。
*/
private void sendAudioPart(String targetId, MessageContentPart part, WeComReplyContext ctx) {
byte[] audioBytes = resolveFileBytes(part);
@@ -983,15 +1716,30 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
String fileName = part.getFileName() != null ? part.getFileName() : "voice_reply.mp3";
boolean isAmr = fileName.toLowerCase().endsWith(".amr");
+ // Pre-decide native voice vs file based on extension; the limits
+ // checker can still downgrade voice→file if size exceeds 2MB.
+ String requestedType = isAmr ? "voice" : "file";
+ String contentTypeHint = part.getContentType();
+ if (contentTypeHint == null && isAmr) contentTypeHint = "audio/amr";
- // AMR 格式：以原生 voice 类型发送（语音气泡）
- // 其他格式（MP3 等）：以 file 类型发送（文件卡片，可点击播放）
- String uploadType = isAmr ? "voice" : "file";
- String mediaId = uploadMedia(audioBytes, fileName, uploadType);
+ WeComUploadLimitDecision decision = applyWeComUploadLimits(
+ audioBytes.length, requestedType, contentTypeHint);
+ if (decision.rejected()) {
+ log.warn("[wecom] Audio upload rejected: {} ({} bytes) — {}",
+ fileName, audioBytes.length, decision.rejectReason());
+ sendMessageToChat(targetId, "⚠️ " + decision.rejectReason());
+ return;
+ }
+
+ String mediaId = uploadMedia(audioBytes, fileName, decision.finalMediaType());
if (mediaId != null) {
String frameReqId = ctx != null ? ctx.frameReqId() : null;
- sendMediaMessage(targetId, mediaId, uploadType, frameReqId);
- log.info("[wecom] Audio sent as {}: {} ({}KB)", uploadType, fileName, audioBytes.length / 1024);
+ sendMediaMessage(targetId, mediaId, decision.finalMediaType(), frameReqId);
+ log.info("[wecom] Audio sent as {}: {} ({}KB)",
+ decision.finalMediaType(), fileName, audioBytes.length / 1024);
+ if (decision.downgraded()) {
+ sendMessageToChat(targetId, "ℹ️ " + decision.downgradeNote());
+ }
} else {
sendFallbackText(targetId, part);
}
@@ -1063,10 +1811,51 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
* @param finish 是否结束流式消息
*/
private void replyStream(String originalReqId, String streamId, String content, boolean finish) {
+ replyStream(originalReqId, streamId, content, finish, null);
+ }
+
+ /**
+ * Streaming reply with optional WeCom feedback id attached on the
+ * final chunk (PR-2 hook installed in PR-0 so the protocol surface
+ * is stable).
+ *
+ * Per WeCom AI Bot protocol (verified against the langbot
+ * reference implementation), {@code feedback.id} is only meaningful
+ * on the chunk where {@code finish=true}. We accept the parameter
+ * on every chunk for ergonomics but only emit the JSON field on
+ * the finishing chunk to avoid surfacing it where the server
+ * would ignore it.
+ *
+ * Callers that don't need feedback collection pass {@code null}
+ * for {@code feedbackId} (or use the legacy 4-arg overload).
+ */
+ private void replyStream(String originalReqId, String streamId, String content,
+ boolean finish, String feedbackId) {
+ // PR-1 chunk dedup: skip the network round-trip when a non-final chunk
+ // has the exact same content as the previous one for the same streamId.
+ // Tool-call argument streaming in particular emits many redundant chunks
+ // (each token re-flushes the partial JSON args) that would otherwise
+ // flicker the IM client. The final chunk (finish=true) ALWAYS goes
+ // through so WeCom closes the slot cleanly. RFC-32 §2.1.3.
+ if (!finish) {
+ String content_safe = content == null ? "" : content;
+ String prev = streamLastContent.get(streamId);
+ if (content_safe.equals(prev)) {
+ return;
+ }
+ streamLastContent.put(streamId, content_safe);
+ } else {
+ // Final chunk consumes the dedup slot.
+ streamLastContent.remove(streamId);
+ }
+
Map
+ * Package-private for unit-test access.
+ */
+ static WeComUploadLimitDecision applyWeComUploadLimits(long fileSize, String mediaType,
+ String contentType) {
+ String type = mediaType == null ? "file" : mediaType.toLowerCase();
+ String mime = contentType == null ? "" : contentType.toLowerCase().trim();
+
+ if (fileSize > FILE_MAX_BYTES) {
+ double mb = fileSize / 1024.0 / 1024.0;
+ return new WeComUploadLimitDecision(
+ type, true,
+ String.format(java.util.Locale.ROOT,
+ "文件大小 %.2fMB 超过企业微信 20MB 上限，无法发送。请压缩或拆分后再发。", mb),
+ false, null);
+ }
+ if ("image".equals(type) && fileSize > IMAGE_MAX_BYTES) {
+ double mb = fileSize / 1024.0 / 1024.0;
+ return new WeComUploadLimitDecision(
+ "file", false, null,
+ true,
+ String.format(java.util.Locale.ROOT,
+ "图片 %.2fMB 超过 10MB 限制，已转为文件形式发送", mb));
+ }
+ if ("video".equals(type) && fileSize > VIDEO_MAX_BYTES) {
+ double mb = fileSize / 1024.0 / 1024.0;
+ return new WeComUploadLimitDecision(
+ "file", false, null,
+ true,
+ String.format(java.util.Locale.ROOT,
+ "视频 %.2fMB 超过 10MB 限制，已转为文件形式发送", mb));
+ }
+ if ("voice".equals(type)) {
+ if (!mime.isEmpty() && !VOICE_SUPPORTED_MIMES.contains(mime)) {
+ return new WeComUploadLimitDecision(
+ "file", false, null, true,
+ "语音格式 " + mime + " 不支持（企微仅支持 AMR），已转为文件形式发送");
+ }
+ if (fileSize > VOICE_MAX_BYTES) {
+ double mb = fileSize / 1024.0 / 1024.0;
+ return new WeComUploadLimitDecision(
+ "file", false, null,
+ true,
+ String.format(java.util.Locale.ROOT,
+ "语音 %.2fMB 超过 2MB 限制，已转为文件形式发送", mb));
+ }
+ }
+ return WeComUploadLimitDecision.pass(type);
+ }
+
/**
* 发送欢迎消息
*/
@@ -1098,6 +1993,139 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
sendFrameWithAck(reqId, frame);
}
+ /**
+ * Send an interactive template card (e.g. button_interaction approval card).
+ *
+ * Wraps the card payload in {@code msgtype=template_card} and routes via
+ * the existing reply channel ({@code aibot_respond_msg}, bound to the inbound
+ * frame's req_id). Source-verified against aibot SDK
+ * {@code client.py:188-207 reply_template_card}.
+ *
+ * Caller must have an active reply context for {@code reqId} — i.e. the
+ * card is sent in response to a previously received message frame, not as a
+ * proactive group push (which WeCom rejects for AI Bots, see RFC-32 G-12).
+ *
+ * @param reqId the original inbound frame's {@code headers.req_id}
+ * @param templateCard the WeCom template_card payload (card_type / task_id /
+ * main_title / button_list / etc.)
+ */
+ public void replyTemplateCard(String reqId, Map 5-second window: per the aibot protocol, the response must be
+ * sent within 5s of receiving the {@code template_card_event} frame —
+ * otherwise the update is silently dropped. The handler path therefore
+ * has to validate identity + render the new card synchronously (fast
+ * DB lookup + map construction, well under 1ms) and only enqueue the
+ * inject-command on the agent thread afterwards.
+ *
+ * Source-verified against aibot SDK {@code client.py:260-284 update_template_card}.
+ *
+ * @param eventReqId the inbound {@code template_card_event} frame's req_id
+ * (DIFFERENT from the original card-posting req_id)
+ * @param templateCard the replacement card payload (same task_id as the
+ * original card)
+ */
+ public void updateTemplateCard(String eventReqId, Map Public so the scheduler in this same package can invoke it; the
+ * scheduler is itself a singleton bean and outside callers should
+ * not be triggering refresh ticks.
+ */
+ public void replyStreamRefreshForKeepalive(String reqId, String streamId, String text) {
+ replyStream(reqId, streamId, text, false);
+ }
+
+ /**
+ * Force-finish the keepalive stream (180s ceiling reached). Sends
+ * {@code finish=true} so WeCom closes the slot cleanly. The
+ * scheduler immediately follows this with
+ * {@link #invalidateReplyContext} so the eventual real reply takes
+ * the fresh-stream path.
+ */
+ public void replyStreamFinishForKeepalive(String reqId, String streamId, String text) {
+ replyStream(reqId, streamId, text, true);
+ }
+
+ /**
+ * Drop the {@link WeComReplyContext} entry for a {@code targetId}
+ * if (and only if) its current {@code processingStreamId} matches
+ * the supplied {@code streamId}. Idempotent and safe to call from
+ * any thread.
+ *
+ * Used by {@link WeComKeepaliveScheduler} after force-finishing
+ * a stuck stream — RFC-32 §2.1.2 invariant: the next
+ * {@link #renderAndSend} call must NOT reuse a finished
+ * {@code processingStreamId}.
+ *
+ * The match-and-remove uses {@link
+ * java.util.concurrent.ConcurrentHashMap#computeIfPresent} so a
+ * concurrent {@code renderAndSend} that already swapped the
+ * context for a fresh stream is left untouched.
+ */
+ public void invalidateReplyContext(String targetId, String streamId) {
+ if (targetId == null || streamId == null) return;
+ replyContexts.computeIfPresent(targetId, (k, ctx) -> {
+ if (streamId.equals(ctx.processingStreamId())) {
+ log.debug("[wecom] invalidateReplyContext: cleared {} (stream={})", targetId, streamId);
+ return null; // remove entry
+ }
+ return ctx;
+ });
+ }
+
+ /**
+ * Route a synthetic message into the standard
+ * {@link ChannelMessageRouter} pipeline as if the user had typed it.
+ *
+ * Bypasses {@link AbstractChannelAdapter#onMessage} so the
+ * pre-flight bot-prefix filter and access-control check are SKIPPED
+ * — appropriate for events that already represent an explicit user
+ * intent (e.g. a button click on an approval card). The router still
+ * runs its own approval validation in
+ * {@link ChannelMessageRouter#processMessage}, so the identity check
+ * for "only original requester can approve" still fires.
+ *
+ * Currently used by tool-guard card handler. Package-private (no
+ * modifier) so only sibling classes in the wecom package can inject;
+ * external code must go through {@link ChannelAdapter#onMessage}.
+ */
+ public void injectSyntheticMessage(ChannelMessage message) {
+ messageRouter.enqueue(message, this, channelEntity);
+ }
+
// ==================== 媒体上传协议 ====================
/**
@@ -1118,6 +2146,18 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
if (webSocket == null || fileBytes == null || fileBytes.length == 0) {
return null;
}
+ // Pre-flight chunk count guard. WeCom's chunked upload protocol caps
+ // out near 100 chunks (~50 MB at 512 KB / chunk) but we already
+ // reject anything over FILE_MAX_BYTES (20 MB ≈ 40 chunks) before
+ // reaching here, so this is a defence-in-depth log line rather
+ // than a routine path.
+ int totalChunks = (int) Math.ceil((double) fileBytes.length / UPLOAD_CHUNK_SIZE);
+ if (totalChunks > 100) {
+ log.warn("[wecom] Upload would require {} chunks (>100 cap), rejecting: {} ({} bytes)",
+ totalChunks, fileName, fileBytes.length);
+ return null;
+ }
+
boolean acquired = false;
try {
acquired = uploadLock.tryAcquire(60, TimeUnit.SECONDS);
@@ -1127,7 +2167,6 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
}
String md5 = md5Hex(fileBytes);
- int totalChunks = (int) Math.ceil((double) fileBytes.length / UPLOAD_CHUNK_SIZE);
// Phase 1: Init
String initReqId = generateReqId(CMD_UPLOAD_INIT);
@@ -1163,7 +2202,13 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
Map
- * 同一 reqId 的消息按顺序发送，每条等待 ACK 后再发下一条。
+ * Serially send a frame on the WS and wait (in a per-reqId worker)
+ * for its ACK. Same {@code reqId} messages are guaranteed to be
+ * dispatched in arrival order: the worker reads from the queue,
+ * registers {@link #pendingAcks} only after the previous ACK
+ * settled, sends, then blocks on the future until the ACK arrives
+ * or {@link #REPLY_ACK_TIMEOUT_MS} elapses.
+ *
+ * RFC-32 §2.4.1 a-2 / R-5/R-6/R-7 invariants this implements:
+ * Returns the ACK future for callers that want to chain on
+ * success (e.g. extract {@code body} fields from the ACK frame).
+ * Existing fire-and-forget callers can ignore the return value;
+ * timeout/error handling lives inside the worker.
*/
- private void sendFrameWithAck(String reqId, Map The router's identifier is {@code chatId} when present (group context)
+ * and {@code senderId} otherwise (1:1) — there is no {@code group:} infix
+ * because the router doesn't add one. An earlier divergence here ({@code
+ * "wecom:group:" + chatId}) persisted media at
+ * {@code data/chat-uploads/wecom:group:{chatId}/} and stamped that path
+ * onto {@link MessageContentPart#getFileUrl()}, but the message's actual
+ * conversationId in {@code mate_conversation} was {@code wecom:{chatId}}
+ * (no {@code group:} infix). The {@code /api/v1/chat/files/...} endpoint
+ * then ran {@code isConversationOwner("wecom:group:...")}, found no
+ * matching row, returned 403, and the IM client rendered every
+ * group-quoted image as a broken icon.
+ */
+ private static String inboundConversationId(String senderId, String chatId, String chatType) {
+ boolean isGroup = "group".equals(chatType);
+ return isGroup ? "wecom:" + chatId : "wecom:" + senderId;
+ }
+
+ // ==================== 引用消息（quote）解析 ====================
+
+ /**
+ * Quoted-message extraction result: a human-readable prefix string the
+ * agent prompt prepends, plus any media (image / file) that was quoted
+ * and needs to be available as a content part. Either field may be
+ * empty; {@link #isEmpty()} returns true only when both are.
+ */
+ private record QuoteContext(String prefix, List
+ *
+ *
+ *
+ *
*/
public Set
+ *
*/
public void setToolBindings(Long agentId, List
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ * Both cases are normalized to {@code "{}"} so the chat-completions
+ * round-trip stays valid. Tool execution downstream still re-validates
+ * arguments and surfaces a per-tool error if the empty payload is wrong
+ * for that tool.
+ */
+ private static String sanitizeToolCallArguments(String toolName, String arguments) {
+ if (arguments == null || arguments.isBlank()) {
+ return "{}";
+ }
+ try {
+ TOOL_ARG_JSON_MAPPER.readTree(arguments);
+ return arguments;
+ } catch (Exception e) {
+ log.warn("Tool '{}' arguments are not valid JSON after stream aggregation "
+ + "(len={}, head={}); replacing with empty object so the "
+ + "follow-up chat-completions request stays well-formed. "
+ + "Parse error: {}",
+ toolName,
+ arguments.length(),
+ arguments.substring(0, Math.min(80, arguments.length())),
+ e.getMessage());
+ return "{}";
+ }
+ }
+
private static class ToolCallAccumulator {
String id;
String type;
diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java
index aa040600..f845d9a3 100644
--- a/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java
+++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/StateGraphReActAgent.java
@@ -198,7 +198,7 @@ public class StateGraphReActAgent extends BaseAgent implements StructuredStreamC
AtomicInteger lastSoftCap = new AtomicInteger(0);
AtomicBoolean sawLegitimateExit = new AtomicBoolean(false);
- return compiledGraph.stream(inputs, config)
+ return BaseAgent.routingStartupDelta(inputs).concatWith(compiledGraph.stream(inputs, config)
.flatMapIterable(output -> {
List
- * raw tool result
- * → truncateToolResult(..., MAX_TOOL_RESULT_CHARS=8000) // Layer 1: hard cap
- * → persistIfOversized(..., perResultThresholdChars=16000) // Layer 2: spill to disk
- * → enforceTurnBudget(..., perTurnBudgetChars=32000) // Layer 3: per-turn aggregate
+ * raw tool result (full bytes)
+ * → spillRawOrTruncate(...)
+ * ├─ persistIfOversized(...) tries to write the raw body to disk
+ * │ when size > perResultThresholdChars and tool is not
+ * │ in the spill exclusion list. Returns a SPILL_MARKER preview
+ * │ on success, or the original string otherwise.
+ * └─ if no SPILL_MARKER on the return, truncateToolResult(...)
+ * caps inline to MAX_TOOL_RESULT_CHARS so a multi-MB raw
+ * body never enters the model prompt.
+ * → enforceTurnBudget(..., perTurnBudgetChars=32000) // per-turn aggregate
*
- * Layer 1 runs first and is intentionally kept at 8000 to prevent oversized
- * results from inflating the prompt. Layers 2/3 thresholds are configured in
- * {@link ToolResultProperties} and application.yml.
+ * Spill must see the RAW result so the full output is preserved on disk
+ * and the model can call {@code read_file} on the spill path. Truncating
+ * before spilling would write a pre-shortened blob to disk, defeating the
+ * "ground truth on disk" guarantee. {@link ToolResultProperties} controls
+ * the thresholds; this constant stays in code because it is the safety
+ * net for the failure case and should not vary by deployment.
*/
private static final int MAX_TOOL_RESULT_CHARS = 8000;
+ /**
+ * Raw-first spill: try to write the full result to disk via the spill
+ * store; only fall back to inline hard-truncate when no spill marker
+ * comes back. Caller distinguishes spill success from "returned
+ * unchanged" by checking {@link ToolResultStorage#SPILL_MARKER_PREFIX}
+ * on the returned string — otherwise an IO failure or under-threshold
+ * body would slip through indistinguishable from a successful spill,
+ * and a multi-MB raw body could end up in the model prompt.
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ */
+public record WorkflowApprovalResolvedEvent(
+ long approvalRowId,
+ String pendingId,
+ String decision,
+ Long workspaceId
+) {}
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java
index fbbb3e78..52ea9cd4 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/AbstractChannelAdapter.java
@@ -319,6 +319,27 @@ public abstract class AbstractChannelAdapter implements ChannelAdapter {
}
}
+ /**
+ * Approval notice rendering — primary implementation position.
+ *
+ *
+ * ApprovalNotice notice = approvalNotificationService.buildNotice(pending);
+ * adapter.sendApprovalNotice(replyTarget, notice);
+ *
+ */
+ default void sendApprovalNotice(String targetId,
+ vip.mate.channel.notification.ApprovalNotice notice) {
+ sendMessage(targetId,
+ vip.mate.channel.notification.ApprovalNotificationService.staticBuildText(notice));
+ }
+
// ==================== 主动推送 ====================
/**
@@ -152,6 +199,34 @@ public interface ChannelAdapter {
return getChannelType();
}
+ /**
+ * Whether this adapter must run on exactly one node in a multi-instance
+ * deployment.
+ *
+ *
+ *
+ *
+ *
+ *
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class ChannelLeaderElection {
+
+ /**
+ * How long the lock is held without a renewal. A failed node's lease
+ * stays locked for this long before another node can take over —
+ * so longer values increase failover latency, shorter values increase
+ * the risk of false handover during a GC pause or DB hiccup.
+ */
+ public static final Duration LOCK_AT_MOST_FOR = Duration.ofSeconds(60);
+
+ private final LockProvider lockProvider;
+
+ /**
+ * Attempt to acquire leadership for the given key.
+ *
+ * @param key a stable identifier for the resource (e.g.
+ * {@code "feishu:42"}). Used verbatim as the underlying
+ * lock name (prefixed by this class to avoid collisions
+ * with other lock users).
+ * @return an empty optional if another node already holds the lease,
+ * otherwise a {@link LeaderLease} that the caller is
+ * responsible for periodically extending and finally
+ * releasing.
+ */
+ public Optional
+ *
+ *
+ *
+ *
+ *
+ *
+ *
+ * Without this layer, oversized uploads would chunk-upload for up to
+ * a minute before the platform server rejected them at the finish
+ * step, with the user seeing nothing arrive in their chat.
+ *
+ *
+ *
+ *
+ *
+ * Returns null when {@code body.quote} is absent / empty / malformed —
+ * the caller treats that the same as "no quote context".
+ */
+ private QuoteContext extractQuoteContext(Map