:send a message

channel-plugin/openclaw.plugin.json

@@ -138,6 +138,60 @@
               "selfWxid": {
                 "type": "string",
                 "description": "The bot's own wxid, used to filter self-sent echoes. If unset, the plugin will try to detect via checkOnline."
+              },
+              "technical": {
+                "type": "object",
+                "description": "Optional technical-layer config. Values here override same-name top-level fields.",
+                "additionalProperties": true,
+                "properties": {
+                  "apiBase": { "type": "string" },
+                  "pollIntervalMs": { "type": "integer", "minimum": 3000, "maximum": 120000 },
+                  "batchWindowMs": { "type": "integer", "minimum": 0, "maximum": 30000 },
+                  "replyCooldownSecPerWxid": { "type": "integer", "minimum": 0, "maximum": 3600 },
+                  "outboundRatePerSec": {
+                    "type": "integer",
+                    "minimum": 1,
+                    "maximum": 100,
+                    "description": "Global outbound rate limiter for all send paths."
+                  },
+                  "outboundQueueMax": {
+                    "type": "integer",
+                    "minimum": 10,
+                    "maximum": 5000,
+                    "description": "Maximum buffered outbound tasks before rejecting new ones."
+                  },
+                  "metricsLogIntervalSec": {
+                    "type": "integer",
+                    "minimum": 30,
+                    "maximum": 3600,
+                    "description": "Base interval for summary metrics logs. Recommended: 60 (debug), 120~180 (normal), 300 (stable high throughput)."
+                  },
+                  "highThroughputOutboundAttemptsPer5Min": {
+                    "type": "integer",
+                    "minimum": 50,
+                    "maximum": 50000,
+                    "description": "Auto-switch metrics summary interval to 300s when outbound attempts in 5 minutes exceed this threshold."
+                  }
+                }
+              },
+              "business": {
+                "type": "object",
+                "description": "Optional business-layer config. Values here override same-name top-level fields.",
+                "additionalProperties": true,
+                "properties": {
+                  "groupEnabled": { "type": "boolean" },
+                  "groupKeywords": { "type": "array", "items": { "type": "string" } },
+                  "groupMentionTokens": { "type": "array", "items": { "type": "string" } },
+                  "groupReplyAtSender": { "type": "boolean" },
+                  "dmPolicy": { "type": "string", "enum": ["open", "allowlist", "pairing"] },
+                  "allowFrom": { "type": "array", "items": { "type": "string" } },
+                  "ignoreWxidPrefixes": { "type": "array", "items": { "type": "string" } },
+                  "ignoreWxidExact": { "type": "array", "items": { "type": "string" } },
+                  "ignoreMessageTypes": { "type": "array", "items": { "type": "string" } },
+                  "faq": { "type": "array" },
+                  "defaultTo": { "type": "string" },
+                  "selfWxid": { "type": "string" }
+                }
               }
             }
           }

channel-plugin/src/channel.ts

@@ -31,7 +31,10 @@ import {
   normalizeAllowEntry,
   normalizeAllowFrom,
   resolveAccount,
+  resolveAllowFrom,
   resolveApiBase,
+  resolveDefaultTo,
+  resolveDmPolicy,
 } from "./config.js";
 import { createClient } from "./client.js";
 import type { ResolvedWechatAgentAccount, WechatAgentConfig } from "./types.js";
@@ -118,13 +121,13 @@ export const wechatAgentPlugin: ChannelPlugin<ResolvedWechatAgentAccount> = {
     }),
 
     resolveAllowFrom: ({ cfg }: { cfg: OpenClawConfig }) =>
-      getChannelCfg(cfg)?.allowFrom ?? [],
+      resolveAllowFrom(getChannelCfg(cfg)) ?? [],
 
     formatAllowFrom: ({ allowFrom }: { allowFrom: Array<string | number> }) =>
       normalizeAllowFrom(allowFrom),
 
     resolveDefaultTo: ({ cfg }: { cfg: OpenClawConfig }) =>
-      getChannelCfg(cfg)?.defaultTo?.trim() || undefined,
+      resolveDefaultTo(getChannelCfg(cfg))?.trim() || undefined,
   },
 
   // --- Pairing adapter ---
@@ -150,7 +153,7 @@ export const wechatAgentPlugin: ChannelPlugin<ResolvedWechatAgentAccount> = {
       if (!channelCfg) return [];
 
       const warnings: string[] = [];
-      const dmPolicy = channelCfg.dmPolicy ?? "open";
+      const dmPolicy = resolveDmPolicy(channelCfg) ?? "open";
       if (dmPolicy === "open") {
         warnings.push(
           `- wechat-agent: dmPolicy="open" lets ANY wxid (including strangers who add the bot) trigger the agent. Set channels.${CHANNEL_ID}.dmPolicy="allowlist" + channels.${CHANNEL_ID}.allowFrom to restrict.`
@@ -232,7 +235,7 @@ export const wechatAgentPlugin: ChannelPlugin<ResolvedWechatAgentAccount> = {
     }) => {
       const q = query?.trim().toLowerCase() ?? "";
       const channelCfg = getChannelCfg(cfg);
-      const ids = normalizeAllowFrom(channelCfg?.allowFrom);
+      const ids = normalizeAllowFrom(resolveAllowFrom(channelCfg));
       const cap = typeof limit === "number" && limit > 0 ? limit : undefined;
 
       return ids
@@ -309,8 +312,8 @@ export const wechatAgentPlugin: ChannelPlugin<ResolvedWechatAgentAccount> = {
       lastInboundAt: runtime?.lastInboundAt ?? null,
       lastOutboundAt: runtime?.lastOutboundAt ?? null,
       lastEventAt: runtime?.lastEventAt,
-      dmPolicy: getChannelCfg(cfg)?.dmPolicy ?? runtime?.dmPolicy,
-      allowFrom: normalizeAllowFrom(getChannelCfg(cfg)?.allowFrom),
+      dmPolicy: resolveDmPolicy(getChannelCfg(cfg)) ?? runtime?.dmPolicy,
+      allowFrom: normalizeAllowFrom(resolveAllowFrom(getChannelCfg(cfg))),
       baseUrl: account.apiBase,
       probe,
       audit,

channel-plugin/src/client.test.ts

@@ -148,6 +148,20 @@ describe("createClient", () => {
       expect(msgs[0].isGroup).toBe(true);
       expect(msgs[0].chatroomMemberWxid).toBe("wxid_alice");
     });
+
+    it("passes incremental since cursor query params when provided", async () => {
+      stubFetch(async (input) => {
+        const url = String(input);
+        expect(url).toContain("sinceTimestampMs=1710000123");
+        expect(url).toContain("sinceMsgId=m99");
+        return jsonResponse({ ret: 200, data: [] });
+      });
+      const c = createClient("http://x");
+      await c.getMessages({
+        sinceTimestampMs: 1710000123,
+        sinceMsgId: "m99",
+      });
+    });
   });
 
   describe("sendText", () => {

channel-plugin/src/client.ts

@@ -37,6 +37,9 @@ export type GetMessagesOptions = {
   signal?: AbortSignal;
   /** Per-request timeout (ms). Default 15s. */
   timeoutMs?: number;
+  /** Optional incremental cursor fields (backend may ignore unknown params). */
+  sinceTimestampMs?: number;
+  sinceMsgId?: string;
 };
 
 export type SendTextOptions = {
@@ -139,6 +142,12 @@ export function createClient(apiBase: string): WechatAgentClient {
       const limit = opts.limit ?? 50;
       const direction = opts.direction ?? "received";
       const qs = new URLSearchParams({ limit: String(limit), direction });
+      if (typeof opts.sinceTimestampMs === "number" && Number.isFinite(opts.sinceTimestampMs)) {
+        qs.set("sinceTimestampMs", String(Math.floor(opts.sinceTimestampMs)));
+      }
+      if (opts.sinceMsgId?.trim()) {
+        qs.set("sinceMsgId", opts.sinceMsgId.trim());
+      }
       const body = (await request(`/messages?${qs.toString()}`, {
         method: "GET",
         signal: opts.signal,

channel-plugin/src/config.ts

@@ -12,6 +12,10 @@ export const CHANNEL_ID = "wechat-agent" as const;
 
 /** Default apiBase used when config is missing. Matches the customer's current backend. */
 export const DEFAULT_API_BASE = "http://127.0.0.1:8138";
+export const DEFAULT_OUTBOUND_RATE_PER_SEC = 5;
+export const DEFAULT_OUTBOUND_QUEUE_MAX = 200;
+export const DEFAULT_METRICS_LOG_INTERVAL_SEC = 180;
+export const DEFAULT_HIGH_THROUGHPUT_OUTBOUND_ATTEMPTS_PER_5MIN = 300;
 
 /** Safe read of the wechat-agent channel config block. */
 export function getChannelCfg(cfg: OpenClawConfig): WechatAgentConfig | undefined {
@@ -20,19 +24,140 @@ export function getChannelCfg(cfg: OpenClawConfig): WechatAgentConfig | undefine
     | undefined;
 }
 
+function pick<T>(legacy: T | undefined, layered: T | undefined): T | undefined {
+  return layered ?? legacy;
+}
+
 /** Resolve the effective apiBase (config + fallback). Strips trailing slash. */
 export function resolveApiBase(channelCfg: WechatAgentConfig | undefined): string {
-  const raw = (channelCfg?.apiBase ?? DEFAULT_API_BASE).trim();
+  const raw = (
+    pick(channelCfg?.apiBase, channelCfg?.technical?.apiBase) ?? DEFAULT_API_BASE
+  ).trim();
   return raw.replace(/\/+$/, "");
 }
 
 /** True if all required fields are present. */
 export function isConfigured(channelCfg: WechatAgentConfig | undefined): boolean {
   if (!channelCfg) return false;
-  const apiBase = channelCfg.apiBase?.trim();
+  const apiBase = pick(channelCfg.apiBase, channelCfg.technical?.apiBase)?.trim();
   return Boolean(apiBase);
 }
 
+export function resolvePollIntervalMs(channelCfg: WechatAgentConfig | undefined): number | undefined {
+  return pick(channelCfg?.pollIntervalMs, channelCfg?.technical?.pollIntervalMs);
+}
+
+export function resolveBatchWindowMs(channelCfg: WechatAgentConfig | undefined): number | undefined {
+  return pick(channelCfg?.batchWindowMs, channelCfg?.technical?.batchWindowMs);
+}
+
+export function resolveReplyCooldownSecPerWxid(
+  channelCfg: WechatAgentConfig | undefined
+): number | undefined {
+  return pick(
+    channelCfg?.replyCooldownSecPerWxid,
+    channelCfg?.technical?.replyCooldownSecPerWxid
+  );
+}
+
+export function resolveOutboundRatePerSec(channelCfg: WechatAgentConfig | undefined): number {
+  const rate = pick<number | undefined>(
+    undefined,
+    channelCfg?.technical?.outboundRatePerSec
+  );
+  if (!rate || !Number.isFinite(rate) || rate <= 0) return DEFAULT_OUTBOUND_RATE_PER_SEC;
+  return rate;
+}
+
+export function resolveOutboundQueueMax(channelCfg: WechatAgentConfig | undefined): number {
+  const size = pick<number | undefined>(undefined, channelCfg?.technical?.outboundQueueMax);
+  if (!size || !Number.isFinite(size) || size <= 0) return DEFAULT_OUTBOUND_QUEUE_MAX;
+  return Math.floor(size);
+}
+
+export function resolveMetricsLogIntervalSec(channelCfg: WechatAgentConfig | undefined): number {
+  const sec = pick<number | undefined>(undefined, channelCfg?.technical?.metricsLogIntervalSec);
+  if (!sec || !Number.isFinite(sec) || sec <= 0) return DEFAULT_METRICS_LOG_INTERVAL_SEC;
+  return Math.floor(sec);
+}
+
+export function resolveHighThroughputOutboundAttemptsPer5Min(
+  channelCfg: WechatAgentConfig | undefined
+): number {
+  const count = pick<number | undefined>(
+    undefined,
+    channelCfg?.technical?.highThroughputOutboundAttemptsPer5Min
+  );
+  if (!count || !Number.isFinite(count) || count <= 0) {
+    return DEFAULT_HIGH_THROUGHPUT_OUTBOUND_ATTEMPTS_PER_5MIN;
+  }
+  return Math.floor(count);
+}
+
+export function resolveGroupEnabled(channelCfg: WechatAgentConfig | undefined): boolean | undefined {
+  return pick(channelCfg?.groupEnabled, channelCfg?.business?.groupEnabled);
+}
+
+export function resolveGroupKeywords(
+  channelCfg: WechatAgentConfig | undefined
+): string[] | undefined {
+  return pick(channelCfg?.groupKeywords, channelCfg?.business?.groupKeywords);
+}
+
+export function resolveGroupMentionTokens(
+  channelCfg: WechatAgentConfig | undefined
+): string[] | undefined {
+  return pick(channelCfg?.groupMentionTokens, channelCfg?.business?.groupMentionTokens);
+}
+
+export function resolveGroupReplyAtSender(
+  channelCfg: WechatAgentConfig | undefined
+): boolean | undefined {
+  return pick(channelCfg?.groupReplyAtSender, channelCfg?.business?.groupReplyAtSender);
+}
+
+export function resolveDmPolicy(
+  channelCfg: WechatAgentConfig | undefined
+): "open" | "allowlist" | "pairing" | undefined {
+  return pick(channelCfg?.dmPolicy, channelCfg?.business?.dmPolicy);
+}
+
+export function resolveAllowFrom(
+  channelCfg: WechatAgentConfig | undefined
+): Array<string | number> | undefined {
+  return pick(channelCfg?.allowFrom, channelCfg?.business?.allowFrom);
+}
+
+export function resolveIgnoreWxidPrefixes(
+  channelCfg: WechatAgentConfig | undefined
+): string[] | undefined {
+  return pick(channelCfg?.ignoreWxidPrefixes, channelCfg?.business?.ignoreWxidPrefixes);
+}
+
+export function resolveIgnoreWxidExact(
+  channelCfg: WechatAgentConfig | undefined
+): string[] | undefined {
+  return pick(channelCfg?.ignoreWxidExact, channelCfg?.business?.ignoreWxidExact);
+}
+
+export function resolveIgnoreMessageTypes(
+  channelCfg: WechatAgentConfig | undefined
+): string[] | undefined {
+  return pick(channelCfg?.ignoreMessageTypes, channelCfg?.business?.ignoreMessageTypes);
+}
+
+export function resolveFaq(channelCfg: WechatAgentConfig | undefined) {
+  return pick(channelCfg?.faq, channelCfg?.business?.faq);
+}
+
+export function resolveDefaultTo(channelCfg: WechatAgentConfig | undefined): string | undefined {
+  return pick(channelCfg?.defaultTo, channelCfg?.business?.defaultTo);
+}
+
+export function resolveSelfWxid(channelCfg: WechatAgentConfig | undefined): string | undefined {
+  return pick(channelCfg?.selfWxid, channelCfg?.business?.selfWxid);
+}
+
 /** Normalized allowlist (lowercased, trimmed, deduped). */
 export function normalizeAllowFrom(
   entries: Array<string | number> | undefined

channel-plugin/src/monitor.ts

@@ -37,7 +37,25 @@ import {
   CHANNEL_ID,
   getChannelCfg,
   normalizeAllowFrom,
+  resolveAllowFrom,
   resolveApiBase,
+  resolveBatchWindowMs,
+  resolveDmPolicy,
+  resolveFaq,
+  resolveGroupEnabled,
+  resolveGroupKeywords,
+  resolveGroupMentionTokens,
+  resolveGroupReplyAtSender,
+  resolveIgnoreMessageTypes,
+  resolveIgnoreWxidExact,
+  resolveIgnoreWxidPrefixes,
+  resolveHighThroughputOutboundAttemptsPer5Min,
+  resolveMetricsLogIntervalSec,
+  resolveOutboundQueueMax,
+  resolveOutboundRatePerSec,
+  resolvePollIntervalMs,
+  resolveReplyCooldownSecPerWxid,
+  resolveSelfWxid,
 } from "./config.js";
 import type { Logger, RawWechatMessage, WechatAgentConfig } from "./types.js";
 import {
@@ -51,6 +69,15 @@ import {
 } from "./cursor-store.js";
 import { join } from "node:path";
 import { withRetry } from "./retry.js";
+import { configureOutboundQueue, enqueueOutboundSend } from "./outbound-queue.js";
+import {
+  getOutboundMetricsSnapshot,
+  recordOutboundAttempt,
+  recordOutboundFailure,
+  recordOutboundSuccess,
+  shouldEmitOutboundFailureAlert,
+  shouldSkipDuplicateSend,
+} from "./outbound-observability.js";
 
 export type MonitorOpts = {
   cfg: OpenClawConfig;
@@ -73,6 +100,10 @@ const SEEN_MAX = 2000;
 const SEEN_TRIM_TO = 1000;
 const DEFAULT_GROUP_KEYWORDS = ["帮我", "请问"];
 const DEFAULT_GROUP_MENTION_TOKENS = ["@bot", "@助手"];
+const POLL_ALERT_MIN_TICKS = 10;
+const POLL_ALERT_FAIL_RATE = 0.5;
+const HIGH_THROUGHPUT_LOG_INTERVAL_MS = 300 * 1000;
+const THROUGHPUT_WINDOW_MS = 5 * 60 * 1000;
 
 export async function startMonitor(opts: MonitorOpts): Promise<MonitorResult> {
   const channelCfg = getChannelCfg(opts.cfg);
@@ -95,10 +126,19 @@ export async function startMonitor(opts: MonitorOpts): Promise<MonitorResult> {
   const client = createClient(apiBase);
   const pollIntervalMs = Math.max(
     POLL_MIN_INTERVAL_MS,
-    channelCfg.pollIntervalMs ?? 10000
+    resolvePollIntervalMs(channelCfg) ?? 10000
+  );
+  const batchWindowMs = Math.max(0, resolveBatchWindowMs(channelCfg) ?? 3000);
+  const replyCooldownMs = Math.max(0, (resolveReplyCooldownSecPerWxid(channelCfg) ?? 5) * 1000);
+  configureOutboundQueue(
+    {
+      ratePerSec: resolveOutboundRatePerSec(channelCfg),
+      maxQueue: resolveOutboundQueueMax(channelCfg),
+    },
+    log
   );
-  const batchWindowMs = Math.max(0, channelCfg.batchWindowMs ?? 3000);
-  const replyCooldownMs = Math.max(0, (channelCfg.replyCooldownSecPerWxid ?? 5) * 1000);
+  const baseMetricsLogIntervalMs = Math.max(30 * 1000, resolveMetricsLogIntervalSec(channelCfg) * 1000);
+  const highThroughputAttemptsThreshold = resolveHighThroughputOutboundAttemptsPer5Min(channelCfg);
 
   // Persistent inbound cursor: best-effort resume after restarts.
   // We store it alongside OpenClaw's session store so admins can back it up
@@ -114,7 +154,7 @@ export async function startMonitor(opts: MonitorOpts): Promise<MonitorResult> {
   let cursor: InboundCursor | null = await loadCursor(cursorPath);
 
   // Resolve bot's own wxid so we can filter self-echoes.
-  let selfWxid: string | null = channelCfg.selfWxid?.trim() || null;
+  let selfWxid: string | null = resolveSelfWxid(channelCfg)?.trim() || null;
   if (!selfWxid) {
     try {
       selfWxid = await client.getSelfWxid();
@@ -143,6 +183,10 @@ export async function startMonitor(opts: MonitorOpts): Promise<MonitorResult> {
     outboundCount: 0,
     faqHits: 0,
   };
+  let metricsLastLogAt = Date.now();
+  let throughputWindowStartAt = Date.now();
+  let throughputWindowLastAttempts = 0;
+  let adaptiveMetricsLogIntervalMs = baseMetricsLogIntervalMs;
 
   // Message dedupe window
   const seenMsgIds = new Set<string>();
@@ -203,8 +247,11 @@ export async function startMonitor(opts: MonitorOpts): Promise<MonitorResult> {
         direction: "received",
         limit: 50,
         signal: opts.abortSignal,
+        sinceTimestampMs: cursor?.lastTimestampMs,
+        sinceMsgId: cursor?.lastMsgId,
       });
       counters.pollsOk++;
+      maybeEmitHealthLogs();
 
       // Always process in stable order so agent context is consistent.
       messages.sort((a, b) => {
@@ -235,7 +282,7 @@ export async function startMonitor(opts: MonitorOpts): Promise<MonitorResult> {
         opts.patchStatus({ lastInboundAt: Date.now() });
 
         // FAQ fast-path
-        const match = matchFaq(m.content, channelCfg.faq);
+        const match = matchFaq(m.content, resolveFaq(channelCfg));
         if (match) {
           counters.faqHits++;
           log.info(
@@ -271,11 +318,52 @@ export async function startMonitor(opts: MonitorOpts): Promise<MonitorResult> {
       const msg = err instanceof Error ? err.message : String(err);
       opts.patchStatus({ lastError: msg });
       log.warn(`poll error: ${msg}`);
+      maybeEmitHealthLogs();
     } finally {
       pollInFlight = false;
     }
   };
 
+  const maybeEmitHealthLogs = () => {
+    const now = Date.now();
+    const ticks = counters.pollsOk + counters.pollsFail;
+    const outboundNow = getOutboundMetricsSnapshot();
+    if (now - throughputWindowStartAt >= THROUGHPUT_WINDOW_MS) {
+      const attemptsInWindow = outboundNow.attempts - throughputWindowLastAttempts;
+      throughputWindowStartAt = now;
+      throughputWindowLastAttempts = outboundNow.attempts;
+      const nextInterval =
+        attemptsInWindow >= highThroughputAttemptsThreshold
+          ? Math.max(baseMetricsLogIntervalMs, HIGH_THROUGHPUT_LOG_INTERVAL_MS)
+          : baseMetricsLogIntervalMs;
+      if (nextInterval !== adaptiveMetricsLogIntervalMs) {
+        adaptiveMetricsLogIntervalMs = nextInterval;
+        log.info(
+          `metrics interval adjusted to ${Math.round(adaptiveMetricsLogIntervalMs / 1000)}s (outboundAttempts5m=${attemptsInWindow}, threshold=${highThroughputAttemptsThreshold})`
+        );
+      }
+    }
+    if (now - metricsLastLogAt >= adaptiveMetricsLogIntervalMs) {
+      metricsLastLogAt = now;
+      log.info(
+        `metrics pollsOk=${counters.pollsOk} pollsFail=${counters.pollsFail} inbound=${counters.inboundCount} outbound=${counters.outboundCount} faqHits=${counters.faqHits} outAttempts=${outboundNow.attempts} outSuccess=${outboundNow.success} outFailed=${outboundNow.failed} outDeduped=${outboundNow.deduped} outAvgLatencyMs=${outboundNow.avgLatencyMs} metricsIntervalSec=${Math.round(adaptiveMetricsLogIntervalMs / 1000)}`
+      );
+    }
+    if (ticks >= POLL_ALERT_MIN_TICKS) {
+      const failRate = counters.pollsFail / ticks;
+      if (failRate >= POLL_ALERT_FAIL_RATE) {
+        log.warn(
+          `[ALERT] high poll failure rate fail=${counters.pollsFail} total=${ticks} ratio=${failRate.toFixed(2)}`
+        );
+      }
+    }
+    if (shouldEmitOutboundFailureAlert()) {
+      log.warn(
+        `[ALERT] outbound failure rate high attempts=${outboundNow.attempts} failed=${outboundNow.failed} ratio=${outboundNow.failRate.toFixed(2)}`
+      );
+    }
+  };
+
   const timer = setInterval(() => {
     tick().catch((err) => log.error(`tick crashed: ${String(err)}`));
   }, pollIntervalMs);
@@ -460,32 +548,46 @@ async function dispatchBatchToAgent(params: DispatchParams): Promise<void> {
 
         for (const chunk of parts) {
           if (!chunk) continue;
-          const result = await withRetry(
-            () =>
-              client.sendText({
-                toWxid: conversationId,
-                content: chunk,
-                ats:
-                  isGroup && channelCfg.groupReplyAtSender && senderWxid ? senderWxid : "",
-              }),
-            {
-              tries: 4,
-              baseDelayMs: 500,
-              maxDelayMs: 4000,
-              isRetryable: (err) => {
-                const msg = err instanceof Error ? err.message : String(err);
-                if (/timeout|aborted|ECONN|ENOTFOUND|EAI_AGAIN/i.test(msg)) return true;
-                if (/HTTP 5\d\d/i.test(msg)) return true;
-                return false;
+          if (shouldSkipDuplicateSend("agent-reply", conversationId, chunk)) {
+            log.warn(`skip duplicate agent reply conversation=${conversationId}`);
+            continue;
+          }
+          const startedAt = Date.now();
+          recordOutboundAttempt("agent-reply");
+          try {
+            await enqueueOutboundSend(`agent-reply:${conversationId}`, () =>
+              withRetry(
+              async () => {
+                const res = await client.sendText({
+                  toWxid: conversationId,
+                  content: chunk,
+                  ats:
+                    isGroup && resolveGroupReplyAtSender(channelCfg) && senderWxid
+                      ? senderWxid
+                      : "",
+                });
+                if (!res.ok) {
+                  throw new Error(res.msg ?? `ret=${String(res.ret)}`);
+                }
+                return res;
               },
-            }
-          );
-          if (!result.ok) {
-            log.error(`send-text failed ret=${String(result.ret)} msg=${result.msg ?? ""}`);
-            throw new Error(
-              `send-text failed: ${result.msg ?? `ret=${String(result.ret)}`}`
-            );
+              {
+                tries: 4,
+                baseDelayMs: 500,
+                maxDelayMs: 4000,
+                isRetryable: (err) => {
+                  const msg = err instanceof Error ? err.message : String(err);
+                  if (/timeout|aborted|ECONN|ENOTFOUND|EAI_AGAIN/i.test(msg)) return true;
+                  if (/HTTP 5\d\d/i.test(msg)) return true;
+                  return false;
+                },
+              }
+            ));
+          } catch (err) {
+            recordOutboundFailure("agent-reply");
+            throw err;
           }
+          recordOutboundSuccess("agent-reply", conversationId, chunk, Date.now() - startedAt);
         }
         log.info(`delivered reply to ${conversationId} (${text.length} chars)`);
         onOutboundSent();
@@ -521,14 +623,25 @@ async function sendFaqReply(
 ): Promise<void> {
   const conversationId = resolveConversationId(m);
   const senderWxid = resolveSenderWxid(m);
-  const result = await client.sendText({
-    toWxid: conversationId,
-    content: reply,
-    ats: m.isGroup && cfg.groupReplyAtSender && senderWxid ? senderWxid : "",
-  });
+  if (shouldSkipDuplicateSend("faq", conversationId, reply)) {
+    log.warn(`skip duplicate faq reply to ${conversationId}`);
+    return;
+  }
+  recordOutboundAttempt("faq");
+  const startedAt = Date.now();
+  const result = await enqueueOutboundSend(`faq:${conversationId}`, () =>
+    client.sendText({
+      toWxid: conversationId,
+      content: reply,
+      ats:
+        m.isGroup && resolveGroupReplyAtSender(cfg) && senderWxid ? senderWxid : "",
+    })
+  );
   if (result.ok) {
+    recordOutboundSuccess("faq", conversationId, reply, Date.now() - startedAt);
     log.info(`faq reply sent to ${conversationId}`);
   } else {
+    recordOutboundFailure("faq");
     log.warn(
       `faq reply failed to ${conversationId}: ret=${String(result.ret)} msg=${result.msg ?? ""}`
     );
@@ -547,23 +660,23 @@ function shouldIgnoreMessage(
   if (selfWxid && m.chatroomMemberWxid === selfWxid) return "self-group-echo";
 
   // Group switch
-  if (m.isGroup && cfg.groupEnabled === false) return "group-disabled";
+  if (m.isGroup && resolveGroupEnabled(cfg) === false) return "group-disabled";
 
   // Type filter
-  const ignoreTypes = cfg.ignoreMessageTypes ?? [];
+  const ignoreTypes = resolveIgnoreMessageTypes(cfg) ?? [];
   if (ignoreTypes.includes(m.type)) return `ignored-type:${m.type}`;
 
   // v0.2: inbound still text-only
   if (m.type !== "text") return "non-text";
 
   // Wxid prefix filter (official accounts etc.)
-  const prefixes = cfg.ignoreWxidPrefixes ?? [];
+  const prefixes = resolveIgnoreWxidPrefixes(cfg) ?? [];
   for (const p of prefixes) {
     if (p && m.fromWxid.startsWith(p)) return `ignored-prefix:${p}`;
   }
 
   // Exact wxid filter (system contacts)
-  const exacts = cfg.ignoreWxidExact ?? [];
+  const exacts = resolveIgnoreWxidExact(cfg) ?? [];
   if (exacts.includes(m.fromWxid)) return `ignored-wxid:${m.fromWxid}`;
 
   // Empty content
@@ -573,10 +686,10 @@ function shouldIgnoreMessage(
 }
 
 function passesDmPolicy(m: RawWechatMessage, cfg: WechatAgentConfig): boolean {
-  const dmPolicy = cfg.dmPolicy ?? "open";
+  const dmPolicy = resolveDmPolicy(cfg) ?? "open";
   if (dmPolicy === "open") return true;
 
-  const allowFrom = normalizeAllowFrom(cfg.allowFrom);
+  const allowFrom = normalizeAllowFrom(resolveAllowFrom(cfg));
   if (allowFrom.includes(m.fromWxid.toLowerCase())) return true;
 
   // pairing is handled upstream in core; here we treat it like allowlist.
@@ -603,13 +716,15 @@ function isGroupTriggered(
 ): boolean {
   if (!m.isGroup) return true;
   const content = m.content.toLowerCase();
-  const keywords = (cfg.groupKeywords?.length ? cfg.groupKeywords : DEFAULT_GROUP_KEYWORDS)
+  const groupKeywords = resolveGroupKeywords(cfg);
+  const keywords = (groupKeywords?.length ? groupKeywords : DEFAULT_GROUP_KEYWORDS)
     .map((s) => s.trim().toLowerCase())
     .filter(Boolean);
   if (keywords.some((kw) => content.includes(kw))) return true;
 
+  const groupMentionTokens = resolveGroupMentionTokens(cfg);
   const mentionTokens = (
-    cfg.groupMentionTokens?.length ? cfg.groupMentionTokens : DEFAULT_GROUP_MENTION_TOKENS
+    groupMentionTokens?.length ? groupMentionTokens : DEFAULT_GROUP_MENTION_TOKENS
   )
     .map((s) => s.trim().toLowerCase())
     .filter(Boolean);

channel-plugin/src/outbound-observability.ts

@@ -0,0 +1,121 @@
+import { createHash } from "node:crypto";
+
+export type OutboundScope = "adapter-text" | "adapter-media" | "agent-reply" | "faq";
+
+type OutboundMetrics = {
+  attempts: number;
+  success: number;
+  failed: number;
+  deduped: number;
+  totalLatencyMs: number;
+  byScope: Record<OutboundScope, { attempts: number; success: number; failed: number; deduped: number }>;
+};
+
+const SUCCESS_WINDOW_MS = 8000;
+const ALERT_MIN_ATTEMPTS = 10;
+const ALERT_FAIL_RATE = 0.2;
+const ALERT_COOLDOWN_MS = 5 * 60 * 1000;
+const MAX_SUCCESS_KEYS = 2000;
+
+const metrics: OutboundMetrics = {
+  attempts: 0,
+  success: 0,
+  failed: 0,
+  deduped: 0,
+  totalLatencyMs: 0,
+  byScope: {
+    "adapter-text": { attempts: 0, success: 0, failed: 0, deduped: 0 },
+    "adapter-media": { attempts: 0, success: 0, failed: 0, deduped: 0 },
+    "agent-reply": { attempts: 0, success: 0, failed: 0, deduped: 0 },
+    faq: { attempts: 0, success: 0, failed: 0, deduped: 0 },
+  },
+};
+
+const recentSuccess = new Map<string, number>();
+let lastAlertAt = 0;
+
+function buildDedupeKey(scope: OutboundScope, toWxid: string, content: string): string {
+  const digest = createHash("sha1").update(content).digest("hex").slice(0, 16);
+  return `${scope}:${toWxid}:${digest}`;
+}
+
+function trimRecentSuccess(): void {
+  if (recentSuccess.size <= MAX_SUCCESS_KEYS) return;
+  const overflow = recentSuccess.size - Math.floor(MAX_SUCCESS_KEYS / 2);
+  const iter = recentSuccess.keys();
+  for (let i = 0; i < overflow; i++) {
+    const next = iter.next();
+    if (next.done) break;
+    recentSuccess.delete(next.value);
+  }
+}
+
+export function shouldSkipDuplicateSend(
+  scope: OutboundScope,
+  toWxid: string,
+  content: string,
+  nowMs: number = Date.now()
+): boolean {
+  const key = buildDedupeKey(scope, toWxid, content);
+  const lastSentAt = recentSuccess.get(key);
+  if (lastSentAt && nowMs - lastSentAt <= SUCCESS_WINDOW_MS) {
+    metrics.deduped++;
+    metrics.byScope[scope].deduped++;
+    return true;
+  }
+  return false;
+}
+
+export function recordOutboundAttempt(scope: OutboundScope): void {
+  metrics.attempts++;
+  metrics.byScope[scope].attempts++;
+}
+
+export function recordOutboundSuccess(
+  scope: OutboundScope,
+  toWxid: string,
+  content: string,
+  latencyMs: number
+): void {
+  metrics.success++;
+  metrics.byScope[scope].success++;
+  metrics.totalLatencyMs += Math.max(0, latencyMs);
+  recentSuccess.set(buildDedupeKey(scope, toWxid, content), Date.now());
+  trimRecentSuccess();
+}
+
+export function recordOutboundFailure(scope: OutboundScope): void {
+  metrics.failed++;
+  metrics.byScope[scope].failed++;
+}
+
+export function getOutboundMetricsSnapshot(): {
+  attempts: number;
+  success: number;
+  failed: number;
+  deduped: number;
+  avgLatencyMs: number;
+  failRate: number;
+  byScope: OutboundMetrics["byScope"];
+} {
+  const avgLatencyMs = metrics.success > 0 ? Math.round(metrics.totalLatencyMs / metrics.success) : 0;
+  const failRate = metrics.attempts > 0 ? metrics.failed / metrics.attempts : 0;
+  return {
+    attempts: metrics.attempts,
+    success: metrics.success,
+    failed: metrics.failed,
+    deduped: metrics.deduped,
+    avgLatencyMs,
+    failRate,
+    byScope: metrics.byScope,
+  };
+}
+
+export function shouldEmitOutboundFailureAlert(nowMs: number = Date.now()): boolean {
+  if (metrics.attempts < ALERT_MIN_ATTEMPTS) return false;
+  const failRate = metrics.failed / metrics.attempts;
+  if (failRate < ALERT_FAIL_RATE) return false;
+  if (nowMs - lastAlertAt < ALERT_COOLDOWN_MS) return false;
+  lastAlertAt = nowMs;
+  return true;
+}

channel-plugin/src/outbound-queue.ts

@@ -0,0 +1,91 @@
+import type { Logger } from "./types.js";
+
+type QueueTask<T> = {
+  fn: () => Promise<T>;
+  resolve: (value: T | PromiseLike<T>) => void;
+  reject: (reason?: unknown) => void;
+  enqueuedAt: number;
+  label: string;
+};
+
+type QueueConfig = {
+  ratePerSec: number;
+  maxQueue: number;
+};
+
+const DEFAULT_RATE_PER_SEC = 5;
+const DEFAULT_MAX_QUEUE = 200;
+
+const state: {
+  queue: Array<QueueTask<unknown>>;
+  running: boolean;
+  lastRunAt: number;
+  cfg: QueueConfig;
+  log?: Logger;
+} = {
+  queue: [],
+  running: false,
+  lastRunAt: 0,
+  cfg: { ratePerSec: DEFAULT_RATE_PER_SEC, maxQueue: DEFAULT_MAX_QUEUE },
+  log: undefined,
+};
+
+export function configureOutboundQueue(
+  cfg: Partial<QueueConfig>,
+  log?: Logger
+): void {
+  if (cfg.ratePerSec && Number.isFinite(cfg.ratePerSec) && cfg.ratePerSec > 0) {
+    state.cfg.ratePerSec = cfg.ratePerSec;
+  }
+  if (cfg.maxQueue && Number.isFinite(cfg.maxQueue) && cfg.maxQueue > 0) {
+    state.cfg.maxQueue = Math.floor(cfg.maxQueue);
+  }
+  if (log) state.log = log;
+}
+
+export function enqueueOutboundSend<T>(label: string, fn: () => Promise<T>): Promise<T> {
+  if (state.queue.length >= state.cfg.maxQueue) {
+    return Promise.reject(
+      new Error(`outbound queue overflow: size=${state.queue.length} max=${state.cfg.maxQueue}`)
+    );
+  }
+  return new Promise<T>((resolve, reject) => {
+    state.queue.push({ fn, resolve, reject, enqueuedAt: Date.now(), label });
+    void drainQueue();
+  });
+}
+
+async function drainQueue(): Promise<void> {
+  if (state.running) return;
+  state.running = true;
+  try {
+    while (state.queue.length > 0) {
+      const minIntervalMs = Math.ceil(1000 / state.cfg.ratePerSec);
+      const waitMs = Math.max(0, state.lastRunAt + minIntervalMs - Date.now());
+      if (waitMs > 0) {
+        await sleep(waitMs);
+      }
+      const task = state.queue.shift();
+      if (!task) continue;
+      state.lastRunAt = Date.now();
+      try {
+        const res = await task.fn();
+        const waited = Date.now() - task.enqueuedAt;
+        if (waited > 3000) {
+          state.log?.warn(
+            `outbound queue delayed label=${task.label} waitMs=${waited} queued=${state.queue.length}`
+          );
+        }
+        task.resolve(res);
+      } catch (err) {
+        task.reject(err);
+      }
+    }
+  } finally {
+    state.running = false;
+  }
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}

channel-plugin/src/outbound.ts

@@ -13,8 +13,21 @@
 import type { ChannelOutboundAdapter } from "openclaw/plugin-sdk/channel-contract";
 import { getRuntime } from "./runtime.js";
 import { createClient } from "./client.js";
-import { CHANNEL_ID, getChannelCfg, resolveApiBase } from "./config.js";
+import {
+  CHANNEL_ID,
+  getChannelCfg,
+  resolveApiBase,
+  resolveOutboundQueueMax,
+  resolveOutboundRatePerSec,
+} from "./config.js";
 import { withRetry } from "./retry.js";
+import {
+  recordOutboundAttempt,
+  recordOutboundFailure,
+  recordOutboundSuccess,
+  shouldSkipDuplicateSend,
+} from "./outbound-observability.js";
+import { configureOutboundQueue, enqueueOutboundSend } from "./outbound-queue.js";
 
 /** Strip any provider prefix and return the bare wxid. */
 function toBareWxid(target: string): string {
@@ -36,39 +49,63 @@ export const outbound: ChannelOutboundAdapter = {
   sendText: async (ctx) => {
     const channelCfg = getChannelCfg(ctx.cfg);
     const apiBase = resolveApiBase(channelCfg);
+    configureOutboundQueue({
+      ratePerSec: resolveOutboundRatePerSec(channelCfg),
+      maxQueue: resolveOutboundQueueMax(channelCfg),
+    });
     const bare = toBareWxid(ctx.to);
     if (!bare) {
       throw new Error(
         `[${CHANNEL_ID}] invalid target "${ctx.to}" \u2014 expected wxid:<wxid>`
       );
     }
+    if (shouldSkipDuplicateSend("adapter-text", bare, ctx.text)) {
+      getRuntime().log?.(
+        `[${CHANNEL_ID}:warn] skip duplicate outbound send (adapter-text) to=${bare}`
+      );
+      return {
+        channel: CHANNEL_ID,
+        messageId: `dedupe-${Date.now()}`,
+        conversationId: bare,
+      };
+    }
 
     const client = createClient(apiBase);
-    const result = await withRetry(
-      () =>
-        client.sendText({
-          toWxid: bare,
-          content: ctx.text,
-        }),
-      {
-        tries: 4,
-        baseDelayMs: 500,
-        maxDelayMs: 4000,
-        isRetryable: (err) => {
-          const msg = err instanceof Error ? err.message : String(err);
-          // fetch abort/timeout/network and 5xx from backend should retry
-          if (/timeout|aborted|ECONN|ENOTFOUND|EAI_AGAIN/i.test(msg)) return true;
-          if (/HTTP 5\d\d/i.test(msg)) return true;
-          return false;
-        },
-      }
-    );
-
-    if (!result.ok) {
-      throw new Error(
-        `[${CHANNEL_ID}] send-text failed: ${result.msg ?? `ret=${String(result.ret)}`}`
+    const startedAt = Date.now();
+    recordOutboundAttempt("adapter-text");
+    let result;
+    try {
+      result = await enqueueOutboundSend(`adapter-text:${bare}`, () =>
+        withRetry(
+          async () => {
+            const res = await client.sendText({
+              toWxid: bare,
+              content: ctx.text,
+            });
+            if (!res.ok) {
+              throw new Error(res.msg ?? `ret=${String(res.ret)}`);
+            }
+            return res;
+          },
+          {
+            tries: 4,
+            baseDelayMs: 500,
+            maxDelayMs: 4000,
+            isRetryable: (err) => {
+              const msg = err instanceof Error ? err.message : String(err);
+              // fetch abort/timeout/network and 5xx from backend should retry
+              if (/timeout|aborted|ECONN|ENOTFOUND|EAI_AGAIN/i.test(msg)) return true;
+              if (/HTTP 5\d\d/i.test(msg)) return true;
+              return false;
+            },
+          }
+        )
       );
+    } catch (err) {
+      recordOutboundFailure("adapter-text");
+      throw err;
     }
+    recordOutboundSuccess("adapter-text", bare, ctx.text, Date.now() - startedAt);
 
     return {
       channel: CHANNEL_ID,
@@ -84,6 +121,10 @@ export const outbound: ChannelOutboundAdapter = {
     // will land in v0.2 once we fingerprint the backend build.
     const channelCfg = getChannelCfg(ctx.cfg);
     const apiBase = resolveApiBase(channelCfg);
+    configureOutboundQueue({
+      ratePerSec: resolveOutboundRatePerSec(channelCfg),
+      maxQueue: resolveOutboundQueueMax(channelCfg),
+    });
     const bare = toBareWxid(ctx.to);
     if (!bare) {
       throw new Error(
@@ -94,31 +135,52 @@ export const outbound: ChannelOutboundAdapter = {
     const caption =
       ctx.text?.trim() ||
       "[media message \u2014 wechat-agent v0.1 does not yet support media outbound]";
+    if (shouldSkipDuplicateSend("adapter-media", bare, caption)) {
+      getRuntime().log?.(
+        `[${CHANNEL_ID}:warn] skip duplicate outbound send (adapter-media) to=${bare}`
+      );
+      return {
+        channel: CHANNEL_ID,
+        messageId: `dedupe-${Date.now()}`,
+        conversationId: bare,
+        meta: { mediaFallback: "caption-only", deduped: true },
+      };
+    }
     const client = createClient(apiBase);
-    const result = await withRetry(
-      () =>
-        client.sendText({
-          toWxid: bare,
-          content: caption,
-        }),
-      {
-        tries: 4,
-        baseDelayMs: 500,
-        maxDelayMs: 4000,
-        isRetryable: (err) => {
-          const msg = err instanceof Error ? err.message : String(err);
-          if (/timeout|aborted|ECONN|ENOTFOUND|EAI_AGAIN/i.test(msg)) return true;
-          if (/HTTP 5\d\d/i.test(msg)) return true;
-          return false;
-        },
-      }
-    );
-
-    if (!result.ok) {
-      throw new Error(
-        `[${CHANNEL_ID}] send-media(fallback-to-text) failed: ${result.msg ?? `ret=${String(result.ret)}`}`
+    const startedAt = Date.now();
+    recordOutboundAttempt("adapter-media");
+    let result;
+    try {
+      result = await enqueueOutboundSend(`adapter-media:${bare}`, () =>
+        withRetry(
+          async () => {
+            const res = await client.sendText({
+              toWxid: bare,
+              content: caption,
+            });
+            if (!res.ok) {
+              throw new Error(res.msg ?? `ret=${String(res.ret)}`);
+            }
+            return res;
+          },
+          {
+            tries: 4,
+            baseDelayMs: 500,
+            maxDelayMs: 4000,
+            isRetryable: (err) => {
+              const msg = err instanceof Error ? err.message : String(err);
+              if (/timeout|aborted|ECONN|ENOTFOUND|EAI_AGAIN/i.test(msg)) return true;
+              if (/HTTP 5\d\d/i.test(msg)) return true;
+              return false;
+            },
+          }
+        )
       );
+    } catch (err) {
+      recordOutboundFailure("adapter-media");
+      throw err;
     }
+    recordOutboundSuccess("adapter-media", bare, caption, Date.now() - startedAt);
 
     return {
       channel: CHANNEL_ID,

channel-plugin/src/types.ts

@@ -26,6 +26,38 @@ export type WechatAgentConfig = {
   faq?: FaqRule[];
   defaultTo?: string;
   selfWxid?: string;
+  /**
+   * Optional layered config: technical knobs (runtime/reliability).
+   * When set, these values override the legacy top-level fields.
+   */
+  technical?: {
+    apiBase?: string;
+    pollIntervalMs?: number;
+    batchWindowMs?: number;
+    replyCooldownSecPerWxid?: number;
+    outboundRatePerSec?: number;
+    outboundQueueMax?: number;
+    metricsLogIntervalSec?: number;
+    highThroughputOutboundAttemptsPer5Min?: number;
+  };
+  /**
+   * Optional layered config: business behavior knobs.
+   * When set, these values override the legacy top-level fields.
+   */
+  business?: {
+    groupEnabled?: boolean;
+    groupKeywords?: string[];
+    groupMentionTokens?: string[];
+    groupReplyAtSender?: boolean;
+    dmPolicy?: "open" | "allowlist" | "pairing";
+    allowFrom?: Array<string | number>;
+    ignoreWxidPrefixes?: string[];
+    ignoreWxidExact?: string[];
+    ignoreMessageTypes?: string[];
+    faq?: FaqRule[];
+    defaultTo?: string;
+    selfWxid?: string;
+  };
 };
 
 /** A single FAQ rule evaluated in the plugin layer as a fast path. */