fix(channels): add idempotency guard to nack() callback (#104903)

evan-YM · claude · evan-YM · commit 595c6010a45b · 2026-07-12T11:42:30.000+08:00
The shared nack() implementation in the channel message receive context
lacked an already-nacked guard, so repeated sequential calls to nack()
invoked onNack more than once. Add a guard that preserves the existing
ack-to-nack transition and leaves rejected nack callbacks retryable.

Add regression coverage: duplicate sequential nack idempotency, retry
after a rejected onNack callback, and ack-to-nack transition.

Add a standalone tsx verification script (scripts/verify-nack-idempotency.ts)
that exercises the production receive module directly and reports 16/16
assertions passing as real-behavior proof.

Co-Authored-By: Claude &lt;noreply@anthropic.com&gt;
diff --git a/scripts/verify-nack-idempotency.ts b/scripts/verify-nack-idempotency.ts
@@ -0,0 +1,111 @@
+/**
+ * Real-behavior verification for nack() idempotency guard.
+ *
+ * Runs the production receive module through the tsx loader to prove:
+ * 1. Duplicate sequential nack calls invoke onNack only once.
+ * 2. A rejected onNack callback leaves the context retryable.
+ * 3. The ack-to-nack transition is preserved.
+ *
+ * Usage: node --import tsx scripts/verify-nack-idempotency.ts
+ */
+import { createMessageReceiveContext } from "../src/channels/message/receive.js";
+
+let passed = 0;
+let failed = 0;
+
+function assert(condition: boolean, label: string): void {
+  if (condition) {
+    passed += 1;
+    console.log(`  \x1b[32m✓\x1b[0m ${label}`);
+  } else {
+    failed += 1;
+    console.error(`  \x1b[31m✗ FAIL\x1b[0m: ${label}`);
+  }
+}
+
+async function main() {
+  console.log("\n=== nack() idempotency real-behavior verification ===\n");
+
+  // ── Test 1: duplicate sequential nack ──
+  console.log("Test 1: Duplicate sequential nack calls onNack only once");
+  {
+    let callCount = 0;
+    const ctx = createMessageReceiveContext({
+      id: "real-1",
+      channel: "telegram",
+      message: { text: "hello" },
+      onNack: async () => {
+        callCount += 1;
+      },
+    });
+
+    await ctx.nack(new Error("first"));
+    assert(callCount === 1, "onNack called once after first nack");
+    assert(ctx.ackState === "nacked", "ackState is 'nacked'");
+    assert(ctx.nackErrorMessage === "first", "nackErrorMessage preserved");
+
+    await ctx.nack(new Error("second"));
+    assert(callCount === 1, "onNack still called only once after duplicate nack");
+    assert(ctx.ackState === "nacked", "ackState remains 'nacked'");
+    assert(ctx.nackErrorMessage === "first", "first error message preserved");
+  }
+
+  // ── Test 2: rejected callback remains retryable ──
+  console.log("\nTest 2: Rejected onNack callback leaves context retryable");
+  {
+    let attempts = 0;
+    const ctx = createMessageReceiveContext({
+      id: "real-2",
+      channel: "telegram",
+      message: { text: "hello" },
+      onNack: async () => {
+        attempts += 1;
+        if (attempts === 1) {
+          throw new Error("transient nack failure");
+        }
+      },
+    });
+
+    await ctx.nack(new Error("receive failed")).catch(() => {});
+    assert(attempts === 1, "onNack called once (first attempt rejected)");
+    assert(ctx.ackState === "pending", "ackState stays 'pending' after rejection");
+    assert(ctx.nackErrorMessage === undefined, "nackErrorMessage not set after rejection");
+
+    await ctx.nack(new Error("receive failed"));
+    assert(attempts === 2, "onNack called again on retry");
+    assert(ctx.ackState === "nacked", "ackState transitions to 'nacked' after success");
+    assert(ctx.nackErrorMessage === "receive failed", "nackErrorMessage set after success");
+  }
+
+  // ── Test 3: ack-to-nack transition preserved ──
+  console.log("\nTest 3: ack-to-nack transition preserved");
+  {
+    let nackCalls = 0;
+    const ctx = createMessageReceiveContext({
+      id: "real-3",
+      channel: "telegram",
+      message: { text: "hello" },
+      onAck: async () => {},
+      onNack: async () => {
+        nackCalls += 1;
+      },
+    });
+
+    await ctx.ack();
+    assert(ctx.ackState === "acked", "ackState is 'acked' after ack()");
+
+    await ctx.nack(new Error("post-ack failure"));
+    assert(nackCalls === 1, "onNack called once after ack-to-nack transition");
+    assert(ctx.ackState === "nacked", "ackState transitions to 'nacked' from 'acked'");
+    assert(ctx.nackErrorMessage === "post-ack failure", "nackErrorMessage set");
+  }
+
+  // ── Summary ──
+  console.log(`\n=== Results: ${passed} passed, ${failed} failed ===\n`);
+  process.exit(failed > 0 ? 1 : 0);
+}
+
+main().catch((err: unknown) => {
+  console.error("Verification error:", err);
+  process.exit(1);
+});
diff --git a/src/channels/message/lifecycle.test.ts b/src/channels/message/lifecycle.test.ts
@@ -372,6 +372,60 @@ describe("message lifecycle primitives", () => {
     expect(ctx.nackErrorMessage).toBe("offset failed");
   });
 
+  it("nack is idempotent and does not re-trigger onNack callback", async () => {
+    const onNack = vi.fn(async () => undefined);
+    const ctx = createMessageReceiveContext({
+      id: "rx-nack-idempotent",
+      channel: "telegram",
+      message: { text: "hello" },
+      onNack,
+    });
+
+    const firstError = new Error("first nack");
+    await ctx.nack(firstError);
+    expect(onNack).toHaveBeenCalledTimes(1);
+    expect(onNack).toHaveBeenCalledWith(firstError);
+    expect(ctx.ackState).toBe("nacked");
+    expect(ctx.nackErrorMessage).toBe("first nack");
+
+    const secondError = new Error("second nack");
+    await ctx.nack(secondError);
+    // onNack must not fire again for duplicate nack calls
+    expect(onNack).toHaveBeenCalledTimes(1);
+    // First nack's error and state are preserved
+    expect(ctx.ackState).toBe("nacked");
+    expect(ctx.nackErrorMessage).toBe("first nack");
+  });
+
+  it("retries nack when onNack callback rejects", async () => {
+    let calls = 0;
+    const onNack = vi.fn(async () => {
+      calls += 1;
+      if (calls === 1) {
+        throw new Error("nack callback transient failure");
+      }
+    });
+    const ctx = createMessageReceiveContext({
+      id: "rx-nack-retry",
+      channel: "telegram",
+      message: { text: "hello" },
+      onNack,
+    });
+
+    const nackErr = new Error("receive failed");
+    await ctx.nack(nackErr).catch(() => undefined);
+    // onNack rejected → state must stay unchanged so callers can retry
+    expect(ctx.ackState).toBe("pending");
+    expect(ctx.nackErrorMessage).toBeUndefined();
+    expect(onNack).toHaveBeenCalledTimes(1);
+
+    // Retry: onNack succeeds this time → nack completes
+    await ctx.nack(nackErr);
+    expect(onNack).toHaveBeenCalledTimes(2);
+    expect(ctx.ackState).toBe("nacked");
+    expect(ctx.nackErrorMessage).toBe("receive failed");
+  });
+
   it("maps ack policies to lifecycle stages", () => {
     expect(shouldAckMessageAfterStage("after_receive_record", "receive_record")).toBe(true);
     expect(shouldAckMessageAfterStage("after_receive_record", "agent_dispatch")).toBe(false);
diff --git a/src/channels/message/receive.ts b/src/channels/message/receive.ts
@@ -88,6 +88,11 @@ export function createMessageReceiveContext<TMessage>(params: {
       delete ctx.nackErrorMessage;
     },
     nack: async (error) => {
+      // Nack callbacks must be idempotent because receive pipelines may
+      // revisit completed stages and retry after nack.
+      if (ctx.ackState === "nacked") {
+        return;
+      }
       await params.onNack?.(error);
       ctx.ackState = "nacked";
       ctx.nackErrorMessage = normalizeAckErrorMessage(error);
