fix(signal): avoid replay after ambiguous quote delivery errors (#117134)

Co-authored-by: Peter Steinberger <steipete@macos.shared>

extensions/signal/src/send.test.ts

@@ -710,6 +710,7 @@ describe("sendMessageSignal receipts", () => {
   it.each([
     "Signal RPC -32602: quote rejected",
     'Signal RPC -32602: Unrecognized field "quoteTimestamp"',
+    "Signal REST 400: quote metadata invalid",
   ])("falls back to an ordinary send when native quote metadata fails: %s", async (message) => {
     signalRpcRequestMock
       .mockRejectedValueOnce(new Error(message))
@@ -781,8 +782,12 @@ describe("sendMessageSignal receipts", () => {
     });
   });
 
-  it("does not retry ordinary send failures as quote fallback", async () => {
-    signalRpcRequestMock.mockRejectedValueOnce(new Error("Signal HTTP timed out after 10000ms"));
+  it.each([
+    "Signal HTTP timed out after 10000ms",
+    "quote metadata not found after an unknown transport failure",
+    "Signal RPC -32000: quote metadata was rejected after an ambiguous send",
+  ])("does not retry an unconfirmed quote rejection: %s", async (message) => {
+    signalRpcRequestMock.mockRejectedValueOnce(new Error(message));
 
     await expect(
       sendMessageSignal("+15551234567", "hello", {
@@ -791,12 +796,113 @@ describe("sendMessageSignal receipts", () => {
         replyToAuthor: "+15550002222",
         replyToBody: "original",
       }),
-    ).rejects.toThrow("Signal HTTP timed out");
+    ).rejects.toThrow(message);
 
     expect(signalRpcRequestMock).toHaveBeenCalledTimes(1);
   });
 });
 
+describe("Signal quoted-message provider replay safety", () => {
+  it.each([
+    { label: "HTTP 408", transportKind: "container", status: 408, fallback: false },
+    { label: "HTTP 429", transportKind: "container", status: 429, fallback: false },
+    { label: "HTTP 503", transportKind: "container", status: 503, fallback: false },
+    { label: "HTTP 400", transportKind: "container", status: 400, fallback: true },
+    { label: "RPC -32603", transportKind: "external-native", rpcCode: -32603, fallback: false },
+    { label: "RPC -32602", transportKind: "external-native", rpcCode: -32602, fallback: true },
+  ] as const)("replays a real $label request only after definitive rejection", async (testCase) => {
+    const requests: Array<Record<string, unknown>> = [];
+    const server = http.createServer((request, response) => {
+      let body = "";
+      request.on("data", (chunk: Buffer) => {
+        body += chunk.toString("utf8");
+      });
+      request.on("end", () => {
+        const payload = JSON.parse(body) as Record<string, unknown>;
+        requests.push(payload);
+        if (requests.length === 1) {
+          if (testCase.transportKind === "container") {
+            response.writeHead(testCase.status, { "content-type": "text/plain" });
+            response.end("quote storage not found during upstream outage");
+          } else {
+            response.writeHead(200, { "content-type": "application/json" });
+            response.end(
+              JSON.stringify({
+                jsonrpc: "2.0",
+                id: payload.id,
+                error: {
+                  code: testCase.rpcCode,
+                  message: "quote storage not found during upstream outage",
+                },
+              }),
+            );
+          }
+          return;
+        }
+        response.writeHead(200, { "content-type": "application/json" });
+        response.end(
+          JSON.stringify(
+            testCase.transportKind === "container"
+              ? { timestamp: 1700000000002 }
+              : {
+                  jsonrpc: "2.0",
+                  id: payload.id,
+                  result: { timestamp: 1700000000002 },
+                },
+          ),
+        );
+      });
+    });
+    await new Promise<void>((resolve) => {
+      server.listen(0, "127.0.0.1", resolve);
+    });
+    const { port } = server.address() as { port: number };
+    const cfg = {
+      channels: {
+        signal: {
+          accounts: {
+            default: {
+              account: "+15550001111",
+              transport: { kind: testCase.transportKind, url: `http://127.0.0.1:${port}` },
+            },
+          },
+        },
+      },
+    } satisfies OpenClawConfig;
+
+    vi.doUnmock("./client-adapter.js");
+    vi.resetModules();
+    const { sendMessageSignal: sendWithRealContainer } = await import("./send.js");
+
+    try {
+      const delivery = sendWithRealContainer("+15551234567", "deliver only once", {
+        cfg,
+        textMode: "plain",
+        replyToId: "1700000000001",
+        replyToAuthor: "+15550002222",
+        replyToBody: "original",
+      });
+      if (testCase.fallback) {
+        await expect(delivery).resolves.toMatchObject({ messageId: "1700000000002" });
+      } else {
+        const expectedError =
+          testCase.transportKind === "container"
+            ? `Signal REST ${testCase.status}`
+            : `Signal RPC ${testCase.rpcCode}`;
+        await expect(delivery).rejects.toThrow(expectedError);
+      }
+      expect(requests).toHaveLength(testCase.fallback ? 2 : 1);
+      const quoteTimestampPath =
+        testCase.transportKind === "container" ? "quote_timestamp" : "params.quoteTimestamp";
+      expect(requests[0]).toHaveProperty(quoteTimestampPath, 1700000000001);
+    } finally {
+      await new Promise<void>((resolve) => {
+        server.close(() => resolve());
+      });
+    }
+  });
+});
+
 describe("Signal native JSON-RPC recipient delivery", () => {
   it("rejects real per-recipient failures and preserves successful send receipts", async () => {
     const failureTypes = [

extensions/signal/src/send.ts

@@ -259,6 +259,22 @@ function resolveSignalQuoteParams(opts: SignalSendOpts):
 function isSignalQuoteMetadataRejection(error: unknown): boolean {
   const message = error instanceof Error ? error.message : String(error);
   const normalized = normalizeLowercaseStringOrEmpty(message);
+  const rpcCode = /^signal rpc (-?\d+):/u.exec(normalized)?.[1];
+  if (rpcCode !== undefined) {
+    if (rpcCode !== "-32602") {
+      return false;
+    }
+  } else {
+    const restStatusText = /^signal rest (\d{3}):/u.exec(normalized)?.[1];
+    if (!restStatusText) {
+      return false;
+    }
+    const restStatus = Number(restStatusText);
+    // Only a definitive provider rejection makes replaying the send safe.
+    if (restStatus < 400 || restStatus >= 500 || restStatus === 408 || restStatus === 429) {
+      return false;
+    }
+  }
   if (!normalized.includes("quote")) {
     return false;
   }