fix: use passive QQ event replies

2026-08-18 22:20:55 +08:00
parent 4af5049605
commit ffedf616e3
9 changed files with 560 additions and 48 deletions
@@ -10,16 +10,33 @@ export interface GatewayResult { ok: boolean; reply?: string; ignored?: boolean;
 interface ChatTarget {
  target: MessageTarget;
  adapter: PlatformAdapter;
+  lastInboundMessageId?: string;
+  lastInboundAt?: number;
+  inboundIsGroup: boolean;
+  lastEventDelivery?: "success" | "failed" | "skipped";
+  lastEventError?: string;
+  lastEventAt?: number;
 }

+export interface GatewayOptions {
+  now?: () => number;
+}
+
+// QQ passive-reply windows (with safety margin): 5 minutes for group chats, 60 minutes for C2C.
+const GROUP_PASSIVE_WINDOW_MS = 4.5 * 60_000;
+const C2C_PASSIVE_WINDOW_MS = 55 * 60_000;
+
 class ReplyStream {
-  private sequence = 0;
  private tail = Promise.resolve();

-  constructor(private readonly message: IncomingMessage, private readonly adapter: PlatformAdapter) {}
+  constructor(
+    private readonly message: IncomingMessage,
+    private readonly adapter: PlatformAdapter,
+    private readonly nextSequence: () => number | undefined
+  ) {}

-  enqueue(text: string): Promise<void> {
-    const replySequence = ++this.sequence;
+  enqueue(text: string, onError?: (error: unknown) => void): Promise<void> {
+    const replySequence = this.nextSequence();
    this.tail = this.tail.then(() => this.adapter.sendMessage({
      target: {
        platform: this.message.platform,
@@ -30,8 +47,9 @@ class ReplyStream {
      text,
      replyTo: this.message.messageId,
      replySequence
-    })).catch(() => {
+    })).catch((error: unknown) => {
      console.error(`Gateway delivery send failed (platform=${this.message.platform})`);
+      onError?.(error);
    });
    return this.tail;
  }
@@ -39,13 +57,21 @@ class ReplyStream {

 export class Gateway {
  private readonly chatTargets = new Map<string, ChatTarget>();
+  private readonly pendingEvents = new Map<string, string[]>();
+  // QQ deduplicates passive replies by msg_id + msg_seq: all sends referencing the same inbound
+  // message (normal replies, events, redelivered pending events) share one counter per chat+messageId.
+  private readonly messageSequences = new Map<string, Map<string, number>>();
+  private readonly now: () => number;
  private closed = false;
  readonly commandRouter = new CommandRouter();

  constructor(
    private readonly policy: GatewayPolicy,
-    private readonly runtime: ConversationRuntime
-  ) {}
+    private readonly runtime: ConversationRuntime,
+    options: GatewayOptions = {}
+  ) {
+    this.now = options.now || Date.now;
+  }

  async receive(message: IncomingMessage, adapter: PlatformAdapter, options: { synchronous?: boolean } = {}): Promise<GatewayResult> {
    const policyError = this.checkPolicy(message);
@@ -54,20 +80,29 @@ export class Gateway {
      return { ok: true, ignored: true, error: policyError };
    }

-    this.chatTargets.set(chatKeyFor(message.platform, message.chatId), {
+    const chatKey = chatKeyFor(message.platform, message.chatId);
+    const previous = this.chatTargets.get(chatKey);
+    this.chatTargets.set(chatKey, {
      target: {
        platform: message.platform,
        chatId: message.chatId,
        userId: message.userId,
        raw: message.raw
      },
-      adapter
+      adapter,
+      lastInboundMessageId: message.messageId,
+      lastInboundAt: this.now(),
+      inboundIsGroup: Boolean(message.isGroup),
+      lastEventDelivery: previous?.lastEventDelivery,
+      lastEventError: previous?.lastEventError,
+      lastEventAt: previous?.lastEventAt
    });
+    await this.flushPendingEvents(chatKey, message, adapter, options);

    const command = this.commandRouter.parse(message.text);
    if (command) {
      const result = await this.executeCommandResult(command, message);
-      if (!options.synchronous) await new ReplyStream(message, adapter).enqueue(result.reply || "");
+      if (!options.synchronous) await this.replyStream(chatKey, message, adapter).enqueue(result.reply || "");
      return result;
    }

@@ -81,12 +116,12 @@ export class Gateway {
        text: message.text,
        messageId: message.messageId
      })).text;
-      await new ReplyStream(message, adapter).enqueue(reply);
+      await this.replyStream(chatKey, message, adapter).enqueue(reply);
      return { ok: true, reply };
    } catch (error) {
      const errorText = error instanceof Error ? error.message : String(error);
      const reply = `Agent error: ${errorText}`;
-      await new ReplyStream(message, adapter).enqueue(reply);
+      await this.replyStream(chatKey, message, adapter).enqueue(reply);
      return { ok: false, error: errorText, reply };
    }
  }
@@ -95,13 +130,89 @@ export class Gateway {
    if (this.closed) return;
    const entry = this.chatTargets.get(chatKey);
    if (!entry) return;
+    const windowMs = entry.inboundIsGroup ? GROUP_PASSIVE_WINDOW_MS : C2C_PASSIVE_WINDOW_MS;
+    const freshInbound = entry.lastInboundMessageId
+      && entry.lastInboundAt !== undefined
+      && this.now() - entry.lastInboundAt <= windowMs;
+    if (!freshInbound) {
+      // No fresh passive window and proactive messages may be unauthorized: skip and remind on the next inbound.
+      this.recordEventDelivery(chatKey, "skipped", "no fresh passive window");
+      this.queuePendingEvent(chatKey, text);
+      console.log(`Gateway event skipped: no fresh passive window (platform=${entry.target.platform})`);
+      return;
+    }
    try {
-      await entry.adapter.sendMessage({ target: entry.target, text });
-    } catch {
+      await entry.adapter.sendMessage({
+        target: entry.target,
+        text,
+        replyTo: entry.lastInboundMessageId,
+        replySequence: this.nextReplySequence(chatKey, entry.lastInboundMessageId)
+      });
+      this.recordEventDelivery(chatKey, "success");
+    } catch (error) {
+      this.recordEventDelivery(chatKey, "failed", safeEventError(error));
+      this.queuePendingEvent(chatKey, text);
      console.error(`Gateway event send failed (platform=${entry.target.platform})`);
    }
  }

+  private recordEventDelivery(chatKey: string, delivery: "success" | "failed" | "skipped", error?: string): void {
+    const entry = this.chatTargets.get(chatKey);
+    if (!entry) return;
+    entry.lastEventDelivery = delivery;
+    entry.lastEventError = error;
+    entry.lastEventAt = this.now();
+  }
+
+  private queuePendingEvent(chatKey: string, text: string): void {
+    const pending = this.pendingEvents.get(chatKey) || [];
+    pending.push(text);
+    this.pendingEvents.set(chatKey, pending);
+  }
+
+  private async flushPendingEvents(chatKey: string, message: IncomingMessage, adapter: PlatformAdapter, options: { synchronous?: boolean }): Promise<void> {
+    if (options.synchronous) return;
+    const pending = this.pendingEvents.get(chatKey);
+    if (!pending || pending.length === 0) return;
+    this.pendingEvents.delete(chatKey);
+    const stream = this.replyStream(chatKey, message, adapter);
+    for (const text of pending) {
+      await stream.enqueue(text, (error) => {
+        // Redelivery failed: keep the event queued for the next inbound and record the failure.
+        this.recordEventDelivery(chatKey, "failed", safeEventError(error));
+        this.queuePendingEvent(chatKey, text);
+      });
+    }
+  }
+
+  private replyStream(chatKey: string, message: IncomingMessage, adapter: PlatformAdapter): ReplyStream {
+    return new ReplyStream(message, adapter, () => this.nextReplySequence(chatKey, message.messageId));
+  }
+
+  private nextReplySequence(chatKey: string, messageId: string | undefined): number | undefined {
+    if (!messageId) return undefined;
+    let sequences = this.messageSequences.get(chatKey);
+    if (!sequences) {
+      sequences = new Map();
+      this.messageSequences.set(chatKey, sequences);
+    }
+    const next = (sequences.get(messageId) || 0) + 1;
+    // Re-set to refresh insertion order, then bound memory to the most recent message ids per chat.
+    sequences.delete(messageId);
+    sequences.set(messageId, next);
+    while (sequences.size > 8) sequences.delete(sequences.keys().next().value!);
+    return next;
+  }
+
+  private eventStatus(chatKey: string): Record<string, string> {
+    const entry = this.chatTargets.get(chatKey);
+    return {
+      lastEventDelivery: entry?.lastEventDelivery || "none",
+      lastEventError: entry?.lastEventError || "none",
+      lastEventAt: entry?.lastEventAt ? new Date(entry.lastEventAt).toISOString() : "never"
+    };
+  }
+
  stats(): ReturnType<ConversationRuntime["stats"]> {
    return this.runtime.stats();
  }
@@ -140,7 +251,8 @@ export class Gateway {
      case "help": return HELP_TEXT;
      case "status": {
        const status = this.runtime.status(message.platform, message.chatId, message.userId);
-        return `OK\n${Object.entries(status).map(([key, value]) => `${key}=${value}`).join("\n")}`;
+        const combined = { ...status, ...this.eventStatus(chatKeyFor(message.platform, message.chatId)) };
+        return `OK\n${Object.entries(combined).map(([key, value]) => `${key}=${value}`).join("\n")}`;
      }
      case "list": {
        if (!this.runtime.listProposals) return "当前运行时不支持列出提案。";
@@ -170,6 +282,15 @@ export class Gateway {

 }

+function safeEventError(error: unknown): string {
+  const message = error instanceof Error ? error.message : "";
+  const http = /HTTP (\d{3})/.exec(message);
+  const code = /code=(\d+)/.exec(message);
+  const parts = [http ? `HTTP ${http[1]}` : "", code ? `QQ code ${code[1]}` : ""].filter(Boolean);
+  if (parts.length > 0) return parts.join(" ");
+  return error instanceof Error ? error.name : "error";
+}
+
 const HELP_TEXT = [
  "直接用自然语言告诉我你要做什么即可，我会自己判断并处理。",
  "兜底命令：",