:message

channel-plugin/src/cursor-store.ts

@@ -0,0 +1,70 @@
+import { mkdir, readFile, writeFile } from "node:fs/promises";
+import { dirname } from "node:path";
+
+export type InboundCursor = {
+  lastTimestampMs: number;
+  lastMsgId: string;
+};
+
+export async function loadCursor(path: string): Promise<InboundCursor | null> {
+  try {
+    const raw = await readFile(path, "utf8");
+    const obj = JSON.parse(raw) as Partial<InboundCursor> | null;
+    if (!obj || typeof obj !== "object") return null;
+    if (typeof obj.lastTimestampMs !== "number") return null;
+    if (typeof obj.lastMsgId !== "string" || !obj.lastMsgId) return null;
+    return { lastTimestampMs: obj.lastTimestampMs, lastMsgId: obj.lastMsgId };
+  } catch {
+    return null;
+  }
+}
+
+export async function saveCursor(path: string, cursor: InboundCursor): Promise<void> {
+  await mkdir(dirname(path), { recursive: true });
+  await writeFile(path, JSON.stringify(cursor), "utf8");
+}
+
+export function toEpochMs(ts: string | number | undefined | null): number {
+  if (typeof ts === "number" && Number.isFinite(ts)) return ts;
+  if (typeof ts === "string") {
+    // Numeric string (ms or seconds-ish). Treat 10-digit as seconds.
+    if (/^\d+$/.test(ts)) {
+      const n = Number(ts);
+      if (!Number.isFinite(n)) return Date.now();
+      return ts.length <= 10 ? n * 1000 : n;
+    }
+    const d = Date.parse(ts);
+    if (Number.isFinite(d)) return d;
+  }
+  return Date.now();
+}
+
+export function compareMsgId(a: string, b: string): number {
+  const an = /^\d+$/.test(a) ? Number(a) : NaN;
+  const bn = /^\d+$/.test(b) ? Number(b) : NaN;
+  if (Number.isFinite(an) && Number.isFinite(bn)) return an - bn;
+  return a.localeCompare(b);
+}
+
+export function isAfterCursor(
+  msg: { msgId: string; timestamp: string | number },
+  cursor: InboundCursor | null
+): boolean {
+  if (!cursor) return true;
+  const ms = toEpochMs(msg.timestamp);
+  if (ms > cursor.lastTimestampMs) return true;
+  if (ms < cursor.lastTimestampMs) return false;
+  return compareMsgId(String(msg.msgId), cursor.lastMsgId) > 0;
+}
+
+export function advanceCursor(
+  cursor: InboundCursor | null,
+  msg: { msgId: string; timestamp: string | number }
+): InboundCursor {
+  const next = { lastTimestampMs: toEpochMs(msg.timestamp), lastMsgId: String(msg.msgId) };
+  if (!cursor) return next;
+  if (next.lastTimestampMs > cursor.lastTimestampMs) return next;
+  if (next.lastTimestampMs < cursor.lastTimestampMs) return cursor;
+  return compareMsgId(next.lastMsgId, cursor.lastMsgId) > 0 ? next : cursor;
+}
+

channel-plugin/src/monitor.ts

@@ -40,6 +40,17 @@ import {
   resolveApiBase,
 } from "./config.js";
 import type { Logger, RawWechatMessage, WechatAgentConfig } from "./types.js";
+import {
+  advanceCursor,
+  compareMsgId,
+  isAfterCursor,
+  loadCursor,
+  saveCursor,
+  toEpochMs,
+  type InboundCursor,
+} from "./cursor-store.js";
+import { join } from "node:path";
+import { withRetry } from "./retry.js";
 
 export type MonitorOpts = {
   cfg: OpenClawConfig;
@@ -89,6 +100,19 @@ export async function startMonitor(opts: MonitorOpts): Promise<MonitorResult> {
   const batchWindowMs = Math.max(0, channelCfg.batchWindowMs ?? 3000);
   const replyCooldownMs = Math.max(0, (channelCfg.replyCooldownSecPerWxid ?? 5) * 1000);
 
+  // Persistent inbound cursor: best-effort resume after restarts.
+  // We store it alongside OpenClaw's session store so admins can back it up
+  // together with other channel state.
+  const core = getRuntime();
+  const sessionCfg = (opts.cfg as Record<string, unknown>).session as
+    | { store?: string }
+    | undefined;
+  const cursorStoreDir = core.channel.session.resolveStorePath(sessionCfg?.store, {
+    agentId: CHANNEL_ID,
+  });
+  const cursorPath = join(cursorStoreDir, `${CHANNEL_ID}.inbound-cursor.json`);
+  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;
   if (!selfWxid) {
@@ -182,7 +206,17 @@ export async function startMonitor(opts: MonitorOpts): Promise<MonitorResult> {
       });
       counters.pollsOk++;
 
+      // Always process in stable order so agent context is consistent.
+      messages.sort((a, b) => {
+        const ta = toEpochMs(a.timestamp);
+        const tb = toEpochMs(b.timestamp);
+        if (ta !== tb) return ta - tb;
+        return compareMsgId(String(a.msgId), String(b.msgId));
+      });
+
       for (const m of messages) {
+        // Cursor gate first: don't let old messages fill the dedupe window.
+        if (!isAfterCursor(m, cursor)) continue;
         if (!markSeen(m.msgId)) continue;
         const ignoredReason = shouldIgnoreMessage(m, channelCfg, selfWxid);
         if (ignoredReason) {
@@ -220,6 +254,17 @@ export async function startMonitor(opts: MonitorOpts): Promise<MonitorResult> {
 
         // Batch + dispatch
         cooldown.enqueue(m, resolveConversationId(m));
+
+        // Advance cursor after enqueue. The actual agent dispatch is async,
+        // but enqueueing is the "accepted for processing" point for this
+        // channel. We still keep a bounded dedupe window to tolerate crashes
+        // between enqueue and outbound delivery.
+        cursor = advanceCursor(cursor, m);
+      }
+
+      // Best-effort persist cursor once per tick.
+      if (cursor) {
+        await saveCursor(cursorPath, cursor);
       }
     } catch (err) {
       counters.pollsFail++;
@@ -415,11 +460,26 @@ async function dispatchBatchToAgent(params: DispatchParams): Promise<void> {
 
         for (const chunk of parts) {
           if (!chunk) continue;
-          const result = await client.sendText({
-            toWxid: conversationId,
-            content: chunk,
-            ats: isGroup && channelCfg.groupReplyAtSender && senderWxid ? senderWxid : "",
-          });
+          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 (!result.ok) {
             log.error(`send-text failed ret=${String(result.ret)} msg=${result.msg ?? ""}`);
             throw new Error(

channel-plugin/src/outbound.ts

@@ -14,6 +14,7 @@ import type { ChannelOutboundAdapter } from "openclaw/plugin-sdk/channel-contrac
 import { getRuntime } from "./runtime.js";
 import { createClient } from "./client.js";
 import { CHANNEL_ID, getChannelCfg, resolveApiBase } from "./config.js";
+import { withRetry } from "./retry.js";
 
 /** Strip any provider prefix and return the bare wxid. */
 function toBareWxid(target: string): string {
@@ -43,10 +44,25 @@ export const outbound: ChannelOutboundAdapter = {
     }
 
     const client = createClient(apiBase);
-    const result = await client.sendText({
-      toWxid: bare,
-      content: ctx.text,
-    });
+    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(
@@ -79,10 +95,24 @@ export const outbound: ChannelOutboundAdapter = {
       ctx.text?.trim() ||
       "[media message \u2014 wechat-agent v0.1 does not yet support media outbound]";
     const client = createClient(apiBase);
-    const result = await client.sendText({
-      toWxid: bare,
-      content: caption,
-    });
+    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-plugin/src/retry.ts

@@ -0,0 +1,32 @@
+export type RetryOpts = {
+  tries: number;
+  baseDelayMs: number;
+  maxDelayMs: number;
+  isRetryable: (err: unknown) => boolean;
+};
+
+export async function withRetry<T>(
+  fn: () => Promise<T>,
+  opts: RetryOpts
+): Promise<T> {
+  let attempt = 0;
+  let lastErr: unknown = null;
+  while (attempt < Math.max(1, opts.tries)) {
+    try {
+      return await fn();
+    } catch (err) {
+      lastErr = err;
+      attempt++;
+      if (attempt >= opts.tries) break;
+      if (!opts.isRetryable(err)) break;
+      const delay = Math.min(opts.maxDelayMs, opts.baseDelayMs * 2 ** (attempt - 1));
+      await sleep(delay);
+    }
+  }
+  throw lastErr;
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((r) => setTimeout(r, ms));
+}
+

docs/channel-plugin-design.md

@@ -142,7 +142,7 @@ openclaw-wechat-agent-channel/
     }
   },
   "peerDependencies": {
-    "openclaw": ">=2026.3.22"
+    "openclaw": "2026.4.15"
   }
 }
 ```