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/ChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelAdapter.java index 73f62d1b..fcf6b292 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelAdapter.java @@ -95,6 +95,53 @@ public interface ChannelAdapter { sendMessage(targetId, content); } + /** + * Extended render-and-send overload that carries an optional + * {@link SendContext} side-channel (e.g. WeCom AI Bot + * {@code feedback.id} for like/dislike collection). + * + *
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: + *
+ * 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)); + } + // ==================== 主动推送 ==================== /** 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..3440dc0d 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelManager.java @@ -46,6 +46,17 @@ 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 will switch 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; + /** 运行中的渠道适配器：channelId -> adapter */ private final Map
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 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/wecom/WeComChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java
index 2aefe81f..1395bfb9 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/wecom/WeComChannelAdapter.java
@@ -114,18 +114,54 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
/** 消息去重集合 */
private final Set {@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);
+
+ /**
+ * Per-reqId 回复队列状态。
+ *
+ * @param queue 串行回复任务队列
+ * @param closed 防御性 closed 标志（详见 {@link #replyQueues} 注释）
+ */
+ private record ReplyQueueState(
+ LinkedBlockingQueue Wired in PR-0 for forward compatibility. PR-1 will use this to
+ * build {@code button_interaction} card payloads for tool-guard
+ * approvals; PR-0 keeps the field but does not consume it (the
+ * default text path on {@link AbstractChannelAdapter#sendApprovalNotice}
+ * still handles approvals). {@code null}-tolerant: if Spring DI
+ * cannot find the bean (unit-test contexts, hot-swap edge cases,
+ * etc.) the adapter still functions in PR-0 fashion.
+ */
+ @SuppressWarnings("unused") // consumed in PR-1
+ private final vip.mate.channel.notification.ApprovalNotificationService approvalNotificationService;
+
public WeComChannelAdapter(ChannelEntity channelEntity,
ChannelMessageRouter messageRouter,
- ObjectMapper objectMapper) {
+ ObjectMapper objectMapper,
+ vip.mate.channel.notification.ApprovalNotificationService approvalNotificationService) {
super(channelEntity, messageRouter, objectMapper);
+ this.approvalNotificationService = approvalNotificationService;
// 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 +237,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 +268,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 +291,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,10 +325,49 @@ 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();
missedPongCount.set(0);
@@ -264,6 +375,119 @@ public class WeComChannelAdapter extends AbstractChannelAdapter {
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(60, TimeUnit.SECONDS);
+ } 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 +578,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();
}
/**
@@ -1063,10 +1291,33 @@ 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) {
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
+ *
+ *
+ *
+ *
+ *
+ *