fix: tighten system event dedupe identity

steipete · steipete · commit e2cc8a7e5c15 · 2026-05-11T14:28:04.000+01:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -21,6 +21,7 @@ Docs: https://docs.openclaw.ai
 
 ### Fixes
 
+- System events: dedupe keyed events across the queue while preserving unkeyed, delivery-route, and trust-boundary event identity. (#73040) Thanks @statxc.
 - Agents/UI: compact exec and tool progress rows by hiding redundant shell tool names, replacing known workspace paths with short context markers, and preserving Discord trace scrubbing for compact command lines.
 - ACPX: run and await the embedded ACP backend startup probe by default so the gateway `ready` signal no longer fires before the acpx runtime has either become usable or reported a probe failure; set `OPENCLAW_ACPX_RUNTIME_STARTUP_PROBE=0` to restore lazy startup. Fixes #79596. Thanks @bzelones.
 - OpenAI-compatible models: strip prior assistant reasoning fields from replayed Chat Completions history by default, preventing oMLX/vLLM Qwen follow-up turns from rejecting or stalling on stale `reasoning` payloads. Fixes #46637. Thanks @zipzagster and @lexhoefsloot.
diff --git a/src/infra/system-events.test.ts b/src/infra/system-events.test.ts
@@ -309,6 +309,18 @@ describe("system events (session routing)", () => {
     ]);
   });
 
+  it("allows non-consecutive unkeyed duplicate events", () => {
+    const key = "agent:main:test-unkeyed-noncons-dupe";
+    const first = enqueueSystemEvent("Node connected", { sessionKey: key });
+    const interleaved = enqueueSystemEvent("Heartbeat tick", { sessionKey: key });
+    const retry = enqueueSystemEvent("Node connected", { sessionKey: key });
+
+    expect(first).toBe(true);
+    expect(interleaved).toBe(true);
+    expect(retry).toBe(true);
+    expect(peekSystemEvents(key)).toEqual(["Node connected", "Heartbeat tick", "Node connected"]);
+  });
+
   it("allows the same text under a different context key", () => {
     const key = "agent:main:test-context-disambiguates";
     const reactionA = enqueueSystemEvent("Discord reaction added: ✅", {
@@ -325,6 +337,42 @@ describe("system events (session routing)", () => {
     expect(peekSystemEventEntries(key)).toHaveLength(2);
   });
 
+  it("allows the same text and context under a different delivery route", () => {
+    const key = "agent:main:test-context-route-disambiguates";
+    const first = enqueueSystemEvent("Build completed", {
+      sessionKey: key,
+      contextKey: "build:123",
+      deliveryContext: { channel: "telegram", to: "100" },
+    });
+    const second = enqueueSystemEvent("Build completed", {
+      sessionKey: key,
+      contextKey: "build:123",
+      deliveryContext: { channel: "telegram", to: "200" },
+    });
+
+    expect(first).toBe(true);
+    expect(second).toBe(true);
+    expect(peekSystemEventEntries(key)).toHaveLength(2);
+  });
+
+  it("allows the same text and context under different trust metadata", () => {
+    const key = "agent:main:test-context-trust-disambiguates";
+    const trusted = enqueueSystemEvent("Hook finished", {
+      sessionKey: key,
+      contextKey: "hook:done",
+      trusted: true,
+    });
+    const untrusted = enqueueSystemEvent("Hook finished", {
+      sessionKey: key,
+      contextKey: "hook:done",
+      trusted: false,
+    });
+
+    expect(trusted).toBe(true);
+    expect(untrusted).toBe(true);
+    expect(peekSystemEventEntries(key)).toHaveLength(2);
+  });
+
   it("preserves lastContextKey when a duplicate is skipped", () => {
     const key = "agent:main:test-context-preserved";
     enqueueSystemEvent("Node connected", { sessionKey: key, contextKey: "build:123" });
@@ -345,6 +393,60 @@ describe("system events (session routing)", () => {
 
     expect(isSystemEventContextChanged(key, "build:123")).toBe(false);
   });
+
+  it("preserves lastContextKey from the newest contextful event after partial consume", () => {
+    const key = "agent:main:test-context-preserved-after-consume";
+    enqueueSystemEvent("startup", { sessionKey: key });
+    enqueueSystemEvent("contextful", { sessionKey: key, contextKey: "build:123" });
+    enqueueSystemEvent("unkeyed followup", { sessionKey: key });
+    const inspected = peekSystemEventEntries(key).slice(0, 1);
+
+    expect(consumeSystemEventEntries(key, inspected).map((entry) => entry.text)).toEqual([
+      "startup",
+    ]);
+    expect(isSystemEventContextChanged(key, "build:123")).toBe(false);
+  });
+
+  it("allows a keyed duplicate after the original is evicted", () => {
+    const key = "agent:main:test-keyed-duplicate-after-eviction";
+    enqueueSystemEvent("Build completed", { sessionKey: key, contextKey: "build:123" });
+    for (let index = 0; index < 20; index += 1) {
+      enqueueSystemEvent(`event ${index}`, { sessionKey: key, contextKey: `event:${index}` });
+    }
+
+    expect(
+      enqueueSystemEvent("Build completed", { sessionKey: key, contextKey: "build:123" }),
+    ).toBe(true);
+  });
+
+  it("allows a keyed duplicate after the original is consumed from the prefix", () => {
+    const key = "agent:main:test-keyed-duplicate-after-prefix-consume";
+    enqueueSystemEvent("Build completed", { sessionKey: key, contextKey: "build:123" });
+    const inspected = peekSystemEventEntries(key);
+
+    expect(consumeSystemEventEntries(key, inspected).map((entry) => entry.text)).toEqual([
+      "Build completed",
+    ]);
+    expect(
+      enqueueSystemEvent("Build completed", { sessionKey: key, contextKey: "build:123" }),
+    ).toBe(true);
+  });
+
+  it("allows a keyed duplicate after the original is selectively consumed", () => {
+    const key = "agent:main:test-keyed-duplicate-after-selected-consume";
+    enqueueSystemEvent("Build completed", { sessionKey: key, contextKey: "build:123" });
+    enqueueSystemEvent("Other event", { sessionKey: key, contextKey: "build:other" });
+    const selected = peekSystemEventEntries(key).filter(
+      (entry) => entry.text === "Build completed",
+    );
+
+    expect(consumeSelectedSystemEventEntries(key, selected).map((entry) => entry.text)).toEqual([
+      "Build completed",
+    ]);
+    expect(
+      enqueueSystemEvent("Build completed", { sessionKey: key, contextKey: "build:123" }),
+    ).toBe(true);
+  });
 });
 
 describe("isCronSystemEvent", () => {
diff --git a/src/infra/system-events.ts b/src/infra/system-events.ts
@@ -90,9 +90,17 @@ function findDuplicateInQueue(
   queue: readonly SystemEvent[],
   text: string,
   contextKey: string | null,
+  deliveryContext: DeliveryContext | undefined,
+  trusted: boolean,
 ): SystemEvent | undefined {
+  if (contextKey === null) {
+    const last = queue[queue.length - 1];
+    return last && isDuplicateSystemEvent(last, { text, contextKey, deliveryContext, trusted })
+      ? last
+      : undefined;
+  }
   for (const event of queue) {
-    if (event.text === text && (event.contextKey ?? null) === contextKey) {
+    if (isDuplicateSystemEvent(event, { text, contextKey, deliveryContext, trusted })) {
       return event;
     }
   }
@@ -114,7 +122,16 @@ export function enqueueSystemEvent(text: string, options: SystemEventOptions) {
   }
   const normalizedContextKey = normalizeContextKey(options?.contextKey);
   const normalizedDeliveryContext = normalizeDeliveryContext(options?.deliveryContext);
-  if (findDuplicateInQueue(entry.queue, cleaned, normalizedContextKey)) {
+  const trusted = options.trusted !== false;
+  if (
+    findDuplicateInQueue(
+      entry.queue,
+      cleaned,
+      normalizedContextKey,
+      normalizedDeliveryContext,
+      trusted,
+    )
+  ) {
     return false;
   }
   applyContextKeyPolicy(entry, normalizedContextKey);
@@ -123,7 +140,7 @@ export function enqueueSystemEvent(text: string, options: SystemEventOptions) {
     ts: Date.now(),
     contextKey: normalizedContextKey,
     deliveryContext: normalizedDeliveryContext,
-    trusted: options.trusted !== false,
+    trusted,
   });
   if (entry.queue.length > MAX_EVENTS) {
     entry.queue.shift();
@@ -154,6 +171,18 @@ function areDeliveryContextsEqual(left?: DeliveryContext, right?: DeliveryContex
   return channelRouteDedupeKey(left) === channelRouteDedupeKey(right);
 }
 
+function isDuplicateSystemEvent(
+  existing: SystemEvent,
+  incoming: Pick<SystemEvent, "text" | "contextKey" | "deliveryContext" | "trusted">,
+): boolean {
+  return (
+    existing.text === incoming.text &&
+    (existing.contextKey ?? null) === (incoming.contextKey ?? null) &&
+    (existing.trusted ?? true) === (incoming.trusted ?? true) &&
+    areDeliveryContextsEqual(existing.deliveryContext, incoming.deliveryContext)
+  );
+}
+
 function areSystemEventsEqual(left: SystemEvent, right: SystemEvent): boolean {
   return (
     left.text === right.text &&
@@ -170,8 +199,14 @@ function resetQueueState(key: string, entry: SessionQueue) {
     queues.delete(key);
     return;
   }
-  const newest = entry.queue[entry.queue.length - 1];
-  entry.lastContextKey = newest.contextKey ?? null;
+  for (let index = entry.queue.length - 1; index >= 0; index -= 1) {
+    const contextKey = entry.queue[index].contextKey ?? null;
+    if (contextKey !== null) {
+      entry.lastContextKey = contextKey;
+      return;
+    }
+  }
+  entry.lastContextKey = null;
 }
 
 export function consumeSystemEventEntries(
